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

fix(protocol): a call that FINISHED must not park its thread for the object's life

Browse Source 
The per-call waiters are joinable rather than detached, which is what closed the
use-after-free where release() deleted the object under a still-running waiter
(4f9d824), and they are interruptible, so teardown no longer waits out the call's
timeout (731e579). Both stay. What neither did was retire a waiter that had
FINISHED: m_waiters was only ever swap()ped, in stopAndJoinWaiters(), so an
exited-but-unjoined std::thread — whose stack and pthread struct are not
reclaimed until somebody joins it — stayed parked for the lifetime of the handle.

Measured against a live PlainTransportHost over TCP, every call completing
normally, one handle held throughout, before:

    10000 calls   m_waiters   300 -> 10300   rss +156.56 MiB   16417 B/call
    30000 calls   m_waiters   300 -> 30300   rss +469.28 MiB   16403 B/call

and the same through the production C ABI — one lp_client, N lp_invoke_async —
at +156.53 MiB. That path is why this matters: LogosAPIConsumer caches ONE
handle per module and reuses it for every async call, releasing it only on
eviction or teardown (cpp/logos_api_consumer.cpp:129 and :207), so a
long-lived module leaks per lp_invoke_async. The ~16KB constant is one page on
this 16KiB-page arm64 and will be smaller elsewhere; the UNBOUNDEDNESS is the
platform-independent part, and follows from m_waiters.size() rising 1:1 with
completed calls and only ever falling in teardown. Attribution: the retention
arrived with the join in 4f9d824, not with 731e579 — but 731e579 is what makes
the join permanent.

The registry is now KEYED, because a thread cannot join itself and so a waiter
can never retire its own entry. Each waiter publishes its id as its FINAL act (a
scope guard declared first, so it destructs last, covering all four exit paths),
and the next spawn — plus teardown — joins those ids and erases them. Joining a
thread that has already returned is a couple of syscalls. Same probe, same
workload, after:

    10000 calls   m_waiters    15 -> 16      rss +0.08 MiB         8 B/call
    30000 calls   m_waiters    16 -> 16      rss +0.06 MiB         2 B/call
    10000 calls via lp_invoke_async          rss +0.09 MiB        10 B/call

Retention is now bounded by the waiters that finish after the LAST spawn, i.e.
by peak in-flight concurrency — 16 at the in-flight window above, and exactly 1
when calls are issued sequentially — instead of by call count.

THE DEADLOCK THIS SHAPE INVITES is a reaper that joins while holding m_waiterMu,
against a waiter blocked on m_waiterMu trying to publish. It is avoided by
construction rather than by argument: nothing is joined with a lock held, in the
reaper or in teardown, whatever a waiter does on its way out. Proven by building
the naive variant that does join under the lock — the new hammer wedges it, with
the main thread in reapFinishedWaiters -> pthread_join and a waiter in
publishFinishedWaiter -> mutex wait, and the test's watchdog names the cause
instead of letting CI hang.

Teardown's guarantee is restated rather than weakened. It is not "every waiter
has been joined by the time stopAndJoinWaiters() returns" — a waiter a
concurrent reaper is mid-join on is no longer in the map — but the thing that
guarantee was ever for: NO WAITER TOUCHES THE OBJECT AFTER IT RETURNS. An entry
leaves m_waiters only once its thread has published, and publishing is that
thread's last access.

The TODO above the waiter still stands: the real fix is to fold the wait into
the shared Asio io_context and have no thread per pending RPC at all. This makes
the interim honest; it does not replace that.

Two more things review turned up, folded in here:

  * The two wait sites resolved stop-vs-result in OPPOSITE directions.
    waitForResult tested the stop flag BEFORE polling, so an already-ready
    future was still reported as transport_error, while awaitCompletion
    deliberately preferred a completion that had landed — and both were
    commented as intentional. One rule now, applied to both: AN ANSWER ALREADY
    IN HAND BEATS A CONCURRENT STOP, and the stop only decides what happens when
    there is nothing to hand over. The callback fires either way
    (postToQtEventLoop copies everything it delivers), so the only thing a stop
    can change is what the callback SAYS — and manufacturing transport_error
    while the true answer sits in the future reports a failure that did not
    happen, to callers that re-acquire, retry and log on that code. Preferring
    the answer costs nothing, since it is already there: the flag is still
    checked before every sleep, so the teardown-latency bound is unchanged.

  * CORRECTION to 731e579's message, which claimed it "closes the registration
    window" where a call arriving after the stop would never be joined. That
    branch is unreachable in defined behaviour: m_stopping is raised only by
    teardown, so any thread that can read it inside callMethodAsyncWithError is
    already calling a method on an object whose destructor is running — the load
    is itself the use-after-free, reproduced as a SIGSEGV on that commit and on
    its parent alike, and nothing inside that function can repair it. The guard
    is harmless and stays (one predictable branch, and it fails safe with one
    callback), but its comment now says what it is instead of claiming a fix it
    does not make.

Verified by running, with every check first shown to FAIL on unfixed code:

  * Retention: the probe above, plus a committed regression test that reads
    m_waiters out of the live object through the explicit-instantiation access
    hole ([temp.spec] does not check access on an explicit instantiation's
    template arguments) — so the code under test keeps its private state, with
    no friend, no test-only accessor and no `#define private public`. 200
    sequential completed calls keep 1 waiter; without pruning they keep 200.
  * Exactly-once on all four paths — normal completion, timeout, cancellation
    and the deferred-completion (pending-sentinel) arm — counted PER CALL so a
    dropped one and a doubled one cannot cancel out, plus the 60-round
    release-during-call race. Shown to catch a cancelled path that returns
    silently (3 failures) rather than delivering.
  * Teardown latency unchanged from 731e579: 10-17ms with an in-flight 8000ms
    call and 0-1ms mid-defer, against 15ms / 1ms on that commit.
  * The UAF stays closed: 11 teardown + reaping tests clean under macOS Guard
    Malloc (ASan/TSan remain unusable on this toolchain).
  * Full suite 281/281 twice, `nix build .#tests` green (281/281 in the
    sandbox), CallErrorAfterAcquireTest hammered 40x clean.
This commit is contained in:
Dario Gabriel Lipicar
2026-08-05 20:24:11 -03:00
parent 731e579064
commit 378d889a1d
4 changed files with 778 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/implementations/plain/plain_logos_object.cpp
+151 -35
View File
@@ -44,6 +44,25 @@ constexpr std::chrono::milliseconds kWaitSlice(25);
enum class WaitOutcome { Ready, TimedOut, Cancelled };
// ── the one rule both waits follow ──────────────────────────────────────────
//
// AN ANSWER ALREADY IN HAND BEATS A CONCURRENT STOP. The stop only decides what
// happens when there is nothing to hand over.
//
// The two sites used to resolve this in opposite directions — this one tested
// the flag before polling, so an already-ready future was still reported as
// transport_error, while awaitCompletion deliberately preferred a completion
// that had landed. Both were commented as deliberate, and they cannot both be
// right, so: the callback fires either way (postToQtEventLoop copies everything
// it delivers, precisely so a released handle costs it nothing), which means the
// only thing a stop can change is what the callback SAYS. Reporting
// transport_error while the true answer sits in the future is a failure that did
// not happen, and that code is not inert — callers re-acquire, retry and log on
// it. Preferring the answer is also free: it is already there, so nothing waits
// for it. The teardown-latency bound is untouched, because the flag is still
// checked before every sleep, and a stop with no answer in hand still wins
// immediately.
//
// The interruptible form of `fut.wait_for(milliseconds(timeoutMs))`.
//
// The overall deadline is computed once, so slicing does not stretch the
@@ -54,27 +73,23 @@ WaitOutcome waitForResult(std::future<ResultMessage>& fut, int timeoutMs,
using clock = std::chrono::steady_clock;
const auto deadline = clock::now() + std::chrono::milliseconds(timeoutMs);
for (;;) {
// Checked BEFORE sleeping, so a stop that already happened costs
// Poll FIRST, with a zero wait: an answer in hand beats both a
// concurrent stop and the deadline. This is also what makes a
// non-positive timeout behave as the single unsliced wait_for did —
// one poll, then give up — and what reports a future that went ready
// during the last slice.
if (fut.wait_for(clock::duration::zero()) == std::future_status::ready)
return WaitOutcome::Ready;
// Checked before sleeping, so a stop that already happened costs
// nothing, and after every slice, so one that arrives mid-wait costs at
// most kWaitSlice. A pending stop beats a result that landed in the
// same tick on purpose: the caller has released the handle and is no
// longer interested in the answer.
// most kWaitSlice.
if (stopping.load(std::memory_order_acquire))
return WaitOutcome::Cancelled;
const auto remaining = deadline - clock::now();
const bool expired = remaining <= clock::duration::zero();
// At (or past) the deadline, poll once with a zero wait rather than
// giving up blind — that is what the single unsliced wait_for did for a
// non-positive timeout, and a future that is already ready must still
// be reported as ready.
const auto slice = expired
? clock::duration::zero()
: std::min<clock::duration>(kWaitSlice, remaining);
if (fut.wait_for(slice) == std::future_status::ready)
return WaitOutcome::Ready;
if (expired)
if (remaining <= clock::duration::zero())
return WaitOutcome::TimedOut;
fut.wait_for(std::min<clock::duration>(kWaitSlice, remaining));
}
}
@@ -129,22 +144,87 @@ void PlainLogosObject::stopWaiters()
m_completionCv.notify_all();
}
void PlainLogosObject::publishFinishedWaiter(std::uint64_t id)
{
std::lock_guard<std::mutex> g(m_waiterMu);
m_finishedWaiters.push_back(id);
}
void PlainLogosObject::reapFinishedWaiters()
{
// Only ids a waiter published are taken, and publishing is that waiter's
// last act — so everything moved into `done` has already stopped touching
// this object, and joining it is effectively instant.
std::vector<std::thread> done;
{
std::lock_guard<std::mutex> g(m_waiterMu);
std::vector<std::uint64_t> keep;
for (const std::uint64_t id : m_finishedWaiters) {
const auto it = m_waiters.find(id);
if (it == m_waiters.end())
continue; // teardown already took this one
if (it->second.get_id() == std::this_thread::get_id()) {
// Unreachable today — callbacks are delivered on the Qt event
// loop, so a waiter thread never re-enters this class — but a
// thread that joined itself would terminate the process, and
// this is one comparison. Leave it registered; teardown, which
// runs on somebody else's thread, will collect it.
keep.push_back(id);
continue;
}
done.push_back(std::move(it->second));
m_waiters.erase(it);
}
m_finishedWaiters.swap(keep);
}
// Joined with NO lock held. Not just hygiene — this is THE deadlock this
// whole mechanism can introduce: a reaper holding m_waiterMu while it joins
// a waiter that is itself blocked on m_waiterMu trying to publish would
// wedge the process. Holding no lock across a join makes that impossible by
// construction rather than by argument, whatever a waiter does on its way
// out. stopAndJoinWaiters() keeps the same discipline for the same reason.
for (auto& t : done) {
if (t.joinable())
t.join();
}
}
void PlainLogosObject::stopAndJoinWaiters()
{
stopWaiters();
std::vector<std::thread> waiters;
std::map<std::uint64_t, std::thread> waiters;
{
std::lock_guard<std::mutex> g(m_waiterMu);
waiters.swap(m_waiters);
}
// Joined with NO lock held: a waiter on its way out still takes
// m_completionMu (awaitCompletion) and m_waiterMu is what a concurrent
// callMethodAsyncWithError needs to see the stop flag.
for (auto& t : waiters) {
// m_completionMu (awaitCompletion) and then m_waiterMu (to publish), and
// m_waiterMu is also what a concurrent callMethodAsyncWithError needs in
// order to see the stop flag.
//
// Everything outstanding is joined by id-independent brute force, so this
// needs no cooperation from the reaper: a waiter that publishes while this
// loop runs simply leaves a stale id behind, and its thread is joined here
// anyway.
//
// A waiter that a concurrent reaper is in the middle of joining is NOT in
// this map, and that is still safe. The invariant is not "every waiter has
// been joined by the time this returns" but the thing that invariant was
// ever for: NO WAITER TOUCHES THIS OBJECT AFTER THIS RETURNS. An entry
// leaves m_waiters only once its thread has published, and publishing is
// that thread's last access — all it has left to do is unwind.
for (auto& entry : waiters) {
std::thread& t = entry.second;
if (t.joinable())
t.join();
}
// Cleared after the joins, so the stale ids just described go too. Nothing
// can be added afterwards: m_stopping is set, so no new waiter registers.
{
std::lock_guard<std::mutex> g(m_waiterMu);
m_finishedWaiters.clear();
}
}
QVariant PlainLogosObject::callMethod(const QString& authToken,
@@ -252,8 +332,9 @@ QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs,
|| m_stopping.load(std::memory_order_relaxed);
});
// A completion that actually landed beats a concurrent stop: there is a real
// answer in hand, so hand it over rather than manufacture an error.
// A completion that actually landed beats a concurrent stop — the same rule
// the future wait follows (see waitForResult): there is a real answer in
// hand, so hand it over rather than manufacture an error.
const auto it = m_completions.find(callId);
if (it != m_completions.end()) {
const QVariant result = it->second;
@@ -359,33 +440,66 @@ void PlainLogosObject::callMethodAsyncWithError(const QString& authToken,
// io_context (the connection already runs on it) so we don't spin
// up a thread per pending RPC.
//
// The thread is JOINed in stopAndJoinWaiters() (destructor / release), not
// detached: capturing `this` for awaitCompletion / m_stopping is
// only safe while the object is alive, and release() used to
// The thread is JOINed — in reapFinishedWaiters() once it has finished, or
// in stopAndJoinWaiters() (destructor / release) if teardown gets there
// first — never detached: capturing `this` for awaitCompletion /
// m_stopping is only safe while the object is alive, and release() used to
// `delete this` while a waiter could still be mid-flight.
const std::string objectName = m_objectName;
const std::string method = methodName.toStdString();
// Retire the previous calls' waiters before adding one. Done HERE rather
// than by the waiters themselves because a thread cannot join itself; done
// BEFORE taking m_waiterMu because it joins, and joining under that lock is
// the deadlock described in reapFinishedWaiters().
reapFinishedWaiters();
// Register under the lock BEFORE the thread can outrun release(): a
// detach-then-push left a window where delete this raced the waiter.
{
std::lock_guard<std::mutex> g(m_waiterMu);
if (m_stopping.load(std::memory_order_acquire)) {
// Teardown has already swapped m_waiters out, so a thread pushed
// now would never be joined — exactly the dangling waiter this
// whole mechanism exists to prevent. Answer as a cancelled call
// instead, which keeps the exactly-once contract either way.
// Refuse rather than register: teardown has already swapped
// m_waiters out, so a thread pushed now would never be joined.
//
// To be honest about what this branch is: it is NOT a reachable
// window that got closed. m_stopping is raised only by teardown
// (release() / the destructor), so a thread that can read it as
// true here is already calling a method on an object whose
// destructor is running — this very load is the use-after-free, and
// nothing inside this function can repair that. Reproduced as a
// SIGSEGV, on this branch and on its parent alike. It is kept
// because it costs one predictable branch on a path that already
// does a socket write, and because failing this way — one callback,
// with the same error a cancelled call gets — is strictly better
// than pushing a thread nobody will ever join, should some future
// caller of stopWaiters() make the state legitimately observable.
postToQtEventLoop(std::move(callback), QVariant(),
callErrorReleased(objectName, method));
return;
}
m_waiters.emplace_back([this, objectName, fut, timeoutMs, methodName, method,
callback = std::move(callback)]() mutable {
const std::uint64_t waiterId = m_nextWaiterId++;
std::thread waiter([this, waiterId, objectName, fut, timeoutMs, methodName, method,
callback = std::move(callback)]() mutable {
// Everything reached through `this` below (m_stopping,
// awaitCompletion's m_completionMu / m_completions) is safe only
// because stopAndJoinWaiters() joins this thread before the object
// dies. Everything handed to postToQtEventLoop is a COPY, because
// that delivery happens after this thread has returned — i.e.
// possibly after the object is gone. Keep it that way.
// because this thread is joined before the object dies — by the
// reaper if it finishes first, by stopAndJoinWaiters() otherwise.
// Everything handed to postToQtEventLoop is a COPY, because that
// delivery happens after this thread has returned — i.e. possibly
// after the object is gone. Keep it that way.
//
// Declared FIRST so it destructs LAST: publishing this waiter's id
// is what permits somebody else to join and drop it, so it must
// come after every access to the object, on every exit path
// (four returns below, plus anything that throws). What runs after
// it is the unwinding of the captures above, none of which belongs
// to the object: a string, a shared_ptr to the call's future, and a
// callback that has already been moved out.
struct PublishOnExit {
PlainLogosObject* self;
std::uint64_t id;
~PublishOnExit() { self->publishFinishedWaiter(id); }
} publishOnExit{this, waiterId};
const WaitOutcome outcome = waitForResult(*fut, timeoutMs, m_stopping);
if (outcome == WaitOutcome::Cancelled) {
// A cancelled call still DELIVERS, exactly once. Returning
@@ -425,6 +539,7 @@ void PlainLogosObject::callMethodAsyncWithError(const QString& authToken,
}
postToQtEventLoop(std::move(callback), std::move(value), std::move(err));
});
m_waiters.emplace(waiterId, std::move(waiter));
}
}
@@ -518,7 +633,8 @@ void PlainLogosObject::release()
// object under a still-running waiter — and merely joining them made
// release() block for the rest of the call's timeout, so they are asked to
// stop first. Each abandoned call still delivers its callback, once, with
// callErrorReleased.
// callErrorReleased. Waiters that already finished were reaped as the
// calls after them were issued; this collects whatever is left.
disconnectEvents();
stopAndJoinWaiters();
m_conn.reset();
cpp/implementations/plain/plain_logos_object.h
+31 -1
View File
@@ -7,6 +7,7 @@
#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <map>
#include <memory>
#include <mutex>
@@ -102,6 +103,17 @@ private:
// because the flag has to be published under m_completionMu (see the .cpp).
void stopWaiters();
// Join and drop the waiters that have already FINISHED, so a handle that
// outlives its calls does not accumulate them. Called on every async spawn
// (a call pays for the corpse of an earlier one) and from
// stopAndJoinWaiters(). Cheap: a join on an already-returned thread is a
// couple of syscalls, and only ids a waiter itself published are touched.
void reapFinishedWaiters();
// A waiter's FINAL act — see the scope guard in callMethodAsyncWithError.
// After this returns, that thread never touches the object again, which is
// what makes it safe for someone else to join and drop it.
void publishFinishedWaiter(std::uint64_t id);
std::string m_objectName;
std::shared_ptr<RpcConnectionBase> m_conn;
std::mutex m_mu;
@@ -112,8 +124,26 @@ private:
std::map<QString, QVariant> m_completions;
bool m_completionSubscribed = false;
// The waiter registry. KEYED, not a plain vector, because a thread cannot
// join itself: a waiter can therefore never retire its own entry, and a
// vector left only one moment to clear it — teardown — so every completed
// call parked a finished-but-unjoined thread (~one page of resident memory
// each) for the whole life of the handle. The production shape is one
// cached handle per module reused for every call (logos_api_consumer.cpp),
// so that grew without bound. Now a waiter publishes its id into
// m_finishedWaiters as its last act and the next spawn (or teardown) joins
// and erases it: see reapFinishedWaiters().
//
// Retention is bounded by the waiters that finish after the LAST spawn,
// i.e. by peak in-flight concurrency, not by call count.
//
// The real fix is still the TODO in callMethodAsyncWithError — fold the
// wait into the shared Asio io_context and have no thread per pending RPC
// at all. This makes the interim honest, it does not replace that.
std::mutex m_waiterMu;
std::vector<std::thread> m_waiters;
std::map<std::uint64_t, std::thread> m_waiters;
std::vector<std::uint64_t> m_finishedWaiters;
std::uint64_t m_nextWaiterId = 0;
// Read lock-free by the sliced future wait and under m_completionMu by
// awaitCompletion's predicate; written under m_completionMu so the
// condition-variable side cannot miss it. Never cleared — an object that
tests/protocol/CMakeLists.txt
+9
View File
@@ -59,6 +59,15 @@ add_executable(protocol_tests
# returns turns a stall into a hang), and the join that keeps `this` alive
# under the waiter is still there.
test_plain_object_teardown.cpp
# The other half of the same mechanism: what a COMPLETED call leaves behind.
# A waiter cannot join itself, so while the registry was a plain vector the
# only thing that ever emptied it was teardown, and every finished call
# parked an unjoined thread (~one page of resident memory) for the life of
# the handle — unbounded under the cached-handle shape LogosAPIConsumer
# actually uses. Pins the registry not growing with call count, the
# reap-vs-publish race not deadlocking, and teardown still joining what is
# left.
test_plain_waiter_reaping.cpp
# Component tests that moved here with their code (from logos-cpp-sdk)
test_token_manager.cpp
test_mock_store.cpp
tests/protocol/test_plain_waiter_reaping.cpp
+587
View File
@@ -0,0 +1,587 @@
// A long-lived PlainLogosObject must not accumulate the waiters of calls that
// have already finished.
//
// Its sibling suite (test_plain_object_teardown.cpp) pins what happens to a
// waiter that is still IN FLIGHT when the handle goes away. This one pins the
// other half: what is left behind by a call that COMPLETED NORMALLY.
//
// The defect these tests exist to prevent. Waiter threads are joined rather
// than detached, because they capture `this` and release() deletes it. But a
// thread cannot join itself, so a waiter cannot retire its own entry, and while
// the registry was a plain vector the only code that ever emptied it was
// teardown. Every completed async call therefore parked a finished-but-unjoined
// std::thread for the whole life of the handle — an exited thread whose stack
// and pthread struct are not reclaimed until somebody joins it, measured at
// ~16KB resident per call on 16KiB-page arm64 (one page; expect less on 4KiB
// Linux, but the growth is the platform-independent part). The production shape
// makes that unbounded rather than academic: LogosAPIConsumer caches ONE handle
// per module and reuses it for every async call, releasing it only on eviction
// or teardown (cpp/logos_api_consumer.cpp), so 30k calls on one handle cost
// ~470MB that never comes back.
//
// What is pinned here:
//
// 1. THE REGISTRY DOES NOT GROW WITH CALL COUNT. Counting threads, not bytes:
// RSS is a noisy proxy and its per-call constant is platform-specific,
// whereas "m_waiters.size() rises 1:1 with completed calls and only ever
// falls in teardown" is the defect itself, exactly and portably. The bound
// is peak in-flight concurrency, because the reaper runs on the next
// spawn — so a sequential caller keeps ~1 and NOT ~N.
//
// 2. THE DEADLOCK THE FIX COULD INTRODUCE. Reaping means joining, and a
// reaper that joined while holding the lock a waiter needs in order to
// announce itself would wedge the process. Hammered here with reaps and
// publishes deliberately overlapped, under a watchdog so a regression is a
// named failure rather than a CI job that hangs until its timeout.
//
// 3. REAPING DOES NOT BREAK TEARDOWN. Completed calls being retired early
// must not lose the join for the one still outstanding, and the callback
// contract stays exactly-once across a run where both happen.
//
// m_waiters is private and stays private: the test reads it through the
// explicit-instantiation access hole ([temp.spec] does not check access on the
// template arguments of an explicit instantiation), so the code under test is
// observed exactly as it ships — no `friend`, no test-only accessor, no
// #define private public.
#include <gtest/gtest.h>
#include "logos_call_error.h"
#include "logos_object.h"
#include "logos_provider_interface.h"
#include "logos_transport_config.h"
#include "module_proxy.h"
#include "plain_transport_connection.h"
#include "plain_transport_host.h"
#include "plain_logos_object.h"
#include <QCoreApplication>
#include <QElapsedTimer>
#include <QJsonArray>
#include <QString>
#include <QThread>
#include <QVariant>
#include <QVariantList>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
using namespace logos::plain;
namespace {
// ── reading m_waiters without touching the production header ────────────────
template <typename Tag, typename Tag::type Member>
struct Rob {
friend typename Tag::type get(Tag) { return Member; }
};
struct WaitersTag {
using type = std::map<std::uint64_t, std::thread> PlainLogosObject::*;
friend type get(WaitersTag);
};
template struct Rob<WaitersTag, &PlainLogosObject::m_waiters>;
struct WaiterMuTag {
using type = std::mutex PlainLogosObject::*;
friend type get(WaiterMuTag);
};
template struct Rob<WaiterMuTag, &PlainLogosObject::m_waiterMu>;
// Taken under the object's OWN mutex — the one the registration path holds —
// so this is a consistent read, not a torn one.
size_t waiterCount(PlainLogosObject* obj)
{
auto& mu = obj->*get(WaiterMuTag{});
auto& m = obj->*get(WaitersTag{});
std::lock_guard<std::mutex> g(mu);
return m.size();
}
// 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.
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("block")) {
std::unique_lock<std::mutex> lk(m_mu);
m_cv.wait(lk, [this] { return m_released; });
return QVariant(42);
}
return QVariant();
}
void letGo()
{
{
std::lock_guard<std::mutex> g(m_mu);
m_released = true;
}
m_cv.notify_all();
}
QJsonArray getMethods() override { return QJsonArray{}; }
bool informModuleToken(const QString&, const QString&) override { return true; }
void setEventListener(EventCallback) override {}
void init(void*) override {}
QString providerName() const override { return QStringLiteral("echo"); }
QString providerVersion() const override { return QStringLiteral("1.0.0"); }
private:
std::mutex m_mu;
std::condition_variable m_cv;
bool m_released = false;
};
QCoreApplication* ensureApp()
{
static int argc = 0;
static char* argv[] = { nullptr };
if (!QCoreApplication::instance())
new QCoreApplication(argc, argv);
return QCoreApplication::instance();
}
class LiveHost {
public:
LiveHost()
{
LogosTransportConfig cfg;
cfg.protocol = LogosProtocol::Tcp;
cfg.host = "127.0.0.1";
cfg.port = 0; // ephemeral
m_host = std::make_unique<PlainTransportHost>(cfg);
m_started = m_host->start();
m_proxy = new ModuleProxy(&m_provider);
m_proxy->saveToken(QStringLiteral("origin"), QStringLiteral("live-token"));
m_thread = new QThread;
m_proxy->moveToThread(m_thread);
m_thread->start();
m_published = m_host->publishObject("echo_module", m_proxy);
const QString endpoint = m_host->endpoint();
m_port = endpoint.mid(endpoint.lastIndexOf(':') + 1).toUShort();
}
~LiveHost()
{
// Anything still parked in the provider would deadlock the thread quit
// below; let every blocked (and queued) call finish first.
m_provider.letGo();
QCoreApplication::processEvents(QEventLoop::AllEvents, 100);
m_host.reset();
QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
m_thread->quit();
m_thread->wait();
delete m_proxy;
delete m_thread;
}
bool ok() const { return m_started && m_published && m_port != 0; }
uint16_t port() const { return m_port; }
private:
EchoProvider m_provider;
std::unique_ptr<PlainTransportHost> m_host;
ModuleProxy* m_proxy = nullptr;
QThread* m_thread = nullptr;
bool m_started = false;
bool m_published = false;
uint16_t m_port = 0;
};
std::unique_ptr<PlainTransportConnection> connectTo(uint16_t port)
{
LogosTransportConfig cfg;
cfg.protocol = LogosProtocol::Tcp;
cfg.host = "127.0.0.1";
cfg.port = port;
auto conn = std::make_unique<PlainTransportConnection>(cfg);
if (!conn->connectToHost()) return nullptr;
return conn;
}
LogosObjectErrorChannel* channelFor(LogosObject* obj)
{
return dynamic_cast<LogosObjectErrorChannel*>(obj);
}
// Per-call delivery counts, so "exactly once" is checked per call and not just
// in aggregate — a double delivery on one call plus a dropped one on another
// would balance out in a total.
struct Deliveries {
explicit Deliveries(int n) : counts(n) {}
std::vector<std::atomic<int>> counts;
std::atomic<int> total{0};
std::atomic<int> errors{0};
std::mutex codeMu;
std::string lastCode;
void record(int i, const logos::CallError& e)
{
{
std::lock_guard<std::mutex> g(codeMu);
lastCode = e.code;
}
counts[i].fetch_add(1);
total.fetch_add(1);
if (!e.code.empty()) errors.fetch_add(1);
}
std::string code()
{
std::lock_guard<std::mutex> g(codeMu);
return lastCode;
}
int worst() const // the largest per-call count seen
{
int w = 0;
for (const auto& c : counts) w = std::max(w, c.load());
return w;
}
int missing() const
{
int m = 0;
for (const auto& c : counts) if (c.load() == 0) ++m;
return m;
}
};
void pumpUntilTotal(Deliveries& d, int target, int budgetMs)
{
QElapsedTimer t;
t.start();
while (d.total.load() < target && t.elapsed() < budgetMs)
QCoreApplication::processEvents(QEventLoop::AllEvents, 5);
}
void pump(int ms)
{
QElapsedTimer t;
t.start();
while (t.elapsed() < ms)
QCoreApplication::processEvents(QEventLoop::AllEvents, 5);
}
const char* kToken = "live-token";
// A deadlock does not fail a test, it hangs it — and a hung gtest binary is a
// CI job that dies on a timeout somewhere far from the cause. This turns that
// into a loud, attributable abort. The budget is ~50x the measured runtime of
// the hammer below, so it can only fire on a genuine wedge.
class Watchdog {
public:
Watchdog(const char* what, int budgetMs)
: m_what(what)
{
m_thread = std::thread([this, budgetMs] {
std::unique_lock<std::mutex> lk(m_mu);
if (!m_cv.wait_for(lk, std::chrono::milliseconds(budgetMs),
[this] { return m_done; })) {
std::fprintf(stderr,
"\nWATCHDOG: '%s' made no progress for %dms — the reaper is "
"deadlocked against a waiter trying to publish.\n",
m_what, budgetMs);
std::fflush(stderr);
std::abort();
}
});
}
~Watchdog()
{
{
std::lock_guard<std::mutex> g(m_mu);
m_done = true;
}
m_cv.notify_all();
m_thread.join();
}
private:
const char* m_what;
std::mutex m_mu;
std::condition_variable m_cv;
bool m_done = false;
std::thread m_thread;
};
} // namespace
class PlainWaiterReapingTest : public ::testing::Test {
protected:
void SetUp() override { ensureApp(); }
};
// ── 1. the registry does not grow with call count ───────────────────────────
//
// One handle, N completed calls, issued strictly sequentially: each callback is
// awaited before the next call goes out, which is both the realistic shape and
// the harshest one for the claim — with at most one call ever in flight, a
// correct implementation keeps ~1 waiter no matter how large N is.
//
// Pre-fix this ends at N.
TEST_F(PlainWaiterReapingTest, SequentialCompletedCallsDoNotAccumulateWaiters)
{
LiveHost host;
ASSERT_TRUE(host.ok());
auto conn = connectTo(host.port());
ASSERT_NE(conn, nullptr);
LogosObject* obj = conn->requestObject(QStringLiteral("echo_module"), 5000);
ASSERT_NE(obj, nullptr);
auto* ch = channelFor(obj);
ASSERT_NE(ch, nullptr);
auto* plain = dynamic_cast<PlainLogosObject*>(obj);
ASSERT_NE(plain, nullptr) << "the plain transport must hand back a PlainLogosObject";
constexpr int kCalls = 200;
Deliveries d(kCalls);
size_t peak = 0;
for (int i = 0; i < kCalls; ++i) {
ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
QVariantList{ QVariant(i) }, 5000,
[&d, i](QVariant, const logos::CallError& e) {
d.record(i, e);
});
pumpUntilTotal(d, i + 1, 10000);
ASSERT_EQ(d.total.load(), i + 1) << "call " << i << " never delivered";
peak = std::max(peak, waiterCount(plain));
}
const size_t finalCount = waiterCount(plain);
std::cout << " " << kCalls << " sequential completed calls -> m_waiters peak="
<< peak << " final=" << finalCount << std::endl;
EXPECT_EQ(d.errors.load(), 0) << "a completed call reported an error";
EXPECT_EQ(d.worst(), 1) << "a callback fired more than once";
EXPECT_EQ(d.missing(), 0) << "a callback never fired";
// The bound that matters is "does not scale with kCalls". 8 is generous
// headroom over the 1-2 this actually keeps (the current call's waiter, and
// at most the previous one if it published after the current spawn reaped),
// while still being 25x below the kCalls this fails at when nothing prunes.
EXPECT_LE(finalCount, 8u)
<< "finished waiters are accumulating: " << finalCount << " left after "
<< kCalls << " completed calls";
EXPECT_LE(peak, 8u) << "the registry grew during the run";
obj->release();
pump(50);
}
// The same claim with calls in flight concurrently: the bound is then peak
// concurrency, since a waiter can only be reaped once it has finished. What must
// still hold is that it does not scale with the number of CALLS.
TEST_F(PlainWaiterReapingTest, ConcurrentCompletedCallsStayBoundedByInFlight)
{
LiveHost host;
ASSERT_TRUE(host.ok());
auto conn = connectTo(host.port());
ASSERT_NE(conn, nullptr);
LogosObject* obj = conn->requestObject(QStringLiteral("echo_module"), 5000);
ASSERT_NE(obj, nullptr);
auto* ch = channelFor(obj);
ASSERT_NE(ch, nullptr);
auto* plain = dynamic_cast<PlainLogosObject*>(obj);
ASSERT_NE(plain, nullptr);
constexpr int kCalls = 600;
constexpr int kInflight = 8;
Deliveries d(kCalls);
size_t peak = 0;
int issued = 0;
while (issued < kCalls) {
while (issued < kCalls && (issued - d.total.load()) < kInflight) {
const int i = issued++;
ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
QVariantList{ QVariant(i) }, 5000,
[&d, i](QVariant, const logos::CallError& e) {
d.record(i, e);
});
}
peak = std::max(peak, waiterCount(plain));
pumpUntilTotal(d, issued - kInflight + 1, 10000);
}
pumpUntilTotal(d, kCalls, 20000);
pump(200); // a duplicate delivery would land here
const size_t finalCount = waiterCount(plain);
std::cout << " " << kCalls << " calls at " << kInflight
<< " in flight -> m_waiters peak=" << peak
<< " final=" << finalCount << std::endl;
EXPECT_EQ(d.total.load(), kCalls);
EXPECT_EQ(d.worst(), 1);
EXPECT_EQ(d.missing(), 0);
EXPECT_EQ(d.errors.load(), 0);
// Bounded by the in-flight window plus the slack of one reap cycle — not by
// kCalls, which is what it reaches when finished waiters are never dropped.
EXPECT_LE(finalCount, size_t(4 * kInflight))
<< "finished waiters accumulated past the in-flight window";
EXPECT_LE(peak, size_t(4 * kInflight));
obj->release();
pump(50);
}
// ── 2. the deadlock the fix could introduce ─────────────────────────────────
//
// A waiter announces itself as finished under m_waiterMu; the reaper takes that
// list under the same lock and then JOINS. If it joined while still holding the
// lock, a waiter blocked on that lock trying to announce itself would never
// return and the join would never complete — a two-thread deadlock, taking out
// the caller's thread (in production, usually the Qt event loop).
//
// So the two are deliberately overlapped: every spawn reaps, and the calls are
// short enough that waiters are finishing while later ones are being registered.
// The watchdog turns a wedge into an abort that names the cause.
TEST_F(PlainWaiterReapingTest, ReapingRacesPublishingWithoutDeadlocking)
{
LiveHost host;
ASSERT_TRUE(host.ok());
auto conn = connectTo(host.port());
ASSERT_NE(conn, nullptr);
constexpr int kRounds = 40;
constexpr int kPerRound = 40;
constexpr int kCalls = kRounds * kPerRound;
// ~2s in practice; 60s can only be reached by a genuine wedge.
Watchdog watchdog("ReapingRacesPublishingWithoutDeadlocking", 60000);
LogosObject* obj = conn->requestObject(QStringLiteral("echo_module"), 5000);
ASSERT_NE(obj, nullptr);
auto* ch = channelFor(obj);
ASSERT_NE(ch, nullptr);
auto* plain = dynamic_cast<PlainLogosObject*>(obj);
ASSERT_NE(plain, nullptr);
Deliveries d(kCalls);
QElapsedTimer total;
total.start();
int issued = 0;
for (int r = 0; r < kRounds; ++r) {
// A burst with no pumping between the calls: the earlier waiters of the
// burst finish (and publish) while the later ones are still being
// registered and reaping, so publish and reap collide inside the burst.
for (int k = 0; k < kPerRound; ++k) {
const int i = issued++;
ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
QVariantList{ QVariant(i) }, 5000,
[&d, i](QVariant, const logos::CallError& e) {
d.record(i, e);
});
}
// Varying drift, so the burst boundary lands at different points of the
// previous burst's completion — sometimes reaping nothing, sometimes
// reaping a batch that is still growing under it.
if (r % 4 != 0)
QThread::usleep(static_cast<unsigned long>((r % 17) * 60));
pumpUntilTotal(d, issued - kPerRound, 20000);
}
pumpUntilTotal(d, kCalls, 30000);
pump(200);
const qint64 elapsed = total.elapsed();
std::cout << " " << kCalls << " calls across " << kRounds
<< " bursts in " << elapsed << "ms, m_waiters="
<< waiterCount(plain) << std::endl;
EXPECT_EQ(d.total.load(), kCalls) << "callbacks went missing under the race";
EXPECT_EQ(d.worst(), 1) << "a callback fired more than once under the race";
EXPECT_EQ(d.missing(), 0);
EXPECT_LE(waiterCount(plain), size_t(4 * kPerRound))
<< "the registry grew across the bursts";
obj->release();
pump(50);
}
// ── 3. reaping must not cost teardown its join ──────────────────────────────
//
// Completed calls being retired early must not disturb the outstanding one: the
// object is released while a call is still in flight, after many others have
// already been reaped. release() must stay fast (it cancels rather than waiting
// the timeout out) and the abandoned call must still deliver, once.
TEST_F(PlainWaiterReapingTest, TeardownAfterReapingStillJoinsAndDeliversOnce)
{
LiveHost host;
ASSERT_TRUE(host.ok());
auto conn = connectTo(host.port());
ASSERT_NE(conn, nullptr);
LogosObject* obj = conn->requestObject(QStringLiteral("echo_module"), 5000);
ASSERT_NE(obj, nullptr);
auto* ch = channelFor(obj);
ASSERT_NE(ch, nullptr);
auto* plain = dynamic_cast<PlainLogosObject*>(obj);
ASSERT_NE(plain, nullptr);
constexpr int kWarm = 100;
Deliveries warm(kWarm);
for (int i = 0; i < kWarm; ++i) {
ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
QVariantList{ QVariant(i) }, 5000,
[&warm, i](QVariant, const logos::CallError& e) {
warm.record(i, e);
});
pumpUntilTotal(warm, i + 1, 10000);
}
ASSERT_EQ(warm.total.load(), kWarm);
ASSERT_LE(waiterCount(plain), 8u) << "the warmup calls were not reaped";
// Now release with a call that REALLY is outstanding: `block` parks in the
// provider until the host is torn down, so the waiter is unambiguously
// mid-wait when release() lands, rather than racing a `ping` that may have
// already answered.
Deliveries last(1);
ch->callMethodAsyncWithError(kToken, QStringLiteral("block"), {}, 8000,
[&last](QVariant, const logos::CallError& e) {
last.record(0, e);
});
pump(200);
ASSERT_EQ(last.total.load(), 0) << "the provider answered; nothing was in flight";
QElapsedTimer timer;
timer.start();
obj->release();
const qint64 releaseMs = timer.elapsed();
pumpUntilTotal(last, 1, 3000);
pump(300); // a second delivery would land here
std::cout << " release() after " << kWarm << " reaped calls took "
<< releaseMs << "ms, in-flight call delivered "
<< last.total.load() << " time(s) code='" << last.code() << "'"
<< std::endl;
EXPECT_LT(releaseMs, 750)
<< "release() waited out the in-flight call instead of cancelling it";
// The join that keeps `this` alive under the waiter is still there, and the
// abandoned call is still told — once. Reaping the finished waiters must
// change neither.
EXPECT_EQ(last.total.load(), 1) << "the in-flight call did not deliver exactly once";
EXPECT_EQ(last.code(), "transport_error");
EXPECT_EQ(warm.worst(), 1);
}