Catcrafts/Crafter.Network
Watch
2
0
Fork
You've already forked Crafter.Network
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(quic): QUICStream::OnRecieve and Abort, callbacks that hold no thread

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Jorijn van der Graaf 2026-11-01 02:32:48 +01:00
commit c5b4e18eb8
Signed by: jorijnvdgraaf
GPG key ID: 2937E59CDCC1BCFB
2 changed files with 316 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

293
implementations/Crafter.Network-ClientQUIC.cpp
View file

@ -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<void(std::span<const char>)> onData;
std::function<void(bool)> 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<std::pair<std::function<void(bool)>, 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<std::mutex>& lk) {
for (;;) {
if (!completions.empty()) {
std::function<void(bool)> 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<char> chunk = std::move(pending.front());
pending.pop_front();
lk.unlock();
onData(std::span<const char>(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<void(std::span<const char>)> dataHandler;
std::function<void(bool)> 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<std::mutex>& lk) {
if (BeginDelivery()) {
DeliverReady(lk);
} else {
lk.unlock();
}
}
// Queues onSent(sent) behind the stream's other callbacks.
void Complete(std::function<void(bool)> onSent, bool sent, std::unique_lock<std::mutex>& 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<void(bool sent)> 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<Impl>& self, std::function<void(bool sent)> 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<void(std::span<const char>)> dataHandler;
std::function<void(bool)> 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<char> 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<std::size_t>(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<char> 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<std::size_t>(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<const char>(reinterpret_cast<const char*>(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<void(bool)> 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<void(bool)> 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<v
void QUICStream::PrependReceived(std::vector<char> 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<void(std::span<const char> data)> onData,
std::function<void(bool clean)> 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<Impl> 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<const char>) {};
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<Impl> 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<std::uint64_t>(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