pubsub: move polling task to alarm executor

pongad · pongad · commit 40bba8095f78 · 2017-05-22T15:53:19.000+10:00
"alarm executor".
This commit moves polling tasks to the same executor,
and give the said executor more threads
to handle the extra workload.

This commit puts all client's "house keeping" tasks in one executor,
and all user code in another.
In this way, long running user code cannot starve the client of CPU.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java
@@ -17,10 +17,9 @@
 package com.google.cloud.pubsub.spi.v1;
 
 import com.google.api.core.ApiClock;
-import com.google.api.gax.core.Distribution;
 import com.google.api.gax.batching.FlowController;
 import com.google.api.gax.batching.FlowController.FlowControlException;
-import com.google.api.gax.grpc.InstantiatingExecutorProvider;
+import com.google.api.gax.core.Distribution;
 import com.google.cloud.pubsub.spi.v1.MessageDispatcher.OutstandingMessagesBatch.OutstandingMessage;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
@@ -46,7 +45,6 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import javax.annotation.Nullable;
 import org.threeten.bp.Duration;
 import org.threeten.bp.Instant;
 
@@ -61,9 +59,6 @@ class MessageDispatcher {
   @VisibleForTesting static final Duration PENDING_ACKS_SEND_DELAY = Duration.ofMillis(100);
   private static final int MAX_ACK_DEADLINE_EXTENSION_SECS = 10 * 60; // 10m
 
-  private static final ScheduledExecutorService SHARED_ALARMS_EXECUTOR =
-      InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(2).build().getExecutor();
-
   private final ScheduledExecutorService executor;
   private final ScheduledExecutorService alarmsExecutor;
   private final ApiClock clock;
@@ -245,10 +240,10 @@ void sendAckOperations(
       Distribution ackLatencyDistribution,
       FlowController flowController,
       ScheduledExecutorService executor,
-      @Nullable ScheduledExecutorService alarmsExecutor,
+      ScheduledExecutorService alarmsExecutor,
       ApiClock clock) {
     this.executor = executor;
-    this.alarmsExecutor = alarmsExecutor == null ? SHARED_ALARMS_EXECUTOR : alarmsExecutor;
+    this.alarmsExecutor = alarmsExecutor;
     this.ackExpirationPadding = ackExpirationPadding;
     this.maxAckExtensionPeriod = maxAckExtensionPeriod;
     this.receiver = receiver;
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
@@ -73,10 +73,10 @@ public PollingSubscriberConnection(
       FlowController flowController,
       @Nullable Long maxDesiredPulledMessages,
       ScheduledExecutorService executor,
-      @Nullable ScheduledExecutorService alarmsExecutor,
+      ScheduledExecutorService alarmsExecutor,
       ApiClock clock) {
     this.subscription = subscription;
-    this.executor = executor;
+    this.executor = alarmsExecutor;
     stub = SubscriberGrpc.newFutureStub(channel);
     messageDispatcher =
         new MessageDispatcher(
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
@@ -26,6 +26,7 @@
 import com.google.api.gax.core.Distribution;
 import com.google.api.gax.grpc.ChannelProvider;
 import com.google.api.gax.grpc.ExecutorProvider;
+import com.google.api.gax.grpc.FixedExecutorProvider;
 import com.google.api.gax.grpc.InstantiatingExecutorProvider;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.annotations.VisibleForTesting;
@@ -85,6 +86,9 @@ public class Subscriber extends AbstractApiService {
   private static final Duration ACK_DEADLINE_UPDATE_PERIOD = Duration.ofMinutes(1);
   private static final double PERCENTILE_FOR_ACK_DEADLINE_UPDATES = 99.9;
 
+  private static final ScheduledExecutorService SHARED_ALARMS_EXECUTOR =
+      InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(6).build().getExecutor();
+
   private static final Logger logger = Logger.getLogger(Subscriber.class.getName());
 
   private final SubscriptionName subscriptionName;
@@ -140,7 +144,6 @@ public void close() throws IOException {
             }
           });
     }
-    if (builder.alarmsExecutorProvider != null) {
       alarmsExecutor = builder.alarmsExecutorProvider.getExecutor();
       if (builder.alarmsExecutorProvider.shouldAutoClose()) {
         closeables.add(
@@ -151,9 +154,6 @@ public void close() throws IOException {
               }
             });
       }
-    } else {
-      alarmsExecutor = null;
-    }
 
     channelProvider = builder.channelProvider;
 
@@ -455,7 +455,7 @@ public static final class Builder {
             .build();
 
     ExecutorProvider executorProvider = DEFAULT_EXECUTOR_PROVIDER;
-    @Nullable ExecutorProvider alarmsExecutorProvider;
+    ExecutorProvider alarmsExecutorProvider = FixedExecutorProvider.create(SHARED_ALARMS_EXECUTOR);
     ChannelProvider channelProvider =
         SubscriptionAdminSettings.defaultChannelProviderBuilder()
             .setMaxInboundMessageSize(MAX_INBOUND_MESSAGE_SIZE)
