Skip to content

Commit 4d06d80

Browse files
jasnelltargos
jasnell
authored and
targos
committed
quic: various additional cleanups, fixes in Endpoint
PR-URL: #51310 Reviewed-By: Yagiz Nizipli <[email protected]>
1 parent 17c554f commit 4d06d80

25 files changed

+2108
-481
lines changed

‎node.gyp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -356,6 +356,7 @@
356356
'src/quic/cid.cc',
357357
'src/quic/data.cc',
358358
'src/quic/endpoint.cc',
359+
'src/quic/http3.cc',
359360
'src/quic/logstream.cc',
360361
'src/quic/packet.cc',
361362
'src/quic/preferredaddress.cc',
@@ -370,6 +371,7 @@
370371
'src/quic/cid.h',
371372
'src/quic/data.h',
372373
'src/quic/endpoint.h',
374+
'src/quic/http3.h',
373375
'src/quic/logstream.h',
374376
'src/quic/packet.h',
375377
'src/quic/preferredaddress.h',

‎src/debug_utils.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,8 @@ void NODE_EXTERN_PRIVATE FWrite(FILE* file, const std::string& str);
5252
V(WASI) \
5353
V(MKSNAPSHOT) \
5454
V(SNAPSHOT_SERDES) \
55-
V(PERMISSION_MODEL)
55+
V(PERMISSION_MODEL) \
56+
V(QUIC)
5657

5758
enum class DebugCategory : unsigned int {
5859
#define V(name) name,

‎src/quic/application.cc‎

Lines changed: 91 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,15 @@
11
#if HAVE_OPENSSL && NODE_OPENSSL_HAS_QUIC
22

33
#include "application.h"
4+
#include <async_wrap-inl.h>
5+
#include <debug_utils-inl.h>
46
#include <node_bob.h>
57
#include <node_sockaddr-inl.h>
68
#include <uv.h>
79
#include <v8.h>
810
#include "defs.h"
911
#include "endpoint.h"
12+
#include "http3.h"
1013
#include "packet.h"
1114
#include "session.h"
1215

@@ -21,24 +24,48 @@ using v8::Value;
2124

2225
namespace quic {
2326

24-
struct Session::Application::StreamData final {
25-
// The actual number of vectors in the struct, up to kMaxVectorCount.
26-
size_t count = 0;
27-
size_t remaining = 0;
28-
// The stream identifier. If this is a negative value then no stream is
29-
// identified.
30-
int64_t id = -1;
31-
int fin = 0;
32-
ngtcp2_vec data[kMaxVectorCount]{};
33-
ngtcp2_vec* buf = data;
34-
BaseObjectPtr<Stream> stream;
35-
};
36-
27+
// ============================================================================
28+
// Session::Application_Options
3729
const Session::Application_Options Session::Application_Options::kDefault = {};
3830

31+
Session::Application_Options::operator const nghttp3_settings() const {
32+
// In theory, Application_Options might contain options for more than just
33+
// HTTP/3. Here we extract only the properties that are relevant to HTTP/3.
34+
return nghttp3_settings{
35+
max_field_section_size,
36+
static_cast<size_t>(qpack_max_dtable_capacity),
37+
static_cast<size_t>(qpack_encoder_max_dtable_capacity),
38+
static_cast<size_t>(qpack_blocked_streams),
39+
enable_connect_protocol,
40+
enable_datagrams,
41+
};
42+
}
43+
44+
std::string Session::Application_Options::ToString() const {
45+
DebugIndentScope indent;
46+
auto prefix = indent.Prefix();
47+
std::string res("{");
48+
res += prefix + "max header pairs: " + std::to_string(max_header_pairs);
49+
res += prefix + "max header length: " + std::to_string(max_header_length);
50+
res += prefix +
51+
"max field section size: " + std::to_string(max_field_section_size);
52+
res += prefix + "qpack max dtable capacity: " +
53+
std::to_string(qpack_max_dtable_capacity);
54+
res += prefix + "qpack encoder max dtable capacity: " +
55+
std::to_string(qpack_encoder_max_dtable_capacity);
56+
res += prefix +
57+
"qpack blocked streams: " + std::to_string(qpack_blocked_streams);
58+
res += prefix + "enable connect protocol: " +
59+
(enable_connect_protocol ? std::string("yes") : std::string("no"));
60+
res += prefix + "enable datagrams: " +
61+
(enable_datagrams ? std::string("yes") : std::string("no"));
62+
res += indent.Close();
63+
return res;
64+
}
65+
3966
Maybe<Session::Application_Options> Session::Application_Options::From(
4067
Environment* env, Local<Value> value) {
41-
if (value.IsEmpty()) {
68+
if (value.IsEmpty() || (!value->IsUndefined() && !value->IsObject())) {
4269
THROW_ERR_INVALID_ARG_TYPE(env, "options must be an object");
4370
return Nothing<Application_Options>();
4471
}
@@ -49,11 +76,6 @@ Maybe<Session::Application_Options> Session::Application_Options::From(
4976
return Just<Application_Options>(options);
5077
}
5178

52-
if (!value->IsObject()) {
53-
THROW_ERR_INVALID_ARG_TYPE(env, "options must be an object");
54-
return Nothing<Application_Options>();
55-
}
56-
5779
auto params = value.As<Object>();
5880

5981
#define SET(name) \
@@ -63,7 +85,8 @@ Maybe<Session::Application_Options> Session::Application_Options::From(
6385

6486
if (!SET(max_header_pairs) || !SET(max_header_length) ||
6587
!SET(max_field_section_size) || !SET(qpack_max_dtable_capacity) ||
66-
!SET(qpack_encoder_max_dtable_capacity) || !SET(qpack_blocked_streams)) {
88+
!SET(qpack_encoder_max_dtable_capacity) || !SET(qpack_blocked_streams) ||
89+
!SET(enable_connect_protocol) || !SET(enable_datagrams)) {
6790
return Nothing<Application_Options>();
6891
}
6992

@@ -78,16 +101,22 @@ Session::Application::Application(Session* session, const Options& options)
78101
bool Session::Application::Start() {
79102
// By default there is nothing to do. Specific implementations may
80103
// override to perform more actions.
104+
Debug(session_, "Session application started");
81105
return true;
82106
}
83107

84108
void Session::Application::AcknowledgeStreamData(Stream* stream,
85109
size_t datalen) {
110+
Debug(session_,
111+
"Application acknowledging stream %" PRIi64 " data: %zu",
112+
stream->id(),
113+
datalen);
86114
DCHECK_NOT_NULL(stream);
87115
stream->Acknowledge(datalen);
88116
}
89117

90118
void Session::Application::BlockStream(int64_t id) {
119+
Debug(session_, "Application blocking stream %" PRIi64, id);
91120
auto stream = session().FindStream(id);
92121
if (stream) stream->EmitBlocked();
93122
}
@@ -96,6 +125,7 @@ bool Session::Application::CanAddHeader(size_t current_count,
96125
size_t current_headers_length,
97126
size_t this_header_length) {
98127
// By default headers are not supported.
128+
Debug(session_, "Application cannot add header");
99129
return false;
100130
}
101131

@@ -104,33 +134,39 @@ bool Session::Application::SendHeaders(const Stream& stream,
104134
const v8::Local<v8::Array>& headers,
105135
HeadersFlags flags) {
106136
// By default do nothing.
137+
Debug(session_, "Application cannot send headers");
107138
return false;
108139
}
109140

110141
void Session::Application::ResumeStream(int64_t id) {
142+
Debug(session_, "Application resuming stream %" PRIi64, id);
111143
// By default do nothing.
112144
}
113145

114146
void Session::Application::ExtendMaxStreams(EndpointLabel label,
115147
Direction direction,
116148
uint64_t max_streams) {
149+
Debug(session_, "Application extending max streams");
117150
// By default do nothing.
118151
}
119152

120153
void Session::Application::ExtendMaxStreamData(Stream* stream,
121154
uint64_t max_data) {
155+
Debug(session_, "Application extending max stream data");
122156
// By default do nothing.
123157
}
124158

125159
void Session::Application::CollectSessionTicketAppData(
126160
SessionTicket::AppData* app_data) const {
161+
Debug(session_, "Application collecting session ticket app data");
127162
// By default do nothing.
128163
}
129164

130165
SessionTicket::AppData::Status
131166
Session::Application::ExtractSessionTicketAppData(
132167
const SessionTicket::AppData& app_data,
133168
SessionTicket::AppData::Source::Flag flag) {
169+
Debug(session_, "Application extracting session ticket app data");
134170
// By default we do not have any application data to retrieve.
135171
return flag == SessionTicket::AppData::Source::Flag::STATUS_RENEW
136172
? SessionTicket::AppData::Status::TICKET_USE_RENEW
@@ -140,14 +176,16 @@ Session::Application::ExtractSessionTicketAppData(
140176
void Session::Application::SetStreamPriority(const Stream& stream,
141177
StreamPriority priority,
142178
StreamPriorityFlags flags) {
179+
Debug(
180+
session_, "Application setting stream %" PRIi64 " priority", stream.id());
143181
// By default do nothing.
144182
}
145183

146184
StreamPriority Session::Application::GetStreamPriority(const Stream& stream) {
147185
return StreamPriority::DEFAULT;
148186
}
149187

150-
BaseObjectPtr<Packet> Session::Application::CreateStreamDataPacket() {
188+
Packet* Session::Application::CreateStreamDataPacket() {
151189
return Packet::Create(env(),
152190
session_->endpoint_.get(),
153191
session_->remote_address_,
@@ -156,24 +194,37 @@ BaseObjectPtr<Packet> Session::Application::CreateStreamDataPacket() {
156194
}
157195

158196
void Session::Application::StreamClose(Stream* stream, QuicError error) {
197+
Debug(session_,
198+
"Application closing stream %" PRIi64 " with error %s",
199+
stream->id(),
200+
error);
159201
stream->Destroy(error);
160202
}
161203

162204
void Session::Application::StreamStopSending(Stream* stream, QuicError error) {
205+
Debug(session_,
206+
"Application stopping sending on stream %" PRIi64 " with error %s",
207+
stream->id(),
208+
error);
163209
DCHECK_NOT_NULL(stream);
164210
stream->ReceiveStopSending(error);
165211
}
166212

167213
void Session::Application::StreamReset(Stream* stream,
168214
uint64_t final_size,
169215
QuicError error) {
216+
Debug(session_,
217+
"Application resetting stream %" PRIi64 " with error %s",
218+
stream->id(),
219+
error);
170220
stream->ReceiveStreamReset(final_size, error);
171221
}
172222

173223
void Session::Application::SendPendingData() {
224+
Debug(session_, "Application sending pending data");
174225
PathStorage path;
175226

176-
BaseObjectPtr<Packet> packet;
227+
Packet* packet = nullptr;
177228
uint8_t* pos = nullptr;
178229
int err = 0;
179230

@@ -182,6 +233,7 @@ void Session::Application::SendPendingData() {
182233
size_t packetSendCount = 0;
183234

184235
const auto updateTimer = [&] {
236+
Debug(session_, "Application updating the session timer");
185237
ngtcp2_conn_update_pkt_tx_time(*session_, uv_hrtime());
186238
session_->UpdateTimer();
187239
};
@@ -209,9 +261,9 @@ void Session::Application::SendPendingData() {
209261
return session_->Close(Session::CloseMethod::SILENT);
210262
}
211263

212-
if (!packet) {
264+
if (packet == nullptr) {
213265
packet = CreateStreamDataPacket();
214-
if (!packet) {
266+
if (packet == nullptr) {
215267
session_->last_error_ = QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL);
216268
return session_->Close(Session::CloseMethod::SILENT);
217269
}
@@ -319,12 +371,14 @@ class DefaultApplication final : public Session::Application {
319371
const uint8_t* data,
320372
size_t datalen,
321373
Stream::ReceiveDataFlags flags) override {
374+
Debug(&session(), "Default application receiving stream data");
322375
DCHECK_NOT_NULL(stream);
323376
if (!stream->is_destroyed()) stream->ReceiveData(data, datalen, flags);
324377
return true;
325378
}
326379

327380
int GetStreamData(StreamData* stream_data) override {
381+
Debug(&session(), "Default application getting stream data");
328382
DCHECK_NOT_NULL(stream_data);
329383
// If the queue is empty, there aren't any streams with data yet
330384
if (stream_queue_.IsEmpty()) return 0;
@@ -380,7 +434,10 @@ class DefaultApplication final : public Session::Application {
380434
return 0;
381435
}
382436

383-
void ResumeStream(int64_t id) override { ScheduleStream(id); }
437+
void ResumeStream(int64_t id) override {
438+
Debug(&session(), "Default application resuming stream %" PRIi64, id);
439+
ScheduleStream(id);
440+
}
384441

385442
bool ShouldSetFin(const StreamData& stream_data) override {
386443
auto const is_empty = [](auto vec, size_t cnt) {
@@ -394,6 +451,7 @@ class DefaultApplication final : public Session::Application {
394451
}
395452

396453
bool StreamCommit(StreamData* stream_data, size_t datalen) override {
454+
Debug(&session(), "Default application committing stream data");
397455
DCHECK_NOT_NULL(stream_data);
398456
const auto consume = [](ngtcp2_vec** pvec, size_t* pcnt, size_t len) {
399457
ngtcp2_vec* v = *pvec;
@@ -425,13 +483,15 @@ class DefaultApplication final : public Session::Application {
425483

426484
private:
427485
void ScheduleStream(int64_t id) {
486+
Debug(&session(), "Default application scheduling stream %" PRIi64, id);
428487
auto stream = session().FindStream(id);
429488
if (stream && !stream->is_destroyed()) {
430489
stream->Schedule(&stream_queue_);
431490
}
432491
}
433492

434493
void UnscheduleStream(int64_t id) {
494+
Debug(&session(), "Default application unscheduling stream %" PRIi64, id);
435495
auto stream = session().FindStream(id);
436496
if (stream && !stream->is_destroyed()) stream->Unschedule();
437497
}
@@ -440,13 +500,15 @@ class DefaultApplication final : public Session::Application {
440500
};
441501

442502
std::unique_ptr<Session::Application> Session::select_application() {
443-
// if (config.options.crypto_options.alpn == NGHTTP3_ALPN_H3)
444-
// return std::make_unique<Http3>(session,
445-
// config.options.application_options);
446-
447503
// In the future, we may end up supporting additional QUIC protocols. As they
448504
// are added, extend the cases here to create and return them.
449505

506+
if (config_.options.tls_options.alpn == NGHTTP3_ALPN_H3) {
507+
Debug(this, "Selecting HTTP/3 application");
508+
return createHttp3Application(this, config_.options.application_options);
509+
}
510+
511+
Debug(this, "Selecting default application");
450512
return std::make_unique<DefaultApplication>(
451513
this, config_.options.application_options);
452514
}

‎src/quic/application.h‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,8 +118,9 @@ class Session::Application : public MemoryRetainer {
118118
protected:
119119
inline Environment* env() const { return session_->env(); }
120120
inline Session& session() { return *session_; }
121+
inline const Session& session() const { return *session_; }
121122

122-
BaseObjectPtr<Packet> CreateStreamDataPacket();
123+
Packet* CreateStreamDataPacket();
123124

124125
struct StreamData;
125126

@@ -137,6 +138,21 @@ class Session::Application : public MemoryRetainer {
137138
Session* session_;
138139
};
139140

141+
struct Session::Application::StreamData final {
142+
// The actual number of vectors in the struct, up to kMaxVectorCount.
143+
size_t count = 0;
144+
size_t remaining = 0;
145+
// The stream identifier. If this is a negative value then no stream is
146+
// identified.
147+
int64_t id = -1;
148+
int fin = 0;
149+
ngtcp2_vec data[kMaxVectorCount]{};
150+
ngtcp2_vec* buf = data;
151+
BaseObjectPtr<Stream> stream;
152+
153+
inline operator nghttp3_vec() const { return {data[0].base, data[0].len}; }
154+
};
155+
140156
} // namespace quic
141157
} // namespace node
142158

0 commit comments

Comments
 (0)