From c5b4e18eb89be0556f0b486dc5e1527da5f8916d Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sun, 1 Nov 2026 02:32:48 +0100 Subject: [PATCH] feat(quic): QUICStream::OnRecieve and Abort, callbacks that hold no thread Every way to wait for a stream's bytes held a thread: RecieveSync and RecieveUntilCloseSync block their caller with no bound, and RecieveAsync and RecieveUntilCloseAsync are those run as pool tasks, so they block a worker of the pool the host shares, for good on a stream a client keeps open. A server embedded in an app, answering requests from the app's main loop and streaming logs to followers, can afford neither. OnRecieve(onData(span), onEnd(clean)) delivers the bytes as they arrive and the end once: clean for the peer's FIN (PEER_SEND_SHUTDOWN, which msquic raises only once every byte before it has been taken), not clean for a reset by either side or a connection that went first. Bytes that arrived before the call are delivered by it, before it returns, and so is the end; from then on the callbacks run on the msquic worker inside the event that made them due, with msquic's own buffers handed over when nothing is queued, so nothing is copied and no thread waits. The callbacks of one stream, SendAsync's onSent included, never run at once. Whichever thread finds nobody delivering becomes the delivering thread and runs what is ready, picking up what others queue meanwhile; the others queue and return, the worker never waiting. A mutex held across the callbacks was the alternative: it would make the worker wait whenever the registering thread was delivering, and deadlock a callback that called into msquic and waited for that worker. A failed SendAsync now reports after the callback that made it, or from a pool task. Abort(errorCode) resets both directions with an application error code (ABORT_SEND | ABORT_RECEIVE | IMMEDIATE, so SHUTDOWN_COMPLETE, and onEnd(false), follow without waiting for the peer's acknowledgement). From inside onData it is how a server enforces a byte cap: no onData follows, what was queued is dropped, onEnd(false) comes once. Destroying the wrapper, or Stop(), unregisters the callbacks the way std::stop_callback does: once it returns neither runs again, one running on another thread has finished, and one on the destructor's own thread, a callback destroying its own wrapper, is not waited for. After the end or the unregistering the callbacks are released, so a callback holding its own stream does not keep it forever. The callbacks use the shared Impl only, so moving the wrapper under them is safe. Breaking: none. Existing receives are unchanged, and must not be mixed with OnRecieve on one stream. Co-Authored-By: Claude Opus 5.5 --- .../Crafter.Network-ClientQUIC.cpp | 293 ++++++++++++++++-- interfaces/Crafter.Network-ClientQUIC.cppm | 51 ++- 2 files changed, 316 insertions(+), 28 deletions(-) diff --git a/implementations/Crafter.Network-ClientQUIC.cpp b/implementations/Crafter.Network-ClientQUIC.cpp index 886f18a..a74e71f 100644 --- a/implementations/Crafter.Network-ClientQUIC.cpp +++ b/implementations/Crafter.Network-ClientQUIC.cpp @@ -251,12 +251,151 @@ struct QUICStream::Impl { return false; } + // ---- Callbacks: OnRecieve's onData and onEnd, SendAsync's onSent ---- + // They run one at a time per stream, in order, and never under `mtx`. + // Whichever thread finds nobody delivering becomes the delivering thread + // and runs what is ready, picking up whatever others queue meanwhile; the + // others only queue and return. Normally that thread is the msquic + // worker, inside the event that made something ready, so no thread is + // held while nothing is; it is the caller's thread for the bytes + // OnRecieve finds waiting, and a pool task for a send that never reached + // msquic. Nothing blocks to wait for the delivering thread, the worker + // least of all, except Unregister, as std::stop_callback does. + // + // A blocking alternative, one mutex held across every callback, would + // make the worker wait whenever the caller's thread was delivering, and a + // callback that called into msquic and waited for that worker + // (GetStreamId does) would deadlock; a pool task per delivery would make + // every byte depend on a pool the host shares. + std::function)> onData; + std::function onEnd; + bool handlerSet = false; // OnRecieve was called + bool unregistered = false; // Stop(): onData and onEnd never again + bool aborted = false; // Abort(): no more onData, what waits is dropped + enum class End { Open, Clean, Unclean }; + End end = End::Open; // how the receive direction ended, once it has + bool endDelivered = false; + bool delivering = false; + std::thread::id deliveringThread; + std::condition_variable deliveryIdle; + std::deque, bool>> completions; // onSent calls waiting their turn + + // Becomes the delivering thread if there is none. Under mtx. + bool BeginDelivery() { + if (delivering) return false; + delivering = true; + deliveringThread = std::this_thread::get_id(); + return true; + } + + // Runs what is ready, one callback at a time and none under the lock, + // until nothing is, then stops delivering. Entered as the delivering + // thread with `lk` held; returns with it released. + void DeliverReady(std::unique_lock& lk) { + for (;;) { + if (!completions.empty()) { + std::function onSent = std::move(completions.front().first); + const bool sent = completions.front().second; + completions.pop_front(); + lk.unlock(); + onSent(sent); + onSent = nullptr; + lk.lock(); + continue; + } + if (!handlerSet || unregistered) break; + if (aborted) pending.clear(); + if (!pending.empty()) { + std::vector chunk = std::move(pending.front()); + pending.pop_front(); + lk.unlock(); + onData(std::span(chunk.data(), chunk.size())); + lk.lock(); + continue; + } + // Only once everything before it has been handed over. + if (end != End::Open && !endDelivered) { + endDelivered = true; + const bool clean = end == End::Clean; + lk.unlock(); + onEnd(clean); + lk.lock(); + continue; + } + break; + } + // Nothing calls onData or onEnd again once the end is out or the + // owner unregistered, so let them go, with what they captured: a + // callback holding its own stream would otherwise keep it forever. + // Destroyed outside the lock, where a capture's destructor may do + // anything. + std::function)> dataHandler; + std::function endHandler; + if (handlerSet && (endDelivered || unregistered)) { + dataHandler.swap(onData); + endHandler.swap(onEnd); + } + delivering = false; + lk.unlock(); + deliveryIdle.notify_all(); + } + + // Delivers what is ready if no thread is, and releases `lk` either way. + void DeliverIfIdle(std::unique_lock& lk) { + if (BeginDelivery()) { + DeliverReady(lk); + } else { + lk.unlock(); + } + } + + // Queues onSent(sent) behind the stream's other callbacks. + void Complete(std::function onSent, bool sent, std::unique_lock& lk) { + completions.emplace_back(std::move(onSent), sent); + DeliverIfIdle(lk); + } + // A SendAsync that never reached msquic still reports, exactly once, but - // from a pool task: running it inside SendAsync would deadlock a caller - // that holds a lock its own callback takes. - static void FailSend(std::function onSent) { + // never from inside SendAsync, which would deadlock a caller holding a + // lock its own callback takes: after the callback that made it, when one + // is being delivered, else from a pool task. + static void FailSend(const std::shared_ptr& self, std::function onSent) { if (!onSent) return; - ThreadPool::Enqueue([onSent = std::move(onSent)] { onSent(false); }); + if (self) { + std::lock_guard lk(self->mtx); + if (self->delivering) { + self->completions.emplace_back(std::move(onSent), false); + return; + } + } + ThreadPool::Enqueue([self, onSent = std::move(onSent)]() mutable { + if (!self) { + onSent(false); + return; + } + std::unique_lock lk(self->mtx); + self->Complete(std::move(onSent), false, lk); + }); + } + + // Stop()'s half: onData and onEnd never run once this returns, and one + // running on another thread has finished. One running on this thread is + // the caller, a callback destroying its own wrapper; it cannot be waited + // for, and the delivering loop lets the handlers go when it unwinds. + void Unregister() { + std::function)> dataHandler; + std::function endHandler; + std::unique_lock lk(mtx); + unregistered = true; + if (!handlerSet) return; + if (delivering && deliveringThread != std::this_thread::get_id()) { + deliveryIdle.wait(lk, [&] { return !delivering; }); + } + if (!delivering) { + dataHandler.swap(onData); + endHandler.swap(onEnd); + } + lk.unlock(); } // Up to `count` received bytes, waiting for them; fewer once the peer @@ -402,21 +541,54 @@ struct QUICStream::Impl { Impl* self = keepAlive.get(); switch (ev->Type) { case QUIC_STREAM_EVENT_RECEIVE: { - std::vector chunk; - std::uint64_t total = 0; - for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) { - total += ev->RECEIVE.Buffers[i].Length; + // Everything is consumed here (SUCCESS), whoever ends up with + // it: a pending receive would hold the stream's shutdown back + // until a StreamReceiveComplete that nothing would ever call. + std::unique_lock lk(self->mtx); + // What Abort() has not stopped yet is dropped, and so is + // whatever comes once nothing will hand it over. + if (self->aborted + || (self->handlerSet && (self->unregistered || self->endDelivered))) { + return QUIC_STATUS_SUCCESS; } - chunk.reserve(static_cast(total)); + if (!self->handlerSet || self->delivering || !self->pending.empty()) { + // For a Recieve* call, or behind what another thread is + // still handing over: a copy, in order. + std::vector chunk; + std::uint64_t total = 0; + for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) { + total += ev->RECEIVE.Buffers[i].Length; + } + chunk.reserve(static_cast(total)); + for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) { + const QUIC_BUFFER& b = ev->RECEIVE.Buffers[i]; + chunk.insert(chunk.end(), b.Buffer, b.Buffer + b.Length); + } + if (!chunk.empty()) self->pending.push_back(std::move(chunk)); + if (!self->handlerSet) { + lk.unlock(); + self->cv.notify_all(); + return QUIC_STATUS_SUCCESS; + } + self->DeliverIfIdle(lk); + return QUIC_STATUS_SUCCESS; + } + // Nothing waiting and nobody delivering: hand msquic's own + // buffers over, valid until this event returns, without a copy. + self->BeginDelivery(); + lk.unlock(); for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) { const QUIC_BUFFER& b = ev->RECEIVE.Buffers[i]; - chunk.insert(chunk.end(), b.Buffer, b.Buffer + b.Length); + if (b.Length == 0) continue; + if (i > 0) { + // The last onData may have aborted or dropped the stream. + std::lock_guard check(self->mtx); + if (self->aborted || self->unregistered) break; + } + self->onData(std::span(reinterpret_cast(b.Buffer), b.Length)); } - { - std::lock_guard lk(self->mtx); - if (!chunk.empty()) self->pending.push_back(std::move(chunk)); - } - self->cv.notify_all(); + lk.lock(); + self->DeliverReady(lk); return QUIC_STATUS_SUCCESS; } case QUIC_STREAM_EVENT_SEND_COMPLETE: { @@ -435,27 +607,44 @@ struct QUICStream::Impl { if (request->outcome) self->cv.notify_all(); std::function onSent = std::move(request->onSent); delete request; - // SendAsync's, here on the worker: msquic is done with the - // copy, and no thread was held while it had it. - if (onSent) onSent(sent); + // SendAsync's, on the worker, in turn with the stream's other + // callbacks: msquic is done with the copy, and no thread was + // held while it had it. + if (onSent) { + std::unique_lock lk(self->mtx); + self->Complete(std::move(onSent), sent, lk); + } return QUIC_STATUS_SUCCESS; } case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN: case QUIC_STREAM_EVENT_PEER_SEND_ABORTED: { + // PEER_SEND_SHUTDOWN comes once every byte up to the FIN has + // been taken from RECEIVE events, so onEnd(true) follows the + // last onData. { - std::lock_guard lk(self->mtx); + std::unique_lock lk(self->mtx); self->peerSendClosed = true; + if (self->end == End::Open) { + self->end = ev->Type == QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN ? End::Clean : End::Unclean; + } + self->DeliverIfIdle(lk); } self->cv.notify_all(); return QUIC_STATUS_SUCCESS; } case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE: { { - std::lock_guard lk(self->mtx); + std::unique_lock lk(self->mtx); self->peerSendClosed = true; // Every send msquic took has had its SEND_COMPLETE by now; // a SendSync still waiting is only woken to find out. self->shutdownComplete = true; + // Neither a FIN nor the peer's reset ended the receive + // side: the connection went, or this side aborted it. + if (self->end == End::Open) self->end = End::Unclean; + // Before Retire, so the handle is still valid for the + // callbacks, as in every other event. + self->DeliverIfIdle(lk); } self->cv.notify_all(); self->Retire(); @@ -511,6 +700,8 @@ void QUICStream::Stop() { if (!handle) return; handle = nullptr; // the wrapper is done with the stream either way if (!impl) return; + // The owner is done with OnRecieve's callbacks too. + impl->Unregister(); // Only INITIATE the shutdown here; the async SHUTDOWN_COMPLETE callback // releases the handle, and the Impl is a shared_ptr held by the callback's // selfRef, so it safely outlives this wrapper until that callback runs. @@ -607,7 +798,7 @@ void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish, // callback, and a stopped stream refuses or cancels the send anyway. HQUIC live = impl && canSend ? impl->AcquireHandle() : nullptr; if (live == nullptr) { - Impl::FailSend(std::move(onSent)); + Impl::FailSend(impl, std::move(onSent)); return; } auto* request = new Impl::SendRequest(buffer, size); @@ -618,7 +809,7 @@ void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish, if (!taken) { std::function callback = std::move(request->onSent); delete request; - Impl::FailSend(std::move(callback)); + Impl::FailSend(impl, std::move(callback)); } } @@ -662,13 +853,65 @@ void QUICStream::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function bytes) { if (bytes.empty() || !impl) return; - { - std::lock_guard lk(impl->mtx); - impl->pending.push_front(std::move(bytes)); + std::unique_lock lk(impl->mtx); + impl->pending.push_front(std::move(bytes)); + if (impl->handlerSet && !impl->unregistered) { + // Handed to onData next, by this thread or the one delivering. + impl->DeliverIfIdle(lk); + return; } + lk.unlock(); impl->cv.notify_all(); } +void QUICStream::OnRecieve(std::function data)> onData, + std::function onEnd) { + // Our own reference: the callbacks delivered below may destroy this + // wrapper, and nothing of `this` is touched once they can run. + const std::shared_ptr self = impl; + if (!self || !handle) { + if (onEnd) onEnd(false); + return; + } + std::unique_lock lk(self->mtx); + if (self->handlerSet) { + throw QUICException("OnRecieve: this stream's callbacks are already set"); + } + self->handlerSet = true; + self->onData = onData ? std::move(onData) : [](std::span) {}; + self->onEnd = onEnd ? std::move(onEnd) : [](bool) {}; + // A stream only we send on gets no data and no FIN to end cleanly with. + if (!canReceive && self->end == Impl::End::Open) self->end = Impl::End::Unclean; + // What arrived before now, and the end if it came: delivered here, unless + // another thread is delivering this stream's callbacks (a send's onSent), + // which then hands these over after its own. + self->DeliverIfIdle(lk); +} + +void QUICStream::Abort(std::uint64_t errorCode) { + const std::shared_ptr self = impl; + if (!self) return; + { + std::lock_guard lk(self->mtx); + self->aborted = true; + self->pending.clear(); + if (self->end == Impl::End::Open) self->end = Impl::End::Unclean; + } + // Inside a callback, the delivering loop hands onEnd(false) over when the + // callback returns; otherwise SHUTDOWN_COMPLETE does, soon: IMMEDIATE + // makes msquic report it as soon as it processes the abort, instead of + // once the peer has acknowledged the reset, which a peer that is gone + // never does. + HQUIC live = self->AcquireHandle(); + if (live == nullptr) return; + Runtime().api->StreamShutdown(live, + QUIC_STREAM_SHUTDOWN_FLAG_ABORT_SEND + | QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE + | QUIC_STREAM_SHUTDOWN_FLAG_IMMEDIATE, + std::min(errorCode, QUIC_UINT62_MAX)); + self->ReleaseHandle(); +} + std::uint64_t QUICStream::GetStreamId() const { if (!handle) throw QUICException("GetStreamId: stream is not open"); // GetParam blocks on the connection's msquic worker, so it must not be the diff --git a/interfaces/Crafter.Network-ClientQUIC.cppm b/interfaces/Crafter.Network-ClientQUIC.cppm index 12948b6..cf1b89a 100644 --- a/interfaces/Crafter.Network-ClientQUIC.cppm +++ b/interfaces/Crafter.Network-ClientQUIC.cppm @@ -150,9 +150,11 @@ namespace Crafter { // thread, inside this stream's own callbacks included, and as often // as you like: the sends queue in msquic in call order, none waits for // another, and BytesInFlight says how far behind they are. - // - onSent runs on the connection's msquic worker thread. A send - // that fails before msquic takes it (a closed stream, a moved-from - // wrapper) reports from a pool task instead, never from inside + // - onSent runs on the connection's msquic worker thread, never at + // the same time as another callback of this stream (see + // OnRecieve). A send that fails before msquic takes it (a closed + // stream, a moved-from wrapper) reports from a pool task, or + // after the callback that made it returns, never from inside // SendAsync, so the caller may hold a lock onSent takes. // - In onSent: SendAsync and the other non-blocking calls are fine; // nothing that blocks, no *Sync call (it waits for this very @@ -191,6 +193,47 @@ namespace Crafter { // to a follower can drop one whose backlog passes a cap. Any thread. std::uint64_t BytesInFlight() const; std::size_t SendsInFlight() const; + + // Receive by callback, holding no thread while nothing arrives. + // onData gets each run of bytes as it arrives, as a span valid only + // for the call (copy what you keep); onEnd fires once, after the last + // onData: clean=true when the peer finished its side (FIN), false + // when either side aborted the stream (Abort(), the peer's reset) or + // the connection went first. + // - What arrived before the call is delivered first, on the calling + // thread before OnRecieve returns, and so is onEnd if the stream + // had already ended; from then on the callbacks run on the + // connection's msquic worker thread. No two callbacks of one + // stream (onData, onEnd, SendAsync's onSent) ever run at once. + // - Inside them: SendAsync, Abort, BytesInFlight, SendsInFlight, + // and destroying or moving this wrapper are fine. Nothing that + // blocks: no *Sync call on any stream (it waits for this very + // worker), no ~ClientQUIC (it waits for its streams to drain), no + // waiting on a thread that may be waiting on this one. A callback + // that throws ends the program. + // - Destroying the wrapper, or Stop(), unregisters both: once that + // returns neither runs again, a callback running on another + // thread has finished (as with std::stop_callback; one on this + // thread, destroying its own wrapper, is not waited for), and an + // onEnd not delivered by then never is. So do not destroy it + // while holding a lock its callbacks take. Sends still complete. + // - Once per stream: a second call throws QUICException. Do not mix + // with the Recieve* calls, which then get nothing. On a stream + // that cannot receive, or a closed or moved-from wrapper, onEnd + // (false) fires at once. + void OnRecieve(std::function data)> onData, + std::function onEnd); + + // Reset both directions at once with a QUIC application error code + // (62 bits; larger values are clamped): the peer's receive side sees + // a reset and its send side a stop-sending. Returns at once. Sends + // not yet delivered report false, no onData follows (not even the + // rest of a run being delivered when called from inside one), and + // onEnd(false) follows unless the end was delivered already or the + // wrapper is destroyed first. Callable from any thread and from + // inside this stream's callbacks, which is how a server enforces a + // byte cap. A no-op on a closed or moved-from wrapper. + void Abort(std::uint64_t errorCode = 0); #endif // Shut the stream down in both directions: our send side gracefully @@ -200,6 +243,8 @@ namespace Crafter { // releases its transport handle. Returns immediately; completion is // asynchronous. Called by the destructor, so an explicit call is only // needed to shut a stream down ahead of its wrapper going away. + // Natively it also unregisters OnRecieve's callbacks (see there); + // sends already made still go out, ahead of the FIN. void Stop(); private: