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: