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

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

/ node Public

Commit 80e6bf9

Browse files
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
Original file line numberDiff line numberDiff line change
@@ -329,14 +329,35 @@ void Session::Application::SendPendingData() {
329329
if (!session().can_send_packets()) [[unlikely]] {
330330
return;
331331
}
332-
static constexpr size_t kMaxPackets = 32;
332+
// Upper bound on packets per SendPendingData call. ngtcp2's send quantum
333+
// is typically 64 KB, which at 1200-byte minimum packet size is ~53
334+
// packets. 64 covers the worst case with headroom. The actual count per
335+
// call is dynamically capped by ngtcp2_conn_get_send_quantum().
336+
static constexpr size_t kMaxPackets = 64;
333337
Debug(session_, "Application sending pending data");
334338
PathStorage path;
335339
StreamData stream_data;
336340

337341
bool closed = false;
342+
343+
// Batch accumulation: packets are collected here and flushed via
344+
// Session::SendBatch when the loop exits, the batch is full, or
345+
// on early return. This enables synchronous batched delivery via
346+
// uv_udp_try_send2 (sendmmsg) from the deferred flush path.
347+
Packet::Ptr batch[kMaxPackets];
348+
PathStorage batch_paths[kMaxPackets];
349+
size_t batch_count = 0;
350+
351+
auto flush_batch = [&] {
352+
if (batch_count == 0) return;
353+
session_->SendBatch(batch, batch_paths, batch_count);
354+
batch_count = 0;
355+
};
356+
338357
auto update_stats = OnScopeLeave([&] {
339358
if (closed) return;
359+
// Flush any remaining accumulated packets before updating stats.
360+
flush_batch();
340361
auto& s = session();
341362
if (!s.is_destroyed()) [[likely]] {
342363
s.UpdatePacketTxTime();
@@ -353,7 +374,7 @@ void Session::Application::SendPendingData() {
353374
kMaxPackets, ngtcp2_conn_get_send_quantum(*session_) / max_packet_size);
354375
if (max_packet_count == 0) return;
355376

356-
// The number of packets that have been sent in this call to SendPendingData.
377+
// The number of packets that have been prepared in this call.
357378
size_t packet_send_count = 0;
358379

359380
Packet::Ptr packet;
@@ -368,6 +389,14 @@ void Session::Application::SendPendingData() {
368389
return true;
369390
};
370391

392+
// Accumulate a completed packet into the batch.
393+
auto enqueue_packet = [&](Packet::Ptr& pkt, size_t len) {
394+
Debug(session_, "Enqueuing packet with %zu bytes into batch", len);
395+
pkt->Truncate(len);
396+
path.CopyTo(&batch_paths[batch_count]);
397+
batch[batch_count++] = std::move(pkt);
398+
};
399+
371400
// We're going to enter a loop here to prepare and send no more than
372401
// max_packet_count packets.
373402
for (;;) {
@@ -502,8 +531,7 @@ void Session::Application::SendPendingData() {
502531
if (result > 0) {
503532
size_t len = result;
504533
Debug(session_, "Sending packet with %zu bytes", len);
505-
packet->Truncate(len);
506-
session_->Send(std::move(packet), path);
534+
enqueue_packet(packet, len);
507535
if (++packet_send_count == max_packet_count) return;
508536
} else if (result < 0) {
509537
// Any negative result other than NGTCP2_ERR_WRITE_MORE
@@ -540,8 +568,7 @@ void Session::Application::SendPendingData() {
540568
// is the size of the packet we are sending.
541569
size_t len = nwrite;
542570
Debug(session_, "Sending packet with %zu bytes", len);
543-
packet->Truncate(len);
544-
session_->Send(std::move(packet), path);
571+
enqueue_packet(packet, len);
545572
if (++packet_send_count == max_packet_count) return;
546573

547574
// If there are pending datagrams, try sending them in a fresh packet.
@@ -560,8 +587,7 @@ void Session::Application::SendPendingData() {
560587
TryWritePendingDatagram(&path, packet->data(), packet->length());
561588
if (result > 0) {
562589
Debug(session_, "Sending datagram packet with %zd bytes", result);
563-
packet->Truncate(static_cast<size_t>(result));
564-
session_->Send(std::move(packet), path);
590+
enqueue_packet(packet, static_cast<size_t>(result));
565591
if (++packet_send_count == max_packet_count) return;
566592
} else if (result < 0 && result != NGTCP2_ERR_WRITE_MORE) {
567593
// Fatal error — session already closed by TryWritePendingDatagram.

‎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
Original file line numberDiff line numberDiff line change
@@ -374,6 +374,13 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
374374
// bindingdata.cc doesn't need the full Application type definition.
375375
void FlushPendingData();
376376

377+
// Send a batch of packets accumulated by SendPendingData. Uses
378+
// Endpoint::SendBatch (uv_udp_try_send2 / sendmmsg) for synchronous
379+
// batched delivery when called from the deferred flush path.
380+
// Handles per-packet path updates and cross-endpoint redirects.
381+
// All Ptr entries are consumed (released or moved) on return.
382+
void SendBatch(Packet::Ptr* packets, PathStorage* paths, size_t count);
383+
377384
void Send(Packet::Ptr packet);
378385
void Send(Packet::Ptr packet, const PathStorage& path);
379386
datagram_id SendDatagram(Store&& data);

0 commit comments

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