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(); +}