pubsub: start connections slowly

pongad · pongad · commit 4c988ec44c66 · 2017-05-12T15:34:05.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,17 @@ public void close() {
       throw new IllegalStateException(e);
     }
 
-    // Streaming pull is not enabled on the service yet.
-    // startStreamingConnections();
-    startPollingConnections();
-    notifyStarted();
+    new Thread(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          startPollingConnections();
+          notifyStarted();
+        } catch (Throwable t) {
+          notifyFailed(t);
+        }
+      }
+    }).start();
   }
 
   @Override
@@ -387,25 +393,9 @@ 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);
+      subscriber.startAsync().awaitRunning();
     }
   }
 
@@ -415,26 +405,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.
+      }
     }
   }
 
