From 6e48a0ff9ad72127069da277823e03feb7a8945b Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew@weaselab.dev>
Date: Tue, 26 Aug 2025 15:37:04 -0400
Subject: [PATCH] Let WaitIfUpstreamIdle spin long enough to stay saturated by
 load_tester

---
 benchmarks/bench_volatile_loop.cpp | 14 ++++++++------
 src/http_handler.hpp               |  5 +++--
 src/loop_iterations.h              |  3 +++
 src/thread_pipeline.hpp            | 13 ++++++++++---
 4 files changed, 24 insertions(+), 11 deletions(-)
 create mode 100644 src/loop_iterations.h

diff --git a/benchmarks/bench_volatile_loop.cpp b/benchmarks/bench_volatile_loop.cpp
index 92ed026..c6f6109 100644
--- a/benchmarks/bench_volatile_loop.cpp
+++ b/benchmarks/bench_volatile_loop.cpp
@@ -1,12 +1,14 @@
 #include <nanobench.h>
 
+#include "../src/loop_iterations.h"
+
 int main() {
-  constexpr int loopIterations = 800;
-  ankerl::nanobench::Bench().run(
-      "volatile loop to " + std::to_string(loopIterations), [&] {
-        for (volatile int i = 0; i < loopIterations; i = i + 1)
-          ;
-      });
+  ankerl::nanobench::Bench bench;
+  bench.minEpochIterations(100000);
+  bench.run("volatile loop to " + std::to_string(loopIterations), [&] {
+    for (volatile int i = 0; i < loopIterations; i = i + 1)
+      ;
+  });
 
   return 0;
 }
diff --git a/src/http_handler.hpp b/src/http_handler.hpp
index f1fd3c8..5930d90 100644
--- a/src/http_handler.hpp
+++ b/src/http_handler.hpp
@@ -8,6 +8,7 @@
 
 #include "connection.hpp"
 #include "connection_handler.hpp"
+#include "loop_iterations.h"
 #include "perfetto_categories.hpp"
 #include "server.hpp"
 #include "thread_pipeline.hpp"
@@ -108,7 +109,7 @@ struct HttpHandler : ConnectionHandler {
           if (!c) {
             return;
           }
-          for (volatile int i = 0; i < 1200; i = i + 1)
+          for (volatile int i = 0; i < loopIterations; i = i + 1)
             ;
         }
       }
@@ -152,7 +153,7 @@ private:
   static constexpr int kFinalStageThreads = 2;
   static constexpr int kLogSize = 12;
   StaticThreadPipeline<std::unique_ptr<Connection>,
-                       WaitStrategy::WaitIfStageEmpty, 1, 2>
+                       WaitStrategy::WaitIfUpstreamIdle, 1, 2>
       pipeline{kLogSize};
   std::thread stage0Thread;
   std::vector<std::thread> finalStageThreads;
diff --git a/src/loop_iterations.h b/src/loop_iterations.h
new file mode 100644
index 0000000..0eaf3f2
--- /dev/null
+++ b/src/loop_iterations.h
@@ -0,0 +1,3 @@
+#pragma once
+
+constexpr int loopIterations = 1550;
diff --git a/src/thread_pipeline.hpp b/src/thread_pipeline.hpp
index b6652b5..a68fab5 100644
--- a/src/thread_pipeline.hpp
+++ b/src/thread_pipeline.hpp
@@ -114,10 +114,17 @@ uint32_t calculate_safe_len(
                 // Empty - busy wait
               } else if constexpr (wait_strategy ==
                                    WaitStrategy::WaitIfUpstreamIdle) {
-                auto push = pushes.load(std::memory_order_relaxed);
-                if (push == thread.local_pops) {
-                  pushes.wait(push, std::memory_order_relaxed);
+                // We're allowed to spin as long as we eventually go to 0% cpu
+                // usage on idle
+                uint32_t push;
+                for (int i = 0; i < 100000; ++i) {
+                  push = pushes.load(std::memory_order_relaxed);
+                  if (push != thread.local_pops) {
+                    goto dont_wait;
+                  }
                 }
+                pushes.wait(push, std::memory_order_relaxed);
+              dont_wait:;
               } else {
                 static_assert(wait_strategy == WaitStrategy::WaitIfStageEmpty);
                 last_push.wait(thread.last_push_read[Is],