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

fix(quic): SendSync throws only when its own send failed

Browse source 
SendSync waited on one sendInFlight flag per stream and, once woken,
threw QUICClosedException whenever the stream had finished shutting
down by then, even though its own send had completed. A peer that
answers at once and finishes the stream (echo, FIN, drop it, its Stop()
also stopping our sending side) completes the stream within a round
trip, which on loopback is shorter than a busy machine can take to wake
the sending thread, so SendSync threw for a send the peer had already
answered. The shared flag also let two SendSync calls on one stream wake
on each other's completion.

Each send now carries its own request: the copy msquic reads, the
QUIC_BUFFER describing it (msquic keeps a pointer to that, not a copy,
which the old code satisfied only because the waiter's stack outlived
the send), and an outcome shared with the waiter. SEND_COMPLETE, which
msquic delivers exactly once for every send it took and always before
the stream's SHUTDOWN_COMPLETE, records whether it was Canceled and
frees the request; SendSync returns unless its own send was canceled or
never completed.

Measured with ShouldReturnFromSendSyncQUIC, 2000 exchanges on fresh
connections four at a time: before, 1 or 2 spurious throws in 6 of 7
runs, the echo of each failed send arriving intact; after, none in 9
runs. It surfaced as one lost echo in a hundred in the OnClosed test
being written.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jorijn van der Graaf 2026-11-01 02:16:18 +01:00
commit b1b0633a27
Signed by: jorijnvdgraaf
GPG key ID: 2937E59CDCC1BCFB
1 changed files with 53 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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<char> bytes;
QUIC_BUFFER buffer{};
std::shared_ptr<SendOutcome> outcome;
SendRequest(const void* data, std::uint32_t size)
: bytes(static_cast<const char*>(data), static_cast<const char*>(data) + size) {
buffer.Buffer = reinterpret_cast<std::uint8_t*>(bytes.data());
buffer.Length = size;
}
};
}
// ---------------- QUICStream::Impl ----------------
@ -136,7 +162,6 @@ struct QUICStream::Impl {
std::deque<std::vector<char>> 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<SendRequest*>(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<char*>(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<std::uint8_t*>(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<SendOutcome>();
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<unsigned>(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<char> QUICStream::RecieveSync() {