From 25ddabd0f396ea213a4d4d9de1c1606f9ccacb4f Mon Sep 17 00:00:00 2001 From: Coldwings Date: Tue, 15 Sep 2026 11:43:25 +0800 Subject: [PATCH] perf(common): notify RingChannel consumers via a lock-free park stack RingChannel woke its idle consumers through a photon::semaphore, whose signal() holds one per-channel spinlock across try_resume(), which in turn calls prelocked_thread_interrupt() -- so a cross-vCPU wake-up costs an eventfd write syscall inside the critical section. Every consumer returning from wait_defer() re-acquires that same lock, and every recv() miss did two seq_cst RMWs on the shared `idler` counter. All notifications of a channel thus serialize behind one lock whose holder may be preempted while everybody else spins on it, so dispatch throughput fell as vCPUs were added instead of rising. Replace it with a park slot per sleeping consumer, linked into a lock-free idle stack. A producer's fast path is a fence and a relaxed load of the stack top: no store to shared state, no lock. A wake-up is one exchange() of the whole stack plus a thread_interrupt() issued outside of any lock, so N producers can wake N consumers in parallel. Not losing a notification rests on three invariants: - Dekker: the producer fences between its push and its idle() load, while the consumer publishes its slot with a seq_cst RMW and re-checks the queue right afterwards. The two can not miss each other. - Publish-in-defer: prepare_usleep() does not inspect error_number, so an interrupt arriving before the sleep commits would be dropped. The slot is therefore promoted to COMMITTED from the defer callback of thread_usleep_defer(), and only a COMMITTED slot may be interrupted. - Slot lifetime: a slot is a stack frame, so its owner never returns from park() while a claimer might still touch it. The claimer's last write is its CLAIMED store, and the owner waits for it, escalating from pause to thread_yield to sched_yield: if that store has not landed yet, the claimer is not on a CPU, and spinning would only keep it away. The escalation alone cuts hand-off stalls of ~1.5ms (0.3% of all parks, yet 15-33% of the wall time of the synchronous WorkPool cases) down to microseconds. The 100ms self-wake of a parked consumer is kept as a safety net only: it re-checks the queue and re-arms the slot. Correctness does not depend on it, which is what lets the new test disable it. Interleaved A/B on one host, perf_workpool --fires=40000, the sync/StdContext case, median of 4 runs: vCPUs 4 8 16 32 before 40413 18416 9162 6898 QPS after 66126 111856 141728 134764 QPS before 9042 15860 36472 49332 ns/dispatch after 5638 2478 1572 1512 ns/dispatch Cross-vCPU wake-up latency improves 4.6x, burst fan-out 1.2x with 4 consumers and 14x with 32, and the whole benchmark spends 39% less user CPU. Fire-and-forget dispatch on few vCPUs is the one case that regresses (~30% at 4 vCPUs), because the semaphore version hardly ever really slept there; it turns into a 1.5x to 3.4x gain by 16 vCPUs. Add test-ringchannel-notify, which drives the channel with its safety net disabled, so that a lost notification hangs the case and is reported, instead of hiding as a 100ms hiccup. Its teeth were verified by mutation: dropping the re-check after publish, or committing the slot before the sleep instead of from the defer callback, are both caught within a few thousand rounds. Add perf-ringchannel to measure the notification paths in isolation. No public interface changes: send(), recv() and notification_pending() keep their signatures, and notification_pending() keeps its meaning of wake-ups issued but not yet observed. Being header-only, the class does change layout, so consumers must be rebuilt against matching headers, as any photon upgrade already requires. --- common/lockfree_queue.h | 368 +++++++++++++++++------- common/test/CMakeLists.txt | 2 + common/test/perf_ringchannel.cpp | 256 +++++++++++++++++ common/test/test_ringchannel_notify.cpp | 303 +++++++++++++++++++ 4 files changed, 822 insertions(+), 107 deletions(-) create mode 100644 common/test/perf_ringchannel.cpp create mode 100644 common/test/test_ringchannel_notify.cpp diff --git a/common/lockfree_queue.h b/common/lockfree_queue.h index a16f7aa0..1a511809 100644 --- a/common/lockfree_queue.h +++ b/common/lockfree_queue.h @@ -599,23 +599,177 @@ using FlexLockfreeSPSCRingQueue = FlexQueue>; namespace photon { namespace common { +/** + * @brief A lock-free stack of the park slots of idle (sleeping) consumers. + * + * A consumer that finds the queue empty publishes a `Slot`, which lives on its + * own stack frame, and then sleeps on it. A producer that has just pushed an + * item claims one slot and wakes its owner up. Claiming is exclusive *by + * construction*: the claimer takes the whole stack with a single `exchange()`, + * keeps the top slot and gives the rest back. Hence + * - a wake-up can not accumulate: one push wakes at most one consumer, and a + * published slot can be claimed exactly once, so the number of in-flight + * wake-ups is capped by the number of sleeping consumers -- structurally, + * without any counter to maintain; + * - the stack needs no version tag: unlike a Treiber pop, `exchange()` never + * CAS-es on a slot's `next` field, which is where the ABA hazard would be + * (slots are re-published by their owners, so the same address does come + * back). + * + * Why no notification can be lost: + * - `publish()` is a seq_cst RMW, and a producer has a seq_cst fence between + * its push and its `idle()` load, so at least one of the two sides sees + * the other (Dekker). The consumer also re-checks the queue right after + * publishing. + * - The owner of a slot never returns from `park()` while the slot is still + * linked into the stack. A claimer therefore never dereferences a dead + * slot, and its `thread_interrupt()` can never hit a thread that has + * already left `park()` and gone on to do something else. + * - A claimer holding the whole stack makes it look empty to everybody else, + * so a concurrent producer may skip its notification. Whoever gives slots + * back is responsible for re-checking the queue afterwards; see + * `RingChannel::notify_recvers()`. + */ +class ParkStack { +public: + struct Slot { + photon::thread* th = photon::CURRENT; + Slot* next = nullptr; + std::atomic st{PARKED}; + bool interrupted = false; // a wake-up interrupt was issued to `th` + }; + + // Is any consumer parked? This is the only load a producer pays for on its + // fast path, and the line is read-mostly while consumers are busy. + bool idle() const { return top.load(std::memory_order_relaxed) != nullptr; } + + // wake-ups that have been issued but not yet observed by their target + uint64_t inflight() const { + return _inflight.load(std::memory_order_acquire); + } + + void publish(Slot* s) { + auto head = top.load(std::memory_order_relaxed); + do { s->next = head; } + while (!top.compare_exchange_weak(head, s, std::memory_order_seq_cst, + std::memory_order_relaxed)); + } + + // Claims one parked consumer and wakes it up. Returns false if and only if + // no slot was published at the moment of the exchange. + bool unpark_one() { + auto s = top.exchange(nullptr, std::memory_order_seq_cst); + if (!s) return false; + if (s->next) give_back(s->next); + wake(s); + return true; + } + + // Sleeps until `s`, which must have been published, gets claimed. `on_idle` + // is invoked every time the safety-net timeout expires (or an unrelated + // thread_interrupt() arrives) without a claim; it must not sleep, because + // until the slot is re-armed a claimer still believes we are sleeping. + template + void park(Slot* s, uint64_t timeout_usec, OnIdle on_idle) { + bool consumed = false; // did we consume the wake-up interrupt? + while (s->st.load(std::memory_order_seq_cst) == PARKED) { + int r = photon::thread_usleep_defer(timeout_usec, &commit, s); + if (r < 0 && errno == -1) consumed = true; + if (s->st.load(std::memory_order_seq_cst) != COMMITTED) + break; // claimed, before or while we slept + on_idle(); + uint32_t expect = COMMITTED; // re-arm and sleep once more + if (!s->st.compare_exchange_strong(expect, PARKED, + std::memory_order_seq_cst)) + break; + } + // The claimer's last touch of the slot is its CLAIMED store. Wait for + // it, or we would let the slot -- a stack frame -- die under its feet. + // It is normally already there (a wake-up interrupt costs the claimer a + // syscall, and only two instructions follow it), so the first pauses + // almost always suffice. If they do not, the claimer is not running on + // any CPU, and spinning is then exactly the wrong thing to do: it keeps + // a core away from the only thread that can end the wait. Hence the + // escalation to sched_yield(), which blocks this vCPU no more than the + // spinning already did, and cuts stalls of milliseconds down to + // microseconds under CPU pressure. + for (uint64_t i = 0; s->st.load(std::memory_order_acquire) != CLAIMED; ++i) { + if (i < 64) CPUPause::pause(); + else if (i < 128) photon::thread_yield(); + else ThreadPause::pause(); + } + _inflight.fetch_sub(1, std::memory_order_release); + // The claimer found us COMMITTED but the safety-net timeout had already + // woken us up: its interrupt landed on a running thread and is still + // pending. Absorb it here, or it would surface at whatever the caller + // of recv() does next. thread_yield() clears the pending error. + if (s->interrupted && !consumed) photon::thread_yield(); + } + +protected: + enum : uint32_t { + PARKED = 0, // published, the owner is sleeping or about to + COMMITTED = 1, // the owner is provably asleep: a claim must interrupt + CLAIMING = 2, // claimed, the claimer is still working on the slot + CLAIMED = 3, // claimed and released: the owner may return + }; + std::atomic top{nullptr}; + std::atomic _inflight{0}; + + // Hands a sub-stack back after having taken all of it. + void give_back(Slot* first) { + Slot* head = nullptr; + if (top.compare_exchange_strong(head, first, std::memory_order_seq_cst, + std::memory_order_relaxed)) + return; // common case: nobody published meanwhile + auto tail = first; + while (tail->next) tail = tail->next; + do { tail->next = head; } + while (!top.compare_exchange_weak(head, first, std::memory_order_seq_cst, + std::memory_order_relaxed)); + } + + // `s` is exclusively ours, as it has been unlinked by the exchange(). + void wake(Slot* s) { + auto th = s->th; // the slot is unreachable after the hand-off below + _inflight.fetch_add(1, std::memory_order_relaxed); + if (s->st.exchange(CLAIMING, std::memory_order_seq_cst) == COMMITTED) { + // It is provably asleep on this slot. Both stores are still safe: + // the owner may not leave park() before our CLAIMED store, and it + // reads `interrupted` only after having seen it. + s->interrupted = true; + photon::thread_interrupt(th, -1); + } // else it is awake and wakes itself up, see commit() below + s->st.store(CLAIMED, std::memory_order_release); + } + + // Runs right after the owner of `s` has committed itself to sleeping, so + // from now on a claimer is allowed to interrupt it. + static void commit(void* arg) { + auto s = (Slot*)arg; + uint32_t expect = PARKED; + if (s->st.compare_exchange_strong(expect, COMMITTED, + std::memory_order_seq_cst)) + return; + // Already claimed, and the claimer saw us not committed yet, so nobody + // else is going to wake us up. + s->interrupted = true; + photon::thread_interrupt(s->th, -1); + } +}; + /** * @brief RingChannel is a photon wrapper to make LockfreeQueue send/recv * efficiently wait and spin using photon style sync mechanism. * - * Notification model (multi-producer, multi-consumer safe): - * - Each consumer entering the slow path bumps `idler`. - * - Each producer, after `push`, may issue at most one `signal(1)` per push, - * and only when `pending < idler`. `pending` mirrors the in-flight - * `queue_sem.m_count`, so the semaphore counter is hard-capped by the - * observed number of idle consumers, never accumulating with burst size. - * - A `seq_cst` fence in send() and a `seq_cst` RMW on `idler` in recv() - * close the Dekker-style window: at any time at least one of - * (consumer sees the push) or (producer signals) must hold, so the queue - * can never end up non-empty while every consumer sleeps. + * Notification model (multi-producer, multi-consumer safe): a consumer that + * runs out of both items and spin budget publishes a park slot and sleeps on + * it; a producer that has just pushed an item claims one park slot and wakes + * its owner up. See ParkStack for why this neither loses nor accumulates + * notifications. * - * Watch out that `recv` should run in photon environment (because it has to) - * use photon semaphore to be notified that new item has sended. `send` could + * Watch out that `recv` should run in photon environment (because it has to + * sleep on a park slot to be notified that new item has sended). `send` could * running in photon or std::thread environment (needs to set template `Pause` * as `ThreadPause`). * @@ -627,10 +781,14 @@ namespace common { // send_pending) are declared in each channel class; only the logic is shared. template struct SendBackoff { - static void notify_senders(photon::semaphore& send_sem, + // Wakes one blocked sender up, unless enough wake-ups are already in + // flight. The caller must have issued a seq_cst fence after the pop that + // freed a queue slot -- hence the name -- so that the load of + // `send_waiters` below can not be reordered before it. That is the Dekker + // half paired with the seq_cst RMW on `send_waiters` in push_backoff(). + static void notify_senders_fenced(photon::semaphore& send_sem, std::atomic& send_waiters, std::atomic& send_pending) { - std::atomic_thread_fence(std::memory_order_seq_cst); auto cur_waiters = send_waiters.load(std::memory_order_seq_cst); if (cur_waiters == 0) return; auto sp = send_pending.load(std::memory_order_acquire); @@ -716,14 +874,16 @@ struct SendBackoff { template class RingChannel : public QueueType { protected: - photon::semaphore queue_sem; - std::atomic idler{0}; // # consumers in idle/wait - std::atomic pending{0}; // mirror of queue_sem.m_count + ParkStack idlers; // park slots of the idle consumers photon::semaphore send_sem; std::atomic send_waiters{0}; std::atomic send_pending{0}; uint64_t default_yield_turn = 1024; uint64_t default_yield_usec = 1024; + // Safety net only: a parked consumer wakes itself up this often to + // re-check the queue and re-arm its slot. Correctness does not rely on + // it, it merely bounds the damage of a hypothetically lost wake-up. + uint64_t default_park_usec = 100UL * 1000; using T = decltype(std::declval().recv()); @@ -745,46 +905,16 @@ class RingChannel : public QueueType { SendBackoff::template push_backoff(x, [this](const T& v) { return push(v); }, default_yield_turn, default_yield_usec, send_sem, send_waiters, send_pending); - // Dekker barrier: ensure the prior push (mark.store release) is - // ordered before the following idler load, paired with the seq_cst - // RMW on `idler` in recv(). This guarantees that we cannot - // simultaneously miss the consumer's idler++ AND have the consumer - // miss our push. - std::atomic_thread_fence(std::memory_order_seq_cst); - auto cur_idler = idler.load(std::memory_order_seq_cst); - if (cur_idler == 0) return; - - // Cap pending (== m_count) at cur_idler so a long burst can never - // accumulate stale wake-up tokens. Multiple producers may each succeed - // up to the observed idler count, preserving fan-out for N consumers. - auto p = pending.load(std::memory_order_acquire); - for (;;) { - if (p >= cur_idler) { - auto fresh = idler.load(std::memory_order_relaxed); - if (fresh <= cur_idler) return; - cur_idler = fresh; - continue; - } - if (pending.compare_exchange_weak(p, p + 1, - std::memory_order_acq_rel, - std::memory_order_acquire)) { - queue_sem.signal(1); - return; - } - // CAS failed: `p` was refreshed automatically; retry. - } + notify_recvers(); } T recv(uint64_t max_yield_turn, uint64_t max_yield_usec) { T x; if (pop(x)) { - SendBackoff::notify_senders(send_sem, send_waiters, send_pending); + after_recv(); return x; } // yield once if failed, so photon::now will be updated photon::thread_yield(); - // seq_cst on idler is the other half of the Dekker barrier (see send). - idler.fetch_add(1, std::memory_order_seq_cst); - DEFER(idler.fetch_sub(1, std::memory_order_seq_cst)); Timeout yield_timeout(max_yield_usec); uint64_t yield_turn = max_yield_turn; while (!pop(x)) { @@ -792,36 +922,72 @@ class RingChannel : public QueueType { yield_turn--; photon::thread_yield(); } else { - // wait for 100ms - int r = queue_sem.wait(1, 100ULL * 1000); - // r == 0 means we actually consumed one m_count token; mirror - // it on `pending`. r < 0 (timeout/interrupt) does not touch - // m_count, so we must not touch `pending` either. - if (r == 0) - pending.fetch_sub(1, std::memory_order_acq_rel); + park(); // reset yield mark and set into busy wait yield_turn = max_yield_turn; yield_timeout.timeout(max_yield_usec); } } - SendBackoff::notify_senders(send_sem, send_waiters, send_pending); + after_recv(); return x; } T recv() { return recv(default_yield_turn, default_yield_usec); } - // Diagnostic accessor: returns the current count of in-flight wake-up - // tokens (mirrors `queue_sem.m_count`). Tests use this to assert that no - // stale signals accumulate across a producer burst. - uint64_t notification_pending() const { - return pending.load(std::memory_order_acquire); + // Diagnostic accessor: wake-ups that have been issued to a parked consumer + // but not yet observed by it. Unlike the semaphore counter that it + // replaces, this can not accumulate over a producer burst: a park slot is + // claimed exactly once, so the count is bounded by the number of parked + // consumers, however long the burst is. + uint64_t notification_pending() const { return idlers.inflight(); } + +protected: + void unpark_if_ready() { if (!empty()) idlers.unpark_one(); } + + // Called by a producer right after its push. + void notify_recvers() { + // Dekker barrier: order the push before the idle() load below, paired + // with the seq_cst RMW on the stack top in ParkStack::publish(). Hence + // we can not both miss a parked consumer and be missed by it. + std::atomic_thread_fence(std::memory_order_seq_cst); + if (idlers.idle()) unpark_if_ready(); + } + + // Called by a consumer right after a successful pop: hand the freed queue + // slot to a blocked sender, and pass the baton on if there is still work + // and somebody to do it. + // + // The baton matters because a claimer holds the whole idle stack while it + // hands the surplus back, so a producer pushing right then finds nobody + // parked and skips its wake-up. Whoever gets woken up is therefore + // responsible for re-checking here, and the fence below makes sure it sees + // that producer's item. Without it, such an item would still be consumed + // -- by us, on our next recv() -- but serially instead of in parallel, and + // it would be left in the queue if we happened not to come back for more. + void after_recv() { + std::atomic_thread_fence(std::memory_order_seq_cst); + SendBackoff::notify_senders_fenced(send_sem, send_waiters, send_pending); + if (idlers.idle()) unpark_if_ready(); + } + + // Publishes a park slot and sleeps on it until a producer claims it. + void park() { + ParkStack::Slot slot; + idlers.publish(&slot); + // The consumer half of the Dekker barrier in notify_recvers(): a + // producer that has missed our slot can not be missed here. This is + // what makes the channel live -- the last consumer to fall asleep is + // the one that can not afford to miss an item. + std::atomic_thread_fence(std::memory_order_seq_cst); + unpark_if_ready(); // may well claim our own slot, which is fine + idlers.park(&slot, default_park_usec, [this] { unpark_if_ready(); }); } }; // FlexRingChannel: composition-based wrapper for FlexQueue types. // Unlike RingChannel (which inherits from QueueType), FlexRingChannel holds // a pointer to a dynamically-allocated FlexQueue. This avoids the memory -// layout conflict where RingChannel's members (semaphore, idler, etc.) would -// overlap with the zero-length slots[] array in the base queue class. +// layout conflict where RingChannel's members (park stack, semaphore, etc.) +// would overlap with the zero-length slots[] array in the base queue class. // // Usage: // using FlexRing = FlexLockfreeMPMCRingQueue>; @@ -832,14 +998,13 @@ class RingChannel : public QueueType { template class FlexRingChannel { FlexQueueType* queue; - photon::semaphore queue_sem; - std::atomic idler{0}; // # consumers in idle/wait - std::atomic pending{0}; // mirror of queue_sem.m_count + ParkStack idlers; // park slots of the idle consumers photon::semaphore send_sem; std::atomic send_waiters{0}; std::atomic send_pending{0}; uint64_t default_yield_turn = 1024; uint64_t default_yield_usec = 1024; + uint64_t default_park_usec = 100UL * 1000; // safety net, see RingChannel using T = decltype(std::declval().recv()); @@ -872,43 +1037,17 @@ class FlexRingChannel { SendBackoff::template push_backoff(x, [this](const T& v) { return queue->push(v); }, default_yield_turn, default_yield_usec, send_sem, send_waiters, send_pending); - // Dekker barrier: ensure the prior push is ordered before the - // following idler load, paired with the seq_cst RMW on `idler` - // in recv(). - std::atomic_thread_fence(std::memory_order_seq_cst); - auto cur_idler = idler.load(std::memory_order_seq_cst); - if (cur_idler == 0) return; - - // Cap pending (== m_count) at cur_idler so a long burst can never - // accumulate stale wake-up tokens. - auto p = pending.load(std::memory_order_acquire); - for (;;) { - if (p >= cur_idler) { - auto fresh = idler.load(std::memory_order_relaxed); - if (fresh <= cur_idler) return; - cur_idler = fresh; - continue; - } - if (pending.compare_exchange_weak(p, p + 1, - std::memory_order_acq_rel, - std::memory_order_acquire)) { - queue_sem.signal(1); - return; - } - } + notify_recvers(); } T recv(uint64_t max_yield_turn, uint64_t max_yield_usec) { T x; if (queue->pop(x)) { - SendBackoff::notify_senders(send_sem, send_waiters, send_pending); + after_recv(); return x; } // yield once if failed, so photon::now will be updated photon::thread_yield(); - // seq_cst on idler is the other half of the Dekker barrier (see send). - idler.fetch_add(1, std::memory_order_seq_cst); - DEFER(idler.fetch_sub(1, std::memory_order_seq_cst)); Timeout yield_timeout(max_yield_usec); uint64_t yield_turn = max_yield_turn; while (!queue->pop(x)) { @@ -916,19 +1055,13 @@ class FlexRingChannel { yield_turn--; photon::thread_yield(); } else { - // wait for 100ms - int r = queue_sem.wait(1, 100UL * 1000); - // r == 0 means we actually consumed one m_count token; mirror - // it on `pending`. r < 0 (timeout/interrupt) does not touch - // m_count, so we must not touch `pending` either. - if (r == 0) - pending.fetch_sub(1, std::memory_order_acq_rel); + park(); // reset yield mark and set into busy wait yield_turn = max_yield_turn; yield_timeout.timeout(max_yield_usec); } } - SendBackoff::notify_senders(send_sem, send_waiters, send_pending); + after_recv(); return x; } @@ -939,10 +1072,31 @@ class FlexRingChannel { size_t read_available() const { return queue->read_available(); } size_t write_available() const { return queue->write_available(); } - // Diagnostic accessor: returns the current count of in-flight wake-up - // tokens (mirrors `queue_sem.m_count`). - uint64_t notification_pending() const { - return pending.load(std::memory_order_acquire); + // Diagnostic accessor: wake-ups that have been issued to a parked consumer + // but not yet observed by it; see RingChannel. + uint64_t notification_pending() const { return idlers.inflight(); } + +protected: + void unpark_if_ready() { if (!queue->empty()) idlers.unpark_one(); } + + // See RingChannel for why each of the three is needed and sufficient. + void notify_recvers() { + std::atomic_thread_fence(std::memory_order_seq_cst); + if (idlers.idle()) unpark_if_ready(); + } + + void after_recv() { + std::atomic_thread_fence(std::memory_order_seq_cst); + SendBackoff::notify_senders_fenced(send_sem, send_waiters, send_pending); + if (idlers.idle()) unpark_if_ready(); + } + + void park() { + ParkStack::Slot slot; + idlers.publish(&slot); + std::atomic_thread_fence(std::memory_order_seq_cst); + unpark_if_ready(); // may well claim our own slot, which is fine + idlers.park(&slot, default_park_usec, [this] { unpark_if_ready(); }); } }; diff --git a/common/test/CMakeLists.txt b/common/test/CMakeLists.txt index c4718afc..55791413 100644 --- a/common/test/CMakeLists.txt +++ b/common/test/CMakeLists.txt @@ -12,6 +12,8 @@ photon_add_test(test-scalepool test_scalepool.cpp) photon_add_test(test-throttle test_throttle.cpp) photon_add_test(test-constexprstr test_constexprstr.cpp) photon_add_test(test-lockfree test_lockfree.cpp) +photon_add_test(test-ringchannel-notify test_ringchannel_notify.cpp) +photon_add_test(perf-ringchannel perf_ringchannel.cpp NO_REGISTER) photon_add_test(test-alog test_alog.cpp x.cpp) photon_add_test(perf-rcuptr perf_rcuptr.cpp NO_REGISTER) photon_add_test(perf-alog perf_alog.cpp NO_REGISTER) diff --git a/common/test/perf_ringchannel.cpp b/common/test/perf_ringchannel.cpp new file mode 100644 index 00000000..3d6495b8 --- /dev/null +++ b/common/test/perf_ringchannel.cpp @@ -0,0 +1,256 @@ +/* +Copyright 2022 The Photon Authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Micro-benchmark of the RingChannel *notification* path, i.e. what happens +// when a consumer has to go idle and be woken up again. The spin budget is set +// to zero in most cases (recv(0, 0)) so that every single item pays the full +// park + wake-up cost, which is exactly the part that a park-slot design +// replaces. +// +// Cases: +// 1. wake_latency -- one producer, one always-idle consumer, strict +// ping-pong. Measures the round-trip of +// "push + notify + wake + pop". +// 2. fanout -- N idle consumers on N vCPUs, bursts of N items. +// Measures how fast a burst spreads over all consumers. +// 3. hot_send -- no consumer at all: measures the producer's fast path +// (the barrier + idle check that every send pays). +// 4. steady -- one producer, one consumer with the default spin +// budget, no artificial pacing: end-to-end throughput. +// 5. same_vcpu -- producer and consumer are photon threads of the *same* +// vCPU, so a wake-up needs no eventfd/epoll kick at all. +// This is the purest measure of the notification +// bookkeeping itself. +// 6. mp_contend -- P producer OS threads against C consumers that park on +// every item: measures how well the notification path +// scales when many producers notify concurrently. + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +DEFINE_uint64(rounds, 100000, "iterations per case"); +DEFINE_uint64(consumers, 4, "consumer num of the fan-out case"); +DEFINE_uint64(producers, 4, "producer num of the contention case"); + +using Queue = LockfreeMPMCRingQueue; +using Channel = photon::common::RingChannel; + +static uint64_t now_ns() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + +// 1. strict ping-pong against an always-idle consumer. +static void case_wake_latency() { + Channel ch; + std::atomic acked{0}; + std::atomic stop{false}; + + std::thread consumer([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + auto x = ch.recv(0, 0); // no spinning: always park + if (x == 0) break; + acked.store(x, std::memory_order_release); + } + }); + + auto start = now_ns(); + for (uint64_t i = 1; i <= FLAGS_rounds; ++i) { + ch.send(i); + // Spin (not sleep) on the ack, so the measured time is the channel's + // wake-up path and not this thread's own scheduling. + while (acked.load(std::memory_order_acquire) != i) CPUPause::pause(); + } + auto cost = now_ns() - start; + stop.store(true); + ch.send(0); + consumer.join(); + + LOG_INFO("wake_latency : ` round-trips, ` ns/round-trip", + FLAGS_rounds, cost / FLAGS_rounds); +} + +// 2. burst of N items against N idle consumers, N vCPUs. +static void case_fanout() { + Channel ch; + auto n = FLAGS_consumers; + std::atomic done{0}; + std::vector consumers; + for (uint64_t i = 0; i < n; ++i) { + consumers.emplace_back([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + auto x = ch.recv(0, 0); + if (x == 0) break; + done.fetch_add(1, std::memory_order_acq_rel); + } + }); + } + // let every consumer reach its idle state + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + uint64_t bursts = FLAGS_rounds / n; + auto start = now_ns(); + for (uint64_t b = 0; b < bursts; ++b) { + auto target = (b + 1) * n; + for (uint64_t i = 0; i < n; ++i) ch.send(b * n + i + 1); + while (done.load(std::memory_order_acquire) < target) CPUPause::pause(); + } + auto cost = now_ns() - start; + for (uint64_t i = 0; i < n; ++i) ch.send(0); + for (auto& t : consumers) t.join(); + + LOG_INFO("fanout : ` bursts of `, ` ns/burst, ` ns/item", + bursts, n, cost / bursts, cost / (bursts * n)); +} + +// 3. producer fast path, nobody is waiting on the other end. +static void case_hot_send() { + Channel ch; + uint64_t rounds = FLAGS_rounds; + auto start = now_ns(); + // The queue holds 4096 entries; drain it in place (single thread, so no + // notification is ever needed) to keep measuring send() only. + for (uint64_t i = 0; i < rounds; ++i) { + ch.send(i + 1); + uint64_t x; + ch.pop(x); + } + auto cost = now_ns() - start; + LOG_INFO("hot_send : ` sends, ` ns/send (no idle consumer)", + rounds, cost / rounds); +} + +// 4. steady state with the default spin budget. +static void case_steady() { + Channel ch; + std::atomic received{0}; + std::thread consumer([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + auto x = ch.recv(); + if (x == 0) break; + received.fetch_add(1, std::memory_order_relaxed); + } + }); + + auto start = now_ns(); + for (uint64_t i = 1; i <= FLAGS_rounds; ++i) ch.send(i); + while (received.load(std::memory_order_relaxed) < FLAGS_rounds) + CPUPause::pause(); + auto cost = now_ns() - start; + ch.send(0); + consumer.join(); + + LOG_INFO("steady : ` items, ` ns/item, QPS `", FLAGS_rounds, + cost / FLAGS_rounds, FLAGS_rounds * 1000000000ULL / cost); +} + +// 5. producer and consumer live on the same vCPU: no OS-level wake-up is +// involved, so what remains is exactly the notification bookkeeping plus two +// coroutine context switches. +static void case_same_vcpu() { + Channel ping, pong; + uint64_t rounds = FLAGS_rounds; + + auto consumer = photon::thread_create11([&] { + for (;;) { + auto x = ping.recv(0, 0); // no spinning: always park + pong.send(x); + if (x == 0) break; + } + }); + photon::thread_enable_join(consumer); + + auto start = now_ns(); + for (uint64_t i = 1; i <= rounds; ++i) { + ping.send(i); + auto x = pong.recv(0, 0); + if (x != i) LOG_ERROR("unexpected `, want `", x, i); + } + auto cost = now_ns() - start; + ping.send(0); + pong.recv(0, 0); + photon::thread_join((photon::join_handle*)consumer); + + LOG_INFO("same_vcpu : ` round-trips, ` ns/round-trip (2 park+wake each)", + rounds, cost / rounds); +} + +// 6. many producers notifying concurrently, consumers park on every item. +static void case_mp_contend() { + Channel ch; + auto np = FLAGS_producers, nc = FLAGS_consumers; + std::atomic done{0}; + std::vector consumers, producers; + for (uint64_t i = 0; i < nc; ++i) { + consumers.emplace_back([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + auto x = ch.recv(0, 0); + if (x == 0) break; + done.fetch_add(1, std::memory_order_relaxed); + } + }); + } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + uint64_t per = FLAGS_rounds / np, total = per * np; + auto start = now_ns(); + for (uint64_t p = 0; p < np; ++p) { + producers.emplace_back([&, p] { + for (uint64_t i = 0; i < per; ++i) ch.send(p * per + i + 1); + }); + } + for (auto& t : producers) t.join(); + while (done.load(std::memory_order_relaxed) < total) CPUPause::pause(); + auto cost = now_ns() - start; + for (uint64_t i = 0; i < nc; ++i) ch.send(0); + for (auto& t : consumers) t.join(); + + LOG_INFO("mp_contend : ` producers x ` consumers, ` items, ` ns/item", + np, nc, total, cost / total); +} + +int main(int argc, char** argv) { + gflags::ParseCommandLineFlags(&argc, &argv, true); + set_log_output_level(ALOG_INFO); + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + + case_hot_send(); + case_same_vcpu(); + case_wake_latency(); + case_steady(); + case_fanout(); + case_mp_contend(); + return 0; +} diff --git a/common/test/test_ringchannel_notify.cpp b/common/test/test_ringchannel_notify.cpp new file mode 100644 index 00000000..939db6ac --- /dev/null +++ b/common/test/test_ringchannel_notify.cpp @@ -0,0 +1,303 @@ +/* +Copyright 2022 The Photon Authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Hunts for lost notifications in RingChannel's park-slot notification path. A +// lost notification is not a wrong number, it is a stall, and in production the +// safety-net timeout inside ParkStack::park() hides it as a 100ms hiccup. Every +// case here therefore runs the channel with an explicit park timeout: +// +// - NO_NET (longer than the whole test): the safety net can never fire, so a +// single lost notification turns into a hang, which the watchdog reports. +// - a few tens of microseconds: every claim races against a timeout wake-up. +// That is the hardest window of the state machine -- the claimer finds the +// slot COMMITTED and interrupts a thread that is already running again -- +// and the interrupt it issued must be absorbed inside park() instead of +// surfacing at whatever the caller of recv() does next. +// +// The windows that have to stay closed: +// - a consumer published its slot but is not asleep yet, so a claimer may not +// interrupt it (the interrupt would be dropped and the sleep lost); +// - a claimer holds the whole idle stack while it hands the rest back, so a +// producer pushing right then finds nobody parked and skips its wake-up: +// the consumer that does get woken has to pass the baton on; +// - a consumer that goes back to sleep after an unrelated interrupt or a +// timeout must re-arm its slot before sleeping again. + +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "../../test/gtest.h" + +using Queue = LockfreeMPMCRingQueue; + +// A channel whose safety net is under the test's control. `default_park_usec` +// is protected exactly so that it can be reached from here. +struct TestChannel : photon::common::RingChannel { + explicit TestChannel(uint64_t park_usec) { + default_park_usec = park_usec; + } + size_t pending_items() { return read_available(); } + + // A producer is an order of magnitude faster than a wake-up, so left alone + // it would keep the queue non-empty and the consumers would never park at + // all. Waiting for the queue to drain first is what makes every item pay + // the notification path -- and it aims the push at the very moment when + // the consumers are publishing their slots. + void send_paced(uint64_t x) { + while (pending_items() > 0) std::this_thread::yield(); + send(x); + } +}; + +// longer than any of the cases below: the safety net must never fire +static const uint64_t NO_NET = 3600ULL * 1000 * 1000; +static const uint64_t DEADLINE_SEC = 30; +static const uint64_t SENTINEL = 0; // tells a consumer to leave + +static std::atomic g_progress{0}; // any item received, anywhere +static const char* g_case = ""; +static std::function g_state; + +// A stalled channel means every thread is asleep, so nothing photon-based can +// be trusted to report it: the watchdog is a plain OS thread, and it writes +// with write() because ALOG may be blocked behind a lock. +static void start_watchdog() { + std::thread([] { + for (uint64_t last = 0, stalls = 0;;) { + std::this_thread::sleep_for(std::chrono::seconds(2)); + auto p = g_progress.load(); + if (p != last) { last = p; stalls = 0; continue; } + if (++stalls * 2 < DEADLINE_SEC) continue; + auto msg = std::string("\nWATCHDOG: case '") + g_case + + "' made no progress in " + std::to_string(stalls * 2) + + "s, received=" + std::to_string(p) + + (g_state ? ", " + g_state() : std::string()) + + "\nlost notification\n"; + auto r = write(2, msg.data(), msg.size()); + (void)r; + _exit(2); + } + }).detach(); +} + +struct Deadline { + std::chrono::steady_clock::time_point end = + std::chrono::steady_clock::now() + std::chrono::seconds(DEADLINE_SEC); + bool expired() const { return std::chrono::steady_clock::now() > end; } +}; + +// Waits for `done` while failing the test rather than stalling forever. The +// caller is never a parked consumer, so blocking the OS thread is fine. +#define AWAIT(done, what) do { \ + Deadline dl; \ + while (!(done)) { \ + ASSERT_FALSE(dl.expired()) << "lost notification: " << what; \ + std::this_thread::yield(); \ + } \ +} while (0) + +// Strict ping-pong: the channel is empty again before the next push, so every +// single item has to travel through publish -> claim -> wake. `check_leak` +// makes the consumer verify after each item that no wake-up interrupt of its +// own is still pending -- a leaked one would hit the next sleep of whoever +// called recv(). +static void run_pingpong(uint64_t rounds, uint64_t park_usec, bool check_leak) { + TestChannel ch(park_usec); + std::atomic acked{0}; + std::atomic leaked{0}; + + std::thread consumer([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + auto x = ch.recv(0, 0); // no spin budget: park on every item + if (check_leak && photon::thread_usleep(1) < 0) + leaked.fetch_add(1, std::memory_order_relaxed); + acked.store(x, std::memory_order_release); + if (x == SENTINEL) break; + g_progress.fetch_add(1, std::memory_order_relaxed); + } + }); + + for (uint64_t i = 1; i <= rounds; ++i) { + ch.send(i); + AWAIT(acked.load(std::memory_order_acquire) == i, "ping-pong ack " << i); + } + ch.send(SENTINEL); + consumer.join(); + + EXPECT_EQ(0UL, leaked.load()); + EXPECT_EQ(0UL, ch.notification_pending()); +} + +TEST(ring_channel, pingpong_no_safety_net) { + g_case = "pingpong_no_safety_net"; + run_pingpong(50000, NO_NET, false); +} + +TEST(ring_channel, pingpong_racing_safety_net) { + g_case = "pingpong_racing_safety_net"; + // 20us: most items are delivered by a claim, but often to a consumer that + // the safety net has just woken up + run_pingpong(5000, 20, true); +} + +// Many producers against many parked consumers. This is where a claimer, which +// holds the whole idle stack while it gives the surplus back, makes a +// concurrent producer believe that nobody is parked. +static void run_mpmc(uint64_t nprod, uint64_t ncons, uint64_t per_prod) { + TestChannel ch(NO_NET); + auto total = nprod * per_prod; + std::atomic received{0}; + std::vector cons, prod; + g_state = [&] { + return "received=" + std::to_string(received.load()) + "/" + + std::to_string(total) + " queued=" + + std::to_string(ch.pending_items()); + }; + DEFER(g_state = nullptr); + + for (uint64_t i = 0; i < ncons; ++i) { + cons.emplace_back([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + if (ch.recv(0, 0) == SENTINEL) break; + received.fetch_add(1, std::memory_order_relaxed); + g_progress.fetch_add(1, std::memory_order_relaxed); + } + }); + } + // give every consumer the time to park, or the burst would be consumed + // without any notification at all + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + for (uint64_t p = 0; p < nprod; ++p) { + prod.emplace_back([&] { + for (uint64_t i = 0; i < per_prod; ++i) ch.send_paced(i + 1); + }); + } + for (auto& t : prod) t.join(); + AWAIT(received.load(std::memory_order_relaxed) >= total, "mpmc drain"); + + for (uint64_t i = 0; i < ncons; ++i) ch.send(SENTINEL); + for (auto& t : cons) t.join(); + + EXPECT_EQ(total, received.load()); + EXPECT_EQ(0UL, ch.notification_pending()); +} + +TEST(ring_channel, mpmc_burst_no_safety_net) { + g_case = "mpmc_burst_no_safety_net"; + run_mpmc(4, 4, 8000); +} + +TEST(ring_channel, single_producer_many_consumers) { + g_case = "single_producer_many_consumers"; + // one producer can only ever hand out one wake-up at a time, so the baton + // has to be passed along the consumers + run_mpmc(1, 8, 20000); +} + +// An unrelated thread_interrupt() on a parked consumer must not swallow an +// item: the consumer has to re-arm its slot before it sleeps again, otherwise +// the next producer would find it parked and wake nobody. +TEST(ring_channel, external_interrupt_keeps_slot_armed) { + g_case = "external_interrupt_keeps_slot_armed"; + TestChannel ch(NO_NET); + constexpr uint64_t kItems = 20000; + std::atomic received{0}; + std::atomic stop{false}; + g_state = [&] { return "received=" + std::to_string(received.load()); }; + DEFER(g_state = nullptr); + + std::thread consumer([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + auto self = photon::CURRENT; + auto pest = photon::thread_create11([&, self] { + while (!stop.load(std::memory_order_relaxed)) { + photon::thread_interrupt(self, EINTR); + photon::thread_usleep(10); + } + }); + photon::thread_enable_join(pest); + for (;;) { + if (ch.recv(0, 0) == SENTINEL) break; + received.fetch_add(1, std::memory_order_relaxed); + g_progress.fetch_add(1, std::memory_order_relaxed); + } + stop.store(true, std::memory_order_relaxed); + photon::thread_join((photon::join_handle*)pest); + }); + + for (uint64_t i = 1; i <= kItems; ++i) ch.send_paced(i); + AWAIT(received.load(std::memory_order_relaxed) >= kItems, "interrupted recv"); + ch.send(SENTINEL); + consumer.join(); + + EXPECT_EQ(kItems, received.load()); + EXPECT_EQ(0UL, ch.notification_pending()); +} + +// Producer and consumer on the same vCPU: a claim never needs to kick an event +// engine, and the producer regularly claims the slot of a consumer that is not +// asleep yet -- or even its own slot, when the queue turns out to be non-empty +// right after publishing it. +TEST(ring_channel, same_vcpu_pairing) { + g_case = "same_vcpu_pairing"; + TestChannel ping(NO_NET), pong(NO_NET); + constexpr uint64_t kRounds = 20000; + + auto consumer = photon::thread_create11([&] { + for (;;) { + auto x = ping.recv(0, 0); + pong.send(x); + if (x == SENTINEL) break; + } + }); + photon::thread_enable_join(consumer); + + for (uint64_t i = 1; i <= kRounds; ++i) { + ping.send(i); + ASSERT_EQ(i, pong.recv(0, 0)); + g_progress.fetch_add(1, std::memory_order_relaxed); + } + ping.send(SENTINEL); + EXPECT_EQ(SENTINEL, pong.recv(0, 0)); + photon::thread_join((photon::join_handle*)consumer); + + EXPECT_EQ(0UL, ping.notification_pending()); + EXPECT_EQ(0UL, pong.notification_pending()); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + start_watchdog(); + return RUN_ALL_TESTS(); +}