feat(quic): QUICStream::SendAsync holds no thread and reports from SEND_COMPLETE
Native SendAsync was SendSync run as a pool task. That held a worker of Crafter.Thread's pool, which the host shares and which runs each worker's queue in order, for as long as msquic took to accept the buffer, and a peer that reads slowly stretches that to its own pace: a server streaming to a slow follower stalled the tasks queued behind it. The task also reached back through `this`, so moving or destroying the wrapper before it ran was a use-after-free, and two sends could reach msquic out of order from two workers. Now SendAsync copies the buffer into a SendRequest, hands it to msquic and returns. SEND_COMPLETE, which msquic delivers exactly once for every send it took and before the stream's SHUTDOWN_COMPLETE, frees the copy and calls onSent on the msquic worker: true when msquic took the whole buffer (in its send buffer or acknowledged, which is what SendSync's return has always meant), false when it was canceled because the stream or the connection ended first. Sends queue in msquic in call order and none waits for another. A send msquic never took (a closed stream, a moved-from wrapper, StreamSend refusing it) reports false from a pool task, never from inside SendAsync, so a caller may hold a lock its own callback takes. Only `impl` is read, not `handle`, so SendAsync is callable from any thread, a callback included; on a stopped stream msquic cancels the send. BytesInFlight() and SendsInFlight() count what SendAsync and SendSync handed msquic whose SEND_COMPLETE has not come, so a server can drop a follower that falls too far behind. Rewritten rather than added beside it: nothing in this repository calls the native SendAsync (ListenerHTTP, WebTransport and ClientHTTP use SendSync; the example's SendAsync is the browser build's), nor does any consumer found here (3DForts uses SendSync natively); its contract is unchanged, copy now and onSent once with whether it went; and the old one's pool thread, `this` capture and reordering were defects, not behaviour to keep. What changes is onSent's thread, msquic's worker rather than a pool task, so a callback that blocks, a *Sync call above all, now stalls its connection rather than a pool worker. The browser build is untouched. Breaking: native onSent runs on msquic's worker thread, not as a pool task, and must not block. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
3be8642d32
commit
1b7417c2ba
2 changed files with 118 additions and 27 deletions
|
|
@ -186,7 +186,10 @@ namespace {
|
||||||
b.Buffer = out.data();
|
b.Buffer = out.data();
|
||||||
return b;
|
return b;
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------- QUICStream::Impl ----------------
|
||||||
|
struct QUICStream::Impl {
|
||||||
// How one SendSync went, written by its SEND_COMPLETE under the stream's
|
// 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
|
// mutex. Shared with the waiting thread so the request can be freed by
|
||||||
// whichever side finishes with it, in either order.
|
// whichever side finishes with it, in either order.
|
||||||
|
|
@ -201,10 +204,13 @@ namespace {
|
||||||
// SEND_COMPLETE, which msquic delivers exactly once for every send it
|
// SEND_COMPLETE, which msquic delivers exactly once for every send it
|
||||||
// took (Canceled when the stream or the connection ended first) and
|
// took (Canceled when the stream or the connection ended first) and
|
||||||
// always before the stream's SHUTDOWN_COMPLETE; that handler frees it.
|
// 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 {
|
struct SendRequest {
|
||||||
std::vector<char> bytes;
|
std::vector<char> bytes;
|
||||||
QUIC_BUFFER buffer{};
|
QUIC_BUFFER buffer{};
|
||||||
std::shared_ptr<SendOutcome> outcome;
|
std::shared_ptr<SendOutcome> outcome;
|
||||||
|
std::function<void(bool sent)> onSent;
|
||||||
|
|
||||||
SendRequest(const void* data, std::uint32_t size)
|
SendRequest(const void* data, std::uint32_t size)
|
||||||
: bytes(static_cast<const char*>(data), static_cast<const char*>(data) + size) {
|
: bytes(static_cast<const char*>(data), static_cast<const char*>(data) + size) {
|
||||||
|
|
@ -212,10 +218,7 @@ namespace {
|
||||||
buffer.Length = size;
|
buffer.Length = size;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
}
|
|
||||||
|
|
||||||
// ---------------- QUICStream::Impl ----------------
|
|
||||||
struct QUICStream::Impl {
|
|
||||||
ClientQUIC* connection = nullptr;
|
ClientQUIC* connection = nullptr;
|
||||||
|
|
||||||
std::mutex mtx;
|
std::mutex mtx;
|
||||||
|
|
@ -224,6 +227,38 @@ struct QUICStream::Impl {
|
||||||
bool peerSendClosed = false;
|
bool peerSendClosed = false;
|
||||||
bool shutdownComplete = 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<void(bool sent)> onSent) {
|
||||||
|
if (!onSent) return;
|
||||||
|
ThreadPool::Enqueue([onSent = std::move(onSent)] { onSent(false); });
|
||||||
|
}
|
||||||
|
|
||||||
// Up to `count` received bytes, waiting for them; fewer once the peer
|
// Up to `count` received bytes, waiting for them; fewer once the peer
|
||||||
// has closed its side and the rest is taken.
|
// has closed its side and the rest is taken.
|
||||||
std::vector<char> TakeUpTo(std::uint32_t count) {
|
std::vector<char> TakeUpTo(std::uint32_t count) {
|
||||||
|
|
@ -387,15 +422,22 @@ struct QUICStream::Impl {
|
||||||
case QUIC_STREAM_EVENT_SEND_COMPLETE: {
|
case QUIC_STREAM_EVENT_SEND_COMPLETE: {
|
||||||
auto* request = static_cast<SendRequest*>(ev->SEND_COMPLETE.ClientContext);
|
auto* request = static_cast<SendRequest*>(ev->SEND_COMPLETE.ClientContext);
|
||||||
if (request == nullptr) return QUIC_STATUS_SUCCESS;
|
if (request == nullptr) return QUIC_STATUS_SUCCESS;
|
||||||
if (request->outcome) {
|
const bool sent = !ev->SEND_COMPLETE.Canceled;
|
||||||
{
|
{
|
||||||
std::lock_guard lk(self->mtx);
|
std::lock_guard lk(self->mtx);
|
||||||
|
self->bytesInFlight -= request->bytes.size();
|
||||||
|
--self->sendsInFlight;
|
||||||
|
if (request->outcome) {
|
||||||
request->outcome->done = true;
|
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<void(bool)> onSent = std::move(request->onSent);
|
||||||
delete request;
|
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;
|
return QUIC_STATUS_SUCCESS;
|
||||||
}
|
}
|
||||||
case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN:
|
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
|
// 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
|
// answered and finished the stream within a round trip, quicker than a
|
||||||
// busy machine wakes this thread, made a send that went out throw.
|
// busy machine wakes this thread, made a send that went out throw.
|
||||||
auto outcome = std::make_shared<SendOutcome>();
|
auto outcome = std::make_shared<Impl::SendOutcome>();
|
||||||
auto* request = new SendRequest(buffer, size);
|
auto* request = new Impl::SendRequest(buffer, size);
|
||||||
request->outcome = outcome;
|
request->outcome = outcome;
|
||||||
QUIC_SEND_FLAGS flags = finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE;
|
QUIC_STATUS s = QUIC_STATUS_SUCCESS;
|
||||||
QUIC_STATUS s = Runtime().api->StreamSend(live, &request->buffer, 1, flags, request);
|
const bool taken = impl->Send(live, request, finish, s);
|
||||||
impl->ReleaseHandle();
|
impl->ReleaseHandle();
|
||||||
if (QUIC_FAILED(s)) {
|
if (!taken) {
|
||||||
// No SEND_COMPLETE will arrive for a send msquic never took.
|
// No SEND_COMPLETE will arrive for a send msquic never took.
|
||||||
delete request;
|
delete request;
|
||||||
throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast<unsigned>(s)));
|
throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast<unsigned>(s)));
|
||||||
|
|
@ -555,19 +597,41 @@ std::vector<char> QUICStream::RecieveUntilFullSync(std::uint32_t bufferSize) {
|
||||||
|
|
||||||
void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish,
|
void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish,
|
||||||
std::function<void(bool sent)> onSent) {
|
std::function<void(bool sent)> onSent) {
|
||||||
// Copy now: the caller's buffer may not outlive the enqueued task.
|
// This used to be SendSync run as a pool task: a pool thread held for as
|
||||||
std::vector<char> copy(static_cast<const char*>(buffer),
|
// long as msquic took to accept the buffer (a peer that reads slowly
|
||||||
static_cast<const char*>(buffer) + size);
|
// stretches that to its pace), and a task that reached back through
|
||||||
ThreadPool::Enqueue([this, copy = std::move(copy), finish, onSent = std::move(onSent)]() mutable {
|
// `this`, so moving or destroying the wrapper before it ran was a
|
||||||
// Nothing may leave a pool task; the callback hears how it went.
|
// use-after-free. Now the send goes to msquic here and its SEND_COMPLETE
|
||||||
bool sent = true;
|
// reports, on the worker. Only `impl` is read, not `handle`, which Stop()
|
||||||
try {
|
// clears: SendAsync may run on another thread than the owner's, inside a
|
||||||
this->SendSync(copy.data(), static_cast<std::uint32_t>(copy.size()), finish);
|
// callback, and a stopped stream refuses or cancels the send anyway.
|
||||||
} catch (...) {
|
HQUIC live = impl && canSend ? impl->AcquireHandle() : nullptr;
|
||||||
sent = false;
|
if (live == nullptr) {
|
||||||
}
|
Impl::FailSend(std::move(onSent));
|
||||||
if (onSent) onSent(sent);
|
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<void(bool)> 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
|
// The async receives deliver a closed stream as the browser build does: an
|
||||||
|
|
|
||||||
|
|
@ -117,7 +117,9 @@ namespace Crafter {
|
||||||
#ifndef CRAFTER_NETWORK_BROWSER
|
#ifndef CRAFTER_NETWORK_BROWSER
|
||||||
// Send a buffer. If finish=true, the send-side of the stream is closed
|
// Send a buffer. If finish=true, the send-side of the stream is closed
|
||||||
// after the buffer is delivered (peer will see graceful shutdown).
|
// 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);
|
void SendSync(const void* buffer, std::uint32_t size, bool finish = false);
|
||||||
|
|
||||||
// Block until at least one byte is received; returns the received
|
// 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
|
// accepted the buffer (native) or the WritableStream writer has
|
||||||
// resolved (browser), and with false when the stream was closed or
|
// resolved (browser), and with false when the stream was closed or
|
||||||
// the send failed. Available on both native and browser builds.
|
// 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,
|
void SendAsync(const void* buffer, std::uint32_t size, bool finish,
|
||||||
std::function<void(bool sent)> onSent);
|
std::function<void(bool sent)> onSent);
|
||||||
|
|
||||||
|
|
@ -164,6 +185,12 @@ namespace Crafter {
|
||||||
// so the server has to query and remember it at session creation
|
// so the server has to query and remember it at session creation
|
||||||
// time. Throws if the stream is not yet started.
|
// time. Throws if the stream is not yet started.
|
||||||
std::uint64_t GetStreamId() const;
|
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
|
#endif
|
||||||
|
|
||||||
// Shut the stream down in both directions: our send side gracefully
|
// Shut the stream down in both directions: our send side gracefully
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue