FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

quic: eliminate per-received datagram allocation · nodejs/node@098e3d7 · GitHub

/ node Public

Commit 098e3d7

Browse files
jasnell
authored and
aduh95
committed
quic: eliminate per-received datagram allocation
Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode:Opus 4.6 PR-URL: #63267 Backport-PR-URL: #64675 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 53b05e2 commit 098e3d7

5 files changed

Lines changed: 68 additions & 68 deletions

File tree

‎src/quic/application.cc‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -448,9 +448,13 @@ void Session::Application::SendPendingData() {
448448

449449
// Awesome, let's write our packet!
450450
PacketInfo pi;
451-
ssize_t nwrite = WriteVStream(
452-
&path, &pi, packet->data(), &ndatalen, packet->length(),
453-
stream_data, ts);
451+
ssize_t nwrite = WriteVStream(&path,
452+
&pi,
453+
packet->data(),
454+
&ndatalen,
455+
packet->length(),
456+
stream_data,
457+
ts);
454458

455459
// When ndatalen is > 0, that's our indication that stream data was accepted
456460
// in to the packet. Yay!

‎src/quic/endpoint.cc‎

Lines changed: 43 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -311,10 +311,18 @@ class Endpoint::UDP::Impl final : public HandleWrap {
311311
SET_SELF_SIZE(Impl)
312312

313313
private:
314+
// Pre-allocated receive buffer. Reused across all datagrams because
315+
// ngtcp2_conn_read_pkt is synchronous — it copies what it needs and
316+
// does not retain a reference to the buffer after returning. This
317+
// eliminates a malloc(64KB)/free(64KB) cycle per received datagram.
318+
static constexpr size_t kRecvBufferSize = 65536; // UV__UDP_DGRAM_MAXSIZE
319+
char recv_buf_[kRecvBufferSize];
320+
314321
static void OnAlloc(uv_handle_t* handle,
315322
size_t suggested_size,
316323
uv_buf_t* buf) {
317-
*buf = From(handle)->env()->allocate_managed_buffer(suggested_size);
324+
auto* impl = From(handle);
325+
*buf = uv_buf_init(impl->recv_buf_, kRecvBufferSize);
318326
}
319327

320328
static void OnReceive(uv_udp_t* handle,
@@ -326,26 +334,22 @@ class Endpoint::UDP::Impl final : public HandleWrap {
326334
DCHECK_NOT_NULL(impl);
327335
DCHECK_NOT_NULL(impl->endpoint_);
328336

329-
auto release_buf = [&]() {
330-
if (buf->base != nullptr) impl->env()->release_managed_buffer(*buf);
331-
};
332-
333337
// Nothing to do in these cases. Specifically, if the nread
334338
// is zero or we have received a partial packet, we are just
335-
// going to ignore it.
339+
// going to ignore it. No buffer release needed — recv_buf_
340+
// is pre-allocated and reused.
336341
if (nread == 0 || flags & UV_UDP_PARTIAL) {
337-
release_buf();
338342
return;
339343
}
340344

341345
if (nread < 0) {
342-
release_buf();
343346
impl->endpoint_->Destroy(CloseContext::RECEIVE_FAILURE,
344347
static_cast<int>(nread));
345348
return;
346349
}
347350

348-
impl->endpoint_->Receive(uv_buf_init(buf->base, static_cast<size_t>(nread)),
351+
impl->endpoint_->Receive(reinterpret_cast<const uint8_t*>(buf->base),
352+
static_cast<size_t>(nread),
349353
SocketAddress(addr));
350354
}
351355

@@ -1264,24 +1268,25 @@ void Endpoint::CloseGracefully() {
12641268
MaybeDestroy();
12651269
}
12661270

1267-
void Endpoint::Receive(const uv_buf_t& buf,
1271+
void Endpoint::Receive(const uint8_t* data,
1272+
size_t len,
12681273
const SocketAddress& remote_address) {
12691274
const auto receive = [&](Session* session,
1270-
Store&& store,
1275+
const uint8_t* pkt_data,
1276+
size_t pkt_len,
12711277
const SocketAddress& local_address,
12721278
const SocketAddress& remote_address,
12731279
const CID& dcid,
12741280
const CID& scid) {
12751281
DCHECK_NOT_NULL(session);
12761282
if (session->is_destroyed()) return;
1277-
size_t len = store.length();
12781283
// Use ReadPacket (no SendPendingDataScope) so that multiple packets
12791284
// received in the same I/O burst are processed before any responses
12801285
// are generated. The deferred flush via BindingData's uv_check
12811286
// callback calls SendPendingData once per dirty session after all
12821287
// packets in the burst have been read.
1283-
if (session->ReadPacket(std::move(store), local_address, remote_address)) {
1284-
STAT_INCREMENT_N(Stats, bytes_received, len);
1288+
if (session->ReadPacket(pkt_data, pkt_len, local_address, remote_address)) {
1289+
STAT_INCREMENT_N(Stats, bytes_received, pkt_len);
12851290
STAT_INCREMENT(Stats, packets_received);
12861291
}
12871292
// Schedule the session for deferred SendPendingData if it hasn't
@@ -1293,7 +1298,9 @@ void Endpoint::Receive(const uv_buf_t& buf,
12931298
}
12941299
};
12951300

1296-
const auto accept = [&](const Session::Config& config, Store&& store) {
1301+
const auto accept = [&](const Session::Config& config,
1302+
const uint8_t* pkt_data,
1303+
size_t pkt_len) {
12971304
// One final check. If the endpoint is closed, closing, or is not listening
12981305
// as a server, then we cannot accept the initial packet.
12991306
if (is_closed() || is_closing() || !is_listening()) return;
@@ -1323,7 +1330,8 @@ void Endpoint::Receive(const uv_buf_t& buf,
13231330
return;
13241331

13251332
receive(session.get(),
1326-
std::move(store),
1333+
pkt_data,
1334+
pkt_len,
13271335
config.local_address,
13281336
config.remote_address,
13291337
config.dcid,
@@ -1333,7 +1341,8 @@ void Endpoint::Receive(const uv_buf_t& buf,
13331341
const auto acceptInitialPacket = [&](const uint32_t version,
13341342
const CID& dcid,
13351343
const CID& scid,
1336-
Store&& store,
1344+
const uint8_t* pkt_data,
1345+
size_t pkt_len,
13371346
const SocketAddress& local_address,
13381347
const SocketAddress& remote_address) {
13391348
// If we're not listening as a server, do not accept an initial packet.
@@ -1343,8 +1352,7 @@ void Endpoint::Receive(const uv_buf_t& buf,
13431352

13441353
// This is our first condition check... A minimal check to see if ngtcp2 can
13451354
// even recognize this packet as a quic packet.
1346-
ngtcp2_vec vec = store;
1347-
if (ngtcp2_accept(&hd, vec.base, vec.len) != NGTCP2_SUCCESS) {
1355+
if (ngtcp2_accept(&hd, pkt_data, pkt_len) != NGTCP2_SUCCESS) {
13481356
// Per the ngtcp2 docs, ngtcp2_accept returns 0 if the check was
13491357
// successful, or an error code if it was not. Currently there's only one
13501358
// documented error code (NGTCP2_ERR_INVALID_ARGUMENT) but we'll handle
@@ -1582,7 +1590,7 @@ void Endpoint::Receive(const uv_buf_t& buf,
15821590
}
15831591
}
15841592

1585-
accept(config, std::move(store));
1593+
accept(config, pkt_data, pkt_len);
15861594
};
15871595

15881596
// When a received packet contains a QUIC short header but cannot be matched
@@ -1598,35 +1606,37 @@ void Endpoint::Receive(const uv_buf_t& buf,
15981606
// possible to avoid a DOS vector.
15991607
const auto maybeStatelessReset = [&](const CID& dcid,
16001608
const CID& scid,
1601-
Store& store,
1609+
const uint8_t* pkt_data,
1610+
size_t pkt_len,
16021611
const SocketAddress& local_address,
16031612
const SocketAddress& remote_address) {
16041613
// Support for stateless resets can be disabled by the application. If that
16051614
// case, or if the packet is too short to contain a reset token, then we
16061615
// skip the remaining checks.
16071616
if (options_.disable_stateless_reset ||
1608-
store.length() < NGTCP2_STATELESS_RESET_TOKENLEN) {
1617+
pkt_len < NGTCP2_STATELESS_RESET_TOKENLEN) {
16091618
return false;
16101619
}
16111620

16121621
// The stateless reset token itself is the *final*
16131622
// NGTCP2_STATELESS_RESET_TOKENLEN bytes in the received packet. If it is a
16141623
// stateless reset then then rest of the bytes in the packet are garbage
16151624
// that we'll ignore.
1616-
ngtcp2_vec vec = store;
1617-
vec.base += (vec.len - NGTCP2_STATELESS_RESET_TOKENLEN);
1625+
const uint8_t* token_pos =
1626+
pkt_data + (pkt_len - NGTCP2_STATELESS_RESET_TOKENLEN);
16181627

16191628
// If a Session has been associated with the token, then it is a valid
16201629
// stateless reset token. We need to dispatch it to the session to be
16211630
// processed.
16221631
auto* session = session_manager().FindSessionByStatelessResetToken(
1623-
StatelessResetToken(vec.base));
1632+
StatelessResetToken(token_pos));
16241633
if (session != nullptr) {
16251634
// If the session happens to have been destroyed already, we'll
16261635
// just ignore the packet.
16271636
if (!session->is_destroyed()) [[likely]] {
16281637
receive(session,
1629-
std::move(store),
1638+
pkt_data,
1639+
pkt_len,
16301640
local_address,
16311641
remote_address,
16321642
dcid,
@@ -1654,22 +1664,8 @@ void Endpoint::Receive(const uv_buf_t& buf,
16541664
// return;
16551665
// }
16561666

1657-
Debug(this, "Received %zu-byte packet from %s", buf.len, remote_address);
1658-
1659-
// The managed buffer here contains the received packet. We do not yet know
1660-
// at this point if it is a valid QUIC packet. We need to do some basic
1661-
// checks. It is critical at this point that we do as little work as possible
1662-
// to avoid a DOS vector.
1663-
std::shared_ptr<BackingStore> backing = env()->release_managed_buffer(buf);
1664-
if (!backing) [[unlikely]] {
1665-
// At this point something bad happened and we need to treat this as a fatal
1666-
// case. There's likely no way to test this specific condition reliably.
1667-
return Destroy(CloseContext::RECEIVE_FAILURE, UV_ENOMEM);
1668-
}
1669-
1670-
Store store(std::move(backing), buf.len, 0);
1667+
Debug(this, "Received %zu-byte packet from %s", len, remote_address);
16711668

1672-
ngtcp2_vec vec = store;
16731669
ngtcp2_version_cid pversion_cid;
16741670

16751671
// This is our first check to see if the received data can be processed as a
@@ -1678,7 +1674,7 @@ void Endpoint::Receive(const uv_buf_t& buf,
16781674
// valid QUIC header but there is still no guarantee that the packet can be
16791675
// successfully processed.
16801676
switch (ngtcp2_pkt_decode_version_cid(
1681-
&pversion_cid, vec.base, vec.len, NGTCP2_MAX_CIDLEN)) {
1677+
&pversion_cid, data, len, NGTCP2_MAX_CIDLEN)) {
16821678
case 0:
16831679
break; // Supported version, continue processing.
16841680
case NGTCP2_ERR_VERSION_NEGOTIATION: {
@@ -1756,7 +1752,7 @@ void Endpoint::Receive(const uv_buf_t& buf,
17561752
// necessary here. We want to return immediately without committing any
17571753
// further resources.
17581754
if (pversion_cid.version == 0 &&
1759-
maybeStatelessReset(dcid, scid, store, addr, remote_address)) {
1755+
maybeStatelessReset(dcid, scid, data, len, addr, remote_address)) {
17601756
Debug(this, "Packet was a stateless reset");
17611757
return; // Stateless reset! Don't do any further processing.
17621758
}
@@ -1771,17 +1767,13 @@ void Endpoint::Receive(const uv_buf_t& buf,
17711767
SendStatelessReset(
17721768
PathDescriptor{
17731769
pversion_cid.version, dcid, scid, addr, remote_address},
1774-
store.length());
1770+
len);
17751771
return;
17761772
}
17771773

17781774
// Process the packet as an initial packet...
1779-
return acceptInitialPacket(pversion_cid.version,
1780-
dcid,
1781-
scid,
1782-
std::move(store),
1783-
addr,
1784-
remote_address);
1775+
return acceptInitialPacket(
1776+
pversion_cid.version, dcid, scid, data, len, addr, remote_address);
17851777
}
17861778

17871779
if (session->is_destroyed()) [[unlikely]] {
@@ -1793,7 +1785,7 @@ void Endpoint::Receive(const uv_buf_t& buf,
17931785
// If we got here, the dcid matched the scid of a known local session. Yay!
17941786
// The session will take over any further processing of the packet.
17951787
Debug(this, "Dispatching packet to known session");
1796-
receive(session.get(), std::move(store), addr, remote_address, dcid, scid);
1788+
receive(session.get(), data, len, addr, remote_address, dcid, scid);
17971789

17981790
// It is important to note that the session may have been destroyed during
17991791
// the call to receive(...). If that's the case, the session object still

‎src/quic/endpoint.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -429,7 +429,7 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
429429
// Ref() causes a listening Endpoint to keep the event loop active.
430430
JS_METHOD(Ref);
431431

432-
void Receive(const uv_buf_t& buf, const SocketAddress& from);
432+
void Receive(const uint8_t* data, size_t len, const SocketAddress& from);
433433

434434
AliasedStruct<Stats> stats_;
435435
AliasedStruct<State> state_;

‎src/quic/session.cc‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2109,7 +2109,8 @@ void Session::SetLastError(QuicError&& error) {
21092109
impl_->last_error_ = std::move(error);
21102110
}
21112111

2112-
bool Session::Receive(Store&& store,
2112+
bool Session::Receive(const uint8_t* data,
2113+
size_t len,
21132114
const SocketAddress& local_address,
21142115
const SocketAddress& remote_address,
21152116
const PacketInfo& pkt_info,
@@ -2120,24 +2121,23 @@ bool Session::Receive(Store&& store,
21202121
// The hot receive path uses ReadPacket() directly with deferred
21212122
// flush via BindingData's uv_check callback.
21222123
SendPendingDataScope send_scope(this);
2123-
return ReadPacket(
2124-
std::move(store), local_address, remote_address, pkt_info, ts);
2124+
return ReadPacket(data, len, local_address, remote_address, pkt_info, ts);
21252125
}
21262126

2127-
bool Session::ReadPacket(Store&& store,
2127+
bool Session::ReadPacket(const uint8_t* data,
2128+
size_t len,
21282129
const SocketAddress& local_address,
21292130
const SocketAddress& remote_address,
21302131
const PacketInfo& pkt_info,
21312132
uint64_t ts) {
21322133
DCHECK(!is_destroyed());
21332134
impl_->remote_address_ = remote_address;
21342135

2135-
ngtcp2_vec vec = store;
21362136
Path path(local_address, remote_address);
21372137

21382138
Debug(this,
21392139
"Session is receiving %zu-byte packet received along path %s",
2140-
vec.len,
2140+
len,
21412141
path);
21422142

21432143
// It is important to understand that reading the packet will cause
@@ -2158,19 +2158,18 @@ bool Session::ReadPacket(Store&& store,
21582158
// receive path caches a timestamp and passes it to all ReadPacket()
21592159
// calls in the same I/O burst.
21602160
if (ts == 0) ts = uv_hrtime();
2161-
err = ngtcp2_conn_read_pkt(
2162-
*this, &path, pkt_info, vec.base, vec.len, ts);
2161+
err = ngtcp2_conn_read_pkt(*this, &path, pkt_info, data, len, ts);
21632162
}
21642163
if (is_destroyed()) return false;
21652164

2166-
Debug(this, "Session receiving %zu-byte packet with result %d", vec.len, err);
2165+
Debug(this, "Session receiving %zu-byte packet with result %d", len, err);
21672166

21682167
switch (err) {
21692168
case 0: {
2170-
Debug(this, "Session successfully received %zu-byte packet", vec.len);
2169+
Debug(this, "Session successfully received %zu-byte packet", len);
21712170
if (!is_destroyed()) [[likely]] {
21722171
auto& stats_ = impl_->stats_;
2173-
STAT_INCREMENT_N(Stats, bytes_received, vec.len);
2172+
STAT_INCREMENT_N(Stats, bytes_received, len);
21742173
// Process deferred operations that couldn't run inside callback
21752174
// scopes (e.g., HTTP/3 GOAWAY handling that calls into JS).
21762175
application().PostReceive();

‎src/quic/session.h‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -353,7 +353,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
353353
bool early = false;
354354
};
355355

356-
bool Receive(Store&& store,
356+
bool Receive(const uint8_t* data,
357+
size_t len,
357358
const SocketAddress& local_address,
358359
const SocketAddress& remote_address,
359360
const PacketInfo& pkt_info = PacketInfo(),
@@ -367,10 +368,14 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
367368
// Receive() is kept as a convenience wrapper that calls ReadPacket()
368369
// then triggers SendPendingData (for paths like Connect that need
369370
// immediate response).
371+
// The data pointer is used synchronously — ngtcp2_conn_read_pkt does
372+
// not retain a reference after returning, so the caller's buffer can
373+
// be reused immediately.
370374
// When ts is 0 (the default), uv_hrtime() is called internally.
371375
// The batched receive path caches a timestamp and passes it to all
372376
// ReadPacket() calls in the same I/O burst.
373-
bool ReadPacket(Store&& store,
377+
bool ReadPacket(const uint8_t* data,
378+
size_t len,
374379
const SocketAddress& local_address,
375380
const SocketAddress& remote_address,
376381
const PacketInfo& pkt_info = PacketInfo(),

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL