Strengthen remote polling coverage

Author: fallintoplace

examples/benchmark/common.hpp

@@ -23,7 +23,6 @@
 #include <barrier>
 #include <chrono>
 #include <cmath>
-#include <cstdlib>
 #include <iomanip>
 #include <iostream>
 #include <memory>
@@ -121,10 +120,6 @@ void my_main(int argc, char** argv, exec::numa_policy policy = exec::get_numa_po
     nthreads = std::atoi(argv[1]);
   }
   std::size_t total_scheds = 10'000'000;
-  if (argc > 2)
-  {
-    total_scheds = static_cast<std::size_t>(std::strtoull(argv[2], nullptr, 10));
-  }
 #ifndef STDEXEC_NO_MONOTONIC_BUFFER_RESOURCE
   std::vector<std::unique_ptr<char, numa_deleter>> buffers;
 #endif

test/exec/test_static_thread_pool.cpp

@@ -26,6 +26,7 @@
 #include <exception>
 #include <latch>
 #include <mutex>
+#include <optional>
 #include <ranges>
 #include <stdexcept>
 #include <thread>
@@ -234,86 +235,106 @@ TEST_CASE("bulk on static_thread_pool executes on multiple threads, take 2",
   REQUIRE(thread_ids.size() == num_of_threads);
 }
 
-TEST_CASE("static_thread_pool drains remote work after idle transitions",
-          "[types][static_thread_pool][stress]")
+namespace
 {
-  constexpr std::size_t num_producers = 4;
-  constexpr std::size_t rounds        = 10'000;
-
-  std::latch                            ready{num_producers};
-  std::atomic<bool>                     start{false};
-  std::atomic<bool>                     stop{false};
-  std::vector<std::atomic<std::size_t>> completed(num_producers);
-  std::vector<std::thread>              producers;
-  producers.reserve(num_producers);
-  for (auto& count: completed)
+  void run_remote_poll_stress(bool separate_schedulers)
   {
-    count.store(0, std::memory_order_relaxed);
-  }
-
-  exec::static_thread_pool pool{1};
-  auto                     scheduler = pool.get_scheduler();
+    constexpr std::size_t num_producers = 4;
+    constexpr std::size_t rounds        = 10'000;
+
+    std::latch                            ready{num_producers};
+    std::atomic<bool>                     start{false};
+    std::atomic<bool>                     stop{false};
+    std::vector<std::atomic<std::size_t>> completed(num_producers);
+    std::vector<std::thread>              producers;
+    producers.reserve(num_producers);
+    for (auto& count: completed)
+    {
+      count.store(0, std::memory_order_relaxed);
+    }
 
-  for (std::size_t producer = 0; producer < num_producers; ++producer)
-  {
-    producers.emplace_back(
-      [&, producer]
-      {
-        ready.count_down();
-        while (!start.load(std::memory_order_acquire))
-        {
-          std::this_thread::yield();
-        }
+    exec::static_thread_pool pool{1};
+    using scheduler_t = decltype(pool.get_scheduler());
+    std::optional<scheduler_t> shared_scheduler;
+    if (!separate_schedulers)
+    {
+      shared_scheduler.emplace(pool.get_scheduler());
+    }
 
-        auto* const producer_completed = &completed[producer];
-        std::size_t expected           = 0;
-        for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed);
-             ++round)
+    for (std::size_t producer = 0; producer < num_producers; ++producer)
+    {
+      producers.emplace_back(
+        [&, producer]
         {
-          std::size_t const batch_size = (round % 4 == 0) ? 2 : 1;
-          expected += batch_size;
-          for (std::size_t i = 0; i < batch_size; ++i)
+          auto scheduler = separate_schedulers ? pool.get_scheduler() : *shared_scheduler;
+          ready.count_down();
+          while (!start.load(std::memory_order_acquire))
           {
-            exec::start_detached(
-              ex::schedule(scheduler)
-              | ex::then([producer_completed]
-                         { producer_completed->fetch_add(1, std::memory_order_relaxed); }));
+            std::this_thread::yield();
           }
 
-          while (!stop.load(std::memory_order_relaxed)
-                 && producer_completed->load(std::memory_order_relaxed) < expected)
+          auto* const producer_completed = &completed[producer];
+          std::size_t expected           = 0;
+          for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed);
+               ++round)
           {
+            std::size_t const batch_size = (round % 4 == 0) ? 2 : 1;
+            expected += batch_size;
+            for (std::size_t i = 0; i < batch_size; ++i)
+            {
+              exec::start_detached(
+                ex::schedule(scheduler)
+                | ex::then([producer_completed]
+                           { producer_completed->fetch_add(1, std::memory_order_relaxed); }));
+            }
+
+            while (!stop.load(std::memory_order_relaxed)
+                   && producer_completed->load(std::memory_order_relaxed) < expected)
+            {
+              std::this_thread::yield();
+            }
             std::this_thread::yield();
           }
-          std::this_thread::yield();
-        }
-      });
-  }
+        });
+    }
 
-  ready.wait();
-  start.store(true, std::memory_order_release);
+    ready.wait();
+    start.store(true, std::memory_order_release);
 
-  auto const expected        = num_producers * rounds + num_producers * ((rounds + 3) / 4);
-  auto const deadline        = std::chrono::steady_clock::now() + std::chrono::seconds(10);
-  auto       completed_total = [&]
-  {
-    std::size_t result = 0;
-    for (auto const & count: completed)
+    auto const expected        = num_producers * rounds + num_producers * ((rounds + 3) / 4);
+    auto const deadline        = std::chrono::steady_clock::now() + std::chrono::seconds(10);
+    auto       completed_total = [&]
     {
-      result += count.load(std::memory_order_relaxed);
+      std::size_t result = 0;
+      for (auto const & count: completed)
+      {
+        result += count.load(std::memory_order_relaxed);
+      }
+      return result;
+    };
+
+    while (completed_total() < expected && std::chrono::steady_clock::now() < deadline)
+    {
+      std::this_thread::yield();
+    }
+    stop.store(true, std::memory_order_release);
+    for (auto& producer: producers)
+    {
+      producer.join();
     }
-    return result;
-  };
 
-  while (completed_total() < expected && std::chrono::steady_clock::now() < deadline)
-  {
-    std::this_thread::yield();
-  }
-  stop.store(true, std::memory_order_release);
-  for (auto& producer: producers)
-  {
-    producer.join();
+    CHECK(completed_total() == expected);
   }
+}  // namespace
+
+TEST_CASE("static_thread_pool drains remote work from a shared scheduler",
+          "[types][static_thread_pool][stress]")
+{
+  run_remote_poll_stress(false);
+}
 
-  CHECK(completed_total() == expected);
+TEST_CASE("static_thread_pool drains remote work from producer schedulers",
+          "[types][static_thread_pool][stress]")
+{
+  run_remote_poll_stress(true);
 }