xDS: implement ADS stream flow control mechanism

xds/src/main/java/io/grpc/xds/CdsLoadBalancer2.java

@@ -320,7 +320,7 @@
 
       private void start() {
         shutdown = false;
-        xdsClient.watchXdsResource(XdsClusterResource.getInstance(), name, this);
+        xdsClient.watchXdsResource(XdsClusterResource.getInstance(), name, this, syncContext);
       }
 
       void shutdown() {
@@ -341,102 +341,85 @@
                 String.format("Unable to load CDS %s. xDS server returned: %s: %s",
                   name, error.getCode(), error.getDescription()))
             .withCause(error.getCause());
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (shutdown) {
-              return;
-            }
-            // All watchers should receive the same error, so we only propagate it once.
-            if (ClusterState.this == root) {
-              handleClusterDiscoveryError(status);
-            }
-          }
-        });
+        if (shutdown) {
+          return;
+        }
+        // All watchers should receive the same error, so we only propagate it once.
+        if (ClusterState.this == root) {
+          handleClusterDiscoveryError(status);
+        }
       }
 
       @Override
       public void onResourceDoesNotExist(String resourceName) {
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (shutdown) {
-              return;
-            }
-            discovered = true;
-            result = null;
-            if (childClusterStates != null) {
-              for (ClusterState state : childClusterStates.values()) {
-                state.shutdown();
-              }
-              childClusterStates = null;
-            }
-            handleClusterDiscovered();
+        if (shutdown) {
+          return;
+        }
+        discovered = true;
+        result = null;
+        if (childClusterStates != null) {
+          for (ClusterState state : childClusterStates.values()) {
+            state.shutdown();
           }
-        });
+          childClusterStates = null;
+        }
+        handleClusterDiscovered();
       }
 
       @Override
       public void onChanged(final CdsUpdate update) {
-        class ClusterDiscovered implements Runnable {
-          @Override
-          public void run() {
-            if (shutdown) {
-              return;
+        if (shutdown) {
+          return;
+        }
+        logger.log(XdsLogLevel.DEBUG, "Received cluster update {0}", update);
+        discovered = true;
+        result = update;
+        if (update.clusterType() == ClusterType.AGGREGATE) {
+          isLeaf = false;
+          logger.log(XdsLogLevel.INFO, "Aggregate cluster {0}, underlying clusters: {1}",
+              update.clusterName(), update.prioritizedClusterNames());
+          Map<String, ClusterState> newChildStates = new LinkedHashMap<>();
+          for (String cluster : update.prioritizedClusterNames()) {
+            if (newChildStates.containsKey(cluster)) {
+              logger.log(XdsLogLevel.WARNING,
+                  String.format("duplicate cluster name %s in aggregate %s is being ignored",
+                      cluster, update.clusterName()));
+              continue;
             }
-
-            logger.log(XdsLogLevel.DEBUG, "Received cluster update {0}", update);
-            discovered = true;
-            result = update;
-            if (update.clusterType() == ClusterType.AGGREGATE) {
-              isLeaf = false;
-              logger.log(XdsLogLevel.INFO, "Aggregate cluster {0}, underlying clusters: {1}",
-                  update.clusterName(), update.prioritizedClusterNames());
-              Map<String, ClusterState> newChildStates = new LinkedHashMap<>();
-              for (String cluster : update.prioritizedClusterNames()) {
-                if (newChildStates.containsKey(cluster)) {
-                  logger.log(XdsLogLevel.WARNING,
-                      String.format("duplicate cluster name %s in aggregate %s is being ignored",
-                          cluster, update.clusterName()));
-                  continue;
-                }
-                if (childClusterStates == null || !childClusterStates.containsKey(cluster)) {
-                  ClusterState childState;
-                  if (clusterStates.containsKey(cluster)) {
-                    childState = clusterStates.get(cluster);
-                    if (childState.shutdown) {
-                      childState.start();
-                    }
-                  } else {
-                    childState = new ClusterState(cluster);
-                    clusterStates.put(cluster, childState);
-                    childState.start();
-                  }
-                  newChildStates.put(cluster, childState);
-                } else {
-                  newChildStates.put(cluster, childClusterStates.remove(cluster));
-                }
-              }
-              if (childClusterStates != null) {  // stop subscribing to revoked child clusters
-                for (ClusterState watcher : childClusterStates.values()) {
-                  watcher.shutdown();
+            if (childClusterStates == null || !childClusterStates.containsKey(cluster)) {
+              ClusterState childState;
+              if (clusterStates.containsKey(cluster)) {
+                childState = clusterStates.get(cluster);
+                if (childState.shutdown) {
+                  childState.start();
                 }
+              } else {
+                childState = new ClusterState(cluster);
+                clusterStates.put(cluster, childState);
+                childState.start();
               }
-              childClusterStates = newChildStates;
-            } else if (update.clusterType() == ClusterType.EDS) {
-              isLeaf = true;
-              logger.log(XdsLogLevel.INFO, "EDS cluster {0}, edsServiceName: {1}",
-                  update.clusterName(), update.edsServiceName());
-            } else {  // logical DNS
-              isLeaf = true;
-              logger.log(XdsLogLevel.INFO, "Logical DNS cluster {0}", update.clusterName());
+              newChildStates.put(cluster, childState);
+            } else {
+              newChildStates.put(cluster, childClusterStates.remove(cluster));
+            }
+          }
+          if (childClusterStates != null) {  // stop subscribing to revoked child clusters
+            for (ClusterState watcher : childClusterStates.values()) {
+              watcher.shutdown();
             }
-            handleClusterDiscovered();
           }
+          childClusterStates = newChildStates;
+        } else if (update.clusterType() == ClusterType.EDS) {
+          isLeaf = true;
+          logger.log(XdsLogLevel.INFO, "EDS cluster {0}, edsServiceName: {1}",
+              update.clusterName(), update.edsServiceName());
+        } else {  // logical DNS
+          isLeaf = true;
+          logger.log(XdsLogLevel.INFO, "Logical DNS cluster {0}", update.clusterName());
         }
-
-        syncContext.execute(new ClusterDiscovered());
+        handleClusterDiscovered();
       }
+
     }
   }
 }

xds/src/main/java/io/grpc/xds/ClusterResolverLoadBalancer.java

@@ -366,7 +366,8 @@
       void start() {
         String resourceName = edsServiceName != null ? edsServiceName : name;
         logger.log(XdsLogLevel.INFO, "Start watching EDS resource {0}", resourceName);
-        xdsClient.watchXdsResource(XdsEndpointResource.getInstance(), resourceName, this);
+        xdsClient.watchXdsResource(XdsEndpointResource.getInstance(),
+            resourceName, this, syncContext);
       }
 
       @Override
@@ -452,7 +453,7 @@
           }
         }
 
-        syncContext.execute(new EndpointsUpdated());
+        new EndpointsUpdated().run();
       }
 
       private List<String> generatePriorityNames(String name,
@@ -491,38 +492,28 @@
 
       @Override
       public void onResourceDoesNotExist(final String resourceName) {
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (shutdown) {
-              return;
-            }
-            logger.log(XdsLogLevel.INFO, "Resource {0} unavailable", resourceName);
-            status = Status.OK;
-            resolved = true;
-            result = null;  // resource revoked
-            handleEndpointResourceUpdate();
-          }
-        });
+        if (shutdown) {
+          return;
+        }
+        logger.log(XdsLogLevel.INFO, "Resource {0} unavailable", resourceName);
+        status = Status.OK;
+        resolved = true;
+        result = null;  // resource revoked
+        handleEndpointResourceUpdate();
       }
 
       @Override
       public void onError(final Status error) {
-        syncContext.execute(new Runnable() {
-          @Override
-          public void run() {
-            if (shutdown) {
-              return;
-            }
-            String resourceName = edsServiceName != null ? edsServiceName : name;
-            status = Status.UNAVAILABLE
-                .withDescription(String.format("Unable to load EDS %s. xDS server returned: %s: %s",
-                      resourceName, error.getCode(), error.getDescription()))
-                .withCause(error.getCause());
-            logger.log(XdsLogLevel.WARNING, "Received EDS error: {0}", error);
-            handleEndpointResolutionError();
-          }
-        });
+        if (shutdown) {
+          return;
+        }
+        String resourceName = edsServiceName != null ? edsServiceName : name;
+        status = Status.UNAVAILABLE
+            .withDescription(String.format("Unable to load EDS %s. xDS server returned: %s: %s",
+                  resourceName, error.getCode(), error.getDescription()))
+            .withCause(error.getCause());
+        logger.log(XdsLogLevel.WARNING, "Received EDS error: {0}", error);
+        handleEndpointResolutionError();
       }
     }