speed up net pumps

Author: ZeroBitsTech

src/modules/net/ZeroHttpPump.h

@@ -64,6 +64,7 @@ class ZeroHttpPump {
   };
 
   static const uint8_t kQueueCapacity = ZEROKERNEL_HTTP_PUMP_QUEUE_CAPACITY;
+  static const uint8_t kImmediatePhaseBudget = 4;
 
   ZeroHttpPump()
       : kernel_(NULL),
@@ -164,29 +165,34 @@ class ZeroHttpPump {
     hasTicked_ = true;
     lastTickAtMs_ = nowMs;
 
-    if (!active_) {
-      if (queueCount_ == 0) {
-        return;
+    for (uint8_t stepBudget = 0; stepBudget < kImmediatePhaseBudget; ++stepBudget) {
+      if (!active_) {
+        if (queueCount_ == 0) {
+          return;
+        }
+
+        startNextRequest_(nowMs);
       }
 
-      startNextRequest_(nowMs);
-    }
+      if (phase_ == kPhaseConnecting && nextRetryAtMs_ != 0 && nowMs < nextRetryAtMs_) {
+        return;
+      }
 
-    if (phase_ == kPhaseConnecting && nextRetryAtMs_ != 0 && nowMs < nextRetryAtMs_) {
-      return;
-    }
+      const StepResult result = stepCurrentPhase_();
+      if (result == kStepPending) {
+        return;
+      }
 
-    const StepResult result = stepCurrentPhase_();
-    if (result == kStepPending) {
-      return;
-    }
+      if (result == kStepFailed) {
+        handlePhaseFailure_(nowMs);
+        return;
+      }
 
-    if (result == kStepFailed) {
-      handlePhaseFailure_(nowMs);
-      return;
+      handlePhaseSuccess_(nowMs);
+      if (!active_) {
+        return;
+      }
     }
-
-    handlePhaseSuccess_(nowMs);
   }
 
   bool isBusy() const {

src/modules/net/ZeroMqttPump.h

@@ -49,6 +49,7 @@ class ZeroMqttPump {
   };
 
   static const uint8_t kQueueCapacity = ZEROKERNEL_MQTT_PUMP_QUEUE_CAPACITY;
+  static const uint8_t kPublishBurstPerTick = 2;
 
   ZeroMqttPump()
       : kernel_(NULL),
@@ -182,21 +183,24 @@ class ZeroMqttPump {
     }
 
     pendingRetryAtMs_ = 0;
-    metrics_.recordSendAttempt();
-    if (publishStep_(queue_[0].topicKey, queue_[0].value, context_)) {
-      metrics_.recordSendResult(true, 0, nowMs - queue_[0].queuedAtMs);
-      popFront_();
-      return;
-    }
+    for (uint8_t burst = 0; burst < kPublishBurstPerTick && queueCount_ > 0; ++burst) {
+      metrics_.recordSendAttempt();
+      if (publishStep_(queue_[0].topicKey, queue_[0].value, context_)) {
+        metrics_.recordSendResult(true, 0, nowMs - queue_[0].queuedAtMs);
+        popFront_();
+        continue;
+      }
 
-    metrics_.recordSendResult(false, 0, nowMs - queue_[0].queuedAtMs);
-    if (queue_[0].retriesRemaining == 0) {
-      popFront_();
+      metrics_.recordSendResult(false, 0, nowMs - queue_[0].queuedAtMs);
+      if (queue_[0].retriesRemaining == 0) {
+        popFront_();
+        return;
+      }
+
+      --queue_[0].retriesRemaining;
+      scheduleBackoff_(nowMs);
       return;
     }
-
-    --queue_[0].retriesRemaining;
-    scheduleBackoff_(nowMs);
   }
 
   bool isConnected() const {

tests/desktop/KernelTests.cpp

@@ -940,20 +940,10 @@ int testHttpPumpModule() {
   g_fakeNowMs = 2;
   isolatedKernel.tick();
   pump.tick();
-  expectTrue(pump.phase() == zerokernel::modules::net::ZeroHttpPump::kPhaseWriting,
-             "http pump advances to write phase");
-
-  g_fakeNowMs = 3;
-  isolatedKernel.tick();
-  pump.tick();
-  expectTrue(pump.phase() == zerokernel::modules::net::ZeroHttpPump::kPhaseReading,
-             "http pump advances to read phase");
-
-  g_fakeNowMs = 4;
-  isolatedKernel.tick();
-  pump.tick();
   expectTrue(pump.phase() == zerokernel::modules::net::ZeroHttpPump::kPhaseConnecting,
-             "http pump retries after read failure");
+             "http pump retries after collapsed phase failure");
+  expectTrue(g_httpWriteCalls == 1, "http pump collapses write phase in same tick");
+  expectTrue(g_httpReadCalls == 1, "http pump collapses read phase in same tick");
 
   g_fakeNowMs = 10;
   isolatedKernel.tick();
@@ -963,15 +953,6 @@ int testHttpPumpModule() {
   g_fakeNowMs = 24;
   isolatedKernel.tick();
   pump.tick();
-  g_fakeNowMs = 25;
-  isolatedKernel.tick();
-  pump.tick();
-  g_fakeNowMs = 26;
-  isolatedKernel.tick();
-  pump.tick();
-  g_fakeNowMs = 27;
-  isolatedKernel.tick();
-  pump.tick();
 
   const zerokernel::modules::net::ZeroTransportMetrics::Snapshot httpMetrics =
       pump.metrics().snapshot();