pubsub: start connections slowly (#2055)

pongad · web-flow · commit 5af4a462bfc4 · 2017-05-17T20:36:15.000+10:00
Starting a new connection seems to add a task which blocks
a thread in our thread pool.
It does not block for long, but Subscriber starts many connections,
slamming the pool.
If the number of threads the pool is low, the pool is blocked for
a long time, causing strange delays and RPC failures.

This PR alleviates this problem by starting connections one at a time.
- Each connection now make one successful call before declaring itself running.
- The Subscriber waits for a connection to be in running state
  before starting another.

Together, these changes limit the number of blocking tasks
registered to the pool.

Since starting the Subscriber can now take a significant amount of time,
and `doStart` should return promptly,
a new thread is used to start up the connections.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java
@@ -99,11 +99,6 @@ public PollingSubscriberConnection(
   @Override
   protected void doStart() {
     logger.config("Starting subscriber.");
-    initialize();
-    notifyStarted();
-  }
-
-  private void initialize() {
     ListenableFuture<Subscription> subscriptionInfo =
         stub.withDeadlineAfter(DEFAULT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)
             .getSubscription(
@@ -116,6 +111,7 @@ private void initialize() {
           public void onSuccess(Subscription result) {
             messageDispatcher.setMessageDeadlineSeconds(result.getAckDeadlineSeconds());
             pullMessages(INITIAL_BACKOFF);
+            notifyStarted();
           }
 
           @Override
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java
@@ -39,7 +39,6 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -244,10 +243,23 @@ public void close() {
       throw new IllegalStateException(e);
     }
 
-    // Streaming pull is not enabled on the service yet.
-    // startStreamingConnections();
-    startPollingConnections();
-    notifyStarted();
+    // When started, connections submit tasks to the executor.
+    // These tasks must finish before the connections can declare themselves running.
+    // If we have a single-thread executor and call startPollingConnections from the
+    // same executor, it will deadlock: the thread will be stuck waiting for connections
+    // to start but cannot start the connections.
+    // For this reason, we spawn a dedicated thread. Starting subscriber should be rare.
+    new Thread(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          startPollingConnections();
+          notifyStarted();
+        } catch (Throwable t) {
+          notifyFailed(t);
+        }
+      }
+    }).start();
   }
 
   @Override
@@ -387,25 +399,11 @@ private void stopAllPollingConnections() {
 
   private void startConnections(
       List<? extends ApiService> connections, final ApiService.Listener connectionsListener) {
-    final CountDownLatch subscribersStarting = new CountDownLatch(numChannels);
-    for (final ApiService subscriber : connections) {
-      executor.submit(
-          new Runnable() {
-            @Override
-            public void run() {
-              subscriber.addListener(connectionsListener, executor);
-              try {
-                subscriber.startAsync().awaitRunning();
-              } finally {
-                subscribersStarting.countDown();
-              }
-            }
-          });
-    }
-    try {
-      subscribersStarting.await();
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
+    for (ApiService subscriber : connections) {
+      subscriber.addListener(connectionsListener, executor);
+      // Starting each connection submits a blocking task to the executor.
+      // We start connections one at a time to avoid swamping executor with blocking tasks.
+      subscriber.startAsync().awaitRunning();
     }
   }
 
@@ -415,26 +413,17 @@ private void stopConnections(List<? extends ApiService> connections) {
       liveConnections = new ArrayList<ApiService>(connections);
       connections.clear();
     }
-    final CountDownLatch connectionsStopping = new CountDownLatch(liveConnections.size());
-    for (final ApiService subscriberConnection : liveConnections) {
-      executor.submit(
-          new Runnable() {
-            @Override
-            public void run() {
-              try {
-                subscriberConnection.stopAsync().awaitTerminated();
-              } catch (IllegalStateException ignored) {
-                // It is expected for some connections to be already in state failed so stop will
-                // throw this expection.
-              }
-              connectionsStopping.countDown();
-            }
-          });
+    for (ApiService subscriber : liveConnections) {
+      subscriber.stopAsync();
     }
-    try {
-      connectionsStopping.await();
-    } catch (InterruptedException e) {
-      throw new IllegalStateException(e);
+    for (ApiService subscriber : liveConnections) {
+      try {
+        subscriber.awaitTerminated();
+      } catch (IllegalStateException e) {
+        // If the service fails, awaitTerminated will throw an exception.
+        // However, we could be stopping services because at least one
+        // has already failed, so we just ignore this exception.
+      }
     }
   }
 
