diff --git a/implementations/Crafter.Network-ClientQUIC.cpp b/implementations/Crafter.Network-ClientQUIC.cpp index a57d857..886f18a 100644 --- a/implementations/Crafter.Network-ClientQUIC.cpp +++ b/implementations/Crafter.Network-ClientQUIC.cpp @@ -186,7 +186,10 @@ namespace { b.Buffer = out.data(); return b; } +} +// ---------------- QUICStream::Impl ---------------- +struct QUICStream::Impl { // How one SendSync went, written by its SEND_COMPLETE under the stream's // mutex. Shared with the waiting thread so the request can be freed by // whichever side finishes with it, in either order. @@ -201,10 +204,13 @@ namespace { // SEND_COMPLETE, which msquic delivers exactly once for every send it // took (Canceled when the stream or the connection ended first) and // always before the stream's SHUTDOWN_COMPLETE; that handler frees it. + // Who hears how it went: SendSync's waiting thread through `outcome`, + // or SendAsync's `onSent`. struct SendRequest { std::vector bytes; QUIC_BUFFER buffer{}; std::shared_ptr outcome; + std::function onSent; SendRequest(const void* data, std::uint32_t size) : bytes(static_cast(data), static_cast(data) + size) { @@ -212,10 +218,7 @@ namespace { buffer.Length = size; } }; -} -// ---------------- QUICStream::Impl ---------------- -struct QUICStream::Impl { ClientQUIC* connection = nullptr; std::mutex mtx; @@ -224,6 +227,38 @@ struct QUICStream::Impl { bool peerSendClosed = false; bool shutdownComplete = false; + // Sends handed to msquic whose SEND_COMPLETE has not come yet, under mtx. + std::uint64_t bytesInFlight = 0; + std::size_t sendsInFlight = 0; + + // StreamSend for one request, counted in flight while msquic has it. + // False, with the request still the caller's, when msquic refused it: + // then no SEND_COMPLETE will come. + bool Send(HQUIC live, SendRequest* request, bool finish, QUIC_STATUS& status) { + { + std::lock_guard lk(mtx); + bytesInFlight += request->bytes.size(); + ++sendsInFlight; + } + // Counted first: SEND_COMPLETE can run on the worker before + // StreamSend has returned here. + status = Runtime().api->StreamSend(live, &request->buffer, 1, + finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE, request); + if (QUIC_SUCCEEDED(status)) return true; + std::lock_guard lk(mtx); + bytesInFlight -= request->bytes.size(); + --sendsInFlight; + return false; + } + + // 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) { + if (!onSent) return; + ThreadPool::Enqueue([onSent = std::move(onSent)] { onSent(false); }); + } + // Up to `count` received bytes, waiting for them; fewer once the peer // has closed its side and the rest is taken. std::vector TakeUpTo(std::uint32_t count) { @@ -387,15 +422,22 @@ struct QUICStream::Impl { case QUIC_STREAM_EVENT_SEND_COMPLETE: { auto* request = static_cast(ev->SEND_COMPLETE.ClientContext); if (request == nullptr) return QUIC_STATUS_SUCCESS; - if (request->outcome) { - { - std::lock_guard lk(self->mtx); + const bool sent = !ev->SEND_COMPLETE.Canceled; + { + std::lock_guard lk(self->mtx); + self->bytesInFlight -= request->bytes.size(); + --self->sendsInFlight; + if (request->outcome) { request->outcome->done = true; - request->outcome->sent = !ev->SEND_COMPLETE.Canceled; + request->outcome->sent = sent; } - self->cv.notify_all(); } + 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); return QUIC_STATUS_SUCCESS; } case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN: @@ -500,13 +542,13 @@ void QUICStream::SendSync(const void* buffer, std::uint32_t size, bool finish) { // stream had finished shutting down by the time it woke: a peer that // answered and finished the stream within a round trip, quicker than a // busy machine wakes this thread, made a send that went out throw. - auto outcome = std::make_shared(); - auto* request = new SendRequest(buffer, size); + auto outcome = std::make_shared(); + auto* request = new Impl::SendRequest(buffer, size); request->outcome = outcome; - QUIC_SEND_FLAGS flags = finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE; - QUIC_STATUS s = Runtime().api->StreamSend(live, &request->buffer, 1, flags, request); + QUIC_STATUS s = QUIC_STATUS_SUCCESS; + const bool taken = impl->Send(live, request, finish, s); impl->ReleaseHandle(); - if (QUIC_FAILED(s)) { + if (!taken) { // No SEND_COMPLETE will arrive for a send msquic never took. delete request; throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast(s))); @@ -555,19 +597,41 @@ std::vector QUICStream::RecieveUntilFullSync(std::uint32_t bufferSize) { void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish, std::function onSent) { - // Copy now: the caller's buffer may not outlive the enqueued task. - std::vector copy(static_cast(buffer), - static_cast(buffer) + size); - ThreadPool::Enqueue([this, copy = std::move(copy), finish, onSent = std::move(onSent)]() mutable { - // Nothing may leave a pool task; the callback hears how it went. - bool sent = true; - try { - this->SendSync(copy.data(), static_cast(copy.size()), finish); - } catch (...) { - sent = false; - } - if (onSent) onSent(sent); - }); + // This used to be SendSync run as a pool task: a pool thread held for as + // long as msquic took to accept the buffer (a peer that reads slowly + // stretches that to its pace), and a task that reached back through + // `this`, so moving or destroying the wrapper before it ran was a + // use-after-free. Now the send goes to msquic here and its SEND_COMPLETE + // reports, on the worker. Only `impl` is read, not `handle`, which Stop() + // clears: SendAsync may run on another thread than the owner's, inside a + // 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)); + return; + } + auto* request = new Impl::SendRequest(buffer, size); + request->onSent = std::move(onSent); + QUIC_STATUS status = QUIC_STATUS_SUCCESS; + const bool taken = impl->Send(live, request, finish, status); + impl->ReleaseHandle(); + if (!taken) { + std::function callback = std::move(request->onSent); + delete request; + Impl::FailSend(std::move(callback)); + } +} + +std::uint64_t QUICStream::BytesInFlight() const { + if (!impl) return 0; + std::lock_guard lk(impl->mtx); + return impl->bytesInFlight; +} + +std::size_t QUICStream::SendsInFlight() const { + if (!impl) return 0; + std::lock_guard lk(impl->mtx); + return impl->sendsInFlight; } // The async receives deliver a closed stream as the browser build does: an diff --git a/interfaces/Crafter.Network-ClientQUIC.cppm b/interfaces/Crafter.Network-ClientQUIC.cppm index 5c837c7..12948b6 100644 --- a/interfaces/Crafter.Network-ClientQUIC.cppm +++ b/interfaces/Crafter.Network-ClientQUIC.cppm @@ -117,7 +117,9 @@ namespace Crafter { #ifndef CRAFTER_NETWORK_BROWSER // Send a buffer. If finish=true, the send-side of the stream is closed // after the buffer is delivered (peer will see graceful shutdown). - // Blocks until msquic accepts the buffer; throws on stream/conn close. + // Blocks until msquic accepts the buffer; throws QUICClosedException + // when the stream or connection closed before it did. SendAsync is + // the same without the thread. void SendSync(const void* buffer, std::uint32_t size, bool finish = false); // Block until at least one byte is received; returns the received @@ -138,6 +140,25 @@ namespace Crafter { // accepted the buffer (native) or the WritableStream writer has // resolved (browser), and with false when the stream was closed or // the send failed. Available on both native and browser builds. + // + // Native: holds no thread. It copies the buffer, hands the copy to + // msquic and returns; msquic's SEND_COMPLETE for it frees the copy and + // fires onSent, exactly once: true when msquic took the whole buffer + // (it is in msquic's send buffer or acknowledged, what SendSync's + // return means; not a receipt from the peer's application), false + // when the stream or the connection closed first. Callable from any + // 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 + // 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 + // worker), no ~ClientQUIC; it must not throw. + // - It fires even after the wrapper is gone: destroying it lets the + // sends already made go out, then the FIN. void SendAsync(const void* buffer, std::uint32_t size, bool finish, std::function onSent); @@ -164,6 +185,12 @@ namespace Crafter { // so the server has to query and remember it at session creation // time. Throws if the stream is not yet started. std::uint64_t GetStreamId() const; + + // Bytes, and sends, handed to SendAsync or SendSync whose completion + // has not come yet: how far behind a slow peer is. A server streaming + // to a follower can drop one whose backlog passes a cap. Any thread. + std::uint64_t BytesInFlight() const; + std::size_t SendsInFlight() const; #endif // Shut the stream down in both directions: our send side gracefully