From e39b9169b9f6e3b852de5f1d964edc12a28dceba Mon Sep 17 00:00:00 2001
From: Matthew Chen <charlesmchen@gmail.com>
Date: Thu, 14 Sep 2017 10:00:34 -0400
Subject: [PATCH] Decrypt and process messages in batches.

// FREEBIE
---
 .../src/Messages/OWSBatchMessageProcessor.m   |  3 +--
 .../src/Messages/OWSMessageReceiver.m         |  3 +--
 .../src/Messages/TSMessagesManager.h          |  2 ++
 .../src/Messages/TSMessagesManager.m          | 19 ++++++++++++++-----
 4 files changed, 18 insertions(+), 9 deletions(-)

diff --git a/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m b/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m
index 46a0bd0a3..dc51b9a9a 100644
--- a/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m
+++ b/SignalServiceKit/src/Messages/OWSBatchMessageProcessor.m
@@ -299,8 +299,7 @@ NSString *const OWSBatchMessageProcessingJobFinderExtensionGroup = @"OWSBatchMes
 {
     AssertIsOnMainThread();
 
-    const NSUInteger kMaxBatchSize = 10;
-    NSArray<OWSBatchMessageProcessingJob *> *jobs = [self.finder nextJobsForBatchSize:kMaxBatchSize];
+    NSArray<OWSBatchMessageProcessingJob *> *jobs = [self.finder nextJobsForBatchSize:kIncomingMessageBatchSize];
     OWSAssert(jobs);
     if (jobs.count < 1) {
         self.isDrainingQueue = NO;
diff --git a/SignalServiceKit/src/Messages/OWSMessageReceiver.m b/SignalServiceKit/src/Messages/OWSMessageReceiver.m
index c7762f302..fa5a0ddc5 100644
--- a/SignalServiceKit/src/Messages/OWSMessageReceiver.m
+++ b/SignalServiceKit/src/Messages/OWSMessageReceiver.m
@@ -294,8 +294,7 @@ NSString *const OWSMessageProcessingJobFinderExtensionGroup = @"OWSMessageProces
 {
     AssertIsOnMainThread();
 
-    const NSUInteger kMaxBatchSize = 10;
-    NSArray<OWSMessageProcessingJob *> *jobs = [self.finder nextJobsForBatchSize:kMaxBatchSize];
+    NSArray<OWSMessageProcessingJob *> *jobs = [self.finder nextJobsForBatchSize:kIncomingMessageBatchSize];
     OWSAssert(jobs);
     if (jobs.count < 1) {
         self.isDrainingQueue = NO;
diff --git a/SignalServiceKit/src/Messages/TSMessagesManager.h b/SignalServiceKit/src/Messages/TSMessagesManager.h
index ce1e88c92..5ac185083 100644
--- a/SignalServiceKit/src/Messages/TSMessagesManager.h
+++ b/SignalServiceKit/src/Messages/TSMessagesManager.h
@@ -8,6 +8,8 @@
 
 NS_ASSUME_NONNULL_BEGIN
 
+extern const NSUInteger kIncomingMessageBatchSize;
+
 @class TSNetworkManager;
 @class TSStorageManager;
 @class OWSSignalServiceProtosEnvelope;
diff --git a/SignalServiceKit/src/Messages/TSMessagesManager.m b/SignalServiceKit/src/Messages/TSMessagesManager.m
index c86afeaeb..a34c06dc3 100644
--- a/SignalServiceKit/src/Messages/TSMessagesManager.m
+++ b/SignalServiceKit/src/Messages/TSMessagesManager.m
@@ -45,6 +45,15 @@
 
 NS_ASSUME_NONNULL_BEGIN
 
+// We need to use a consistent batch size throughout
+// the incoming message pipeline (i.e. in the
+// "decrypt" and "process" steps), or the pipeline
+// doesn't flow smoothly.
+//
+// We want a value that is just high enough to yield
+// perf benefits.  The right value is probably 5-15.
+const NSUInteger kIncomingMessageBatchSize = 10;
+
 @interface TSMessagesManager ()
 
 @property (nonatomic, readonly) id<OWSCallMessageHandler> callMessageHandler;
@@ -1087,13 +1096,13 @@ NS_ASSUME_NONNULL_BEGIN
             }
             case OWSSignalServiceProtosGroupContextTypeDeliver: {
                 if (body.length == 0 && attachmentIds.count < 1) {
-                    DDLogWarn(@"%@ ignoring empty incoming message from: %@ for group: %@ with timestampe: %lu",
+                    DDLogWarn(@"%@ ignoring empty incoming message from: %@ for group: %@ with timestamp: %lu",
                         self.tag,
                         envelopeAddress(envelope),
                         groupId,
                         (unsigned long)timestamp);
                 } else {
-                    DDLogDebug(@"%@ incoming message from: %@ for group: %@ with timestampe: %lu",
+                    DDLogDebug(@"%@ incoming message from: %@ for group: %@ with timestamp: %lu",
                         self.tag,
                         envelopeAddress(envelope),
                         groupId,
@@ -1111,19 +1120,19 @@ NS_ASSUME_NONNULL_BEGIN
                 break;
             }
             default: {
-                DDLogWarn(@"%@ Ignoring unknown group message type:%d", self.tag, (int)dataMessage.group.type);
+                DDLogWarn(@"%@ Ignoring unknown group message type: %d", self.tag, (int)dataMessage.group.type);
             }
         }
 
         thread = gThread;
     } else {
         if (body.length == 0 && attachmentIds.count < 1) {
-            DDLogWarn(@"%@ ignoring empty incoming message from: %@ with timestampe: %lu",
+            DDLogWarn(@"%@ ignoring empty incoming message from: %@ with timestamp: %lu",
                 self.tag,
                 envelopeAddress(envelope),
                 (unsigned long)timestamp);
         } else {
-            DDLogDebug(@"%@ incoming message from: %@ with timestampe: %lu",
+            DDLogDebug(@"%@ incoming message from: %@ with timestamp: %lu",
                 self.tag,
                 envelopeAddress(envelope),
                 (unsigned long)timestamp);