quic: improve batching of packet sending · nodejs/node@80e6bf9 · GitHub
Skip to content

Commit 80e6bf9

Browse files
jasnelladuh95
jasnell
authored and
aduh95
committed
quic: improve batching of packet sending
Improves overall performance and sets us up for eventual support for GRO/GSO batching in libuv in the future. 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 f200aac commit 80e6bf9

5 files changed

Lines changed: 201 additions & 26 deletions

File tree

‎src/quic/application.cc‎

Lines changed: 34 additions & 8 deletions

‎src/quic/endpoint.cc‎

Lines changed: 83 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -492,14 +492,24 @@ int Endpoint::UDP::Send(Packet::Ptr packet) {
492492
return err;
493493
}
494494

495-
int Endpoint::UDP::TrySend(Packet* packet) {
496-
DCHECK_NOT_NULL(packet);
495+
int Endpoint::UDP::TrySend(const Packet::Ptr& packet) {
496+
DCHECK(packet);
497497
if (is_closed_or_closing()) return UV_EBADF;
498498
uv_buf_t buf = *packet;
499499
return uv_udp_try_send(
500500
&impl_->handle_, &buf, 1, packet->destination().data());
501501
}
502502

503+
int Endpoint::UDP::TrySendBatch(uv_buf_t* bufs[],
504+
unsigned int nbufs[],
505+
struct sockaddr* addrs[],
506+
size_t count) {
507+
DCHECK_GT(count, 0);
508+
if (is_closed_or_closing()) return UV_EBADF;
509+
return uv_udp_try_send2(
510+
&impl_->handle_, static_cast<unsigned int>(count), bufs, nbufs, addrs, 0);
511+
}
512+
503513
void Endpoint::UDP::MemoryInfo(MemoryTracker* tracker) const {
504514
if (impl_) tracker->TrackField("impl", impl_);
505515
}
@@ -827,20 +837,19 @@ void Endpoint::SendOrTrySend(Packet::Ptr packet) {
827837
}
828838
#endif
829839

830-
if (is_closed() || is_closing() || packet->length() == 0) {
840+
if (is_closed() || is_closing() || !packet || packet->length() == 0) {
831841
return;
832842
}
833843

834844
Debug(this, "TrySend %s", packet->ToString());
835-
size_t packet_length = packet->length();
836845

837846
// Attempt synchronous send. On success (returns number of bytes sent),
838847
// the packet is delivered immediately — no callback overhead, no
839848
// waiting for the next poll cycle.
840-
int err = udp_.TrySend(packet.get());
849+
int err = udp_.TrySend(packet);
841850
if (err >= 0) {
842-
// Synchronous send succeeded. Release the packet immediately.
843-
STAT_INCREMENT_N(Stats, bytes_sent, packet_length);
851+
// Synchronous send succeeded.
852+
STAT_INCREMENT_N(Stats, bytes_sent, packet->length());
844853
STAT_INCREMENT(Stats, packets_sent);
845854
// Ptr destructor releases back to arena pool.
846855
return;
@@ -859,6 +868,73 @@ void Endpoint::SendOrTrySend(Packet::Ptr packet) {
859868
Destroy(CloseContext::SEND_FAILURE, err);
860869
}
861870

871+
void Endpoint::SendBatch(Packet::Ptr* packets, size_t count) {
872+
if (count == 0) return;
873+
874+
#ifdef DEBUG
875+
if (is_diagnostic_packet_loss(options_.tx_loss)) [[unlikely]] {
876+
for (size_t i = 0; i < count; i++) packets[i].reset();
877+
return;
878+
}
879+
#endif
880+
881+
if (is_closed() || is_closing()) {
882+
for (size_t i = 0; i < count; i++) packets[i].reset();
883+
return;
884+
}
885+
886+
static constexpr size_t kMaxBatch = 64;
887+
DCHECK_LE(count, kMaxBatch);
888+
889+
// Build libuv argument arrays directly from the Ptr array.
890+
// Packets with zero length are released and skipped.
891+
uv_buf_t bufs[kMaxBatch];
892+
uv_buf_t* buf_ptrs[kMaxBatch];
893+
unsigned int nbufs[kMaxBatch];
894+
struct sockaddr* addrs[kMaxBatch];
895+
// Map from valid-index back to the original packets[] index.
896+
size_t index_map[kMaxBatch];
897+
size_t valid_count = 0;
898+
899+
for (size_t i = 0; i < count; i++) {
900+
if (!packets[i] || packets[i]->length() == 0) {
901+
packets[i].reset();
902+
continue;
903+
}
904+
bufs[valid_count] = *packets[i];
905+
buf_ptrs[valid_count] = &bufs[valid_count];
906+
nbufs[valid_count] = 1;
907+
addrs[valid_count] =
908+
const_cast<struct sockaddr*>(packets[i]->destination().data());
909+
index_map[valid_count] = i;
910+
valid_count++;
911+
}
912+
913+
if (valid_count == 0) return;
914+
915+
// Attempt synchronous batched send via sendmmsg.
916+
int sent = udp_.TrySendBatch(buf_ptrs, nbufs, addrs, valid_count);
917+
918+
if (sent > 0) {
919+
// Packets [0, sent) were delivered synchronously.
920+
// Release them immediately — no async callback needed.
921+
for (size_t i = 0; i < static_cast<size_t>(sent); i++) {
922+
size_t idx = index_map[i];
923+
STAT_INCREMENT_N(Stats, bytes_sent, packets[idx]->length());
924+
STAT_INCREMENT(Stats, packets_sent);
925+
packets[idx].reset();
926+
}
927+
}
928+
929+
// Any unsent packets (EAGAIN, partial send, or total failure) fall
930+
// back to async uv_udp_send.
931+
size_t start = (sent > 0) ? static_cast<size_t>(sent) : 0;
932+
for (size_t i = start; i < valid_count; i++) {
933+
size_t idx = index_map[i];
934+
Send(std::move(packets[idx]));
935+
}
936+
}
937+
862938
void Endpoint::SendRetry(const PathDescriptor& options) {
863939
// Generating and sending retry packets does consume some system resources,
864940
// and it is possible for a malicious peer to trigger sending a large number

‎src/quic/endpoint.h‎

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -235,6 +235,13 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
235235
// the one-tick latency of async uv_udp_send.
236236
void SendOrTrySend(Packet::Ptr packet);
237237

238+
// Send a batch of packets using uv_udp_try_send2 (sendmmsg) for
239+
// synchronous batched delivery. Packets successfully sent are released
240+
// immediately. On EAGAIN or partial send, remaining packets fall back
241+
// to async uv_udp_send. The Packet::Ptr array is consumed: all entries
242+
// will be empty (released or moved) on return.
243+
void SendBatch(Packet::Ptr* packets, size_t count);
244+
238245
// Acquire a Packet from the pool. length sets the initial working
239246
// size (must be <= pool capacity). The slot is always allocated at
240247
// full capacity to avoid fragmentation.
@@ -308,11 +315,19 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
308315
void Close();
309316
int Send(Packet::Ptr packet);
310317

311-
// Synchronous send using uv_udp_try_send. Returns 0 on success,
312-
// UV_EAGAIN if the socket is not writable or the send queue is
313-
// non-empty, or another negative error code on failure.
314-
// On success, the caller is responsible for releasing the packet.
315-
int TrySend(Packet* packet);
318+
// Synchronous send using uv_udp_try_send. Returns the number of
319+
// bytes sent on success, UV_EAGAIN if the socket is not writable
320+
// or the send queue is non-empty, or another negative error code.
321+
// The Ptr is not consumed — the caller manages the lifecycle.
322+
int TrySend(const Packet::Ptr& packet);
323+
324+
// Synchronous batched send using uv_udp_try_send2 (sendmmsg).
325+
// Takes pre-built libuv argument arrays. Returns the number of
326+
// messages successfully sent (>= 0), or a negative error code.
327+
int TrySendBatch(uv_buf_t* bufs[],
328+
unsigned int nbufs[],
329+
struct sockaddr* addrs[],
330+
size_t count);
316331

317332
// Returns the local UDP socket address to which we are bound,
318333
// or fail with an assert if we are not bound.

‎src/quic/session.cc‎

Lines changed: 57 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2149,12 +2149,8 @@ bool Session::ReadPacket(Store&& store,
21492149
// libuv does not currently deliver per-packet ECN metadata. When
21502150
// libuv gains ECN receive reporting, the pkt_info should be
21512151
// populated from the per-packet metadata and passed through here.
2152-
err = ngtcp2_conn_read_pkt(*this,
2153-
&path,
2154-
nullptr,
2155-
vec.base,
2156-
vec.len,
2157-
uv_hrtime());
2152+
err = ngtcp2_conn_read_pkt(
2153+
*this, &path, nullptr, vec.base, vec.len, uv_hrtime());
21582154
}
21592155
if (is_destroyed()) return false;
21602156

@@ -2263,6 +2259,61 @@ bool Session::ReadPacket(Store&& store,
22632259
return false;
22642260
}
22652261

2262+
void Session::SendBatch(Packet::Ptr* packets,
2263+
PathStorage* paths,
2264+
size_t count) {
2265+
DCHECK(!is_destroyed());
2266+
if (count == 0) return;
2267+
2268+
// Separate packets into those going to the primary endpoint and those
2269+
// redirected to other endpoints (rare: path validation, preferred address).
2270+
// Redirected packets are sent individually via the target endpoint.
2271+
static constexpr size_t kMaxBatch = 64;
2272+
DCHECK_LE(count, kMaxBatch);
2273+
Packet::Ptr primary_packets[kMaxBatch];
2274+
size_t primary_count = 0;
2275+
2276+
for (size_t i = 0; i < count; i++) {
2277+
if (!packets[i] || !can_send_packets()) {
2278+
packets[i].reset();
2279+
continue;
2280+
}
2281+
2282+
UpdatePath(paths[i]);
2283+
2284+
// Check for cross-endpoint redirect.
2285+
bool redirected = false;
2286+
if (paths[i].path.local.addrlen > 0) {
2287+
SocketAddress local_addr(paths[i].path.local.addr);
2288+
auto& mgr = BindingData::Get(env()).session_manager();
2289+
Endpoint* target = mgr.FindEndpointForAddress(local_addr);
2290+
if (target != nullptr && target != &endpoint()) {
2291+
SocketAddress remote_addr(paths[i].path.remote.addr);
2292+
packets[i]->Redirect(static_cast<Packet::Listener*>(target),
2293+
remote_addr);
2294+
target->Send(std::move(packets[i]));
2295+
redirected = true;
2296+
}
2297+
}
2298+
2299+
if (!redirected) {
2300+
primary_packets[primary_count++] = std::move(packets[i]);
2301+
}
2302+
}
2303+
2304+
if (primary_count == 0) return;
2305+
2306+
// Use batched send for the primary endpoint.
2307+
if (prefer_try_send_) {
2308+
endpoint().SendBatch(primary_packets, primary_count);
2309+
} else {
2310+
// Non-flush path: send individually via async uv_udp_send.
2311+
for (size_t i = 0; i < primary_count; i++) {
2312+
Send(std::move(primary_packets[i]));
2313+
}
2314+
}
2315+
}
2316+
22662317
void Session::FlushPendingData() {
22672318
DCHECK(!is_destroyed());
22682319
if (impl_->application_) {

‎src/quic/session.h‎

Lines changed: 7 additions & 0 deletions

0 commit comments

Comments
 (0)