diff --git a/implementations/Crafter.Network-ClientQUIC.cpp b/implementations/Crafter.Network-ClientQUIC.cpp index b00f323..5ba6e2e 100644 --- a/implementations/Crafter.Network-ClientQUIC.cpp +++ b/implementations/Crafter.Network-ClientQUIC.cpp @@ -125,6 +125,32 @@ namespace { b.Buffer = out.data(); return b; } + + // 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. + struct SendOutcome { + bool done = false; + bool sent = false; + }; + + // One send handed to msquic: the copy it reads from, and the QUIC_BUFFER + // describing it, which msquic keeps a pointer to rather than copying. + // Owned by msquic from a successful StreamSend until that send's + // 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. + struct SendRequest { + std::vector bytes; + QUIC_BUFFER buffer{}; + std::shared_ptr outcome; + + SendRequest(const void* data, std::uint32_t size) + : bytes(static_cast(data), static_cast(data) + size) { + buffer.Buffer = reinterpret_cast(bytes.data()); + buffer.Length = size; + } + }; } // ---------------- QUICStream::Impl ---------------- @@ -136,7 +162,6 @@ struct QUICStream::Impl { std::deque> pending; bool peerSendClosed = false; bool shutdownComplete = false; - bool sendInFlight = false; // Up to `count` received bytes, waiting for them; fewer once the peer // has closed its side and the rest is taken. @@ -299,14 +324,17 @@ struct QUICStream::Impl { return QUIC_STATUS_SUCCESS; } case QUIC_STREAM_EVENT_SEND_COMPLETE: { - { - std::lock_guard lk(self->mtx); - self->sendInFlight = false; + auto* request = static_cast(ev->SEND_COMPLETE.ClientContext); + if (request == nullptr) return QUIC_STATUS_SUCCESS; + if (request->outcome) { + { + std::lock_guard lk(self->mtx); + request->outcome->done = true; + request->outcome->sent = !ev->SEND_COMPLETE.Canceled; + } + self->cv.notify_all(); } - if (ev->SEND_COMPLETE.ClientContext) { - delete[] static_cast(ev->SEND_COMPLETE.ClientContext); - } - self->cv.notify_all(); + delete request; return QUIC_STATUS_SUCCESS; } case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN: @@ -322,10 +350,9 @@ struct QUICStream::Impl { { std::lock_guard 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; - // No further SEND_COMPLETE can arrive, so release anyone - // waiting on one — they throw on shutdownComplete instead. - self->sendInFlight = false; } self->cv.notify_all(); self->Retire(); @@ -407,31 +434,27 @@ void QUICStream::SendSync(const void* buffer, std::uint32_t size, bool finish) { // call below, and msquic bugchecks on the freed handle. HQUIC live = impl->AcquireHandle(); if (!live) throw QUICClosedException(); - auto* copy = new char[size]; - std::memcpy(copy, buffer, size); - QUIC_BUFFER quicBuf{}; - quicBuf.Buffer = reinterpret_cast(copy); - quicBuf.Length = size; - { - std::lock_guard lk(impl->mtx); - impl->sendInFlight = true; - } + // This send's own outcome, not the stream's state. The flag this used to + // wait on was one per stream, and the wait then threw whenever the + // 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); + request->outcome = outcome; QUIC_SEND_FLAGS flags = finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE; - QUIC_STATUS s = Runtime().api->StreamSend(live, &quicBuf, 1, flags, copy); + QUIC_STATUS s = Runtime().api->StreamSend(live, &request->buffer, 1, flags, request); impl->ReleaseHandle(); if (QUIC_FAILED(s)) { - { - // No SEND_COMPLETE will arrive for a send msquic never took. - std::lock_guard lk(impl->mtx); - impl->sendInFlight = false; - } - impl->cv.notify_all(); - delete[] copy; + // No SEND_COMPLETE will arrive for a send msquic never took. + delete request; throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast(s))); } std::unique_lock lk(impl->mtx); - impl->cv.wait(lk, [&]{ return !impl->sendInFlight || impl->shutdownComplete; }); - if (impl->shutdownComplete) throw QUICClosedException(); + // SEND_COMPLETE always precedes the stream's SHUTDOWN_COMPLETE; waking on + // the latter only keeps an msquic surprise from hanging this thread. + impl->cv.wait(lk, [&]{ return outcome->done || impl->shutdownComplete; }); + if (!outcome->done || !outcome->sent) throw QUICClosedException(); } std::vector QUICStream::RecieveSync() {