From 2031920137e92847186d404c919421c3dbf9af66 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Wed, 3 Jun 2026 18:39:08 +0800 Subject: [PATCH] more tests --- common/executor/test/test_burst_drain.cpp | 110 ++++++++++++++++++++++ thread/test/test-workpool-fanout.cpp | 78 +++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 common/executor/test/test_burst_drain.cpp create mode 100644 thread/test/test-workpool-fanout.cpp diff --git a/common/executor/test/test_burst_drain.cpp b/common/executor/test/test_burst_drain.cpp new file mode 100644 index 000000000..23080ccf8 --- /dev/null +++ b/common/executor/test/test_burst_drain.cpp @@ -0,0 +1,110 @@ +/* +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. +*/ + +#include +#include +#include + +#include +#include +#include +#include + +#include "../../../test/gtest.h" + +// Regression test for the RingChannel notification refactor. +// +// Failure mode being guarded: +// When N producers fire a burst, the previous design issued one +// `queue_sem.signal(1)` per push every time the consumer happened to be in +// its `idler` window. The semaphore counter accumulated linearly with the +// burst size, so after the burst the consumer would loop forever between +// 1ms busy-yield and a `wait()` that returned immediately, burning a full +// core for as long as the accumulated counter took to drain. +// +// Verification strategy (deterministic, no wall-clock assumptions): +// - Drive a real `RingChannel` directly, with `kProducers` std::thread +// producers and a single std::thread consumer that owns its own photon +// vCPU. +// - Synchronize using a producer-side `received` counter and a sentinel +// value to terminate the consumer. All "wait" operations are +// event-driven (`std::thread::join`, atomic counter spin), never +// wall-clock sleeps. +// - After the consumer thread has fully joined, all RingChannel state is +// quiescent and visible to the test thread. Assert the *invariant* +// `notification_pending() <= idler_peak`. With a single consumer, +// `idler_peak == 1`, so `pending` must remain ≤1 regardless of burst +// size. Without the fix, `pending` (== `queue_sem.m_count`) would carry +// leftover tokens proportional to how often the consumer dipped into +// the idler window during the burst. +TEST(ring_channel, burst_drain_no_signal_accumulation) { + using QT = LockfreeMPMCRingQueue; + photon::common::RingChannel ch(1024, 1024); + + constexpr int kProducers = 4; + constexpr int kPerProducer = 25000; + constexpr int kBurst = kProducers * kPerProducer; + constexpr int kSentinel = 0; // sentinel value to terminate consumer + + std::atomic received{0}; + + // Consumer owns its own photon vCPU, so it can be scheduled + // independently of the producers and of this test thread. + std::thread consumer([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + for (;;) { + int x = ch.recv(); + if (x == kSentinel) break; + received.fetch_add(1, std::memory_order_acq_rel); + } + }); + + std::vector producers; + producers.reserve(kProducers); + for (int p = 0; p < kProducers; ++p) { + producers.emplace_back([&] { + for (int i = 0; i < kPerProducer; ++i) { + // Producers are plain std::threads; ThreadPause yields the + // OS thread on the (extremely unlikely) full-queue spin. + ch.template send(/*non-zero*/ 1); + } + }); + } + for (auto& t : producers) t.join(); + + // Deterministic synchronization point: spin until the consumer has + // drained exactly the burst. No wall-clock sleeps. + while (received.load(std::memory_order_acquire) < kBurst) { + std::this_thread::yield(); + } + + // Terminate consumer cleanly, then join. After join() returns, no other + // thread can touch `ch`, so we can sample its state safely. + ch.template send(kSentinel); + consumer.join(); + + auto pending = ch.notification_pending(); + LOG_INFO("after burst: received=`, notification_pending=`", + received.load(), pending); + + // Strong invariant under the fix: in-flight wake-up tokens are capped + // by the historical peak of `idler`. With a single consumer that peak + // is 1, so `pending` must remain ≤1 regardless of burst size. Without + // the fix `pending` would scale with kBurst. + EXPECT_LE(pending, 1u); + EXPECT_EQ(kBurst, received.load()); +} diff --git a/thread/test/test-workpool-fanout.cpp b/thread/test/test-workpool-fanout.cpp new file mode 100644 index 000000000..5e4d76950 --- /dev/null +++ b/thread/test/test-workpool-fanout.cpp @@ -0,0 +1,78 @@ +/* +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. +*/ + +#include + +#include +#include +#include +#include + +#include "../../test/gtest.h" + +// Regression test for the RingChannel notification refactor's multi-consumer +// behavior: when N idle workers share a RingChannel, a burst of N tasks must +// fan out to all of them concurrently. A binary (single-token) notification +// scheme would serialize the burst onto a single worker. +// +// Verification strategy (deterministic, no wall-clock assumptions): +// - Each task signals an `arrived` photon::semaphore on entry and then +// blocks on `release` until the harness allows it to finish. +// - The harness calls `arrived.wait(N, deadline_us)`. If fan-out works, +// all N tasks reach the wait point; if the notification path serializes, +// only one worker can run at a time (the others are stuck holding their +// vCPU on `release.wait`), so `arrived` will never reach N and the wait +// will return non-zero on timeout. The deadline is generous (5s) to +// tolerate CI noise without weakening the assertion. +TEST(workpool, fanout_wakeup) { + // WorkPool::async_call uses PhotonPause and signals a photon::semaphore; + // the caller side must already be a photon vcpu. + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE); + DEFER(photon::fini()); + + constexpr int N = 4; + photon::WorkPool pool(N, photon::INIT_EVENT_DEFAULT, + photon::INIT_IO_NONE, -1); + + photon::semaphore arrived(0); + photon::semaphore release(0); + std::atomic done{0}; + + for (int i = 0; i < N; ++i) { + pool.async_call(new auto([&] { + // Mark this worker as woken-and-running. Hold the vCPU until the + // harness lets every other worker also reach this point. + arrived.signal(1); + release.wait(1); + done.fetch_add(1, std::memory_order_release); + })); + } + + // Deterministic fan-out check. With the fix, all N tasks must reach + // `arrived.signal(1)` concurrently. Without it, only one worker would + // ever be released by the channel, so `arrived` saturates at 1 and the + // wait times out. + int r = arrived.wait(N, /*timeout_us=*/5UL * 1000 * 1000); + EXPECT_EQ(0, r); + LOG_INFO("fan-out check: arrived.wait(`) returned ` (vcpu_num=`)", + N, r, N); + + // Release everyone so the WorkPool can be cleanly destroyed. + release.signal(N); + while (done.load(std::memory_order_acquire) < N) { + photon::thread_yield(); + } +}