From fad5b8b5b9e143ce40b61688e64f9e184f830207 Mon Sep 17 00:00:00 2001 From: Vadim Skipin Date: Mon, 29 Jun 2026 09:41:18 +0000 Subject: [PATCH 1/3] Fix comments --- include/silk/util/bounded-queue.h | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/include/silk/util/bounded-queue.h b/include/silk/util/bounded-queue.h index a9d8317..d6c91ad 100644 --- a/include/silk/util/bounded-queue.h +++ b/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); } /** From 152d8b3839e8617e024f01e3ff8f60ccbb4671d4 Mon Sep 17 00:00:00 2001 From: Vadim Skipin Date: Mon, 29 Jun 2026 09:38:12 +0000 Subject: [PATCH 2/3] Add Bitmap utility Non-owning view over a caller-owned bitmap of bits packed into 64-bit words, with set/test/clear and findBit for in-order enumeration of set or clear bits. Includes unit tests. --- include/silk/util/bitmap.h | 55 +++++++++ src/util/bitmap.cpp | 35 ++++++ src/util/tests/bitmap-test.cpp | 211 +++++++++++++++++++++++++++++++++ 3 files changed, 301 insertions(+) create mode 100644 include/silk/util/bitmap.h create mode 100644 src/util/bitmap.cpp create mode 100644 src/util/tests/bitmap-test.cpp diff --git a/include/silk/util/bitmap.h b/include/silk/util/bitmap.h new file mode 100644 index 0000000..4a0ea89 --- /dev/null +++ b/include/silk/util/bitmap.h @@ -0,0 +1,55 @@ +#pragma once + +#include + +#include +#include + +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 diff --git a/src/util/bitmap.cpp b/src/util/bitmap.cpp new file mode 100644 index 0000000..20bdc52 --- /dev/null +++ b/src/util/bitmap.cpp @@ -0,0 +1,35 @@ +#include + +#include +#include + +namespace silk +{ + +bool Bitmap::findBit(uint32_t from, bool value, uint32_t * index) const noexcept +{ + uint32_t totalWords = wordCount(bitCount); + uint64_t mask = ~ZERO << (from % WORD_BITS); + + for (uint32_t word = from / WORD_BITS; word < totalWords; ++word) + { + uint64_t matches = (value ? words[word] : ~words[word]) & mask; + if (matches) + { + uint32_t found = word * WORD_BITS + static_cast(std::countr_zero(matches)); + if (found >= bitCount) + { + return false; + } + + *index = found; + return true; + } + + mask = ~ZERO; + } + + return false; +} + +} // namespace silk diff --git a/src/util/tests/bitmap-test.cpp b/src/util/tests/bitmap-test.cpp new file mode 100644 index 0000000..ae47a85 --- /dev/null +++ b/src/util/tests/bitmap-test.cpp @@ -0,0 +1,211 @@ +#include + +#include + +#include +#include + +namespace silk +{ + +// Enumerate every bit equal to value by repeatedly advancing past each hit, exactly as a caller would. +static std::vector collectBits(const Bitmap & bitmap, bool value) +{ + std::vector result; + + for (uint32_t bit = 0; bitmap.findBit(bit, value, &bit); ++bit) + { + result.push_back(bit); + } + + return result; +} + +// wordCount rounds bitCount up to whole 64-bit words. +TEST(BitmapTest, WordCount) +{ + struct Case + { + uint32_t bitCount; + uint32_t expected; + }; + + const Case cases[] = {{0, 0}, {1, 1}, {64, 1}, {65, 2}, {128, 2}, {129, 3}}; + + for (const Case & testCase : cases) + { + uint32_t count = Bitmap::wordCount(testCase.bitCount); + ASSERT_EQ(count, testCase.expected) << "bitCount=" << testCase.bitCount; + } +} + +// set / test / clear address single bits, including the word-boundary bits 63 and 64, without disturbing neighbours. +TEST(BitmapTest, SetTestAndClearIndividualBits) +{ + uint64_t words[2] = {}; + Bitmap bitmap(words, 128); + + const uint32_t indices[] = {0, 63, 64, 127}; + + for (uint32_t index : indices) + { + bool before = bitmap.test(index); + ASSERT_FALSE(before) << "index=" << index; + } + + for (uint32_t index : indices) + { + bitmap.set(index); + } + + for (uint32_t index : indices) + { + bool after = bitmap.test(index); + ASSERT_TRUE(after) << "index=" << index; + } + + bool neighbourLow = bitmap.test(1); + ASSERT_FALSE(neighbourLow); + + bool neighbourHigh = bitmap.test(65); + ASSERT_FALSE(neighbourHigh); + + bitmap.clear(64); + + bool cleared = bitmap.test(64); + ASSERT_FALSE(cleared); + + bool stillSet = bitmap.test(63); + ASSERT_TRUE(stillSet); +} + +// The no-argument clear zeroes every word. +TEST(BitmapTest, ClearZeroesEveryBit) +{ + uint64_t words[2] = {}; + Bitmap bitmap(words, 128); + + bitmap.set(0); + bitmap.set(70); + bitmap.set(127); + + bitmap.clear(); + + std::vector remaining = collectBits(bitmap, true); + bool empty = remaining.empty(); + ASSERT_TRUE(empty); +} + +// findBit walks set bits in ascending order, crossing the word boundary. +TEST(BitmapTest, FindEnumeratesSetBitsInOrderAcrossWords) +{ + uint64_t words[2] = {}; + Bitmap bitmap(words, 128); + + const std::vector expected = {0, 63, 64, 65, 127}; + + for (uint32_t index : expected) + { + bitmap.set(index); + } + + std::vector found = collectBits(bitmap, true); + ASSERT_EQ(found, expected); +} + +// Searching for value false enumerates the clear bits. +TEST(BitmapTest, FindEnumeratesClearBits) +{ + uint64_t words[1] = {}; + Bitmap bitmap(words, 8); + + for (uint32_t index = 0; index < 8; ++index) + { + bitmap.set(index); + } + + bitmap.clear(1); + bitmap.clear(4); + bitmap.clear(7); + + std::vector found = collectBits(bitmap, false); + const std::vector expected = {1, 4, 7}; + ASSERT_EQ(found, expected); +} + +// findBit starts at from inclusive, skipping any earlier matching bit. +TEST(BitmapTest, FindBitHonoursFromLowerBound) +{ + uint64_t words[1] = {}; + Bitmap bitmap(words, 64); + + bitmap.set(2); + bitmap.set(10); + bitmap.set(40); + + uint32_t index; + + bool fromZero = bitmap.findBit(0, true, &index); + ASSERT_TRUE(fromZero); + ASSERT_EQ(index, 2u); + + bool fromThree = bitmap.findBit(3, true, &index); + ASSERT_TRUE(fromThree); + ASSERT_EQ(index, 10u); + + bool fromTen = bitmap.findBit(10, true, &index); + ASSERT_TRUE(fromTen); + ASSERT_EQ(index, 10u); + + bool fromEleven = bitmap.findBit(11, true, &index); + ASSERT_TRUE(fromEleven); + ASSERT_EQ(index, 40u); +} + +// findBit returns false when nothing matches at or after from. +TEST(BitmapTest, FindBitReturnsFalseWhenNoMatchRemains) +{ + uint64_t words[2] = {}; + Bitmap bitmap(words, 128); + + bitmap.set(5); + + uint32_t index; + + bool pastLast = bitmap.findBit(6, true, &index); + ASSERT_FALSE(pastLast); + + bool atEnd = bitmap.findBit(128, true, &index); + ASSERT_FALSE(atEnd); + + bitmap.clear(); + + bool none = bitmap.findBit(0, true, &index); + ASSERT_FALSE(none); +} + +// bitCount 70 spans two words; bits 70..127 are don't-care padding that read as zero. A clear-bit search +// must not report them - only bits below bitCount count. +TEST(BitmapTest, FindClearIgnoresBitsBeyondBitCount) +{ + uint64_t words[2] = {}; + Bitmap bitmap(words, 70); + + for (uint32_t position = 0; position < 70; ++position) + { + bitmap.set(position); + } + + uint32_t index; + + bool foundPadding = bitmap.findBit(0, false, &index); + ASSERT_FALSE(foundPadding); + + bitmap.clear(68); + + bool foundReal = bitmap.findBit(0, false, &index); + ASSERT_TRUE(foundReal); + ASSERT_EQ(index, 68u); +} + +} // namespace silk From d68a535439b3e505e2c2afe8485f852645b61f71 Mon Sep 17 00:00:00 2001 From: Vadim Skipin Date: Mon, 29 Jun 2026 09:40:00 +0000 Subject: [PATCH 3/3] Batch and coalesce mass fiber wakeups Add FiberFuture::setAll and FiberScheduler::scheduleAll to wake many parked fibers at once: enqueue each to its home ready queue, dedup the distinct target processors in a Bitmap, then ring each parked target's doorbell exactly once behind a single seq_cst fence and one submit - instead of a schedule/doorbell/submit per fiber. Cross-ring doorbells post straight into the target's CQ via IORING_OP_MSG_RING; IOSQE_CQE_SKIP_SUCCESS drops the send-side completion and the carrier is tagged CQE_TAG_WAKEUP for the drain to skip. Gated on the new IORING_FEAT_CQE_SKIP feature check at init. signal is split into extractWaitingFiber + schedule so setAll can reuse the extraction. enqueueReady now returns the processor whose doorbell the caller must ring (or null when it self-handled the wake), letting schedule and runFiber ring immediately while scheduleAll defers and coalesces. FiberSequencer drains through setAll; its trivial wait/increment/advance move inline to the header. Adds setAll tests and a SetAllWake benchmark. --- include/silk/fibers/fiber.h | 12 +- include/silk/fibers/future.h | 14 +++ include/silk/fibers/sequencer.h | 70 +++++++++-- src/fibers/benchmarks/future-bench.cpp | 94 +++++++++++++++ src/fibers/fiber.cpp | 154 ++++++++++++++++++++++--- src/fibers/future.cpp | 66 ++++++++--- src/fibers/sequencer.cpp | 70 +++++------ src/fibers/tests/future-test.cpp | 77 +++++++++++++ 8 files changed, 471 insertions(+), 86 deletions(-) create mode 100644 src/fibers/benchmarks/future-bench.cpp diff --git a/include/silk/fibers/fiber.h b/include/silk/fibers/fiber.h index 676f334..a12ddbc 100644 --- a/include/silk/fibers/fiber.h +++ b/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; diff --git a/include/silk/fibers/future.h b/include/silk/fibers/future.h index bf18cb7..b8e9f74 100644 --- a/include/silk/fibers/future.h +++ b/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; diff --git a/include/silk/fibers/sequencer.h b/include/silk/fibers/sequencer.h index dc85020..4a7ab25 100644 --- a/include/silk/fibers/sequencer.h +++ b/include/silk/fibers/sequencer.h @@ -54,8 +54,14 @@ 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(); } @@ -63,31 +69,54 @@ class FiberSequencer * 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; - using WaiterTree = Tree; - using WaitList = Stack; + 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; + using WaiterTree = Tree; + using WaitList = Stack; + // // 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. diff --git a/src/fibers/benchmarks/future-bench.cpp b/src/fibers/benchmarks/future-bench.cpp new file mode 100644 index 0000000..37fc8b2 --- /dev/null +++ b/src/fibers/benchmarks/future-bench.cpp @@ -0,0 +1,94 @@ +#include + +#include +#include + +#include + +#include +#include + +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(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 wake(count); + std::vector ready(count); + std::vector done(count); + std::vector 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 diff --git a/src/fibers/fiber.cpp b/src/fibers/fiber.cpp index 89d0251..3bb04a6 100644 --- a/src/fibers/fiber.cpp +++ b/src/fibers/fiber.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -50,6 +51,7 @@ FiberScheduler::SchedulerState * FiberScheduler::scheduler; static constexpr uint64_t CQE_TAG_CANCEL = 0; static constexpr uint64_t CQE_TAG_DOORBELL = 1; +static constexpr uint64_t CQE_TAG_WAKEUP = 2; // clang-format off #define FIBER_SIMPLE_COUNTERS(x) \ @@ -523,12 +525,14 @@ struct FiberScheduler::ProcessorState FiberId allocateFiberId(uint8_t category) noexcept; void profileEvent(ProfileEventKind kind, uint8_t category, uint64_t durationCycles) noexcept; - void enqueueDoorbell() noexcept; void wakeThread() noexcept; void parkThread(uint64_t waitNs, CpuTimer * timer) noexcept; - bool hasWork() const noexcept; + void enqueueDoorbell() noexcept; + void postWakeup(ProcessorState * target) noexcept; + bool enqueueWakeup(ProcessorState * target) noexcept; + template bool enqueueIo(IoFuture * future, Setup && setup) noexcept; bool submitIo(bool flush) noexcept; @@ -638,8 +642,13 @@ void FiberScheduler::ProcessorState::initialize(uint16_t cpu) noexcept int r = ::io_uring_queue_init_params(options.ioUringQueueSize, &ring, ¶ms); SILK_ASSERT(!r); + // The kernel queues overflowing CQEs (failing submit with EBUSY) rather than dropping completions, + // so a transiently full CQ never silently loses one - submitIo relies on it. SILK_ASSERT(params.features & IORING_FEAT_NODROP); + // postWakeup posts cross-ring doorbells with IOSQE_CQE_SKIP_SUCCESS to drop the send-side completion. + SILK_ASSERT(params.features & IORING_FEAT_CQE_SKIP); + // Arm the wakeup doorbell. The kernel can end the multishot poll on CQ overflow, // so handleCompletionQueueSlow re-arms it through the same path on F_MORE loss. enqueueDoorbell(); @@ -789,6 +798,48 @@ void FiberScheduler::ProcessorState::enqueueDoorbell() noexcept } } +void FiberScheduler::ProcessorState::postWakeup(ProcessorState * target) noexcept +{ + // Same lost-wakeup handshake as wakeThread: this seq_cst fence pairs with parkThread, so a target + // observed awake here cannot have missed the ready-queue store its parkThread hasWork() re-check sees. + std::atomic_thread_fence(std::memory_order_seq_cst); + + if (!target->sleeping.load(std::memory_order_acquire)) + { + return; + } + + Perf::getSimpleCounter(simpleCounters[SCHEDULER_THREAD_WAKED], number).increment(); + + // Fill the doorbell SQE and submit it; retry until the SQE is accepted (the SQ ring may be full). + for (;;) + { + bool enqueued = enqueueWakeup(target); + submitIo(true); + if (enqueued) + { + break; + } + } +} + +bool FiberScheduler::ProcessorState::enqueueWakeup(ProcessorState * target) noexcept +{ + // Fill (do not submit) a doorbell SQE on this (the caller's) ring posting straight into the target's + // CQ via IORING_OP_MSG_RING, waking its io_uring_enter2 - cheaper than the eventfd doorbell. + // IOSQE_CQE_SKIP_SUCCESS drops our send-side completion, so only the target's CQE lands, tagged + // CQE_TAG_WAKEUP for the drain to skip. Returns false if the SQ ring is momentarily full. + int targetRingFd = target->ring.ring_fd; + return enqueueIo( + nullptr, + [targetRingFd](io_uring_sqe * sqe) noexcept + { + ::io_uring_prep_msg_ring(sqe, targetRingFd, 0, CQE_TAG_WAKEUP, 0); + ::io_uring_sqe_set_data64(sqe, CQE_TAG_WAKEUP); + ::io_uring_sqe_set_flags(sqe, IOSQE_CQE_SKIP_SUCCESS); + }); +} + template bool FiberScheduler::ProcessorState::enqueueIo(IoFuture * future, Setup && setup) noexcept { @@ -1226,14 +1277,24 @@ bool FiberScheduler::schedule(Fiber * fiber) noexcept { if (fiber->tryChangeStateToReady()) { - enqueueReady(fiber); + ProcessorState * processor = &scheduler->processorState[getCurrentProcessor()]; + ProcessorState * target = enqueueReady(processor, fiber); + if (target) + { + processor->postWakeup(target); + } return true; } return false; } -void FiberScheduler::enqueueReady(Fiber * fiber) noexcept +// Shared by schedule, scheduleAll, and runFiber: place a ready fiber on its home ready queue and return +// the processor whose doorbell the caller must ring (immediately, or batched). processor is the caller's +// own processor - used as the home default and, by the caller, as the doorbell source - so it must be the +// current processor. A proxy or thread-mode fiber, or a full ready queue, rings its own wakeup here and +// returns null. +FiberScheduler::ProcessorState * FiberScheduler::enqueueReady(ProcessorState * processor, Fiber * fiber) noexcept { if (!fiber->isProxyFiber) { @@ -1246,30 +1307,84 @@ void FiberScheduler::enqueueReady(Fiber * fiber) noexcept { if (fiber->processorNumber == kInvalidProcessorNumber) { - fiber->processorNumber = getCurrentProcessor(); + fiber->processorNumber = processor->number; } - ProcessorState * processor = &scheduler->processorState[fiber->processorNumber]; - if (processor->readyQueue.enqueue(fiber)) + ProcessorState * target = &scheduler->processorState[fiber->processorNumber]; + if (target->readyQueue.enqueue(fiber)) { Perf::getSimpleCounter(simpleCounters[FIBER_ENQUEUED], processor->number).increment(); - processor->wakeThread(); - return; + return target; } - // Ready queue full: fall back to worker thread pool. + // Ready queue full: fall back to the worker thread pool. Perf::getSimpleCounter(simpleCounters[READY_QUEUE_FULL], processor->number).increment(); } scheduler->readyQueue.enqueue(fiber); - - Perf::getSimpleCounter(simpleCounters[FIBER_ENQUEUED_SHARED]).increment(); + Perf::getSimpleCounter(simpleCounters[FIBER_ENQUEUED_SHARED], processor->number).increment(); scheduler->wakeThread(); } else { fiber->wakeThread(); } + + return nullptr; +} + +void FiberScheduler::scheduleAll(Fiber ** fibers, uint64_t count) noexcept +{ + ProcessorState * processor = &scheduler->processorState[getCurrentProcessor()]; + + // Dedup the wake targets in a bitmap over all processors so each parked target is rung exactly once. + uint64_t words[Bitmap::wordCount(kInvalidProcessorNumber)]; + Bitmap wakeTargets(words, scheduler->processorCount); + wakeTargets.clear(); + + for (uint64_t i = 0; i < count; i++) + { + Fiber * fiber = fibers[i]; + if (!fiber->tryChangeStateToReady()) + { + continue; + } + + ProcessorState * target = enqueueReady(processor, fiber); + if (target) + { + wakeTargets.set(target->number); + } + } + + // One StoreLoad barrier for the whole batch (same handshake as postWakeup): the ready-queue stores + // above are ordered before the sleeping loads below, so a target parking concurrently is not missed. + std::atomic_thread_fence(std::memory_order_seq_cst); + + // Ring each distinct parked target once, filling one doorbell SQE per target; a single submit delivers + // them all (and drains the SQ for the retry if it momentarily fills). + bool enqueued = false; + for (uint32_t bit = 0; wakeTargets.findBit(bit, true, &bit); bit++) + { + ProcessorState * target = &scheduler->processorState[bit]; + if (!target->sleeping.load(std::memory_order_acquire)) + { + continue; + } + + Perf::getSimpleCounter(simpleCounters[SCHEDULER_THREAD_WAKED], processor->number).increment(); + + while (!processor->enqueueWakeup(target)) + { + processor->submitIo(true); + } + enqueued = true; + } + + if (enqueued) + { + processor->submitIo(true); + } } void FiberScheduler::yield() noexcept @@ -1823,6 +1938,14 @@ __attribute__((noinline)) bool FiberScheduler::handleCompletionQueueSlow(Process continue; } + if (tag == CQE_TAG_WAKEUP) + { + // Cross-ring wakeup doorbell (IORING_OP_MSG_RING): a pure wakeup carrier - the work is + // already in the ready queue. Covers both an incoming doorbell and the sender-side + // completion of an outgoing one. Nothing to drain or re-arm. + continue; + } + // IO completion. Every IO op is one-shot - only the doorbell is multishot - // so each IoFuture completes exactly once. A multishot IO op added later // (recv/accept multishot) would deliver IORING_CQE_F_MORE here and set the @@ -2138,7 +2261,12 @@ void FiberScheduler::runFiber(Fiber * fiber, CpuTimer * timer) noexcept { fiber->suspendTimestamp = 0; } - enqueueReady(fiber); + + ProcessorState * target = enqueueReady(processor, fiber); + if (target) + { + processor->postWakeup(target); + } } return; } diff --git a/src/fibers/future.cpp b/src/fibers/future.cpp index 42aad73..6ab05e4 100644 --- a/src/fibers/future.cpp +++ b/src/fibers/future.cpp @@ -34,6 +34,15 @@ bool FiberFuture::subscribe(SubscribeCallback * callback) noexcept } void FiberFuture::signal() noexcept +{ + Fiber * waitingFiber = extractWaitingFiber(); + if (waitingFiber) + { + FiberScheduler::schedule(waitingFiber); + } +} + +Fiber * FiberFuture::extractWaitingFiber() noexcept { State currentState; currentState.raw = state.load(std::memory_order_relaxed); @@ -41,7 +50,7 @@ void FiberFuture::signal() noexcept { if (currentState.isSet) { - return; + return nullptr; } State newState(currentState); @@ -49,30 +58,26 @@ void FiberFuture::signal() noexcept if (state.compare_exchange_weak(currentState.raw, newState.raw, std::memory_order_acq_rel, std::memory_order_relaxed)) { + // Wake the waiter first, then increment the counter. waitForMultiple + // spins on the counter only after completionFuture.wait() returns, so + // this order is safe: the drain loop will not start until the wake has + // already been delivered. if (currentState.multipleWait) { - // Wake the waiter first, then increment the counter. waitForMultiple - // spins on the counter only after completionFuture.wait() returns, so - // this order is safe: the drain loop will not start until the wake has - // already been delivered. MultipleWaitState * waitState = reinterpret_cast(currentState.waiter); waitState->completionFuture->signal(); waitState->completionCounter.fetch_add(1, std::memory_order_release); + return nullptr; } - else if (currentState.hasCallback) + + if (currentState.hasCallback) { SubscribeCallback * callback = reinterpret_cast(currentState.waiter); callback(this); + return nullptr; } - else - { - Fiber * waitingFiber = reinterpret_cast(currentState.waiter); - if (waitingFiber) - { - FiberScheduler::schedule(waitingFiber); - } - } - return; + + return reinterpret_cast(currentState.waiter); } } } @@ -277,4 +282,35 @@ int FiberFuture::waitWithTimeout(FiberFuture * future, uint64_t nanoseconds) noe return r; } +void FiberFuture::setAll(int err, FiberFuture ** futures, uint64_t count) noexcept +{ + // Collect the plain-fiber waiters of the just-set futures into a fixed stack buffer and hand each + // chunk to one batched schedule (one doorbell per target processor, one submit). The two rare waiter + // kinds - multiple-wait and callback - are completed inline, exactly as signal() does. + Fiber * wakeBatch[WAKE_BATCH]; + uint64_t batchSize = 0; + + for (uint64_t i = 0; i < count; i++) + { + FiberFuture * future = futures[i]; + future->error = err; + + Fiber * waitingFiber = future->extractWaitingFiber(); + if (waitingFiber) + { + wakeBatch[batchSize++] = waitingFiber; + if (batchSize == WAKE_BATCH) + { + FiberScheduler::scheduleAll(wakeBatch, batchSize); + batchSize = 0; + } + } + } + + if (batchSize) + { + FiberScheduler::scheduleAll(wakeBatch, batchSize); + } +} + } // namespace silk diff --git a/src/fibers/sequencer.cpp b/src/fibers/sequencer.cpp index cca1206..43bbdee 100644 --- a/src/fibers/sequencer.cpp +++ b/src/fibers/sequencer.cpp @@ -7,20 +7,13 @@ namespace silk { -void FiberSequencer::wait(uint64_t token, Future * future) noexcept +void FiberSequencer::registerWaiter(uint64_t token, Future * future) noexcept { + // Slow path: register future in the request queue for the next combiner to process. future->sequencer = this; future->token = token; future->state.store(0, std::memory_order_relaxed); - // Fast path: counter already satisfied. - if (counter.load(std::memory_order_acquire) >= token) - { - future->set(0); - return; - } - - // Slow path: register future in the request queue for the next combiner to process. requestQueue.push(future); // Waiter half of a StoreLoad (Dekker) handshake on requestQueue vs counter (acquire / release cannot do @@ -37,32 +30,6 @@ void FiberSequencer::wait(uint64_t token, Future * future) noexcept } } -uint64_t FiberSequencer::increment() noexcept -{ - uint64_t current = counter.fetch_add(1, std::memory_order_release) + 1; - drain(); - return current; -} - -bool FiberSequencer::advance(uint64_t value) noexcept -{ - 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; - } - } - - drain(); - return true; -} - void FiberSequencer::cancelWait(Future * future) noexcept { uint32_t expected = future->state.load(std::memory_order_relaxed); @@ -207,15 +174,9 @@ void FiberSequencer::drain() noexcept std::atomic_thread_fence(std::memory_order_seq_cst); } - // Wake outside the combiner so set can itself call drain without deadlocking. - while (Future * future = wakeList.pop()) - { - future->set(0); - } - while (Future * future = cancelList.pop()) - { - future->set(ECANCELED); - } + // Wake outside the combiner so a woken fiber re-entering drain cannot deadlock on the combiner. + setAll(&wakeList, 0); + setAll(&cancelList, ECANCELED); } bool FiberSequencer::acquireCombiner() noexcept @@ -266,4 +227,25 @@ bool FiberSequencer::releaseCombiner() noexcept } } +void FiberSequencer::setAll(WaitList * wakeList, int err) noexcept +{ + FiberFuture * wakeBatch[WAKE_BATCH]; + uint64_t batchSize = 0; + + while (Future * future = wakeList->pop()) + { + wakeBatch[batchSize++] = future; + if (batchSize == WAKE_BATCH) + { + FiberFuture::setAll(err, wakeBatch, batchSize); + batchSize = 0; + } + } + + if (batchSize) + { + FiberFuture::setAll(err, wakeBatch, batchSize); + } +} + } // namespace silk diff --git a/src/fibers/tests/future-test.cpp b/src/fibers/tests/future-test.cpp index 58f72a1..dcff0d1 100644 --- a/src/fibers/tests/future-test.cpp +++ b/src/fibers/tests/future-test.cpp @@ -5,6 +5,7 @@ #include #include +#include namespace silk { @@ -377,4 +378,80 @@ TEST(FiberFuture, resetAndReuse) ASSERT_EQ(future.wait(), 2); } +// setAll sets a batch of futures and wakes their fiber waiters together; each woken fiber sees the value. +// N exceeds WAKE_BATCH (32) so the chunking inside setAll is exercised. +TEST(FiberFuture, setAllWakesWaiters) +{ + constexpr uint64_t N = 100; + + struct Params + { + FiberFuture * started; + FiberFuture * wait; + + static int fiberMain(Params * p) noexcept + { + p->started->set(0); + return p->wait->wait(); + } + }; + + std::vector started(N); + std::vector wait(N); + std::vector done(N); + + for (uint64_t i = 0; i < N; ++i) + { + int r = FiberScheduler::run(Params::fiberMain, {&started[i], &wait[i]}, &done[i]); + ASSERT_FALSE(r); + } + + // Let every fiber run up to its wait() before the batched wake. + for (uint64_t i = 0; i < N; ++i) + { + started[i].wait(); + } + + std::vector futures(N); + for (uint64_t i = 0; i < N; ++i) + { + futures[i] = &wait[i]; + } + + FiberFuture::setAll(42, futures.data(), N); + + for (uint64_t i = 0; i < N; ++i) + { + EXPECT_EQ(done[i].wait(), 42); + } +} + +// setAll on futures with no fiber waiter simply sets each; the result is observable afterwards. +TEST(FiberFuture, setAllNoWaiter) +{ + constexpr uint64_t N = 40; + + std::vector futures(N); + std::vector pointers(N); + for (uint64_t i = 0; i < N; ++i) + { + pointers[i] = &futures[i]; + } + + FiberFuture::setAll(7, pointers.data(), N); + + for (uint64_t i = 0; i < N; ++i) + { + int r; + EXPECT_TRUE(futures[i].isSet(&r)); + EXPECT_EQ(r, 7); + } +} + +// setAll with count 0 is a no-op. +TEST(FiberFuture, setAllEmpty) +{ + FiberFuture::setAll(0, nullptr, 0); +} + } // namespace silk