diff --git a/cpp/implementations/plain/plain_logos_object.cpp b/cpp/implementations/plain/plain_logos_object.cpp index fb9b752..6c9b829 100644 --- a/cpp/implementations/plain/plain_logos_object.cpp +++ b/cpp/implementations/plain/plain_logos_object.cpp @@ -1,11 +1,13 @@ #include "plain_logos_object.h" +#include "logos_async_dispatch.h" #include "qvariant_rpc_value.h" #include #include #include #include +#include #include #include @@ -33,6 +35,10 @@ QVariant PlainLogosObject::callMethod(const QString& authToken, { if (!m_conn || !m_conn->isOpen()) return QVariant(); + // Subscribe to the completion channel BEFORE sending, so a "multi" provider's + // completion can't race ahead of the waiter (it's buffered either way). + ensureCompletionSub(); + CallMessage msg; msg.id = m_conn->nextId(); msg.authToken = authToken.toStdString(); @@ -52,7 +58,52 @@ QVariant PlainLogosObject::callMethod(const QString& authToken, << "failed:" << QString::fromStdString(res.err); return QVariant(); } - return rpcValueToQVariant(res.value); + const QVariant value = rpcValueToQVariant(res.value); + // A "multi" provider may have deferred: it returned a pending sentinel and + // pushes the real result as a completion event. Wait for it, keyed by callId. + if (value.typeId() == QMetaType::QVariantMap) { + const QVariantMap m = value.toMap(); + if (m.contains(logos::pendingCallKey())) + return awaitCompletion(m.value(logos::pendingCallKey()).toString(), timeoutMs); + } + return value; +} + +void PlainLogosObject::ensureCompletionSub() +{ + { + std::lock_guard g(m_completionMu); + if (m_completionSubscribed) return; + m_completionSubscribed = true; + } + // Reuse the normal event subscription path (tracked in m_subs, so + // disconnectEvents() tears it down). The handler fires on the connection's + // IO thread; it buffers the result and wakes any waiter. + onEvent(logos::callCompleteEvent(), [this](const QString&, const QVariantList& data) { + if (data.size() != 2) return; + const QString callId = data.at(0).toString(); + { + std::lock_guard g(m_completionMu); + m_completions[callId] = data.at(1); + } + m_completionCv.notify_all(); + }); +} + +QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs) +{ + std::unique_lock lk(m_completionMu); + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(timeoutMs > 0 ? timeoutMs : 30000); + const bool got = m_completionCv.wait_until(lk, deadline, + [&] { return m_completions.count(callId) > 0; }); + if (!got) { + qWarning() << "PlainLogosObject: deferred call" << callId << "timed out"; + return QVariant(); + } + const QVariant result = m_completions[callId]; + m_completions.erase(callId); + return result; } namespace { @@ -95,6 +146,8 @@ void PlainLogosObject::callMethodAsync(const QString& authToken, return; } + ensureCompletionSub(); + CallMessage msg; msg.id = m_conn->nextId(); msg.authToken = authToken.toStdString(); @@ -110,7 +163,7 @@ void PlainLogosObject::callMethodAsync(const QString& authToken, // future iteration can fold this wait into the shared Asio // io_context (the connection already runs on it) so we don't spin // up a thread per pending RPC. - std::thread([fut, timeoutMs, callback = std::move(callback)]() mutable { + std::thread([this, fut, timeoutMs, callback = std::move(callback)]() mutable { if (fut->wait_for(std::chrono::milliseconds(timeoutMs)) != std::future_status::ready) { postToQtEventLoop(std::move(callback), QVariant()); @@ -118,6 +171,13 @@ void PlainLogosObject::callMethodAsync(const QString& authToken, } auto res = fut->get(); QVariant value = res.ok ? rpcValueToQVariant(res.value) : QVariant(); + // Resolve a "multi" provider's deferred completion (sentinel → wait for + // the completion event) right here on the waiter thread. + if (value.typeId() == QMetaType::QVariantMap) { + const QVariantMap m = value.toMap(); + if (m.contains(logos::pendingCallKey())) + value = awaitCompletion(m.value(logos::pendingCallKey()).toString(), timeoutMs); + } postToQtEventLoop(std::move(callback), std::move(value)); }).detach(); } diff --git a/cpp/implementations/plain/plain_logos_object.h b/cpp/implementations/plain/plain_logos_object.h index c088b60..87f6df1 100644 --- a/cpp/implementations/plain/plain_logos_object.h +++ b/cpp/implementations/plain/plain_logos_object.h @@ -5,6 +5,8 @@ #include "rpc_connection.h" +#include +#include #include #include #include @@ -51,10 +53,25 @@ public: quintptr id() const override; private: + // Deferred ("multi") completion rendezvous. A multi provider returns a + // pending sentinel (logos::pendingCallKey) from callMethod and later pushes + // the real result as a logos::callCompleteEvent event keyed by callId. We + // subscribe to that event EAGERLY (before any call can defer) so a completion + // racing ahead of the waiter is buffered, then block the caller until the + // matching callId lands. The completion arrives on the connection's IO + // thread; the caller waits on another thread — m_completionMu/Cv bridge them. + void ensureCompletionSub(); + QVariant awaitCompletion(const QString& callId, int timeoutMs); + std::string m_objectName; std::shared_ptr m_conn; std::mutex m_mu; std::vector> m_subs; + + std::mutex m_completionMu; + std::condition_variable m_completionCv; + std::map m_completions; + bool m_completionSubscribed = false; }; } // namespace logos::plain diff --git a/cpp/implementations/plain/rpc_server.cpp b/cpp/implementations/plain/rpc_server.cpp index 56a565b..7c8a25f 100644 --- a/cpp/implementations/plain/rpc_server.cpp +++ b/cpp/implementations/plain/rpc_server.cpp @@ -44,12 +44,22 @@ bool RpcServerTcp::start() void RpcServerTcp::stop() { - std::lock_guard g(m_mu); - m_stopped = true; - boost::system::error_code ignore; - m_acceptor.close(ignore); - for (auto& c : m_conns) c->stop("server stopped"); - m_conns.clear(); + // Move the connection list OUT under the lock, then release it before + // stopping each connection. conn->stop() fails the connection, which + // synchronously invokes its error handler (set in doAccept) — and that + // handler locks m_mu to erase itself from m_conns. Holding m_mu across the + // stop() call would re-enter this non-recursive mutex on the same thread and + // self-deadlock. The handler's erase is then a harmless no-op (the list it + // scans is already empty). + std::vector> conns; + { + std::lock_guard g(m_mu); + m_stopped = true; + boost::system::error_code ignore; + m_acceptor.close(ignore); + conns.swap(m_conns); + } + for (auto& c : conns) c->stop("server stopped"); } void RpcServerTcp::doAccept() @@ -118,12 +128,17 @@ bool RpcServerSsl::start() void RpcServerSsl::stop() { - std::lock_guard g(m_mu); - m_stopped = true; - boost::system::error_code ignore; - m_acceptor.close(ignore); - for (auto& c : m_conns) c->stop("server stopped"); - m_conns.clear(); + // See RpcServerTcp::stop — release m_mu before stopping connections so the + // per-connection error handler (which re-locks m_mu) can't self-deadlock. + std::vector> conns; + { + std::lock_guard g(m_mu); + m_stopped = true; + boost::system::error_code ignore; + m_acceptor.close(ignore); + conns.swap(m_conns); + } + for (auto& c : conns) c->stop("server stopped"); } void RpcServerSsl::doAccept() diff --git a/cpp/implementations/qt_remote/remote_transport.cpp b/cpp/implementations/qt_remote/remote_transport.cpp index 0233797..8186b74 100644 --- a/cpp/implementations/qt_remote/remote_transport.cpp +++ b/cpp/implementations/qt_remote/remote_transport.cpp @@ -1,15 +1,18 @@ #include "remote_transport.h" +#include "../../logos_async_dispatch.h" #include #include #include #include #include #include +#include #include #include #include #include #include +#include // ── RemoteLogosObject ──────────────────────────────────────────────────────── @@ -51,6 +54,28 @@ public: : m_replica(replica), m_helper(nullptr) { qDebug() << "[LogosObject] Created RemoteLogosObject wrapping QRemoteObjectReplica" << reinterpret_cast(replica); + if (m_replica) { + // Eager event wiring — a deferred ("multi") call's result arrives as a + // completion event, so the channel must be live even when the caller + // never subscribes to a user event. onEvent() reuses this same helper. + m_helper = new RemoteEventHelper(); + QObject::connect(m_replica, SIGNAL(eventResponse(QString,QVariantList)), + m_helper, SLOT(onEventResponse(QString,QVariantList))); + m_helper->addCallback(logos::callCompleteEvent(), + [this](const QString&, const QVariantList& data) { + if (data.size() != 2) return; + const QString id = data.at(0).toString(); + const QVariant result = data.at(1); + m_completions.insert(id, result); + if (QEventLoop* loop = m_completionWaiters.value(id, nullptr)) + loop->quit(); // wake a sync waiter + if (m_asyncCompletionCbs.contains(id)) { // fire an async waiter + auto cb = m_asyncCompletionCbs.take(id); + m_completions.remove(id); + if (cb) cb(result); + } + }); + } } ~RemoteLogosObject() override { @@ -92,7 +117,10 @@ public: return QVariant(); } - return pendingCall.returnValue(); + // A "multi" provider may have deferred the result (returned a pending + // sentinel); resolveDeferred waits for the completion event, or returns + // the value unchanged for an ordinary (synchronous) result. + return resolveDeferred(pendingCall.returnValue(), timeoutMs); } void callMethodAsync(const QString& authToken, @@ -135,7 +163,7 @@ public: // Success handler -- delivers result on the consumer's thread QObject::connect(watcher, &QRemoteObjectPendingCallWatcher::finished, - watcher, [callback, timer](QRemoteObjectPendingCallWatcher* w) { + watcher, [this, callback, timer, timeoutMs](QRemoteObjectPendingCallWatcher* w) { timer->stop(); // cancel timeout QVariant result; if (w->error() == QRemoteObjectPendingCall::NoError) { @@ -143,8 +171,26 @@ public: } else { qWarning() << "RemoteLogosObject: async callMethod error:" << w->error(); } - callback(result); w->deleteLater(); + // A "multi" provider may have deferred the result: wait for the + // completion event instead of delivering the pending sentinel. + if (result.typeId() == QMetaType::QVariantMap) { + const QVariantMap m = result.toMap(); + if (m.contains(logos::pendingCallKey())) { + const QString callId = m.value(logos::pendingCallKey()).toString(); + if (m_completions.contains(callId)) { callback(m_completions.take(callId)); return; } + m_asyncCompletionCbs.insert(callId, callback); + // Bound the wait: deliver an empty result once if it never lands. + QTimer::singleShot(timeoutMs, m_helper, [this, callId]() { + if (m_asyncCompletionCbs.contains(callId)) { + auto cb = m_asyncCompletionCbs.take(callId); + if (cb) cb(QVariant()); + } + }); + return; + } + } + callback(result); }, Qt::QueuedConnection); // Timeout handler -- stops the watcher and delivers empty result @@ -241,8 +287,38 @@ public: quintptr id() const override { return reinterpret_cast(m_replica); } private: + // Resolve a possibly-deferred result. If `rv` is a pending sentinel from a + // "multi" provider, wait (up to timeoutMs) for the completion event keyed by + // callId, pumping the consumer event loop; otherwise return `rv` unchanged. + QVariant resolveDeferred(const QVariant& rv, int timeoutMs) + { + if (rv.typeId() != QMetaType::QVariantMap) return rv; + const QVariantMap m = rv.toMap(); + if (!m.contains(logos::pendingCallKey())) return rv; + const QString callId = m.value(logos::pendingCallKey()).toString(); + // The completion can arrive during the call's own waitForFinished above. + if (m_completions.contains(callId)) return m_completions.take(callId); + QEventLoop loop; + m_completionWaiters.insert(callId, &loop); + QTimer timer; + timer.setSingleShot(true); + QObject::connect(&timer, &QTimer::timeout, &loop, &QEventLoop::quit); + timer.start(timeoutMs > 0 ? timeoutMs : 30000); + loop.exec(); + m_completionWaiters.remove(callId); + if (m_completions.contains(callId)) return m_completions.take(callId); + qWarning() << "RemoteLogosObject: deferred call" << callId << "timed out"; + return QVariant(); + } + QObject* m_replica; RemoteEventHelper* m_helper; + // Deferred ("multi") completions delivered over the event channel, keyed by + // callId: buffered results + sync (QEventLoop) and async (callback) waiters. + // Touched only on the consumer event-loop thread. + QHash m_completions; + QHash m_completionWaiters; + QHash m_asyncCompletionCbs; }; // ── RemoteTransportHost ────────────────────────────────────────────────────── diff --git a/cpp/logos_api_client.cpp b/cpp/logos_api_client.cpp index 55d2df7..9ce36b4 100644 --- a/cpp/logos_api_client.cpp +++ b/cpp/logos_api_client.cpp @@ -168,43 +168,56 @@ void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QS QString token = getToken(objectName); if (token.isEmpty() && objectName != "capability_module" && m_capability_consumer) { - // Async-chain: dispatch the requestModule call asynchronously, - // and only fire the real method's invokeRemoteMethodAsync from - // its callback. The previous version called `requestModule` - // synchronously here, which made the "async" entry point - // block its caller for the full requestModule round-trip - // (a real perf hit when capability_module has any latency). + // Async-chain: dispatch the requestModule call asynchronously, and only + // fire the real method's invokeRemoteMethodAsync from its callback. The + // previous version called `requestModule` synchronously here, which made + // the "async" entry point block its caller for the full round-trip. // - // Lifetime: the inner callback captures m_consumer through a - // QPointer guard. If the LogosAPIClient (and thus its - // QObject-parented m_consumer) is destroyed while the - // requestModule round-trip is still in flight, the QPointer - // goes null and the inner dispatch is suppressed instead of - // dereferencing dangling memory. + // COALESCE concurrent first-calls behind ONE handshake. A driver that + // fans out N async calls to an un-tokened target before any completes + // would otherwise fire N separate requestModule handshakes; each mints a + // distinct token and informs the target, and the later inform OVERWRITES + // the earlier token there (the target stores one token per caller). The + // already-dispatched calls then carry a superseded token and the target + // rejects them as unauthorized. So only the first caller starts the + // handshake; the rest queue and all drain with the single minted token. + // (The sync path can't hit this — it blocks per call, so handshakes + // never overlap.) m_pendingHandshakes is touched only on the owner + // thread, reached above, so no lock is needed. + m_pendingHandshakes[objectName].push_back( + [this, objectName, methodName, args, timeout, cb = std::move(callback)] + (const QString& tok) mutable { + m_consumer->invokeRemoteMethodAsync(tok, objectName, methodName, args, + std::move(cb), timeout); + }); + if (m_pendingHandshakes[objectName].size() > 1) + return; // a handshake for this target is already in flight + const QString capabilityToken = getToken("capability_module"); const QString origin = m_origin_module; - QPointer consumer = m_consumer; - auto outerCallback = std::move(callback); + // Lifetime: capture the client through a QPointer guard. If it (and its + // QObject-parented consumers + the pending queue) is destroyed while the + // requestModule round-trip is in flight, the guard goes null and we drop + // the queued continuations instead of dereferencing dangling memory. + QPointer self(this); m_capability_consumer->invokeRemoteMethodAsync( capabilityToken, QStringLiteral("capability_module"), QStringLiteral("requestModule"), QVariantList() << origin << objectName, - [consumer, objectName, methodName, args, timeout, - outerCallback = std::move(outerCallback)] - (const QVariant& tokenResult) mutable { - if (!consumer) { - // Client was destroyed mid-flight. Honour the - // contract by firing the outer callback with an - // invalid QVariant so callers don't deadlock - // waiting for a result that'll never come. - if (outerCallback) outerCallback(QVariant{}); - return; - } - consumer->invokeRemoteMethodAsync( - tokenResult.toString(), - objectName, methodName, args, - std::move(outerCallback), timeout); + [self, objectName](const QVariant& tokenResult) mutable { + if (!self) return; // client destroyed mid-flight + const QString tok = tokenResult.toString(); + // Drain every continuation queued for this target with the one + // minted token — the target was informed of exactly this token. + // An empty tok (handshake failed) still flows through: the + // consumer call is then rejected and each callback fires with an + // invalid QVariant, so callers never hang. + auto it = self->m_pendingHandshakes.find(objectName); + if (it == self->m_pendingHandshakes.end()) return; + std::vector> calls = std::move(it.value()); + self->m_pendingHandshakes.erase(it); + for (auto& c : calls) c(tok); }, timeout); return; diff --git a/cpp/logos_api_client.h b/cpp/logos_api_client.h index eeefa7d..5422e83 100644 --- a/cpp/logos_api_client.h +++ b/cpp/logos_api_client.h @@ -8,6 +8,7 @@ #include #include #include +#include #include "logos_call_error.h" #include "logos_mode.h" @@ -282,6 +283,17 @@ private: // so any old constructor that doesn't list this field still // leaves a defined value. LogosAPIConsumer* m_capability_consumer = nullptr; + + // Per-target queue of continuations waiting on an in-flight async + // requestModule handshake. The FIRST async call to an un-tokened target + // starts exactly one handshake; concurrent calls to the same target queue + // here and all drain with the single minted token when it resolves. Without + // this coalescing a fan-out of N first-calls fires N racing handshakes whose + // distinct tokens overwrite each other on the target — so already-dispatched + // calls carry a superseded token and get rejected. Touched only on the + // owner thread (invokeRemoteMethodAsync marshals there), so it needs no + // lock. Appended last per the ABI note above; defaults to empty. + QMap>> m_pendingHandshakes; }; #endif // LOGOS_API_CLIENT_H diff --git a/cpp/logos_async_dispatch.h b/cpp/logos_async_dispatch.h new file mode 100644 index 0000000..0f33364 --- /dev/null +++ b/cpp/logos_async_dispatch.h @@ -0,0 +1,32 @@ +#ifndef LOGOS_ASYNC_DISPATCH_H +#define LOGOS_ASYNC_DISPATCH_H + +#include + +// Shared wire constants for "multi" (concurrent) dispatch. Concurrency is a +// MODULE-side concern handled entirely behind the ordinary callMethod entry +// point — no new provider/host vtable method, so the provider ABI is unchanged +// and an old host/daemon loads a "multi" module and forwards its traffic +// without even understanding these markers. +// +// A "multi" module's generated glue does NOT block in callMethod: it hands the +// handler to a worker and returns a PENDING SENTINEL immediately (a QVariantMap +// carrying the call id under pendingCallKey()). When the worker finishes, the +// module pushes the real result back as a COMPLETION event +// (callCompleteEvent(), data = [callId, result]) over the SAME event channel it +// already uses (setEventListener). The host (ModuleProxy / liblogos) is a pure +// forwarder — it returns whatever callMethod returned and forwards whatever +// events the module emits. The CONSUMER transport (RemoteLogosObject for QtRO, +// PlainLogosObject for the plain transport) detects the sentinel, waits for the +// matching completion keyed by callId, and returns the real result — so +// generated clients call transparently. Both transports use this path; the +// version that speaks it is logos-protocol 0.2 (additive minor — see +// logos_protocol.h). +namespace logos { + +inline QString pendingCallKey() { return QStringLiteral("__logos_pending_call__"); } +inline QString callCompleteEvent() { return QStringLiteral("__logos_call_complete__"); } + +} // namespace logos + +#endif // LOGOS_ASYNC_DISPATCH_H diff --git a/cpp/logos_protocol.h b/cpp/logos_protocol.h index bb0a271..83e0352 100644 --- a/cpp/logos_protocol.h +++ b/cpp/logos_protocol.h @@ -50,9 +50,15 @@ * =========================================================================== */ #define LOGOS_PROTOCOL_VERSION_MAJOR 0 -#define LOGOS_PROTOCOL_VERSION_MINOR 1 +// 0.2: per-module concurrent dispatch ("multi"). Additive/back-compatible — a +// multi module returns a deferred-completion sentinel from callMethod and pushes +// the result as a __logos_call_complete__ event (see logos_async_dispatch.h); +// the provider/host ABI is UNCHANGED, so same-MAJOR hosts (incl. 0.1 daemons) +// load and forward multi modules without modification. A pre-0.2 *consumer* +// would see the raw sentinel rather than awaiting it — graceful, not a crash. +#define LOGOS_PROTOCOL_VERSION_MINOR 2 #define LOGOS_PROTOCOL_VERSION_PATCH 0 -#define LOGOS_PROTOCOL_VERSION_STRING "0.1.0" +#define LOGOS_PROTOCOL_VERSION_STRING "0.2.0" #ifdef __cplusplus extern "C" { diff --git a/nix/default.nix b/nix/default.nix index 2711385..ddae4b5 100644 --- a/nix/default.nix +++ b/nix/default.nix @@ -3,7 +3,7 @@ { pname = "logos-protocol"; - version = "0.1.0"; + version = "0.2.0"; # Common native build inputs nativeBuildInputs = [ diff --git a/tests/protocol/CMakeLists.txt b/tests/protocol/CMakeLists.txt index 61b1e50..599b68d 100644 --- a/tests/protocol/CMakeLists.txt +++ b/tests/protocol/CMakeLists.txt @@ -19,6 +19,8 @@ add_executable(protocol_tests test_rpc_framing.cpp test_json_codec.cpp test_cbor_codec.cpp + # Per-module concurrent dispatch (concurrency:"multi") over the plain transport. + test_concurrent_dispatch.cpp # test_plain_transport_tcp.cpp — entirely #if 0 (see the note inside: # in-process Qt-event-loop deadlock between consumer + provider under # nix's test sandbox; exercised cross-process by the integration matrix). diff --git a/tests/protocol/test_concurrent_dispatch.cpp b/tests/protocol/test_concurrent_dispatch.cpp new file mode 100644 index 0000000..a1e5326 --- /dev/null +++ b/tests/protocol/test_concurrent_dispatch.cpp @@ -0,0 +1,247 @@ +// Proves per-module concurrent dispatch (concurrency:"multi") with NO provider +// ABI change — concurrency is owned by the module, behind the ordinary +// callMethod, exactly as a generated "multi" glue does it: +// +// - multi : callMethod does NOT block. It hands slow() to a worker and +// returns a PENDING SENTINEL ({pendingCallKey: callId}) at once, so +// the dispatch thread is free to take the next call. The worker +// records peak overlap, then pushes the result as a +// callCompleteEvent([callId, result]) over the event listener. +// The plain consumer detects the sentinel and awaits the +// completion. ⇒ overlap (peak 2) +// - single : callMethod runs slow() inline, blocking the dispatch thread until +// it returns the result directly (no sentinel). ⇒ serial (peak 1) +// +// The host (ModuleProxy / PlainTransportHost) is unchanged from master: it just +// returns whatever callMethod returned and forwards whatever events the provider +// emits. The callers run on their own threads (blocking in the consumer) while +// the main thread pumps the event loop so the worker's completion event — which +// ModuleProxy marshals onto the source thread — is delivered. + +#include + +#include "logos_async_dispatch.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 +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +using namespace logos::plain; + +namespace { + +// Records the peak number of slow() handlers running at the same time. +class SlowProvider : public LogosProviderObject { +public: + explicit SlowProvider(bool multi) : m_multi(multi) {} + + QVariant callMethod(const QString& method, const QVariantList& args) override + { + if (method != QLatin1String("slow")) return QVariant(); + const int ms = args.value(0).toInt(); + + // single: run inline → blocks the dispatch thread → calls serialize. + if (!m_multi) return runSlow(ms); + + // multi: defer. Spawn a worker, return a pending sentinel immediately so + // the dispatch thread is freed; the worker emits the completion event. + const QString callId = QStringLiteral("lc-%1").arg( + static_cast(m_callCounter.fetch_add(1, std::memory_order_relaxed))); + std::thread([this, callId, ms]() { + const int result = runSlow(ms); + if (m_eventCb) + m_eventCb(logos::callCompleteEvent(), QVariantList{ callId, QVariant(result) }); + }).detach(); + QVariantMap pending; + pending[logos::pendingCallKey()] = callId; + return pending; + } + + QJsonArray getMethods() override { return QJsonArray{}; } + bool informModuleToken(const QString&, const QString&) override { return true; } + void setEventListener(EventCallback cb) override { m_eventCb = std::move(cb); } + void init(void*) override {} + QString providerName() const override { return QStringLiteral("slow"); } + QString providerVersion() const override { return QStringLiteral("1.0.0"); } + + int maxConcurrent() const { return m_maxSeen.load(); } + +private: + int runSlow(int ms) + { + const int now = ++m_inFlight; + int prev = m_maxSeen.load(); + while (now > prev && !m_maxSeen.compare_exchange_weak(prev, now)) { /* retry */ } + std::this_thread::sleep_for(std::chrono::milliseconds(ms)); + --m_inFlight; + return ms; + } + + bool m_multi; + EventCallback m_eventCb; + std::atomic m_callCounter{0}; + std::atomic m_inFlight{0}; + std::atomic m_maxSeen{0}; +}; + +QCoreApplication* ensureApp() +{ + static int argc = 0; + static char* argv[] = { nullptr }; + if (!QCoreApplication::instance()) + new QCoreApplication(argc, argv); + return QCoreApplication::instance(); +} + +} // namespace + +class ConcurrentDispatchTest : public ::testing::Test { +protected: + void SetUp() override { ensureApp(); } + + // Fire two concurrent slow() calls and return the peak observed overlap. + int peakOverlap(bool multi) + { + LogosTransportConfig cfg; + cfg.protocol = LogosProtocol::Tcp; + cfg.host = "127.0.0.1"; + cfg.port = 0; + + auto host = std::make_unique(cfg); + EXPECT_TRUE(host->start()); + + SlowProvider provider(multi); + ModuleProxy proxy(&provider); + proxy.saveToken(QStringLiteral("core"), QStringLiteral("tok")); + EXPECT_TRUE(host->publishObject("slow_mod", &proxy)); + + const QString endpoint = host->endpoint(); + const uint16_t port = endpoint.mid(endpoint.lastIndexOf(':') + 1).toUShort(); + + LogosTransportConfig ccfg = cfg; + ccfg.port = port; + auto conn = std::make_unique(ccfg); + EXPECT_TRUE(conn->connectToHost()); + + LogosObject* obj = conn->requestObject("slow_mod", 2000); + EXPECT_NE(obj, nullptr); + if (!obj) return -1; + + std::atomic done{0}; + auto caller = [&]() { + obj->callMethod(QStringLiteral("tok"), QStringLiteral("slow"), + QVariantList{ 300 }, 5000); + done.fetch_add(1); + }; + std::thread t1(caller), t2(caller); + + // Pump the host event loop until both callers return (or a generous cap). + for (int i = 0; i < 800 && done.load() < 2; ++i) { + QCoreApplication::processEvents(); + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + t1.join(); + t2.join(); + EXPECT_EQ(done.load(), 2); + + const int peak = provider.maxConcurrent(); + obj->release(); + host.reset(); + return peak; + } + + // Fire N concurrent calls from a SINGLE thread via the ASYNC consumer path + // (callMethodAsync) — the fan-out pattern a real driver module uses through + // the generated work_async() client: it fires N non-blocking calls without + // waiting between them, so all N are in flight before any completes. This + // exercises PlainLogosObject::callMethodAsync, which resolves a "multi" + // provider's pending sentinel on its waiter thread (the sync peakOverlap + // above only covers the blocking callMethod path). Returns the peak overlap. + int peakOverlapAsync(bool multi, int n) + { + LogosTransportConfig cfg; + cfg.protocol = LogosProtocol::Tcp; + cfg.host = "127.0.0.1"; + cfg.port = 0; + + auto host = std::make_unique(cfg); + EXPECT_TRUE(host->start()); + + SlowProvider provider(multi); + ModuleProxy proxy(&provider); + proxy.saveToken(QStringLiteral("core"), QStringLiteral("tok")); + EXPECT_TRUE(host->publishObject("slow_mod", &proxy)); + + const QString endpoint = host->endpoint(); + const uint16_t port = endpoint.mid(endpoint.lastIndexOf(':') + 1).toUShort(); + + LogosTransportConfig ccfg = cfg; + ccfg.port = port; + auto conn = std::make_unique(ccfg); + EXPECT_TRUE(conn->connectToHost()); + + LogosObject* obj = conn->requestObject("slow_mod", 2000); + EXPECT_NE(obj, nullptr); + if (!obj) return -1; + + // Fire n async calls back-to-back from this one thread. None blocks, so + // all n reach the provider before any returns — a "multi" provider runs + // them at once; a "single" one serializes them. + std::atomic done{0}; + for (int i = 0; i < n; ++i) { + obj->callMethodAsync(QStringLiteral("tok"), QStringLiteral("slow"), + QVariantList{ 300 }, 5000, + [&done](const QVariant&) { done.fetch_add(1); }); + } + + // Pump the host event loop until every async callback has fired. + for (int i = 0; i < 1000 && done.load() < n; ++i) { + QCoreApplication::processEvents(); + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + EXPECT_EQ(done.load(), n); + + const int peak = provider.maxConcurrent(); + obj->release(); + host.reset(); + return peak; + } +}; + +TEST_F(ConcurrentDispatchTest, MultiProviderOverlaps) +{ + EXPECT_EQ(peakOverlap(/*multi=*/true), 2); +} + +TEST_F(ConcurrentDispatchTest, SingleProviderSerializes) +{ + EXPECT_EQ(peakOverlap(/*multi=*/false), 1); +} + +TEST_F(ConcurrentDispatchTest, MultiProviderOverlapsAsync) +{ + // The fan-out pattern over the async consumer path: one thread fires 4 + // non-blocking calls; the "multi" provider runs all 4 concurrently. + EXPECT_EQ(peakOverlapAsync(/*multi=*/true, /*n=*/4), 4); +} + +TEST_F(ConcurrentDispatchTest, SingleProviderSerializesAsync) +{ + // The same fan-out at a "single" provider serializes — peak 1. + EXPECT_EQ(peakOverlapAsync(/*multi=*/false, /*n=*/4), 1); +}