Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion src/quic/application.cc
Original file line number Diff line number Diff line change
Expand Up @@ -419,7 +419,7 @@ class DefaultApplication final : public Session::Application {
void BlockStream(stream_id id) override {
if (auto stream = session().FindStream(id)) [[likely]] {
// Remove the stream from the send queue. It will be re-scheduled
// via ExtendMaxStreamData when the peer grants more flow control.
// via ExtendMax(Stream)Data when the peer grants more flow control.
// Without this, SendPendingData would repeatedly pop and retry
// the same blocked stream in an infinite loop.
stream->Unschedule();
Expand All @@ -434,6 +434,15 @@ class DefaultApplication final : public Session::Application {
stream->Schedule(&stream_queue_);
}

void ExtendMaxData(uint64_t max_data) override {
// The peer granted more flow control for session. Re-schedule
// all streams so SendPendingData will resume writing.
for (auto& [id, stream] : session().streams()) {
stream->Schedule(&stream_queue_);
}
}


bool StreamCommit(Session::StreamData* stream_data, size_t datalen) override {
DCHECK_NOT_NULL(stream_data);
CHECK(stream_data->stream);
Expand Down
8 changes: 8 additions & 0 deletions src/quic/application.h
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,14 @@ class Session::Application : public MemoryRetainer {
// By default do nothing.
}

// Called when the Session determines that the flow control window for the
// session has been expanded. Not all Application types will require
// this notification so the default is to do nothing.
virtual void ExtendMaxData(uint64_t max_data) {
Debug(session_, "Application extending max data");
// By default do nothing.
}

// Different Applications may wish to set some application data in the
// session ticket (e.g. http/3 would set server settings in the application
// data). The first byte written MUST be the Application::Type enum value.
Expand Down
10 changes: 10 additions & 0 deletions src/quic/http3.cc
Original file line number Diff line number Diff line change
Expand Up @@ -378,6 +378,16 @@ class Http3ApplicationImpl final : public Session::Application {
nghttp3_conn_unblock_stream(*this, stream->id());
}

void ExtendMaxData(uint64_t max_data) override {
Debug(&session(),
"HTTP/3 application extending max data to %" PRIu64,
max_data);
for (auto& [id, stream] : session().streams()) {
stream->UpdateWriteDesiredSize(); // the stream might be blocked on js side
// is unblock stream also required?
}
}

void CollectSessionTicketAppData(
SessionTicket::AppData* app_data) const override {
uint8_t buf[kSessionTicketAppDataSize];
Expand Down
14 changes: 14 additions & 0 deletions src/quic/session.cc
Original file line number Diff line number Diff line change
Expand Up @@ -1358,6 +1358,14 @@ struct Session::Impl final : public MemoryRetainer {
return NGTCP2_SUCCESS;
}

static int on_extend_max_data(ngtcp2_conn* conn,
uint64_t max_data,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->application().ExtendMaxData(max_data);
return NGTCP2_SUCCESS;
}

static int on_get_new_cid(ngtcp2_conn* conn,
ngtcp2_cid* cid,
ngtcp2_stateless_reset_token* token,
Expand Down Expand Up @@ -1700,6 +1708,9 @@ struct Session::Impl final : public MemoryRetainer {
on_receive_stream_stop_sending,
#ifdef NGTCP2_CALLBACKS_V5
nullptr,
#ifdef NGTCP2_CALLBACKS_V6
on_extend_max_data,
#endif
#endif // NGTCP2_CALLBACKS_V5
#endif // NGTCP2_CALLBACKS_V4
};
Expand Down Expand Up @@ -1754,6 +1765,9 @@ struct Session::Impl final : public MemoryRetainer {
on_receive_stream_stop_sending,
#ifdef NGTCP2_CALLBACKS_V5
nullptr,
#ifdef NGTCP2_CALLBACKS_V6
on_extend_max_data,
#endif
#endif // NGTCP2_CALLBACKS_V5
#endif // NGTCP2_CALLBACKS_V4
};
Expand Down
76 changes: 76 additions & 0 deletions test/parallel/test-quic-h3-maxdata-external-buffer-failure.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
// Flags: --experimental-quic --experimental-stream-iter --no-warnings

// Test: Quic maxdata updates on http/3
// Client sends a body that precisely fills the session window size,
// and verifies that it is data transfer is not stalled.

import { hasQuic, skip } from '../common/index.mjs';
import { readFile } from 'node:fs/promises';
import { setTimeout as sleep } from 'node:timers/promises';

if (!hasQuic) {
skip('QUIC is not enabled');
}
const { listen, connect } = await import('node:quic');
const { createPrivateKey } = await import('node:crypto');
const { drainableProtocol } = await import('stream/iter');

const keys = 'test/fixtures/keys';
const key = createPrivateKey(await readFile(`${keys}/agent1-key.pem`));
const cert = await readFile(`${keys}/agent1-cert.pem`);

const WINDOW = 4096;
// Fills the window exactly:
// considers all framing including some initial session capsules
const BODY = WINDOW - 38;

let letServerRead;
const serverMayRead = new Promise((resolve) => { letServerRead = resolve; });

const endpoint = await listen((session) => {
session.onstream = async (stream) => {
await serverMayRead;
// eslint-disable-next-line no-unused-vars
for await (const _ of stream) { /* reading extends the window */ }
};
}, {
sni: { '*': { keys: [key], certs: [cert] } },
transportParams: {
initialMaxStreamDataBidiRemote: 1024 * 1024, // Make sure maxstreamdata does not block
initialMaxData: WINDOW,
},
onheaders() { this.sendHeaders({ ':status': '200' }); },
});

const session = await connect(endpoint.address, {
servername: 'localhost',
verifyPeer: 'manual',
});
await session.opened;

// Budget well above the window, so the window is what stops the writer.
const stream = await session.createBidirectionalStream({ budget: 1024 * 1024 });
stream.sendHeaders({
':method': 'POST',
':path': '/',
':scheme': 'https',
':authority': 'localhost',
}, { terminal: false });

const writer = stream.writer;
writer.writeSync(new Uint8Array(BODY));

// Long enough for every byte to be acked. The peer acks as data arrives,
// whether or not its application has read any of it, so by now the window is
// exhausted, the send buffer is empty, and no further ACK can arrive.
await sleep(500);

const watchdog = setTimeout(() => {
console.error('STALLED: no drain after MAX_STREAM_DATA');
process.exit(1);
}, 5000);
letServerRead(); // Extend the window, with no ack attached
await writer[drainableProtocol]();

clearTimeout(watchdog);
process.exit(0);
Loading