diff --git a/CMakeLists.txt b/CMakeLists.txt index 97ea97d..6525f9a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -709,6 +709,7 @@ set(APP_HEADERS src/data/exchange_info.h src/util/logger.h src/util/async_task_manager.h + src/util/serial_job_lane.h src/util/fetch_backoff.h src/util/coalescing_writer.h src/util/background_sampler.h diff --git a/src/util/serial_job_lane.h b/src/util/serial_job_lane.h new file mode 100644 index 0000000..182f464 --- /dev/null +++ b/src/util/serial_job_lane.h @@ -0,0 +1,144 @@ +// DragonX Wallet - ImGui Edition +// Copyright 2024-2026 The Hush Developers +// Released under the GPLv3 +// +// util::SerialJobLane - a FIFO of blocking jobs run one at a time on a lazily started background thread that can be +// ABANDONED. Each job returns a completion; completions run only inside pump(), on the owner's thread. Unlike +// RPCWorker / AsyncTaskManager nothing here ever joins: a job stuck in a blocking call (a lite-backend save waiting +// out a scan) is abandoned when the lane is destroyed, and its thread finishes on its own holding only the shared State. +// post() is meant to be called from ONE producer thread (the owner); pump() from the owner thread. +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace dragonx { +namespace util { + +class SerialJobLane { +public: + using Completion = std::function; // runs on the thread that calls pump() + using Job = std::function; // runs on the lane thread; may return an empty Completion + + SerialJobLane() : st_(std::make_shared()) {} + ~SerialJobLane() { abandon(); } + SerialJobLane(const SerialJobLane&) = delete; + SerialJobLane& operator=(const SerialJobLane&) = delete; + + // Owner thread. False if the job was dropped: the lane is abandoned, or no thread could be started. + bool post(Job job) + { + if (!job) return false; + bool spawn = false; + { + std::lock_guard lk(st_->m); + if (st_->closed) return false; + st_->queue.push_back(std::move(job)); + if (!st_->threadRunning) { st_->threadRunning = true; spawn = true; } + } + if (!spawn) return true; + try { + std::thread([st = st_] { run(st); }).detach(); + } catch (...) { + std::deque dropped; // destroyed outside the lock (may wipe captured secrets) + { + std::lock_guard lk(st_->m); + st_->threadRunning = false; + dropped.swap(st_->queue); // single producer: holds only the job just posted + } + return false; + } + return true; + } + + // Owner thread: run every completion produced so far, in FIFO order. Returns how many ran. + int pump() + { + std::deque ready; + { + std::lock_guard lk(st_->m); + ready.swap(st_->done); + } + int n = 0; + for (auto& c : ready) { if (c) { c(); ++n; } } + return n; + } + + bool hasCompletions() const { std::lock_guard lk(st_->m); return !st_->done.empty(); } + bool busy() const + { + std::lock_guard lk(st_->m); + return !st_->queue.empty() || st_->executing || !st_->done.empty(); + } + + // Bounded: true once nothing is queued or executing (completions may still await pump()). + bool waitIdle(std::chrono::milliseconds timeout) + { + std::unique_lock lk(st_->m); + return st_->cv.wait_for(lk, timeout, [this] { return st_->queue.empty() && !st_->executing; }); + } + + // Drop queued jobs and undelivered completions; refuse later posts. A running job finishes by itself and its + // completion is discarded. Idempotent. + void abandon() + { + std::deque q; + std::deque d; + { + std::lock_guard lk(st_->m); + st_->closed = true; + q.swap(st_->queue); + d.swap(st_->done); + } // q/d destroyed here, outside the lock + } + +private: + struct State { + mutable std::mutex m; + std::condition_variable cv; + std::deque queue; + std::deque done; + bool threadRunning = false; + bool executing = false; + bool closed = false; + }; + + static void run(std::shared_ptr st) + { + for (;;) { + Job job; + { + std::lock_guard lk(st->m); + if (st->closed || st->queue.empty()) { + st->threadRunning = false; + st->cv.notify_all(); + return; + } + job = std::move(st->queue.front()); + st->queue.pop_front(); + st->executing = true; + } + Completion done; + try { done = job(); } catch (...) {} + job = nullptr; // release captured state (secret holders) BEFORE reporting idle + { + std::lock_guard lk(st->m); + st->executing = false; + if (done && !st->closed) st->done.push_back(std::move(done)); + st->cv.notify_all(); + } + // A completion that was not queued (closed lane) is destroyed here, on this thread, never invoked. + } + } + + std::shared_ptr st_; +}; + +} // namespace util +} // namespace dragonx diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index 843dcb2..1c3677b 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -16,6 +16,7 @@ #include "daemon/embedded_daemon.h" #include "util/connect_stall.h" #include "util/fetch_backoff.h" +#include "util/serial_job_lane.h" #include "util/logger.h" #include "data/transaction_history_cache.h" #include "data/transaction_history_store_policy.h" @@ -7978,6 +7979,81 @@ void testLiteWalletControllerShutdownDoesNotHangDuringSync() std::this_thread::sleep_for(std::chrono::milliseconds(50)); // let it unwind cleanly } +// Tier 4 (g1-lite): SerialJobLane — completions only via pump(), FIFO, abandonable, exception-safe. +void testSerialJobLane() +{ + using dragonx::util::SerialJobLane; + { // completions run only inside pump(), on the pumping thread, in FIFO order + SerialJobLane lane; + std::vector order; + const auto mainId = std::this_thread::get_id(); + bool rightThread = true; + for (int i = 0; i < 3; ++i) + EXPECT_TRUE(lane.post([i, &order, &rightThread, mainId]() -> SerialJobLane::Completion { + return [i, &order, &rightThread, mainId]() { + order.push_back(i); + if (std::this_thread::get_id() != mainId) rightThread = false; + }; + })); + EXPECT_TRUE(lane.waitIdle(std::chrono::milliseconds(3000))); + EXPECT_TRUE(order.empty()); + EXPECT_EQ(lane.pump(), 3); + EXPECT_TRUE(order == std::vector({0, 1, 2})); + EXPECT_TRUE(rightThread); + } + { // post() never waits for a blocked job; waitIdle is bounded + SerialJobLane lane; + auto gate = std::make_shared>(true); + const auto t0 = std::chrono::steady_clock::now(); + lane.post([gate]() -> SerialJobLane::Completion { + while (gate->load()) std::this_thread::sleep_for(std::chrono::milliseconds(2)); + return {}; + }); + EXPECT_TRUE(std::chrono::steady_clock::now() - t0 < std::chrono::milliseconds(500)); + EXPECT_TRUE(lane.busy()); + EXPECT_FALSE(lane.waitIdle(std::chrono::milliseconds(50))); + gate->store(false); + EXPECT_TRUE(lane.waitIdle(std::chrono::milliseconds(3000))); + } + { // destroying the lane with a blocked job returns at once; queued job dropped; late completion never runs + auto gate = std::make_shared>(true); + auto ran = std::make_shared>(0); + auto entered = std::make_shared>(false); + std::chrono::steady_clock::time_point t0; + { + SerialJobLane lane; + lane.post([gate, ran, entered]() -> SerialJobLane::Completion { + entered->store(true); + while (gate->load()) std::this_thread::sleep_for(std::chrono::milliseconds(2)); + ran->fetch_add(1); + return [ran]() { ran->fetch_add(100); }; + }); + lane.post([ran]() -> SerialJobLane::Completion { ran->fetch_add(10); return {}; }); + // The first job must be RUNNING when the lane dies: a job the lane thread hasn't picked up yet is + // dropped unrun like the queued one. + for (int i = 0; i < 400 && !entered->load(); ++i) std::this_thread::sleep_for(std::chrono::milliseconds(5)); + EXPECT_TRUE(entered->load()); + t0 = std::chrono::steady_clock::now(); + } + EXPECT_TRUE(std::chrono::steady_clock::now() - t0 < std::chrono::milliseconds(500)); + gate->store(false); + for (int i = 0; i < 400 && ran->load() == 0; ++i) std::this_thread::sleep_for(std::chrono::milliseconds(5)); + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + EXPECT_EQ(ran->load(), 1); + } + { // a throwing job doesn't kill the lane; post() after abandon() is refused + SerialJobLane lane; + int got = 0; + lane.post([]() -> SerialJobLane::Completion { throw std::runtime_error("boom"); }); + lane.post([&got]() -> SerialJobLane::Completion { return [&got]() { got = 7; }; }); + EXPECT_TRUE(lane.waitIdle(std::chrono::milliseconds(3000))); + EXPECT_EQ(lane.pump(), 1); + EXPECT_EQ(got, 7); + lane.abandon(); + EXPECT_FALSE(lane.post([]() -> SerialJobLane::Completion { return {}; })); + } +} + // M5b: lite-wallet rollout / kill-switch policy (wallet/lite_rollout_policy.h). void testLiteRolloutVersionCompare() { @@ -13107,6 +13183,7 @@ int main() testLiteWalletGatewayRefreshSkipsFailedCommand(); testLiteWalletControllerWorkerProducesModel(); testLiteWalletControllerShutdownDoesNotHangDuringSync(); + testSerialJobLane(); testLiteRolloutVersionCompare(); testLiteRolloutBucketAndOverrideHelpers(); testLiteRolloutPolicyOverridesAndKillSwitch();