feat(util): add SerialJobLane
A FIFO of blocking jobs, run one at a time on a background thread that a post starts when none is running and that exits once the queue is empty. Each job returns a completion, and completions run only in pump(), on the owner's thread. Unlike RPCWorker and AsyncTaskManager nothing ever joins: abandon() (also run by the destructor) drops queued jobs and undelivered completions and refuses later posts, and a job stuck in a blocking call finishes on its own thread. It is meant for backend calls that can wait out a long scan and may have to be given up. post() reports a dropped job (abandoned lane, or no thread could be started), and waitIdle() is bounded. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 <chrono>
|
||||
#include <condition_variable>
|
||||
#include <deque>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
|
||||
namespace dragonx {
|
||||
namespace util {
|
||||
|
||||
class SerialJobLane {
|
||||
public:
|
||||
using Completion = std::function<void()>; // runs on the thread that calls pump()
|
||||
using Job = std::function<Completion()>; // runs on the lane thread; may return an empty Completion
|
||||
|
||||
SerialJobLane() : st_(std::make_shared<State>()) {}
|
||||
~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<std::mutex> 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<Job> dropped; // destroyed outside the lock (may wipe captured secrets)
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<Completion> ready;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> lk(st_->m); return !st_->done.empty(); }
|
||||
bool busy() const
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<Job> q;
|
||||
std::deque<Completion> d;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<Job> queue;
|
||||
std::deque<Completion> done;
|
||||
bool threadRunning = false;
|
||||
bool executing = false;
|
||||
bool closed = false;
|
||||
};
|
||||
|
||||
static void run(std::shared_ptr<State> st)
|
||||
{
|
||||
for (;;) {
|
||||
Job job;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<State> st_;
|
||||
};
|
||||
|
||||
} // namespace util
|
||||
} // namespace dragonx
|
||||
@@ -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<int> 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<int>({0, 1, 2}));
|
||||
EXPECT_TRUE(rightThread);
|
||||
}
|
||||
{ // post() never waits for a blocked job; waitIdle is bounded
|
||||
SerialJobLane lane;
|
||||
auto gate = std::make_shared<std::atomic<bool>>(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<std::atomic<bool>>(true);
|
||||
auto ran = std::make_shared<std::atomic<int>>(0);
|
||||
auto entered = std::make_shared<std::atomic<bool>>(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();
|
||||
|
||||
Reference in New Issue
Block a user