Batch and coalesce mass fiber wakeups

include/silk/fibers/fiber.h

@@ -260,6 +260,13 @@ class FiberScheduler
      */
     static bool schedule(Fiber * fiber) noexcept;
 
+    /**
+     * Resume a batch of suspended fibers, coalescing their cross-CPU wakeups: each is enqueued to its
+     * home processor, then at most one doorbell per distinct target processor is posted and all are
+     * delivered with a single io_uring_submit. Used by FiberFuture::setAll for mass wakeups.
+     */
+    static void scheduleAll(Fiber ** fibers, uint64_t count) noexcept;
+
     /**
      * Suspend the current fiber and immediately reschedule it, giving other
      * fibers a chance to run. No-op when called from a non-fiber thread.
@@ -439,7 +446,8 @@ class FiberScheduler
      * @param bytesRead If not null, receives the number of bytes read on success.
      * @param future    Completion handle.
      */
-    static void readFixed(int fd, void * buf, uint32_t len, uint64_t offset, int bufIndex, uint64_t * bytesRead, IoFuture * future) noexcept;
+    static void
+    readFixed(int fd, void * buf, uint32_t len, uint64_t offset, int bufIndex, uint64_t * bytesRead, IoFuture * future) noexcept;
 
     /**
      * Async fixed-buffer write: IORING_OP_WRITE_FIXED. See readFixed.
@@ -653,7 +661,7 @@ class FiberScheduler
     static Fiber *
     allocateFiber(FiberMain * fiberMain, FiberParametersDtor * parametersDtor, uint8_t category, FiberFuture * future) noexcept;
     static void freeFiber(Fiber * fiber) noexcept;
-    static void enqueueReady(Fiber * fiber) noexcept;
+    static ProcessorState * enqueueReady(ProcessorState * processor, Fiber * fiber) noexcept;
     static void yieldSuspendCallback(Fiber * fiber, void * context) noexcept;
     static void enterThreadModeSuspendCallback(Fiber * fiber, void * context) noexcept;
     static void exitThreadModeSuspendCallback(Fiber * fiber, void * context) noexcept;

include/silk/fibers/future.h

@@ -118,7 +118,20 @@ class FiberFuture
      */
     [[nodiscard]] static int waitWithTimeout(FiberFuture * future, uint64_t nanoseconds) noexcept;
 
+    /**
+     * Set the same result on a batch of futures and wake their waiters together. Equivalent to calling
+     * set(err) on each, but the plain-fiber waiter wakeups are coalesced into a single batched schedule.
+     * Each future must be set at most once.
+     */
+    static void setAll(int err, FiberFuture ** futures, uint64_t count) noexcept;
+
 private:
+    //
+    // Constants.
+    //
+
+    static constexpr uint32_t WAKE_BATCH = 32;
+
     /**
      * Packed future state.
      */
@@ -152,6 +165,7 @@ class FiberFuture
     //
 
     void signal() noexcept;
+    Fiber * extractWaitingFiber() noexcept;
     int suspend() noexcept;
     static void suspendCallback(Fiber * fiber, FiberFuture * future) noexcept;
     bool attachWaiter(MultipleWaitState * waitState) noexcept;

include/silk/fibers/sequencer.h

@@ -54,40 +54,69 @@ class FiberSequencer
     /** Wait until the counter reaches @p token, blocking the calling fiber. */
     void wait(uint64_t token) noexcept
     {
+        // Fast path: counter already satisfied.
+        if (counter.load(std::memory_order_acquire) >= token)
+        {
+            return;
+        }
+
         Future future;
-        wait(token, &future);
+        registerWaiter(token, &future);
         future.wait();
     }
 
     /**
      * Register @p future to be set when the counter reaches @p token.
      * Sets the future immediately if the counter is already >= @p token.
      */
-    void wait(uint64_t token, Future * future) noexcept;
+    void wait(uint64_t token, Future * future) noexcept
+    {
+        // Fast path: counter already satisfied.
+        if (counter.load(std::memory_order_acquire) >= token)
+        {
+            future->set(0);
+            return;
+        }
+
+        registerWaiter(token, future);
+    }
 
     /**
      * Increment the counter and wake all futures whose token has been reached.
      * Returns the new counter value.
      */
-    uint64_t increment() noexcept;
+    uint64_t increment() noexcept
+    {
+        uint64_t current = counter.fetch_add(1, std::memory_order_release) + 1;
+        drain();
+        return current;
+    }
 
     /**
      * Advance the counter to @p value if @p value exceeds the current counter.
      * Wakes all futures whose token is now reached.
      * Returns true if the counter was advanced, false if it was already >= @p value.
      */
-    [[nodiscard]] bool advance(uint64_t value) noexcept;
-
-private:
-    struct FutureCompare
+    [[nodiscard]] bool advance(uint64_t value) noexcept
     {
-        bool operator()(const Future & l, const Future & r) const noexcept { return l.token < r.token; }
-    };
+        uint64_t current = counter.load(std::memory_order_relaxed);
+        for (;;)
+        {
+            if (current >= value)
+            {
+                return false;
+            }
+            if (counter.compare_exchange_weak(current, value, std::memory_order_release, std::memory_order_relaxed))
+            {
+                break;
+            }
+        }
 
-    using RequestQueue = LockFreeStack<Future, &Future::stackEntry>;
-    using WaiterTree = Tree<Future, &Future::treeEntry, FutureCompare, true /* AllowDuplicates */>;
-    using WaitList = Stack<Future, &Future::stackEntry>;
+        drain();
+        return true;
+    }
 
+private:
     //
     // Constants.
     //
@@ -96,14 +125,31 @@ class FiberSequencer
     static constexpr uint32_t BUSY = 1;
     static constexpr uint32_t PENDING = 2;
 
+    static constexpr uint32_t WAKE_BATCH = 32;
+
+    //
+    // Data structures.
+    //
+
+    struct FutureCompare
+    {
+        bool operator()(const Future & l, const Future & r) const noexcept { return l.token < r.token; }
+    };
+
+    using RequestQueue = LockFreeStack<Future, &Future::stackEntry>;
+    using WaiterTree = Tree<Future, &Future::treeEntry, FutureCompare, true /* AllowDuplicates */>;
+    using WaitList = Stack<Future, &Future::stackEntry>;
+
     //
     // Helpers.
     //
 
+    void registerWaiter(uint64_t token, Future * future) noexcept;
     void cancelWait(Future * future) noexcept;
     void drain() noexcept;
     bool acquireCombiner() noexcept;
     bool releaseCombiner() noexcept;
+    static void setAll(WaitList * wakeList, int err) noexcept;
 
     //
     // State.

include/silk/util/bitmap.h

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <silk/util/platform.h>
+
+#include <cstdint>
+#include <cstring>
+
+namespace silk
+{
+
+/**
+ * Non-owning view over a bitmap of bitCount bits packed into 64-bit words. The words are caller-owned;
+ * size that buffer with wordCount.
+ */
+class Bitmap
+{
+public:
+    Bitmap(uint64_t * words, uint32_t bitCount) noexcept
+        : words(words)
+        , bitCount(bitCount)
+    {
+    }
+
+    /** Number of 64-bit words a bitmap of bitCount bits occupies. */
+    static constexpr uint32_t wordCount(uint32_t bitCount) noexcept { return alignUp(bitCount, WORD_BITS) / WORD_BITS; }
+
+    /** True if the bit at index is set. */
+    bool test(uint32_t index) const noexcept { return (words[index / WORD_BITS] >> (index % WORD_BITS)) & 1; }
+
+    /** Set the bit at index. */
+    void set(uint32_t index) noexcept { words[index / WORD_BITS] |= ONE << (index % WORD_BITS); }
+
+    /** Clear the bit at index. */
+    void clear(uint32_t index) noexcept { words[index / WORD_BITS] &= ~(ONE << (index % WORD_BITS)); }
+
+    /** Zero every word. */
+    void clear() noexcept { std::memset(words, 0, wordCount(bitCount) * sizeof(uint64_t)); }
+
+    /**
+     * Lowest bit at or after from whose value is value (set if true, clear if false): writes its index to
+     * index and returns true, or returns false if none remain. Enumerate by advancing from one past each hit.
+     */
+    bool findBit(uint32_t from, bool value, uint32_t * index) const noexcept;
+
+private:
+    static constexpr uint32_t WORD_BITS = 64;
+
+    static constexpr uint64_t ZERO = 0;
+    static constexpr uint64_t ONE = 1;
+
+    uint64_t * words;
+    uint32_t bitCount;
+};
+
+} // namespace silk

include/silk/util/bounded-queue.h

@@ -32,7 +32,10 @@ class BoundedQueue
      */
     BoundedQueue() noexcept = default;
 
-    // TBD
+    /**
+     * Construct and initialize in one step when the capacity is known at construction. Capacity must be a
+     * power of two and >= 2 (see initialize).
+     */
     explicit BoundedQueue(uint64_t capacity) noexcept { initialize(capacity); }
 
     /**

src/fibers/benchmarks/future-bench.cpp

@@ -0,0 +1,94 @@
+#include <silk/fibers/future.h>
+
+#include <silk/fibers/fiber.h>
+#include <silk/util/assert.h>
+
+#include <benchmark/benchmark.h>
+
+#include <cstdint>
+#include <vector>
+
+namespace silk
+{
+
+// Error value the driver wakes a waiter with to make it exit; normal wakeups use 0.
+static constexpr int kStopWaiter = -1;
+
+class FiberFutureBench : public benchmark::Fixture
+{
+};
+
+// Batched wake: count waiter fibers each park on a future; the timed region calls setAll once to wake them
+// all. The re-park barrier (waiting every waiter back onto its future) runs under PauseTiming, so the
+// measurement is setAll's dispatch - extractWaitingFiber per future plus scheduleAll's coalesced doorbell
+// (one msg_ring per distinct parked processor, one submit). Exercises real fiber wakeups, unlike the
+// sequencer FanIn which registers bare futures with no fiber waiter.
+BENCHMARK_DEFINE_F(FiberFutureBench, SetAllWake)(benchmark::State & state)
+{
+    uint64_t count = static_cast<uint64_t>(state.range(0));
+
+    struct Waiter
+    {
+        FiberFuture * wake;
+        FiberFuture * ready;
+
+        static int fiberMain(Waiter * p) noexcept
+        {
+            for (;;)
+            {
+                p->ready->set(0); // re-armed and about to park
+                int r = p->wake->wait();
+                p->wake->reset();
+                if (r == kStopWaiter)
+                {
+                    break;
+                }
+            }
+            return 0;
+        }
+    };
+
+    std::vector<FiberFuture> wake(count);
+    std::vector<FiberFuture> ready(count);
+    std::vector<FiberFuture> done(count);
+    std::vector<FiberFuture *> wakePointers(count);
+
+    for (uint64_t i = 0; i < count; ++i)
+    {
+        wakePointers[i] = &wake[i];
+    }
+    for (uint64_t i = 0; i < count; ++i)
+    {
+        int r = FiberScheduler::run(Waiter::fiberMain, {&wake[i], &ready[i]}, &done[i]);
+        SILK_ASSERT(!r);
+    }
+
+    auto awaitAllParked = [&]() noexcept
+    {
+        for (uint64_t i = 0; i < count; ++i)
+        {
+            ready[i].wait();
+            ready[i].reset();
+        }
+    };
+
+    for (auto _ : state)
+    {
+        state.PauseTiming();
+        awaitAllParked();
+        state.ResumeTiming();
+
+        FiberFuture::setAll(0, wakePointers.data(), count);
+    }
+
+    // Drain: every waiter is parked (barrier), so one stop-wake makes them all exit.
+    awaitAllParked();
+    FiberFuture::setAll(kStopWaiter, wakePointers.data(), count);
+    for (uint64_t i = 0; i < count; ++i)
+    {
+        done[i].wait();
+    }
+}
+BENCHMARK_REGISTER_F(FiberFutureBench, SetAllWake)->Arg(8)->Arg(64)->Arg(512);
+
+} // namespace silk