de84edde1aA paused or admission-blocked poller slept for the full max idle delay instead of waiting on its registered timer, so resumeStage, resumeAll, enabling the knowledge graph, and adding entity or title providers took effect only after the delay elapsed. The poller now waits through awaitPressureLimitedPollerWake and the setters signal the stage timer. Tests: pressure_limited_poller and post_ingest_queue_unit. Signed-off-by: trvon <me@trevon.dev>
798 lines
31 KiB
C++
798 lines
31 KiB
C++
// PressureLimitedPoller unit tests (Catch2)
|
|
// Exercises the generic poller coroutine directly, focusing on shutdown
|
|
// invariants that are otherwise only tested indirectly through the full
|
|
// PostIngestQueue / ServiceManager integration stack.
|
|
//
|
|
// Highest-value coverage: capability-sleep shutdown path, which was traced
|
|
// to the Windows CI SIGSEGV cluster where the title poller (isCapableFn
|
|
// returning false) did not exit cleanly before PostIngestQueue teardown.
|
|
|
|
#include <catch2/catch_test_macros.hpp>
|
|
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <optional>
|
|
#include <vector>
|
|
|
|
#include <boost/asio/co_spawn.hpp>
|
|
#include <boost/asio/detached.hpp>
|
|
#include <boost/asio/io_context.hpp>
|
|
#include <boost/asio/post.hpp>
|
|
#include <boost/asio/steady_timer.hpp>
|
|
|
|
#include <yams/daemon/components/InternalEventBus.h>
|
|
#include <yams/daemon/pressure_limited_poller.h>
|
|
|
|
using namespace yams::daemon;
|
|
using namespace std::chrono_literals;
|
|
|
|
namespace {
|
|
|
|
/// Minimal poller config for testing shutdown paths.
|
|
/// The poller is spawned with an empty queue and isCapableFn returning false
|
|
/// so it immediately enters the capability-sleep branch (250ms timer loop).
|
|
/// Fields that the poller never reaches in this path are left as no-op stubs.
|
|
template <typename Task>
|
|
PressureLimitedPollerConfig<Task> makeCapabilitySleepConfig(
|
|
std::atomic<bool>& stopFlag, std::atomic<bool>& startedFlag, std::atomic<bool>& pauseFlag,
|
|
std::shared_ptr<boost::asio::steady_timer> wakeTimer, std::mutex& wakeMutex) {
|
|
PressureLimitedPollerConfig<Task> cfg{};
|
|
cfg.stageName = "test-capability-sleep";
|
|
cfg.stopFlag = &stopFlag;
|
|
cfg.startedFlag = &startedFlag;
|
|
cfg.pauseFlag = &pauseFlag;
|
|
cfg.isCapableFn = []() { return false; };
|
|
// Stubs — never called in the capability-sleep path but must be present.
|
|
cfg.maxConcurrentFn = []() -> std::size_t { return 1; };
|
|
cfg.tryAcquireFn = [](GradientLimiter*, const std::string&, const std::string&) {
|
|
return false;
|
|
};
|
|
cfg.getHashFn = [](const Task&) -> std::string { return ""; };
|
|
cfg.completeJobFn = [](const std::string&, bool) {};
|
|
cfg.processFn = [](Task&) {};
|
|
// Required even for capability-sleep path — the template checks at entry.
|
|
thread_local std::atomic<bool> tlsWasActive{false};
|
|
thread_local std::atomic<std::size_t> tlsInFlight{0};
|
|
cfg.wasActiveFlag = &tlsWasActive;
|
|
cfg.inFlightCounter = &tlsInFlight;
|
|
cfg.wakeTimer = std::move(wakeTimer);
|
|
cfg.wakeTimerMutex = &wakeMutex;
|
|
return cfg;
|
|
}
|
|
|
|
/// Normal capable poller that reaches the event-driven idle path when the
|
|
/// channel is empty. Requires a wakeTimer so the test can exercise
|
|
/// timer-cancellation-based shutdown (the path taken by extraction, kg,
|
|
/// symbol, and entity pollers in production).
|
|
template <typename Task>
|
|
PressureLimitedPollerConfig<Task>
|
|
makeWakeTimerConfig(std::atomic<bool>& stopFlag, std::atomic<bool>& startedFlag,
|
|
std::atomic<bool>& pauseFlag,
|
|
std::shared_ptr<boost::asio::steady_timer> wakeTimer, std::mutex& wakeMutex) {
|
|
PressureLimitedPollerConfig<Task> cfg{};
|
|
cfg.stageName = "test-wake-timer";
|
|
cfg.stopFlag = &stopFlag;
|
|
cfg.startedFlag = &startedFlag;
|
|
cfg.pauseFlag = &pauseFlag;
|
|
cfg.isCapableFn = []() { return true; };
|
|
cfg.maxConcurrentFn = []() -> std::size_t { return 1; };
|
|
cfg.tryAcquireFn = [](GradientLimiter*, const std::string&, const std::string&) {
|
|
return false;
|
|
};
|
|
cfg.getHashFn = [](const Task&) -> std::string { return ""; };
|
|
cfg.completeJobFn = [](const std::string&, bool) {};
|
|
cfg.processFn = [](Task&) {};
|
|
cfg.wakeTimer = std::move(wakeTimer);
|
|
cfg.wakeTimerMutex = &wakeMutex;
|
|
// Required even for wake-timer path — the template checks at entry.
|
|
thread_local std::atomic<bool> tlsWasActive{false};
|
|
thread_local std::atomic<std::size_t> tlsInFlight{0};
|
|
cfg.wasActiveFlag = &tlsWasActive;
|
|
cfg.inFlightCounter = &tlsInFlight;
|
|
return cfg;
|
|
}
|
|
|
|
class BatchPollerHarness {
|
|
public:
|
|
explicit BatchPollerHarness(std::chrono::milliseconds coalesceWindow,
|
|
bool limiterEnabled = true)
|
|
: coalesceWindow_(coalesceWindow), limiterEnabled_(limiterEnabled),
|
|
wakeTimer_(std::make_shared<boost::asio::steady_timer>(ioContext_)),
|
|
channel_(std::make_shared<SpscQueue<int>>(16)) {}
|
|
|
|
bool pushNormal(int value) { return channel_->try_push(value); }
|
|
|
|
void postNormalAfterAdmission(int admittedValue, std::vector<int> values) {
|
|
postAfterAdmissionValue_ = admittedValue;
|
|
postAfterAdmissionValues_ = std::move(values);
|
|
}
|
|
|
|
bool deferredPushSucceeded() const { return deferredPushSucceeded_; }
|
|
|
|
const std::vector<std::string>& acquiredLimiterIds() const { return acquiredLimiterIds_; }
|
|
const std::vector<std::string>& completedLimiterIds() const { return completedLimiterIds_; }
|
|
std::size_t hashCalls() const { return hashCalls_; }
|
|
|
|
bool pushHighPriority(int value) {
|
|
if (!highPriorityChannel_) {
|
|
highPriorityChannel_ = std::make_shared<SpscQueue<int>>(16);
|
|
}
|
|
return highPriorityChannel_->try_push(value);
|
|
}
|
|
|
|
void scheduleNormal(int value, std::chrono::milliseconds delay) {
|
|
auto timer = std::make_shared<boost::asio::steady_timer>(ioContext_, delay);
|
|
timer->async_wait([this, value](boost::system::error_code ec) {
|
|
if (!ec) {
|
|
(void)channel_->try_push(value);
|
|
cancelWakeTimer();
|
|
}
|
|
});
|
|
arrivals_.push_back(std::move(timer));
|
|
}
|
|
|
|
void run(std::chrono::milliseconds timeout = 100ms) {
|
|
startedAt_ = std::chrono::steady_clock::now();
|
|
auto deadline = std::make_shared<boost::asio::steady_timer>(ioContext_, timeout);
|
|
deadline->async_wait([this](boost::system::error_code ec) {
|
|
if (!ec) {
|
|
timedOut = true;
|
|
stop();
|
|
}
|
|
});
|
|
|
|
auto cfg = makeConfig();
|
|
cfg.batchProcessFn = [this, deadline](std::vector<int>&& batch) {
|
|
observed = std::move(batch);
|
|
elapsed = std::chrono::steady_clock::now() - startedAt_;
|
|
stop();
|
|
deadline->cancel();
|
|
};
|
|
boost::asio::co_spawn(ioContext_, pressureLimitedPoll<int>(channel_, std::move(cfg)),
|
|
boost::asio::detached);
|
|
ioContext_.run();
|
|
}
|
|
|
|
std::vector<int> observed;
|
|
std::chrono::steady_clock::duration elapsed{};
|
|
bool timedOut{false};
|
|
|
|
private:
|
|
PressureLimitedPollerConfig<int> makeConfig() {
|
|
PressureLimitedPollerConfig<int> cfg;
|
|
cfg.stageName = "test-batch";
|
|
cfg.stopFlag = &stopFlag_;
|
|
cfg.startedFlag = &startedFlag_;
|
|
cfg.pauseFlag = &pauseFlag_;
|
|
cfg.wasActiveFlag = &wasActive_;
|
|
cfg.inFlightCounter = &inFlight_;
|
|
cfg.getLimiterFn = [this]() -> GradientLimiter* {
|
|
return limiterEnabled_ ? &limiter_ : nullptr;
|
|
};
|
|
cfg.maxConcurrentFn = []() -> std::size_t { return 4; };
|
|
cfg.tryAcquireFn = [this](GradientLimiter*, const std::string& limiterId,
|
|
const std::string&) {
|
|
acquiredLimiterIds_.push_back(limiterId);
|
|
if (postAfterAdmissionValue_) {
|
|
const auto expected = std::to_string(*postAfterAdmissionValue_);
|
|
if (limiterId == expected || limiterId.ends_with("#" + expected)) {
|
|
postAfterAdmissionValue_.reset();
|
|
auto values = std::move(postAfterAdmissionValues_);
|
|
boost::asio::post(ioContext_, [this, values = std::move(values)]() {
|
|
for (const int value : values) {
|
|
deferredPushSucceeded_ =
|
|
channel_->try_push(value) && deferredPushSucceeded_;
|
|
}
|
|
cancelWakeTimer();
|
|
});
|
|
}
|
|
}
|
|
return true;
|
|
};
|
|
cfg.completeJobFn = [this](const std::string& limiterId, bool) {
|
|
completedLimiterIds_.push_back(limiterId);
|
|
};
|
|
cfg.getHashFn = [this](const int& task) {
|
|
++hashCalls_;
|
|
return std::to_string(task);
|
|
};
|
|
cfg.executor = ioContext_.get_executor();
|
|
cfg.batchMode = true;
|
|
cfg.batchSizeFn = []() -> std::size_t { return 4; };
|
|
cfg.batchCoalesceWindowFn = [this]() { return coalesceWindow_; };
|
|
cfg.highPriorityChannel = highPriorityChannel_;
|
|
cfg.highPriorityMaxPerBatchFn = []() -> std::size_t { return 4; };
|
|
cfg.enableCpuThrottling = false;
|
|
cfg.wakeTimer = wakeTimer_;
|
|
cfg.wakeTimerMutex = &wakeMutex_;
|
|
return cfg;
|
|
}
|
|
|
|
void cancelWakeTimer() {
|
|
std::lock_guard<std::mutex> lock(wakeMutex_);
|
|
wakeTimer_->cancel();
|
|
}
|
|
|
|
void stop() {
|
|
stopFlag_.store(true, std::memory_order_release);
|
|
cancelWakeTimer();
|
|
for (const auto& timer : arrivals_) {
|
|
timer->cancel();
|
|
}
|
|
}
|
|
|
|
std::chrono::milliseconds coalesceWindow_;
|
|
bool limiterEnabled_{true};
|
|
GradientLimiter limiter_{"test-batch"};
|
|
boost::asio::io_context ioContext_;
|
|
std::atomic<bool> stopFlag_{false};
|
|
std::atomic<bool> startedFlag_{false};
|
|
std::atomic<bool> pauseFlag_{false};
|
|
std::atomic<bool> wasActive_{false};
|
|
std::atomic<std::size_t> inFlight_{0};
|
|
std::mutex wakeMutex_;
|
|
std::shared_ptr<boost::asio::steady_timer> wakeTimer_;
|
|
std::shared_ptr<SpscQueue<int>> channel_;
|
|
std::shared_ptr<SpscQueue<int>> highPriorityChannel_;
|
|
std::vector<std::shared_ptr<boost::asio::steady_timer>> arrivals_;
|
|
std::optional<int> postAfterAdmissionValue_;
|
|
std::vector<int> postAfterAdmissionValues_;
|
|
std::vector<std::string> acquiredLimiterIds_;
|
|
std::vector<std::string> completedLimiterIds_;
|
|
std::size_t hashCalls_{0};
|
|
bool deferredPushSucceeded_{true};
|
|
std::chrono::steady_clock::time_point startedAt_{};
|
|
};
|
|
|
|
enum class BlockedPollerReason { Paused, ZeroConcurrency, AdmissionPaused };
|
|
|
|
void checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason reason,
|
|
bool stopInsteadOfResume = false,
|
|
bool enqueueAfterRegistration = false) {
|
|
boost::asio::io_context ioc;
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{reason == BlockedPollerReason::Paused};
|
|
std::atomic<bool> blocked{true};
|
|
std::atomic<bool> wasActive{false};
|
|
std::atomic<std::size_t> inFlight{0};
|
|
std::atomic<std::size_t> callbacksInFlight{0};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
wakeTimer->expires_at(boost::asio::steady_timer::time_point::max());
|
|
auto watchdog = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
if (!enqueueAfterRegistration) {
|
|
REQUIRE(channel->try_push(42));
|
|
}
|
|
|
|
std::vector<int> observed;
|
|
std::size_t drainChecks = 0;
|
|
std::size_t lifecycleNotifications = 0;
|
|
std::size_t cancelledWaits = 0;
|
|
bool timedOut = false;
|
|
|
|
PressureLimitedPollerConfig<int> cfg;
|
|
cfg.stageName = "test-registered-blocked-wake";
|
|
cfg.stopFlag = &stopFlag;
|
|
cfg.startedFlag = &startedFlag;
|
|
cfg.pauseFlag = &pauseFlag;
|
|
cfg.wasActiveFlag = &wasActive;
|
|
cfg.inFlightCounter = &inFlight;
|
|
cfg.callbackInFlightCounter = &callbacksInFlight;
|
|
cfg.maxConcurrentFn = [&blocked, reason]() -> std::size_t {
|
|
return reason == BlockedPollerReason::ZeroConcurrency &&
|
|
blocked.load(std::memory_order_acquire)
|
|
? 0
|
|
: 1;
|
|
};
|
|
cfg.admissionPausedFn = [&blocked, reason]() {
|
|
return reason == BlockedPollerReason::AdmissionPaused &&
|
|
blocked.load(std::memory_order_acquire);
|
|
};
|
|
cfg.tryAcquireFn = [](GradientLimiter*, const std::string&, const std::string&) {
|
|
return true;
|
|
};
|
|
cfg.completeJobFn = [](const std::string&, bool) {};
|
|
cfg.checkDrainFn = [&drainChecks]() { ++drainChecks; };
|
|
cfg.notifyLifecycleFn = [&lifecycleNotifications]() { ++lifecycleNotifications; };
|
|
cfg.getHashFn = [](const int& task) { return std::to_string(task); };
|
|
cfg.processFn = [&observed, &stopFlag, &wakeTimer, &wakeMutex, &watchdog](int& task) {
|
|
observed.push_back(task);
|
|
stopFlag.store(true, std::memory_order_release);
|
|
watchdog->cancel();
|
|
std::lock_guard<std::mutex> lock(wakeMutex);
|
|
wakeTimer->cancel();
|
|
};
|
|
cfg.executor = ioc.get_executor();
|
|
cfg.enableCpuThrottling = false;
|
|
cfg.wakeTimer = wakeTimer;
|
|
cfg.wakeTimerMutex = &wakeMutex;
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
|
|
// Drain ready startup handlers without assuming co_spawn uses exactly one handler.
|
|
// The old coroutine-local timer leaves our explicitly initialized sentinel unchanged.
|
|
REQUIRE(ioc.poll() > 0);
|
|
REQUIRE(startedFlag.load(std::memory_order_acquire));
|
|
CHECK(observed.empty());
|
|
CHECK(channel->size_approx() == (enqueueAfterRegistration ? 0U : 1U));
|
|
CHECK(wakeTimer->expiry() != boost::asio::steady_timer::time_point::max());
|
|
if (enqueueAfterRegistration) {
|
|
REQUIRE(channel->try_push(42));
|
|
}
|
|
|
|
if (stopInsteadOfResume) {
|
|
stopFlag.store(true, std::memory_order_release);
|
|
} else {
|
|
blocked.store(false, std::memory_order_release);
|
|
pauseFlag.store(false, std::memory_order_release);
|
|
}
|
|
auto wakeExecutor = wakeTimer->get_executor();
|
|
boost::asio::post(wakeExecutor, [wakeTimer, watchdog, &cancelledWaits, stopInsteadOfResume]() {
|
|
cancelledWaits = wakeTimer->cancel();
|
|
if (stopInsteadOfResume) {
|
|
watchdog->cancel();
|
|
}
|
|
});
|
|
|
|
watchdog->expires_after(500ms);
|
|
watchdog->async_wait([&](const boost::system::error_code& error) {
|
|
if (!error) {
|
|
timedOut = true;
|
|
stopFlag.store(true, std::memory_order_release);
|
|
std::lock_guard<std::mutex> lock(wakeMutex);
|
|
wakeTimer->cancel();
|
|
}
|
|
});
|
|
ioc.run();
|
|
|
|
CHECK_FALSE(timedOut);
|
|
CHECK(cancelledWaits == 1);
|
|
if (stopInsteadOfResume) {
|
|
CHECK(observed.empty());
|
|
CHECK(channel->size_approx() == 1);
|
|
CHECK(drainChecks == 0);
|
|
CHECK(lifecycleNotifications >= 2);
|
|
} else {
|
|
CHECK((observed == std::vector<int>{42}));
|
|
CHECK(channel->size_approx() == 0);
|
|
CHECK(drainChecks == 1);
|
|
CHECK(lifecycleNotifications >= 3);
|
|
}
|
|
CHECK(inFlight.load(std::memory_order_acquire) == 0);
|
|
CHECK(callbacksInFlight.load(std::memory_order_acquire) == 0);
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
}
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("PressureLimitedPoller coalesces normal-priority trickle into a bounded batch",
|
|
"[daemon][poller][batch][coalesce][catch2]") {
|
|
BatchPollerHarness harness{10ms};
|
|
REQUIRE(harness.pushNormal(1));
|
|
harness.postNormalAfterAdmission(1, {2, 3, 4});
|
|
harness.run();
|
|
|
|
CHECK_FALSE(harness.timedOut);
|
|
CHECK(harness.deferredPushSucceeded());
|
|
CHECK((harness.observed == std::vector<int>{1, 2, 3, 4}));
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller does not coalesce high-priority batch work",
|
|
"[daemon][poller][batch][coalesce][priority][catch2]") {
|
|
BatchPollerHarness harness{20ms};
|
|
REQUIRE(harness.pushHighPriority(42));
|
|
harness.scheduleNormal(7, 2ms);
|
|
harness.run();
|
|
|
|
CHECK((harness.observed == std::vector<int>{42}));
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller bounds lone normal-task coalescing latency",
|
|
"[daemon][poller][batch][coalesce][latency][catch2]") {
|
|
BatchPollerHarness harness{3ms};
|
|
REQUIRE(harness.pushNormal(9));
|
|
harness.run(50ms);
|
|
|
|
CHECK_FALSE(harness.timedOut);
|
|
CHECK((harness.observed == std::vector<int>{9}));
|
|
CHECK((harness.elapsed < 50ms));
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller assigns unique limiter IDs to duplicate task identities",
|
|
"[daemon][poller][limiter][identity][catch2]") {
|
|
BatchPollerHarness harness{0ms};
|
|
REQUIRE(harness.pushNormal(7));
|
|
REQUIRE(harness.pushNormal(7));
|
|
harness.run();
|
|
|
|
REQUIRE((harness.acquiredLimiterIds().size() == 2));
|
|
CHECK((harness.acquiredLimiterIds()[0] != harness.acquiredLimiterIds()[1]));
|
|
CHECK((harness.completedLimiterIds() == harness.acquiredLimiterIds()));
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller batch path skips identities without a limiter",
|
|
"[daemon][poller][batch][limiter][identity][catch2]") {
|
|
BatchPollerHarness harness{0ms, false};
|
|
REQUIRE(harness.pushNormal(7));
|
|
REQUIRE(harness.pushNormal(8));
|
|
harness.run();
|
|
|
|
CHECK((harness.observed == std::vector<int>{7, 8}));
|
|
CHECK((harness.hashCalls() == 0));
|
|
CHECK(harness.acquiredLimiterIds().empty());
|
|
CHECK(harness.completedLimiterIds().empty());
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller single-item path skips identities without a limiter",
|
|
"[daemon][poller][single][limiter][identity][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::atomic<bool> wasActive{false};
|
|
std::atomic<std::size_t> inFlight{0};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
REQUIRE(channel->try_push(9));
|
|
|
|
std::size_t hashCalls = 0;
|
|
std::size_t completedCalls = 0;
|
|
std::vector<int> observed;
|
|
PressureLimitedPollerConfig<int> cfg;
|
|
cfg.stageName = "test-single";
|
|
cfg.stopFlag = &stopFlag;
|
|
cfg.startedFlag = &startedFlag;
|
|
cfg.pauseFlag = &pauseFlag;
|
|
cfg.wasActiveFlag = &wasActive;
|
|
cfg.inFlightCounter = &inFlight;
|
|
cfg.getLimiterFn = []() -> GradientLimiter* { return nullptr; };
|
|
cfg.maxConcurrentFn = []() -> std::size_t { return 1; };
|
|
cfg.tryAcquireFn = [](GradientLimiter*, const std::string&, const std::string&) {
|
|
return true;
|
|
};
|
|
cfg.completeJobFn = [&completedCalls](const std::string&, bool) { ++completedCalls; };
|
|
cfg.getHashFn = [&hashCalls](const int& task) {
|
|
++hashCalls;
|
|
return std::to_string(task);
|
|
};
|
|
cfg.processFn = [&observed, &stopFlag, &wakeTimer, &wakeMutex](int& task) {
|
|
observed.push_back(task);
|
|
stopFlag.store(true, std::memory_order_release);
|
|
std::lock_guard<std::mutex> lock(wakeMutex);
|
|
wakeTimer->cancel();
|
|
};
|
|
cfg.executor = ioc.get_executor();
|
|
cfg.enableCpuThrottling = false;
|
|
cfg.wakeTimer = wakeTimer;
|
|
cfg.wakeTimerMutex = &wakeMutex;
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
ioc.run();
|
|
|
|
CHECK((observed == std::vector<int>{9}));
|
|
CHECK((hashCalls == 0));
|
|
CHECK((completedCalls == 0));
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller blocked branches arm and honor the registered wake timer",
|
|
"[daemon][poller][wake-timer][blocked][catch2]") {
|
|
SECTION("paused") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::Paused);
|
|
}
|
|
SECTION("zero concurrency") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::ZeroConcurrency);
|
|
}
|
|
SECTION("admission paused") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::AdmissionPaused);
|
|
}
|
|
SECTION("enqueue after registered wait") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::Paused, false, true);
|
|
}
|
|
SECTION("stop while paused preserves queued work") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::Paused, true);
|
|
}
|
|
SECTION("stop while concurrency is zero preserves queued work") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::ZeroConcurrency, true);
|
|
}
|
|
SECTION("stop while admission is paused preserves queued work") {
|
|
checkRegisteredWakeResumesBlockedPoller(BlockedPollerReason::AdmissionPaused, true);
|
|
}
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller leaves source work queued while downstream admission is paused",
|
|
"[daemon][poller][backpressure][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::atomic<bool> admissionPaused{true};
|
|
std::atomic<bool> wasActive{false};
|
|
std::atomic<std::size_t> inFlight{0};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
auto watchdog = std::make_shared<boost::asio::steady_timer>(ioc, 250ms);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
REQUIRE(channel->try_push(9));
|
|
|
|
bool processedWhilePaused = false;
|
|
std::vector<int> observed;
|
|
PressureLimitedPollerConfig<int> cfg;
|
|
cfg.stageName = "test-downstream-backpressure";
|
|
cfg.stopFlag = &stopFlag;
|
|
cfg.startedFlag = &startedFlag;
|
|
cfg.pauseFlag = &pauseFlag;
|
|
cfg.wasActiveFlag = &wasActive;
|
|
cfg.inFlightCounter = &inFlight;
|
|
cfg.getLimiterFn = []() -> GradientLimiter* { return nullptr; };
|
|
cfg.maxConcurrentFn = []() -> std::size_t { return 1; };
|
|
cfg.admissionPausedFn = [&admissionPaused] {
|
|
return admissionPaused.load(std::memory_order_acquire);
|
|
};
|
|
cfg.tryAcquireFn = [](GradientLimiter*, const std::string&, const std::string&) {
|
|
return true;
|
|
};
|
|
cfg.completeJobFn = [](const std::string&, bool) {};
|
|
cfg.getHashFn = [](const int& task) { return std::to_string(task); };
|
|
cfg.processFn = [&observed, &processedWhilePaused, &admissionPaused, &stopFlag, &wakeTimer,
|
|
&watchdog, &wakeMutex](int& task) {
|
|
processedWhilePaused = admissionPaused.load(std::memory_order_acquire);
|
|
observed.push_back(task);
|
|
stopFlag.store(true, std::memory_order_release);
|
|
watchdog->cancel();
|
|
std::lock_guard<std::mutex> lock(wakeMutex);
|
|
wakeTimer->cancel();
|
|
};
|
|
cfg.executor = ioc.get_executor();
|
|
cfg.enableCpuThrottling = false;
|
|
cfg.wakeTimer = wakeTimer;
|
|
cfg.wakeTimerMutex = &wakeMutex;
|
|
|
|
boost::asio::steady_timer releaseTimer(ioc, 10ms);
|
|
releaseTimer.async_wait([&admissionPaused](const boost::system::error_code& error) {
|
|
if (!error) {
|
|
admissionPaused.store(false, std::memory_order_release);
|
|
}
|
|
});
|
|
watchdog->async_wait(
|
|
[&stopFlag, &wakeTimer, &wakeMutex](const boost::system::error_code& error) {
|
|
if (!error) {
|
|
stopFlag.store(true, std::memory_order_release);
|
|
std::lock_guard<std::mutex> lock(wakeMutex);
|
|
wakeTimer->cancel();
|
|
}
|
|
});
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
ioc.run();
|
|
|
|
CHECK_FALSE(processedWhilePaused);
|
|
CHECK((observed == std::vector<int>{9}));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Capability-sleep shutdown (the Windows title-poller crash path)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
TEST_CASE("PressureLimitedPoller exits on stop while sleeping in capability check",
|
|
"[daemon][poller][shutdown][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
|
|
auto cfg =
|
|
makeCapabilitySleepConfig<int>(stopFlag, startedFlag, pauseFlag, wakeTimer, wakeMutex);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
|
|
// Step 1: advance the coroutine until it signals started.
|
|
{
|
|
auto deadline = std::chrono::steady_clock::now() + 500ms;
|
|
while (!startedFlag.load(std::memory_order_acquire) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
ioc.poll_one();
|
|
}
|
|
}
|
|
REQUIRE(startedFlag.load(std::memory_order_acquire));
|
|
|
|
// Step 2: request stop and run the io_context until the poller exits.
|
|
stopFlag.store(true, std::memory_order_release);
|
|
|
|
// Deadline timer stops the io_context after 800ms to prevent hangs.
|
|
boost::asio::steady_timer deadline(ioc, 800ms);
|
|
deadline.async_wait([&ioc](boost::system::error_code) { ioc.stop(); });
|
|
|
|
ioc.run();
|
|
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Idempotent stop: calling stop before the poller starts should be safe
|
|
// ---------------------------------------------------------------------------
|
|
|
|
TEST_CASE("PressureLimitedPoller stop before start is safe", "[daemon][poller][shutdown][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
|
|
std::atomic<bool> stopFlag{true}; // already stopped
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
|
|
auto cfg =
|
|
makeCapabilitySleepConfig<int>(stopFlag, startedFlag, pauseFlag, wakeTimer, wakeMutex);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
|
|
// The poller should see stopFlag==true immediately and never set startedFlag.
|
|
boost::asio::steady_timer deadline(ioc, 200ms);
|
|
deadline.async_wait([&ioc](boost::system::error_code) { ioc.stop(); });
|
|
|
|
ioc.run();
|
|
|
|
// startedFlag should never have been set (or was immediately cleared).
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Wake-timer cancellation shutdown (extraction/kg/symbol/entity poller path)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
TEST_CASE("PressureLimitedPoller exits on stop when idling on wake-timer",
|
|
"[daemon][poller][shutdown][wake-timer][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
|
|
auto cfg = makeWakeTimerConfig<int>(stopFlag, startedFlag, pauseFlag, wakeTimer, wakeMutex);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
|
|
// Advance until the poller signals started.
|
|
{
|
|
auto deadline = std::chrono::steady_clock::now() + 500ms;
|
|
while (!startedFlag.load(std::memory_order_acquire) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
ioc.poll_one();
|
|
}
|
|
}
|
|
REQUIRE(startedFlag.load(std::memory_order_acquire));
|
|
|
|
// The poller is now in the event-driven idle loop, waiting on wakeTimer
|
|
// (10ms expiry). Cancel the timer to simulate signalAllWakeTimers(),
|
|
// then request stop.
|
|
{
|
|
std::lock_guard<std::mutex> lock(wakeMutex);
|
|
wakeTimer->cancel();
|
|
}
|
|
stopFlag.store(true, std::memory_order_release);
|
|
|
|
// Run until the poller exits.
|
|
boost::asio::steady_timer deadline(ioc, 500ms);
|
|
deadline.async_wait([&ioc](boost::system::error_code) { ioc.stop(); });
|
|
ioc.run();
|
|
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
}
|
|
|
|
TEST_CASE("PressureLimitedPoller exits when wake-timer expires naturally during stop",
|
|
"[daemon][poller][shutdown][wake-timer][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
|
|
auto cfg = makeWakeTimerConfig<int>(stopFlag, startedFlag, pauseFlag, wakeTimer, wakeMutex);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, std::move(cfg)),
|
|
boost::asio::detached);
|
|
|
|
// Let the poller start.
|
|
{
|
|
auto deadline = std::chrono::steady_clock::now() + 500ms;
|
|
while (!startedFlag.load(std::memory_order_acquire) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
ioc.poll_one();
|
|
}
|
|
}
|
|
REQUIRE(startedFlag.load(std::memory_order_acquire));
|
|
|
|
// Set stopFlag WITHOUT explicitly cancelling the timer — the 10ms
|
|
// wake-timer should expire naturally, the poller sees stopFlag, and exits.
|
|
stopFlag.store(true, std::memory_order_release);
|
|
|
|
boost::asio::steady_timer deadline(ioc, 500ms);
|
|
deadline.async_wait([&ioc](boost::system::error_code) { ioc.stop(); });
|
|
ioc.run();
|
|
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Repeated stop/unstop cycles
|
|
// ---------------------------------------------------------------------------
|
|
|
|
TEST_CASE("PressureLimitedPoller survives start-stop-restart cycle",
|
|
"[daemon][poller][shutdown][catch2]") {
|
|
boost::asio::io_context ioc;
|
|
|
|
std::atomic<bool> stopFlag{false};
|
|
std::atomic<bool> startedFlag{false};
|
|
std::atomic<bool> pauseFlag{false};
|
|
std::mutex wakeMutex;
|
|
auto wakeTimer = std::make_shared<boost::asio::steady_timer>(ioc);
|
|
|
|
auto cfg =
|
|
makeCapabilitySleepConfig<int>(stopFlag, startedFlag, pauseFlag, wakeTimer, wakeMutex);
|
|
auto channel = std::make_shared<SpscQueue<int>>(16);
|
|
|
|
// Cycle 1
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, cfg), boost::asio::detached);
|
|
|
|
{
|
|
auto deadline = std::chrono::steady_clock::now() + 500ms;
|
|
while (!startedFlag.load(std::memory_order_acquire) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
ioc.poll_one();
|
|
}
|
|
}
|
|
REQUIRE(startedFlag.load(std::memory_order_acquire));
|
|
|
|
stopFlag.store(true, std::memory_order_release);
|
|
{
|
|
boost::asio::steady_timer deadline(ioc, 800ms);
|
|
deadline.async_wait([&ioc](boost::system::error_code) { ioc.stop(); });
|
|
ioc.run();
|
|
}
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
|
|
// Cycle 2 — restart after stop
|
|
ioc.restart();
|
|
stopFlag.store(false, std::memory_order_release);
|
|
startedFlag.store(false, std::memory_order_release);
|
|
|
|
boost::asio::co_spawn(ioc, pressureLimitedPoll<int>(channel, cfg), boost::asio::detached);
|
|
|
|
{
|
|
auto deadline = std::chrono::steady_clock::now() + 500ms;
|
|
while (!startedFlag.load(std::memory_order_acquire) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
ioc.poll_one();
|
|
}
|
|
}
|
|
REQUIRE(startedFlag.load(std::memory_order_acquire));
|
|
|
|
stopFlag.store(true, std::memory_order_release);
|
|
{
|
|
boost::asio::steady_timer deadline(ioc, 800ms);
|
|
deadline.async_wait([&ioc](boost::system::error_code) { ioc.stop(); });
|
|
ioc.run();
|
|
}
|
|
CHECK_FALSE(startedFlag.load(std::memory_order_acquire));
|
|
}
|