logos-co/logos-protocol
1
0
Fork 0
mirror of https://github.com/logos-co/logos-protocol.git synced 2026-08-27 12:01:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge base branch: the burst-drain experiment is a burst on both platforms

Browse Source 
Takes the base's gated burst and retargets it to the fold, resolving the two
conflicts by hand.

  * plain_logos_object.h — the base rewrote the paragraph on what an idle handle
    retains, on the waiter registry this branch deletes. Kept deleted.

  * test_plain_waiter_reaping.cpp — took the base's restructure whole (the gated
    provider, the paced release, the measured precondition, the ReleaseOnExit
    guard, the watchdog) and pointed it at CallState::inflight.

WHAT CARRIES OVER UNCHANGED. The experiment's flaw was in the experiment, not in
the mechanism under it: "issue 800 calls in a loop and hope they overlap" is a
race between spawn cost and RPC latency, and on Linux the burst completed while
it was still being issued (0-237 of 800 outstanding at the end of the loop,
median 85). That is just as wrong here, so the gate is just as necessary: every
call parks in the provider until the test releases it, and the burst's own
precondition is now asserted — 0 replies produced, 800/800 registered — in all
760 runs of both arms at every load level on both platforms.

WHAT DOES NOT. The base's residue is the last exit batch, so its bound had to be
paced into determinism and then stated against the release step (4 * kRelease =
64). Here a delivered call has already left the registry and the residue is
ZERO. The bound goes back to a small constant, 8, and it is slack rather than
headroom over a distribution. The pacing stays anyway — the structure is worth
keeping identical while both branches are live — and is documented as not
load-bearing, measured: with kRelease set to kBurst, one release and all 800
answered together, the residue is still 0, 30 runs per platform.

THE DEFECT ARM HAD TO BE RE-CHOSEN, because the exit-guard reap the base's arm
removes does not exist here. The mechanism-appropriate inversion is to drop
`st->inflight.erase(id)` from AsyncCall::deliver(), so a completed call keeps its
registration. Seven cells, 380 runs, both platforms, idle through 64x CPU
oversubscription: 800 every run, and 801 after the follow-up call. All 380 fail
the assertion; all 380 unmodified runs pass it at 0.

AND THE PEAK IS NOW THE HEADLINE. 800 async calls outstanding at once is a
measured fact on this branch rather than a hope, and on this branch it costs no
threads at all — which is the thing the fold exists to do. The same reading on
the base is 800 live std::threads.

Cannot hang: inherited from the base and re-verified here — with the release
deleted, this test FAILS in 20.1s naming the gate.

nix build .#tests: 302/302 on both platforms.
This commit is contained in:
Dario Gabriel Lipicar
2026-08-08 15:22:45 -03:00
parent 94239acea6 abe11cc119
commit cf1b9b0fc7
1 changed files with 281 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/protocol/test_plain_waiter_reaping.cpp
+281 -22
View File
@@ -95,6 +95,7 @@
#include <QVariant>
#include <QVariantList>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
@@ -149,30 +150,67 @@ size_t inflightCount(PlainLogosObject* obj)
}
// Answers `ping` immediately — every call in the retention tests COMPLETES,
// which is the case that leaks — and parks on `block` until the test lets go,
// for the one place that needs a call genuinely still in flight.
// which is the case that leaks — and parks on `gate` / `block` until the test
// lets go. The two parked flavours differ in WHAT releases them: `block` waits
// for a single all-or-nothing release, for the one place that needs a call
// genuinely still in flight, while `gate` waits for its TURN, which is what lets
// the burst below hold every reply until the whole burst is outstanding and then
// hand them back at a rate the test controls.
class EchoProvider : public LogosProviderObject {
public:
QVariant callMethod(const QString& method, const QVariantList& args) override
{
if (method == QLatin1String("ping")) return args.value(0, QVariant(1));
if (method == QLatin1String("ping")) {
m_answered.fetch_add(1);
return args.value(0, QVariant(1));
}
if (method == QLatin1String("gate")) {
if (!awaitTurn()) m_gateExpired.store(true);
m_answered.fetch_add(1);
return args.value(0, QVariant(1));
}
if (method == QLatin1String("block")) {
std::unique_lock<std::mutex> lk(m_mu);
m_cv.wait(lk, [this] { return m_released; });
await();
m_answered.fetch_add(1);
return QVariant(42);
}
return QVariant();
}
// Release everything, parked and future, `gate` and `block` alike.
void letGo()
{
{
std::lock_guard<std::mutex> g(m_mu);
m_released = true;
m_allowed = kAll;
}
m_cv.notify_all();
}
// Let the first `n` gated calls through — the burst's pacing knob. Dispatch
// is single-threaded, so arrival order is call order and this is exactly
// "answer calls 0..n-1".
void allow(int n)
{
{
std::lock_guard<std::mutex> g(m_mu);
if (n > m_allowed) m_allowed = n;
}
m_cv.notify_all();
}
// Replies PRODUCED so far, counted immediately before each one is handed
// back to the host. This is the direct measure of "has anything answered
// yet", direct in the sense that it does not go through the client at all —
// so unlike a delivery count it cannot be fooled by a callback that has been
// posted to the Qt event loop but not yet pumped, and unlike a registry size
// it does not depend on who reaps what.
int answered() const { return m_answered.load(); }
// True if a parked call gave up waiting instead of being released. A test
// that forgets to open the gate must FAIL on this, not hang; see await().
bool gateExpired() const { return m_gateExpired.load(); }
QJsonArray getMethods() override { return QJsonArray{}; }
bool informModuleToken(const QString&, const QString&) override { return true; }
void setEventListener(EventCallback) override {}
@@ -181,9 +219,58 @@ public:
QString providerVersion() const override { return QStringLiteral("1.0.0"); }
private:
// Park until the test lets go / until this call's turn comes. Returns false
// if it gave up first.
//
// BOUNDED, and SELF-RELEASING on the way out, because both halves are what
// keep a mistake here from becoming a hung binary. Provider dispatch is
// single-threaded (PlainTransportHost posts each call to the proxy's thread
// with a QueuedConnection), so a call parked here holds up every call behind
// it: an unbounded wait would wedge the whole host if the test never opened
// the gate, and a bounded wait that did NOT open it on the way out would
// wedge it just as thoroughly, one budget at a time, 800 times over. Opening
// it here means one budget is the WHOLE cost — after which every held call
// answers, every callback fires, and the run ends in a named assertion
// failure (gateExpired) instead of a timeout somewhere in CI. Measured: with
// the release deleted, this test FAILS in 20.2s naming the gate, where the
// unbounded version would have hung.
bool await()
{
std::unique_lock<std::mutex> lk(m_mu);
if (m_cv.wait_for(lk, kGateBudget, [this] { return m_released; }))
return true;
forceOpen();
return false;
}
bool awaitTurn()
{
std::unique_lock<std::mutex> lk(m_mu);
const int mine = m_arrived++;
if (m_cv.wait_for(lk, kGateBudget, [this, mine] { return mine < m_allowed; }))
return true;
forceOpen();
return false;
}
void forceOpen() // called with m_mu held
{
m_released = true;
m_allowed = kAll;
}
// ~40x the slowest complete run of the burst below measured at any load, so
// it can only be reached by a gate that is never opened at all.
static constexpr std::chrono::seconds kGateBudget{20};
static constexpr int kAll = 1 << 30;
std::mutex m_mu;
std::condition_variable m_cv;
bool m_released = false;
int m_arrived = 0; // gated calls seen, under m_mu
int m_allowed = 0; // gated calls permitted, under m_mu
std::atomic<int> m_answered{0};
std::atomic<bool> m_gateExpired{false};
};
QCoreApplication* ensureApp()
@@ -234,6 +321,14 @@ public:
bool ok() const { return m_started && m_published && m_port != 0; }
uint16_t port() const { return m_port; }
// Let every call parked in the provider's gate — and every one that arrives
// after this — answer.
void openGate() { m_provider.letGo(); }
// Let the first `n` gated calls answer, and no more.
void allow(int n) { m_provider.allow(n); }
int answered() const { return m_provider.answered(); }
bool gateExpired() const { return m_provider.gateExpired(); }
private:
EchoProvider m_provider;
std::unique_ptr<PlainTransportHost> m_host;
@@ -489,16 +584,81 @@ TEST_F(PlainWaiterReapingTest, ConcurrentCompletedCallsStayBoundedByInFlight)
// then the handle goes quiet. If reaping only ever happened on the spawn path,
// everything that finished after the LAST spawn would stay parked for the life
// of the handle — measured at 1428 waiters and +24MiB after 2000 completed
// calls, collapsing to 1 the moment one further call was issued. That figure is
// race-dependent, not a constant: a re-measure of the same build gave 1421 (and
// 599 rather than 610 for the 800-call burst below). Same magnitude, different
// number every time — which is the point of asserting a bound and not a value.
// calls, collapsing to 1 the moment one further call was issued. Those figures
// came off a burst that was merely issued in a loop, so they are race-dependent
// and not constants: a re-measure of the same build gave 1421. The burst below
// is gated instead, and that is what turns the defect's side from "some large
// number" into kBurst exactly — see WHY THE PROVIDER IS GATED.
//
// So the bound is read here with NO further call: the burst has to have drained
// itself. What remains is what published after the last reap — at minimum the
// last waiter to finish, which has nobody behind it to collect it (1 in almost
// every run, 2 when a waiter's publish slips past the final reap). The bound
// below is generous against that and still ~100x under the pre-fix number.
// itself. RETARGETED BY THE FOLD, and this is where the two mechanisms differ
// most: there is no last reap and no last exit batch, because an entry's
// lifetime IS its call's. What remains after a burst that all completed is
// therefore ZERO, not "one or two", and the interesting quantity moves to the
// OTHER end of the test — the peak, which the gate below makes a measured 800
// concurrent async calls carrying no threads at all.
//
// WHY THE PROVIDER IS GATED, which is the whole design of this test. "Issue 800
// calls in a loop and hope they overlap" is not an experiment, it is a race
// between two rates: how fast the caller can spawn a std::thread, and how fast a
// loopback RPC can come back. On macOS the first is much cheaper than the second
// and the burst really is concurrent. On Linux they are comparable, so most of
// the burst COMPLETES WHILE IT IS STILL BEING ISSUED and gets collected by the
// SPAWN-path reaper — the very reaper this test is supposed to be doing without.
// Measured on the ungated version, calls still outstanding when the last one of
// the burst went out (n=25 per platform):
//
// in flight at the end of the issue loop, of 800
// macOS 544-612 (68-77% of the burst)
// Linux 0-237 (0-30%, median 85)
//
// On Linux, then, the thing being measured was ~85 concurrent calls with a spawn
// reaper running throughout — one run in 25 had the ENTIRE burst answered before
// the last call was issued — and not an 800-call burst going quiet. The residue
// it left was a draw from the scheduler rather than a property of the code, both
// arms drew from overlapping distributions, and NO bound could separate them:
// correct code reached 152 unloaded and 713 under load, the defect fell as low
// as 5. The bound before this commit (400) scored 25/25 on macOS and 0/40 on an
// idle Linux box; the bound before THAT (8) failed correct code 96 times in 160.
// Neither number was the problem.
//
// The gate removes the race instead of arbitrating it. Every call in the burst
// invokes `gate`, which parks in the provider; provider dispatch is
// single-threaded, so the first one to arrive holds up all 800 and NOT ONE REPLY
// EXISTS until the test opens the gate — asserted below off the provider's own
// counter, not inferred. The burst is then concurrent by construction on every
// platform: 800/800 in flight, measured, on both, at every load level tried.
//
// THE PACED DRAIN BELOW IS INHERITED AND, ON THIS BRANCH, NOT LOAD-BEARING. It
// is carried over from the base because releasing all 800 at once mattered
// THERE: waiters reap and only then publish, so a reaper that collected a large
// batch sat in its join loop while everyone behind it published, and correct
// code left 2-389 on a 6-core Linux box against the defect's 800 — a 2x
// separation, useless as a detector. Here there is nothing to reap and nothing
// to serialize, and the residue is 0 at any pace: measured with kRelease set to
// kBurst — one release, all 800 answered together — it is 0 on all 30 runs per
// platform, the same as at 16. It stays because the structure is worth keeping
// identical to the base's while both branches are live, and because a paced
// drain checks the entries leaving progressively rather than all at the end.
//
// MEASURED, this gated burst, CallState::inflight when it goes idle. Numbers, n
// and margins are at the assertion below; the shape is:
//
// this code a delivery that does not erase its entry
// macOS 0 800 (exactly, every run)
// Linux, any load 0 800 (exactly, every run)
//
// The defect arm here is NOT the base's — the exit-guard reap it removed does
// not exist any more — but the shape is the same one and it is the closest
// mechanism-appropriate inversion: drop `st->inflight.erase(id)` from
// AsyncCall::deliver(), so a completed call keeps its registration. That gives
// kBurst exactly, on both platforms, and 801 after the follow-up call at the end
// of this test. The separation is not statistical on either side.
//
// AND THE PEAK IS THE CLAIM NOW. 800 async calls outstanding at once is what
// this branch exists to make cheap: on the base, that reading is 800 live
// std::threads; here it is 800 shared_ptr<AsyncCall> on the shared io_context
// and no thread per pending RPC at all.
TEST_F(PlainWaiterReapingTest, BurstThatGoesIdleDrainsWithoutAnotherCall)
{
LiveHost host;
@@ -513,18 +673,94 @@ TEST_F(PlainWaiterReapingTest, BurstThatGoesIdleDrainsWithoutAnotherCall)
auto* plain = dynamic_cast<PlainLogosObject*>(obj);
ASSERT_NE(plain, nullptr);
// Issued in one go, with no pumping in between, so they really are
// concurrent and the tail of the burst is large.
// ~1s in practice. Every wait in this test is already bounded — the gate
// self-releases, the pumps have budgets, the calls have timeouts — so this
// is the backstop for a wedge INSIDE the code under test, which none of
// those would catch. Cheap, and kept across the fold for that reason.
Watchdog watchdog("BurstThatGoesIdleDrainsWithoutAnotherCall", 120000);
// Issued in one go, with no pumping in between, against a CLOSED gate: no
// reply can be produced until every one of them is outstanding.
constexpr int kBurst = 800;
// Gated calls released per step of the drain, each step awaited before the
// next. Inherited from the base and not load-bearing here — see the header
// comment; the residue is 0 at any step, including one.
constexpr int kRelease = 16;
// A small CONSTANT, because on this branch the quantity it bounds is a
// registry that a completed call has already left. Sized at the assertion.
constexpr size_t kDrained = 8;
Deliveries d(kBurst);
// Declared AFTER `d`, so it runs BEFORE it. The tail of this test used to be
// a bare `obj->release(); pump(50);` on the happy path, which a FATAL
// assertion skips — and skipping it is not merely untidy here. Every
// outstanding callback holds a reference to `d`, and LiveHost's destructor
// pumps the event loop, so bailing out with deliveries still in flight would
// run them against a destroyed Deliveries. That is the one way this test
// could answer a failure with a crash instead of a verdict, and the
// assertions below (a gate that never opened, a burst that did not all
// complete) are exactly the ones that would trigger it. release() resolves
// every outstanding call, and the pump behind it runs what they posted —
// both while `d` is still alive.
struct ReleaseOnExit {
LogosObject* obj;
~ReleaseOnExit() { obj->release(); pump(50); }
} releaseOnExit{obj};
for (int i = 0; i < kBurst; ++i) {
ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
QVariantList{ QVariant(i) }, 9000,
ch->callMethodAsyncWithError(kToken, QStringLiteral("gate"),
QVariantList{ QVariant(i) }, 45000,
[&d, i](QVariant, const logos::CallError& e) {
d.record(i, e);
});
}
// THE EXPERIMENT'S OWN PRECONDITION, measured rather than assumed — because
// an unmeasured one is exactly how this test spent three revisions asserting
// a bound on a burst that was not a burst. Two independent readings:
//
// * the provider has produced NOTHING. Read from the provider itself, so
// it does not depend on the client, on who reaps what, or on the Qt
// event loop having been pumped (it has not been — a delivery count
// would read 0 here whether or not replies existed).
// * all 800 calls are registered in CallState::inflight. A call that
// completed during the loop would have left it, so this number falling
// short is the ungated behaviour coming back. It is also the fold's own
// headline, measured: 800 concurrent async calls, zero threads.
const int answeredAtIssueEnd = host.answered();
const size_t inflight = inflightCount(plain);
std::cout << " burst issued: provider replies=" << answeredAtIssueEnd
<< " calls in flight=" << inflight << "/" << kBurst << std::endl;
ASSERT_EQ(answeredAtIssueEnd, 0)
<< "the gate leaked: replies were produced while the burst was still "
"being issued, so what this test measures below is not a burst drain";
EXPECT_EQ(inflight, size_t(kBurst))
<< "only " << inflight << " of " << kBurst << " calls were in flight when "
"the burst finished issuing";
// Drain it, kRelease at a time and NEVER issuing another call — which is the
// claim. Every entry that leaves CallState::inflight from here leaves
// because its own call was delivered; nothing else in the object touches
// that map while the handle is alive.
for (int done = 0; done < kBurst; done += kRelease) {
const int upto = std::min(done + kRelease, kBurst);
host.allow(upto);
pumpUntilTotal(d, upto, 30000);
// A step that does not complete is a failure, and carrying on would turn
// one stuck call into fifty budgets back to back. Break; the assertion
// below reports it.
if (d.total.load() < upto) break;
}
// Anything the loop left behind (it broke early, or the provider saw calls
// the client never counted) answers now, so nothing is parked in the host
// while the assertions run.
host.openGate();
pumpUntilTotal(d, kBurst, 60000);
// A gate that was never opened, a call that errored, a callback that went
// missing: all of them arrive here as a FAILED assertion on a bounded run,
// never as a hang. See EchoProvider::await() for why that is true of the
// gate in particular.
ASSERT_FALSE(host.gateExpired())
<< "a gated call gave up waiting: the gate was never opened";
ASSERT_EQ(d.total.load(), kBurst) << "the burst did not all complete";
// Every callback has landed; now let the waiters that delivered them finish
@@ -542,9 +778,29 @@ TEST_F(PlainWaiterReapingTest, BurstThatGoesIdleDrainsWithoutAnotherCall)
EXPECT_EQ(d.worst(), 1);
EXPECT_EQ(d.missing(), 0);
EXPECT_EQ(d.errors.load(), 0);
EXPECT_LE(idle, 8u)
// kDrained is 8 and it is a CONSTANT, which the ungated version of this test
// could not afford (its residue was a draw from the scheduler; see the
// header). Here the measured value is not "small", it is ZERO — a delivered
// call has already left the registry — so 8 is not headroom over a
// distribution, it is slack for a shape this branch does not currently have:
// an entry whose erase is deferred to a later turn of the loop.
//
// MEASURED, this code, 380 runs, worst value per cell:
//
// macOS idle 0 (n=100) 4x 0 (n=40) 16x 0 (n=40)
// Linux idle 0 (n=40) 4x 0 (n=60) 16x 0 (n=60) 64x 0 (n=40)
//
// The same seven cells with `st->inflight.erase(id)` removed from
// AsyncCall::deliver(), 380 more runs, give 800 in every single one, on both
// platforms, at every load level. All 380 of those runs FAILED this
// assertion and all 380 unmodified runs passed it.
//
// MARGINS: the correct-code side never leaves the floor, so the headroom is
// 8 over 0, and 100x below the defect's 800. There is no overlap to report
// and neither side is a distribution.
EXPECT_LE(idle, kDrained)
<< "a burst that went idle left " << idle << " of " << kBurst
<< " waiters parked: they are only being reaped on the spawn path";
<< " calls registered: completed calls are not leaving the registry";
// And the handle still works afterwards — draining from inside the waiters
// must not have disturbed the object they are draining.
@@ -557,10 +813,13 @@ TEST_F(PlainWaiterReapingTest, BurstThatGoesIdleDrainsWithoutAnotherCall)
pumpUntilTotal(after, 1, 10000);
EXPECT_EQ(after.total.load(), 1);
EXPECT_EQ(after.errors.load(), 0);
EXPECT_LE(inflightCount(plain), 8u);
// Same claim, same bound, read once more after a further call has been and
// gone: 0 in practice, and 801 with the erase removed — the retention grows
// by exactly the one call, which is the shape of the bug this pins.
EXPECT_LE(inflightCount(plain), kDrained);
obj->release();
pump(50);
// release() + pump: see ReleaseOnExit above. It runs on every exit path from
// here, not just this one.
}
// ── 2. the deadlock the fix could introduce ─────────────────────────────────