Decouple the processing of new received messages from the dispatching to

davidtorres · davidtorres · commit b555c975c048 · 2017-04-07T13:10:22.000-04:00
user code - Separated the handling of batched received messages from the per-message dispatching of message to the user code. In order to accomplish this I had to keep the batch in memory while the processing of each message will draw from the in-memory batch until completely depleted. - Drawing flow controller permits on a per message basis (used to try to draw permits for the whole batch potentially deadlocking the whole subscriber), this addresses the deadlock condition raised here #1868 - No longer using blocking flow controller, instead pausing and resuming pulls/streamed-messages based on the flow controller feedback and when new permits become available. - A separate executor for alarms (2 threads in it has showed up to scale pretty well with many subscriber, given our ack operations are pretty lightweight) - Setting the maximum of messages to pull per request based on the number requested by the user in the flow controller (if any), this in a best effort addresses #1868 Fixes #1868, fixes #1865 and fixes #1855
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
@@ -16,9 +16,11 @@
 
 package com.google.cloud.pubsub.spi.v1;
 
-import com.google.api.gax.core.FlowController;
 import com.google.api.gax.core.ApiClock;
+import com.google.api.gax.core.FlowController;
+import com.google.api.gax.core.FlowController.FlowControlException;
 import com.google.api.stats.Distribution;
+import com.google.cloud.pubsub.spi.v1.MessageDispatcher.OutstandingMessagesBatch.OutstandingMessage;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.google.common.primitives.Ints;
@@ -29,11 +31,13 @@
 import com.google.pubsub.v1.ReceivedMessage;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Deque;
 import java.util.HashSet;
-import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.PriorityQueue;
 import java.util.Set;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -56,6 +60,9 @@ class MessageDispatcher {
   private static final int INITIAL_ACK_DEADLINE_EXTENSION_SECONDS = 2;
   @VisibleForTesting static final Duration PENDING_ACKS_SEND_DELAY = Duration.millis(100);
   private static final int MAX_ACK_DEADLINE_EXTENSION_SECS = 10 * 60; // 10m
+  
+  private static final ScheduledExecutorService alarmsExecutor =
+      Executors.newScheduledThreadPool(2);
 
   private final ScheduledExecutorService executor;
   private final ApiClock clock;
@@ -184,6 +191,7 @@ public void onFailure(Throwable t) {
       setupPendingAcksAlarm();
       flowController.release(1, outstandingBytes);
       messagesWaiter.incrementPendingMessages(-1);
+      processOutstandingBatches();
     }
 
     @Override
@@ -194,25 +202,23 @@ public void onSuccess(AckReply reply) {
           synchronized (pendingAcks) {
             pendingAcks.add(ackId);
           }
-          setupPendingAcksAlarm();
-          flowController.release(1, outstandingBytes);
           // Record the latency rounded to the next closest integer.
           ackLatencyDistribution.record(
               Ints.saturatedCast(
                   (long) Math.ceil((clock.millisTime() - receivedTime.getMillis()) / 1000D)));
-          messagesWaiter.incrementPendingMessages(-1);
-          return;
+          break;
         case NACK:
           synchronized (pendingNacks) {
             pendingNacks.add(ackId);
           }
-          setupPendingAcksAlarm();
-          flowController.release(1, outstandingBytes);
-          messagesWaiter.incrementPendingMessages(-1);
-          return;
+          break;
         default:
           throw new IllegalArgumentException(String.format("AckReply: %s not supported", reply));
       }
+      setupPendingAcksAlarm();
+      flowController.release(1, outstandingBytes);
+      messagesWaiter.incrementPendingMessages(-1);
+      processOutstandingBatches();
     }
   }
 
@@ -269,28 +275,117 @@ public int getMessageDeadlineSeconds() {
     return messageDeadlineSeconds;
   }
 
-  public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> responseMessages) {
-    int receivedMessagesCount = responseMessages.size();
-    if (receivedMessagesCount == 0) {
+  static class OutstandingMessagesBatch {
+    static class OutstandingMessage {
+      private final com.google.pubsub.v1.ReceivedMessage receivedMessage;
+      private final AckHandler ackHandler;
+
+      public OutstandingMessage(ReceivedMessage receivedMessage, AckHandler ackHandler) {
+        this.receivedMessage = receivedMessage;
+        this.ackHandler = ackHandler;
+      }
+
+      public com.google.pubsub.v1.ReceivedMessage receivedMessage() {
+        return receivedMessage;
+      }
+
+      public AckHandler ackHandler() {
+        return ackHandler;
+      }
+    }
+
+    private final Deque<OutstandingMessage> messages;
+    private final Runnable doneCallback;
+
+    public OutstandingMessagesBatch(Runnable doneCallback) {
+      this.messages = new LinkedList<>();
+      this.doneCallback = doneCallback;
+    }
+
+    public void addMessage(
+        com.google.pubsub.v1.ReceivedMessage receivedMessage, AckHandler ackHandler) {
+      this.messages.add(new OutstandingMessage(receivedMessage, ackHandler));
+    }
+
+    public Deque<OutstandingMessage> messages() {
+      return messages;
+    }
+
+    public void done() {
+      doneCallback.run();
+    }
+  }
+
+  Deque<OutstandingMessagesBatch> outstandingMessageBatches = new LinkedList<>();
+
+  public void processReceivedMessages(
+      List<com.google.pubsub.v1.ReceivedMessage> messages, Runnable doneCallback) {
+    if (messages.size() == 0) {
+      doneCallback.run();
       return;
     }
-    Instant now = new Instant(clock.millisTime());
-    int totalByteCount = 0;
-    final ArrayList<AckHandler> ackHandlers = new ArrayList<>(responseMessages.size());
-    for (ReceivedMessage pubsubMessage : responseMessages) {
-      int messageSize = pubsubMessage.getMessage().getSerializedSize();
-      totalByteCount += messageSize;
-      ackHandlers.add(new AckHandler(pubsubMessage.getAckId(), messageSize));
+    messagesWaiter.incrementPendingMessages(messages.size());
+
+    OutstandingMessagesBatch outstandingBatch = new OutstandingMessagesBatch(doneCallback);
+    final ArrayList<AckHandler> ackHandlers = new ArrayList<>(messages.size());
+    for (ReceivedMessage message : messages) {
+      AckHandler ackHandler =
+          new AckHandler(message.getAckId(), message.getMessage().getSerializedSize());
+      ackHandlers.add(ackHandler);
+      outstandingBatch.addMessage(message, ackHandler);
     }
-    Instant expiration = now.plus(messageDeadlineSeconds * 1000);
-    logger.log(
-        Level.FINER, "Received {0} messages at {1}", new Object[] {responseMessages.size(), now});
-
-    messagesWaiter.incrementPendingMessages(responseMessages.size());
-    Iterator<AckHandler> acksIterator = ackHandlers.iterator();
-    for (ReceivedMessage userMessage : responseMessages) {
-      final PubsubMessage message = userMessage.getMessage();
-      final AckHandler ackHandler = acksIterator.next();
+
+    Instant expiration = new Instant(clock.millisTime()).plus(messageDeadlineSeconds * 1000);
+    synchronized (outstandingAckHandlers) {
+      outstandingAckHandlers.add(
+          new ExtensionJob(
+              new Instant(clock.millisTime()),
+              expiration,
+              INITIAL_ACK_DEADLINE_EXTENSION_SECONDS,
+              ackHandlers));
+    }
+    setupNextAckDeadlineExtensionAlarm(expiration);
+
+    synchronized (outstandingMessageBatches) {
+      outstandingMessageBatches.add(outstandingBatch);
+    }
+    processOutstandingBatches();
+  }
+
+  public void processOutstandingBatches() {
+    while (true) {
+      boolean batchDone = false;
+      Runnable batchCallback = null;
+      OutstandingMessage outstandingMessage;
+      synchronized (outstandingMessageBatches) {
+        OutstandingMessagesBatch nextBatch = outstandingMessageBatches.peek();
+        if (nextBatch == null) {
+          return;
+        }
+        outstandingMessage = nextBatch.messages.peek();
+        if (outstandingMessage == null) {
+          return;
+        }
+        try {
+          // This is a non-blocking flow controller.
+          flowController.reserve(
+              1, outstandingMessage.receivedMessage().getMessage().getSerializedSize());
+        } catch (FlowController.MaxOutstandingElementCountReachedException
+            | FlowController.MaxOutstandingRequestBytesReachedException flowControlException) {
+          return;
+        } catch (FlowControlException unexpectedException) {
+          throw new IllegalStateException("Flow control unexpected exception", unexpectedException);
+        }
+        nextBatch.messages.poll(); // We got a hold to the message already.
+        batchDone = nextBatch.messages.isEmpty();
+        if (batchDone) {
+          outstandingMessageBatches.poll();
+          batchCallback = nextBatch.doneCallback;
+        }
+      }
+
+      final PubsubMessage message = outstandingMessage.receivedMessage().getMessage();
+      final AckHandler ackHandler = outstandingMessage.ackHandler();
       final SettableFuture<AckReply> response = SettableFuture.create();
       final AckReplyConsumer consumer =
           new AckReplyConsumer() {
@@ -311,22 +406,9 @@ public void run() {
               }
             }
           });
-    }
-
-    synchronized (outstandingAckHandlers) {
-      outstandingAckHandlers.add(
-          new ExtensionJob(
-              new Instant(clock.millisTime()),
-              expiration,
-              INITIAL_ACK_DEADLINE_EXTENSION_SECONDS,
-              ackHandlers));
-    }
-    setupNextAckDeadlineExtensionAlarm(expiration);
-
-    try {
-      flowController.reserve(receivedMessagesCount, totalByteCount);
-    } catch (FlowController.FlowControlException unexpectedException) {
-      throw new IllegalStateException("Flow control unexpected exception", unexpectedException);
+      if (batchDone) {
+        batchCallback.run();
+      }
     }
   }
 
@@ -335,7 +417,7 @@ private void setupPendingAcksAlarm() {
     try {
       if (pendingAcksAlarm == null) {
         pendingAcksAlarm =
-            executor.schedule(
+            alarmsExecutor.schedule(
                 new Runnable() {
                   @Override
                   public void run() {
@@ -400,7 +482,7 @@ public void run() {
             // drop it.
             continue;
           }
-          
+
           // If a message has already been acked, remove it, nothing to do.
           for (int i = 0; i < job.ackHandlers.size(); ) {
             if (job.ackHandlers.get(i).acked.get()) {
@@ -464,7 +546,7 @@ private void setupNextAckDeadlineExtensionAlarm(Instant expiration) {
         nextAckDeadlineExtensionAlarmTime = possibleNextAlarmTime;
 
         ackDeadlineExtensionAlarm =
-            executor.schedule(
+            alarmsExecutor.schedule(
                 new AckDeadlineAlarm(),
                 nextAckDeadlineExtensionAlarmTime.getMillis() - clock.millisTime(),
                 TimeUnit.MILLISECONDS);
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
@@ -42,6 +42,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.annotation.Nullable;
 import org.joda.time.Duration;
 
 /**
@@ -61,6 +62,7 @@ final class PollingSubscriberConnection extends AbstractApiService implements Ac
   private final ScheduledExecutorService executor;
   private final SubscriberFutureStub stub;
   private final MessageDispatcher messageDispatcher;
+  private final int maxDesiredPulledMessages;
 
   public PollingSubscriberConnection(
       String subscription,
@@ -71,6 +73,7 @@ public PollingSubscriberConnection(
       Distribution ackLatencyDistribution,
       Channel channel,
       FlowController flowController,
+      @Nullable Integer maxDesiredPulledMessages,
       ScheduledExecutorService executor,
       ApiClock clock) {
     this.subscription = subscription;
@@ -89,6 +92,8 @@ public PollingSubscriberConnection(
             executor,
             clock);
     messageDispatcher.setMessageDeadlineSeconds(Subscriber.MIN_ACK_DEADLINE_SECONDS);
+    this.maxDesiredPulledMessages =
+        maxDesiredPulledMessages != null ? maxDesiredPulledMessages : DEFAULT_MAX_MESSAGES;
   }
 
   @Override
@@ -117,7 +122,8 @@ public void onSuccess(Subscription result) {
           public void onFailure(Throwable cause) {
             notifyFailed(cause);
           }
-        });
+        },
+        executor);
   }
 
   @Override
@@ -132,7 +138,7 @@ private void pullMessages(final Duration backoff) {
             .pull(
                 PullRequest.newBuilder()
                     .setSubscription(subscription)
-                    .setMaxMessages(DEFAULT_MAX_MESSAGES)
+                    .setMaxMessages(maxDesiredPulledMessages)
                     .setReturnImmediately(true)
                     .build());
 
@@ -141,7 +147,6 @@ private void pullMessages(final Duration backoff) {
         new FutureCallback<PullResponse>() {
           @Override
           public void onSuccess(PullResponse pullResponse) {
-            messageDispatcher.processReceivedMessages(pullResponse.getReceivedMessagesList());
             if (pullResponse.getReceivedMessagesCount() == 0) {
               // No messages in response, possibly caught up in backlog, we backoff to avoid
               // slamming the server.
@@ -160,7 +165,14 @@ public void run() {
                   TimeUnit.MILLISECONDS);
               return;
             }
-            pullMessages(INITIAL_BACKOFF);
+            messageDispatcher.processReceivedMessages(
+                pullResponse.getReceivedMessagesList(),
+                new Runnable() {
+                  @Override
+                  public void run() {
+                    pullMessages(INITIAL_BACKOFF);
+                  }
+                });
           }
 
           @Override
@@ -190,7 +202,8 @@ public void run() {
               notifyFailed(cause);
             }
           }
-        });
+        },
+        executor);
   }
 
   private boolean isAlive() {
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StreamingSubscriberConnection.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StreamingSubscriberConnection.java
@@ -125,11 +125,17 @@ public void beforeStart(ClientCallStreamObserver<StreamingPullRequest> requestOb
 
     @Override
     public void onNext(StreamingPullResponse response) {
-      messageDispatcher.processReceivedMessages(response.getReceivedMessagesList());
-      // Only if not shutdown we will request one more batches of messages to be delivered.
-      if (isAlive()) {
-        requestObserver.request(1);
-      }
+      messageDispatcher.processReceivedMessages(
+          response.getReceivedMessagesList(),
+          new Runnable() {
+            @Override
+            public void run() {
+              // Only if not shutdown we will request one more batches of messages to be delivered.
+              if (isAlive()) {
+                requestObserver.request(1);
+              }
+            }
+          });
     }
 
     @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
