add schobi scheduler

build_and_bench_all.py

@@ -15,7 +15,7 @@
 import shutil
 
 runtimes = {
-    "cpp": ["citor", "libfork", "TooManyCooks", "tbb", "taskflow", "cppcoro", "coros", "cobalt",
+    "cpp": ["citor", "libfork", "TooManyCooks", "schobi", "tbb", "taskflow", "cppcoro", "coros", "cobalt",
             # these 4 are quite slow - you can remove them to speed up total runtime
             "folly", "concurrencpp", "HPX", "libcoro"]
 }
@@ -26,6 +26,7 @@
     "citor": "https://github.com/Lallapallooza/citor",
     "libfork": "https://github.com/ConorWilliams/libfork",
     "TooManyCooks": "https://github.com/tzcnt/TooManyCooks",
+    "schobi": "https://codeberg.org/Schobi/Schobi",
     "tbb": "https://www.intel.com/content/www/us/en/developer/tools/oneapi/onetbb.html",
     "taskflow": "https://github.com/taskflow/taskflow",
     "cppcoro": "https://github.com/andreasbuhr/cppcoro",
@@ -65,6 +66,22 @@
 # The runtime name will be suffixed with "_<config>" in output (e.g., "cobalt_st_asio")
 # Format: { "runtime": { "benchmark": ["config1", "config2", ...], ... }, ... }
 benchmark_configs = {
+    # schobi ships three implementations of each fork-join benchmark. The empty
+    # config is the recursive fork/join every runtime here runs, under the plain
+    # name, and is the only one comparable to the other rows. "sequential" is the
+    # same problem on one thread with no scheduler started - the baseline the rest
+    # are measured against. "group-shared" regroups the work into one fiber group
+    # that folds its own partials through group barriers.
+    #
+    # Note that the ranking below takes the best row per benchmark as the
+    # denominator, so schobi_sequential and schobi_group-shared both compete for
+    # it despite running a different program from everyone else.
+    "schobi": {
+        "fib": ["", "sequential", "group-shared"],
+        "skynet": ["", "sequential", "group-shared"],
+        "nqueens": ["", "sequential", "group-shared"],
+        "matmul": ["", "sequential", "group-shared"],
+    },
     "cobalt": {
         "channel": ["st_asio"]
     },

cpp/2common/bench_config.hpp

@@ -0,0 +1,44 @@
+#pragma once
+// Selects which implementation of a benchmark to run. The harness appends the
+// config name after the thread count; an absent or unrecognised name is the
+// default, per-node program that every runtime here implements.
+//
+// group-shared cuts the problem into one chunk per task up front and reduces the
+// partials, instead of spawning a task per node. It is the same arithmetic, so
+// it stays a valid answer to the same benchmark, but it measures bulk dispatch
+// rather than spawn cost.
+
+#include <cstring>
+
+namespace bench {
+
+enum class Config {
+  fork_join,
+  group_shared,
+};
+
+inline constexpr const char* fork_join_name = "fork-join";
+inline constexpr const char* group_shared_name = "group-shared";
+
+// tasks per worker in the group-shared implementations: enough chunks that a
+// late-finishing one can be absorbed by the rest, few enough to stay bulk work
+inline constexpr size_t tasks_per_worker = 8;
+
+inline Config parse_config(const char* name) {
+  if (name != nullptr && std::strcmp(name, group_shared_name) == 0) {
+    return Config::group_shared;
+  }
+  return Config::fork_join;
+}
+
+inline const char* config_name(Config config) {
+  switch (config) {
+  case Config::group_shared:
+    return group_shared_name;
+  case Config::fork_join:
+    break;
+  }
+  return fork_join_name;
+}
+
+} // namespace bench

cpp/libfork/fib.cpp

@@ -13,6 +13,8 @@
 #include <cinttypes>
 #include <cstdio>
 #include <cstdlib>
+#include <vector>
+#include "bench_config.hpp"
 #include "memusage.hpp"
 #include <libfork.hpp>
 
@@ -34,6 +36,79 @@ inline constexpr auto fib = [](auto fib, size_t n) -> lf::task<size_t> {
   co_return x + y;
 };
 
+
+// --------------------------------------------------------------------------
+// group-shared
+// --------------------------------------------------------------------------
+// fib(n) is expanded breadth-first into a frontier of subproblems, one chunk of
+// which each task sums without the scheduler. Replacing v with v-1 and v-2 keeps
+// sum(fib(frontier)) equal to fib(n), so the same tree is walked.
+
+inline constexpr size_t fib_leaf = 2; // below this a node is its own value
+// room for the frontier to overshoot its target on the last expansion pass
+inline constexpr size_t fib_frontier_slots_per_task = 4;
+
+static size_t fib_serial(size_t n) {
+  return n < fib_leaf ? n : fib_serial(n - 1) + fib_serial(n - 2);
+}
+
+static size_t
+build_frontier(std::vector<unsigned>& frontier, size_t n, size_t target) {
+  size_t capacity = frontier.size();
+  size_t count = 1;
+  frontier[0] = static_cast<unsigned>(n);
+  for (;;) {
+    if (count >= target || count * 2 > capacity) {
+      break;
+    }
+    size_t split = 0;
+    for (size_t i = count; i-- > 0;) {
+      unsigned v = frontier[i];
+      if (v < fib_leaf) {
+        continue;
+      }
+      frontier[i] = v - 1;
+      frontier[count + split] = v - 2;
+      ++split;
+    }
+    if (split == 0) {
+      break;
+    }
+    count += split;
+  }
+  return count;
+}
+
+inline constexpr auto fib_chunk =
+  [](auto, unsigned const* frontier, size_t begin, size_t end) -> lf::task<size_t> {
+  size_t sum = 0;
+  for (size_t i = begin; i < end; ++i) {
+    sum += fib_serial(frontier[i]);
+  }
+  co_return sum;
+};
+
+inline constexpr auto fib_grouped =
+  [](auto, size_t n, size_t tasks) -> lf::task<size_t> {
+  std::vector<unsigned> frontier(tasks * fib_frontier_slots_per_task);
+  size_t count = build_frontier(frontier, n, tasks);
+
+  std::vector<size_t> results(tasks);
+  unsigned const* data = frontier.data();
+  for (size_t t = 0; t < tasks; ++t) {
+    co_await lf::fork[&results[t], fib_chunk](
+      data, t * count / tasks, (t + 1) * count / tasks
+    );
+  }
+  co_await lf::join;
+
+  size_t sum = 0;
+  for (size_t t = 0; t < tasks; ++t) {
+    sum += results[t];
+  }
+  co_return sum;
+};
+
 int main(int argc, char* argv[]) {
   if (argc > 2) {
     thread_count = static_cast<size_t>(atoi(argv[2]));
@@ -43,16 +118,25 @@ int main(int argc, char* argv[]) {
     exit(0);
   }
   size_t n = static_cast<size_t>(atoi(argv[1]));
+  bench::Config config = bench::parse_config(argc > 3 ? argv[3] : nullptr);
+  size_t tasks = thread_count * bench::tasks_per_worker;
 
   std::printf("threads: %" PRIu64 "\n", thread_count);
+  std::printf("config: %s\n", bench::config_name(config));
   lf::lazy_pool pool(thread_count);
 
-  auto result = lf::sync_wait(pool, fib, 30); // warmup
+  auto run = [&](size_t value) {
+    return config == bench::Config::group_shared
+             ? lf::sync_wait(pool, fib_grouped, value, tasks)
+             : lf::sync_wait(pool, fib, value);
+  };
+
+  run(30); // warmup
 
   auto startTime = std::chrono::high_resolution_clock::now();
 
   for (size_t i = 0; i < iter_count; ++i) {
-    auto result = lf::sync_wait(pool, fib, n);
+    auto result = run(n);
     std::printf("output: %zu\n", result);
   }
 

cpp/libfork/matmul.cpp

@@ -7,6 +7,7 @@
 // Distributed under the Boost Software License, Version 1.0. (See accompanying
 // file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt)
 
+#include "bench_config.hpp"
 #include "matmul.hpp"
 #include "memusage.hpp"
 #include <libfork.hpp>
@@ -47,7 +48,49 @@ inline constexpr auto matmul =
   }
 };
 
-std::vector<int> run_matmul(lf::lazy_pool& executor, int N) {
+// --------------------------------------------------------------------------
+// group-shared
+// --------------------------------------------------------------------------
+// One task per block of C, each accumulating over the k blocks. Every block of C
+// has a single writer, so the two-phase split the recursion needs disappears and
+// the whole multiply is one dispatch. Same N^3 multiply-accumulates.
+
+// sized for the cache: a task holds a block of C live while streaming a block
+// row of A against a block column of B
+inline constexpr int matmul_blocked_block = 128;
+
+inline constexpr auto matmul_block_row =
+  [](auto, int* a, int* b, int* c, int N, int nb, int begin,
+     int end) -> lf::task<void> {
+  for (int idx = begin; idx < end; ++idx) {
+    int bi = idx / nb;
+    int bj = idx % nb;
+    int* cblock = c + bi * matmul_blocked_block * N + bj * matmul_blocked_block;
+    for (int k = 0; k < nb; ++k) {
+      matmul_small(
+        a + bi * matmul_blocked_block * N + k * matmul_blocked_block,
+        b + k * matmul_blocked_block * N + bj * matmul_blocked_block, cblock,
+        matmul_blocked_block, N
+      );
+    }
+  }
+  co_return;
+};
+
+inline constexpr auto matmul_grouped =
+  [](auto, int* a, int* b, int* c, int N, size_t tasks) -> lf::task<void> {
+  int nb = N / matmul_blocked_block;
+  int blocks = nb * nb;
+  for (size_t t = 0; t < tasks; ++t) {
+    int begin = static_cast<int>(t * blocks / tasks);
+    int end = static_cast<int>((t + 1) * blocks / tasks);
+    co_await lf::fork[matmul_block_row](a, b, c, N, nb, begin, end);
+  }
+  co_await lf::join;
+};
+
+std::vector<int>
+run_matmul(lf::lazy_pool& executor, int N, bench::Config config, size_t tasks) {
   std::vector<int> A(N * N, 1);
   std::vector<int> B(N * N, 1);
   std::vector<int> C(N * N, 0);
@@ -56,7 +99,11 @@ std::vector<int> run_matmul(lf::lazy_pool& executor, int N) {
   int* b = B.data();
   int* c = C.data();
 
-  lf::sync_wait(executor, matmul, a, b, c, N, N);
+  if (config == bench::Config::group_shared) {
+    lf::sync_wait(executor, matmul_grouped, a, b, c, N, tasks);
+  } else {
+    lf::sync_wait(executor, matmul, a, b, c, N, N);
+  }
   return C;
 }
 
@@ -77,9 +124,9 @@ void validate_result(std::vector<int>& C, int N) {
   }
 }
 
-void run_one(lf::lazy_pool& executor, int N) {
+void run_one(lf::lazy_pool& executor, int N, bench::Config config, size_t tasks) {
   auto startTime = std::chrono::high_resolution_clock::now();
-  std::vector<int> result = run_matmul(executor, N);
+  std::vector<int> result = run_matmul(executor, N, config, tasks);
   auto endTime = std::chrono::high_resolution_clock::now();
   validate_result(result, N);
   auto totalTimeUs =
@@ -98,12 +145,16 @@ int main(int argc, char* argv[]) {
     exit(0);
   }
   int n = atoi(argv[1]);
+  bench::Config config = bench::parse_config(argc > 3 ? argv[3] : nullptr);
+  size_t tasks = thread_count * bench::tasks_per_worker;
+
   std::printf("threads: %zu\n", thread_count);
+  std::printf("config: %s\n", bench::config_name(config));
   lf::lazy_pool executor(thread_count);
 
-  run_matmul(executor, n); // warmup
+  run_matmul(executor, n, config, tasks); // warmup
 
   std::printf("runs:\n");
 
-  run_one(executor, n);
+  run_one(executor, n, config, tasks);
 }