quic: use arena allocation for packets

Previously Packets were ReqWrap objects with a shared
free-list. This commit changes to a per-Endpoint arena
with no v8 involvement. This is the design I originally
had in mind but I initially went with the simpler
freelist approach to get something working. There's
too much overhead in the reqrap/freelist approach and
individual packets do not really need to be observable
via async hooks.

This design should eliminate the risk of memory fragmentation
and eliminate a significant bottleneck in the hot path.

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode:Opus 4.6
PR-URL: https://github.com/nodejs/node/pull/62589
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Stephen Belanger <admin@stephenbelanger.com>
This commit is contained in:
James M Snell committed 2026-04-06 11:12:29 -07:00
1 parent 500c3a53e3
commit d581ead032
14 files changed
+854 -454

No files matched your search

+2
View File
@@ -354,6 +354,7 @@
'src/quic/tlscontext.cc',
'src/quic/transportparams.cc',
'src/quic/quic.cc',
'src/quic/arena.h',
'src/quic/bindingdata.h',
'src/quic/cid.h',
'src/quic/data.h',
@@ -440,6 +441,7 @@
'test/cctest/test_node_crypto_env.cc',
],
'node_cctest_quic_sources': [
'test/cctest/test_quic_arena.cc',
'test/cctest/test_quic_cid.cc',
'test/cctest/test_quic_error.cc',
'test/cctest/test_quic_preferredaddress.cc',
-1
View File
@@ -63,7 +63,6 @@ namespace node {
V(QUERYWRAP) \
V(QUIC_ENDPOINT) \
V(QUIC_LOGSTREAM) \
V(QUIC_PACKET) \
V(QUIC_SESSION) \
V(QUIC_STREAM) \
V(QUIC_UDP) \
+10 -19
View File
@@ -1,7 +1,6 @@
#if HAVE_OPENSSL && HAVE_QUIC
#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include "application.h"
#include <async_wrap-inl.h>
#include <debug_utils-inl.h>
#include <nghttp3/nghttp3.h>
@@ -10,6 +9,7 @@
#include <node_sockaddr-inl.h>
#include <uv.h>
#include <v8.h>
#include "application.h"
#include "defs.h"
#include "endpoint.h"
#include "http3.h"
@@ -207,12 +207,9 @@ StreamPriority Session::Application::GetStreamPriority(const Stream& stream) {
return StreamPriority::DEFAULT;
}
BaseObjectPtr<Packet> Session::Application::CreateStreamDataPacket() {
return Packet::Create(env(),
session_->endpoint(),
session_->remote_address(),
session_->max_packet_size(),
"stream data");
Packet::Ptr Session::Application::CreateStreamDataPacket() {
return session_->endpoint().CreatePacket(
session_->remote_address(), session_->max_packet_size(), "stream data");
}
void Session::Application::StreamClose(Stream* stream, QuicError&& error) {
@@ -264,7 +261,7 @@ void Session::Application::SendPendingData() {
// The number of packets that have been sent in this call to SendPendingData.
size_t packet_send_count = 0;
BaseObjectPtr<Packet> packet;
Packet::Ptr packet;
uint8_t* pos = nullptr;
uint8_t* begin = nullptr;
@@ -273,7 +270,7 @@ void Session::Application::SendPendingData() {
packet = CreateStreamDataPacket();
if (!packet) [[unlikely]]
return false;
pos = begin = ngtcp2_vec(*packet).base;
pos = begin = packet->data();
}
DCHECK(packet);
DCHECK_NOT_NULL(pos);
@@ -299,7 +296,6 @@ void Session::Application::SendPendingData() {
// The stream_data is the next block of data from the application stream.
if (GetStreamData(&stream_data) < 0) {
Debug(session_, "Application failed to get stream data");
packet->CancelPacket();
session_->SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return session_->Close(CloseMethod::SILENT);
@@ -367,7 +363,6 @@ void Session::Application::SendPendingData() {
if (ndatalen >= 0 && !StreamCommit(&stream_data, ndatalen)) {
Debug(session_,
"Failed to commit stream data while writing packets");
packet->CancelPacket();
session_->SetLastError(
QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
@@ -380,7 +375,6 @@ void Session::Application::SendPendingData() {
// ngtcp2 callback failed for some reason. This would be a
// bug in our code.
Debug(session_, "Internal failure with ngtcp2 callback");
packet->CancelPacket();
session_->SetLastError(
QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
@@ -393,12 +387,10 @@ void Session::Application::SendPendingData() {
Debug(session_,
"Application encountered error while writing packet: %s",
ngtcp2_strerror(nwrite));
packet->CancelPacket();
session_->SetLastError(QuicError::ForNgtcp2Error(nwrite));
closed = true;
return session_->Close(CloseMethod::SILENT);
} else if (ndatalen >= 0 && !StreamCommit(&stream_data, ndatalen)) {
packet->CancelPacket();
session_->SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return session_->Close(CloseMethod::SILENT);
@@ -416,10 +408,9 @@ void Session::Application::SendPendingData() {
if (datalen) {
Debug(session_, "Sending packet with %zu bytes", datalen);
packet->Truncate(datalen);
session_->Send(packet, path);
} else {
packet->CancelPacket();
session_->Send(std::move(packet), path);
}
// If no data, Ptr destructor releases the packet.
return;
}
@@ -429,7 +420,7 @@ void Session::Application::SendPendingData() {
size_t datalen = pos - begin;
Debug(session_, "Sending packet with %zu bytes", datalen);
packet->Truncate(datalen);
session_->Send(packet, path);
session_->Send(std::move(packet), path);
// If we have sent the maximum number of packets, we're done.
if (++packet_send_count == max_packet_count) {
@@ -437,7 +428,7 @@ void Session::Application::SendPendingData() {
}
// Prepare to loop back around to prepare a new packet.
packet.reset();
// packet is already empty from the std::move above.
pos = begin = nullptr;
}
}
+1 -1
View File
@@ -132,7 +132,7 @@ class Session::Application : public MemoryRetainer {
}
private:
BaseObjectPtr<Packet> CreateStreamDataPacket();
Packet::Ptr CreateStreamDataPacket();
// Write the given stream_data into the buffer.
ssize_t WriteVStream(PathStorage* path,
+373
View File
@@ -0,0 +1,373 @@
#pragma once
#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
#include <memory_tracker-inl.h>
#include <util.h>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <new>
#include <utility>
#include <vector>
namespace node::quic {
// ArenaPool<T> is a block-based arena allocator for fixed-size objects.
//
// Objects are allocated from contiguous memory blocks ("slabs"), reducing
// heap fragmentation and allocation overhead. Each block contains a fixed
// number of equally-sized slots. Free slots form an intrusive singly-linked
// list for O(1) acquire/release.
//
// Each slot can optionally hold extra bytes after T (e.g., a trailing data
// buffer). The extra_bytes parameter is set at pool construction time and
// applies uniformly to all slots. All slots are allocated at the same size
// regardless of how much of the extra space is actually used, to prevent
// fragmentation.
//
// ArenaPool<T>::Ptr is a move-only RAII smart pointer that releases slots
// back to the pool on destruction. Ptr::release() detaches ownership for
// handoff to external systems (e.g., libuv), after which the caller must
// eventually call ArenaPool<T>::Release(raw_ptr) to return the slot.
//
// The pool supports lazy GC: when the ratio of free slots to total slots
// exceeds a threshold, fully-unused blocks are reclaimed. At least one
// block is always retained.
template <typename T>
class ArenaPool final : public MemoryRetainer {
public:
class Ptr;
// extra_bytes: additional memory available after T in each slot.
// All slots are sized identically regardless of how
// much extra space is actually used.
// slots_per_block: number of T slots per allocation block.
explicit ArenaPool(size_t extra_bytes = 0,
size_t slots_per_block = kDefaultSlotsPerBlock);
~ArenaPool();
ArenaPool(const ArenaPool&) = delete;
ArenaPool& operator=(const ArenaPool&) = delete;
ArenaPool(ArenaPool&&) = delete;
ArenaPool& operator=(ArenaPool&&) = delete;
// Construct T in an acquired slot with forwarded args.
// Returns an empty Ptr only on allocation failure.
template <typename... Args>
[[nodiscard]] Ptr Acquire(Args&&... args);
// Construct T with (extra_data_ptr, extra_bytes, ...args).
// Use this for types whose constructor accepts a trailing data
// buffer as its first two parameters.
template <typename... Args>
[[nodiscard]] Ptr AcquireExtra(Args&&... args);
// Release a raw T* previously detached via Ptr::release().
// Calls ~T() and returns the slot to the pool's free list.
// Recovers the pool instance from T*'s slot metadata.
static void Release(T* obj);
// Attempt to reclaim fully-unused blocks. Called automatically
// from Release/ReleaseSlot when the pool is over-provisioned.
void MaybeGC();
// Free all unused blocks immediately, keeping at least one.
void Flush();
size_t extra_bytes() const { return extra_bytes_; }
size_t slot_size() const { return slot_size_; }
size_t total_slots() const { return total_slots_; }
size_t in_use_count() const { return in_use_count_; }
size_t block_count() const { return blocks_.size(); }
void MemoryInfo(MemoryTracker* tracker) const override;
SET_MEMORY_INFO_NAME(ArenaPool)
SET_SELF_SIZE(ArenaPool)
static constexpr size_t kDefaultSlotsPerBlock = 128;
private:
// -------------------------------------------------------------------
// Slot layout in memory:
//
// [0, kObjectOffset) SlotHeader
// [kObjectOffset, kObjectOffset+sizeof(T)) T object
// [kObjectOffset+sizeof(T), slot_size_) extra bytes + padding
//
// SlotHeader is only used when the slot is on the free list.
// When the slot is acquired, the T object occupies its storage
// and the header fields are not accessed.
// -------------------------------------------------------------------
struct Block;
struct SlotHeader {
SlotHeader* next_free;
Block* block;
};
struct Block {
std::unique_ptr<char[]> memory;
ArenaPool* pool;
size_t capacity;
size_t in_use_count;
};
static constexpr size_t kObjectOffset =
(sizeof(SlotHeader) + alignof(T) - 1) & ~(alignof(T) - 1);
// Slot ↔ T* conversion using the compile-time offset.
static T* ObjectFromSlot(SlotHeader* slot) {
return reinterpret_cast<T*>(reinterpret_cast<char*>(slot) + kObjectOffset);
}
static SlotHeader* SlotFromObject(T* obj) {
return reinterpret_cast<SlotHeader*>(reinterpret_cast<char*>(obj) -
kObjectOffset);
}
static uint8_t* ExtraDataFromSlot(SlotHeader* slot) {
return reinterpret_cast<uint8_t*>(reinterpret_cast<char*>(slot) +
kObjectOffset + sizeof(T));
}
SlotHeader* SlotAt(Block* block, size_t index) {
return reinterpret_cast<SlotHeader*>(block->memory.get() +
index * slot_size_);
}
SlotHeader* AcquireSlot();
void ReleaseSlot(SlotHeader* slot);
bool Grow();
void FreeEmptyBlocks();
void RemoveBlockFromFreeList(Block* block);
const size_t extra_bytes_;
const size_t slots_per_block_;
const size_t slot_size_;
SlotHeader* free_list_ = nullptr;
std::vector<std::unique_ptr<Block>> blocks_;
size_t total_slots_ = 0;
size_t in_use_count_ = 0;
};
// =====================================================================
// ArenaPool<T>::Ptr — Move-only RAII smart pointer
// =====================================================================
template <typename T>
class ArenaPool<T>::Ptr final {
public:
Ptr() = default;
~Ptr() { reset(); }
Ptr(Ptr&& other) noexcept : slot_(other.slot_) { other.slot_ = nullptr; }
Ptr& operator=(Ptr&& other) noexcept {
if (this != &other) {
reset();
slot_ = other.slot_;
other.slot_ = nullptr;
}
return *this;
}
Ptr(const Ptr&) = delete;
Ptr& operator=(const Ptr&) = delete;
T* get() const { return slot_ ? ObjectFromSlot(slot_) : nullptr; }
T* operator->() const {
DCHECK(slot_);
return ObjectFromSlot(slot_);
}
T& operator*() const {
DCHECK(slot_);
return *ObjectFromSlot(slot_);
}
explicit operator bool() const { return slot_ != nullptr; }
// Access the extra data region after T in the slot.
uint8_t* extra_data() const {
return slot_ ? ExtraDataFromSlot(slot_) : nullptr;
}
size_t extra_bytes() const {
return slot_ ? slot_->block->pool->extra_bytes_ : 0;
}
// Detach ownership. The caller takes responsibility for eventually
// calling ArenaPool<T>::Release(ptr) to destruct T and return
// the slot to the pool.
[[nodiscard]] T* release() noexcept {
if (!slot_) return nullptr;
T* obj = ObjectFromSlot(slot_);
slot_ = nullptr;
return obj;
}
// Destruct T and return the slot to the pool. Ptr becomes empty.
void reset() {
if (slot_) {
ObjectFromSlot(slot_)->~T();
slot_->block->pool->ReleaseSlot(slot_);
slot_ = nullptr;
}
}
private:
friend class ArenaPool<T>;
explicit Ptr(SlotHeader* slot) : slot_(slot) {}
SlotHeader* slot_ = nullptr;
};
// =====================================================================
// ArenaPool<T> implementation
// =====================================================================
template <typename T>
ArenaPool<T>::ArenaPool(size_t extra_bytes, size_t slots_per_block)
: extra_bytes_(extra_bytes),
slots_per_block_(slots_per_block),
slot_size_(((kObjectOffset + sizeof(T) + extra_bytes) +
alignof(std::max_align_t) - 1) &
~(alignof(std::max_align_t) - 1)) {
DCHECK_GT(slots_per_block, 0);
}
template <typename T>
ArenaPool<T>::~ArenaPool() {
DCHECK_EQ(in_use_count_, 0);
}
template <typename T>
template <typename... Args>
typename ArenaPool<T>::Ptr ArenaPool<T>::Acquire(Args&&... args) {
SlotHeader* slot = AcquireSlot();
if (!slot) return Ptr();
T* obj = new (ObjectFromSlot(slot)) T(std::forward<Args>(args)...);
CHECK_EQ(obj, ObjectFromSlot(slot));
return Ptr(slot);
}
template <typename T>
template <typename... Args>
typename ArenaPool<T>::Ptr ArenaPool<T>::AcquireExtra(Args&&... args) {
SlotHeader* slot = AcquireSlot();
if (!slot) return Ptr();
T* obj = new (ObjectFromSlot(slot))
T(ExtraDataFromSlot(slot), extra_bytes_, std::forward<Args>(args)...);
CHECK_EQ(obj, ObjectFromSlot(slot));
return Ptr(slot);
}
template <typename T>
void ArenaPool<T>::Release(T* obj) {
DCHECK_NOT_NULL(obj);
SlotHeader* slot = SlotFromObject(obj);
DCHECK_NOT_NULL(slot->block);
DCHECK_NOT_NULL(slot->block->pool);
obj->~T();
slot->block->pool->ReleaseSlot(slot);
}
template <typename T>
typename ArenaPool<T>::SlotHeader* ArenaPool<T>::AcquireSlot() {
if (!free_list_) {
if (!Grow()) return nullptr;
}
DCHECK_NOT_NULL(free_list_);
SlotHeader* slot = free_list_;
free_list_ = slot->next_free;
slot->next_free = nullptr;
slot->block->in_use_count++;
in_use_count_++;
return slot;
}
template <typename T>
void ArenaPool<T>::ReleaseSlot(SlotHeader* slot) {
DCHECK_NOT_NULL(slot);
DCHECK_NOT_NULL(slot->block);
DCHECK_GT(slot->block->in_use_count, 0);
DCHECK_GT(in_use_count_, 0);
slot->block->in_use_count--;
in_use_count_--;
slot->next_free = free_list_;
free_list_ = slot;
MaybeGC();
}
template <typename T>
bool ArenaPool<T>::Grow() {
auto block = std::make_unique<Block>();
block->pool = this;
block->capacity = slots_per_block_;
block->in_use_count = 0;
block->memory.reset(new char[slots_per_block_ * slot_size_]());
// Initialize slot headers and chain onto free list.
for (size_t i = 0; i < slots_per_block_; i++) {
SlotHeader* slot = SlotAt(block.get(), i);
slot->block = block.get();
slot->next_free = free_list_;
free_list_ = slot;
}
total_slots_ += slots_per_block_;
blocks_.push_back(std::move(block));
return true;
}
template <typename T>
void ArenaPool<T>::MaybeGC() {
// Only GC when we have excess capacity: more than one block and
// less than half the slots are in use.
if (blocks_.size() <= 1 || in_use_count_ >= total_slots_ / 2) return;
FreeEmptyBlocks();
}
template <typename T>
void ArenaPool<T>::Flush() {
FreeEmptyBlocks();
}
template <typename T>
void ArenaPool<T>::FreeEmptyBlocks() {
for (auto it = blocks_.begin(); it != blocks_.end();) {
if ((*it)->in_use_count == 0 && blocks_.size() > 1) {
RemoveBlockFromFreeList(it->get());
total_slots_ -= (*it)->capacity;
it = blocks_.erase(it);
} else {
++it;
}
}
}
template <typename T>
void ArenaPool<T>::RemoveBlockFromFreeList(Block* block) {
char* block_start = block->memory.get();
char* block_end = block_start + block->capacity * slot_size_;
SlotHeader** pp = &free_list_;
while (*pp) {
char* addr = reinterpret_cast<char*>(*pp);
if (addr >= block_start && addr < block_end) {
*pp = (*pp)->next_free;
} else {
pp = &(*pp)->next_free;
}
}
}
template <typename T>
void ArenaPool<T>::MemoryInfo(MemoryTracker* tracker) const {
tracker->TrackFieldWithSize("blocks", total_slots_ * slot_size_);
}
} // namespace node::quic
#endif // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
+1 -10
View File
@@ -1,7 +1,6 @@
#if HAVE_OPENSSL && HAVE_QUIC
#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include "bindingdata.h"
#include <base_object-inl.h>
#include <env-inl.h>
#include <memory_tracker-inl.h>
@@ -13,6 +12,7 @@
#include <node_mem-inl.h>
#include <node_realm-inl.h>
#include <v8.h>
#include "bindingdata.h"
namespace node {
@@ -61,8 +61,6 @@ void BindingData::DecreaseAllocatedSize(size_t size) {
void BindingData::InitPerContext(Realm* realm, Local<Object> target) {
SetMethod(realm->context(), target, "setCallbacks", SetCallbacks);
SetMethod(
realm->context(), target, "flushPacketFreelist", FlushPacketFreelist);
Realm::GetCurrent(realm->context())->AddBindingData<BindingData>(target);
}
@@ -70,7 +68,6 @@ void BindingData::RegisterExternalReferences(
ExternalReferenceRegistry* registry) {
registry->Register(IllegalConstructor);
registry->Register(SetCallbacks);
registry->Register(FlushPacketFreelist);
}
BindingData::BindingData(Realm* realm, Local<Object> object)
@@ -165,12 +162,6 @@ JS_METHOD_IMPL(BindingData::SetCallbacks) {
#undef V
}
JS_METHOD_IMPL(BindingData::FlushPacketFreelist) {
auto env = Environment::GetCurrent(args);
auto& state = Get(env);
state.packet_freelist.clear();
}
NgTcp2CallbackScope::NgTcp2CallbackScope(Environment* env) : env(env) {
auto& binding = BindingData::Get(env);
CHECK(!binding.in_ngtcp2_callback_scope);
-8
View File
@@ -11,7 +11,6 @@
#include <node.h>
#include <node_mem.h>
#include <v8.h>
#include <list>
#include <unordered_map>
#include "defs.h"
@@ -27,7 +26,6 @@ class Packet;
V(endpoint) \
V(http3application) \
V(logstream) \
V(packet) \
V(session) \
V(stream) \
V(udp)
@@ -106,7 +104,6 @@ class Packet;
V(max_stream_window, "maxStreamWindow") \
V(max_window, "maxWindow") \
V(min_version, "minVersion") \
V(packetwrap, "PacketWrap") \
V(preferred_address_strategy, "preferredAddressPolicy") \
V(protocol, "protocol") \
V(qlog, "qlog") \
@@ -172,11 +169,6 @@ class BindingData final
// bridge out to the JS API.
JS_METHOD(SetCallbacks);
// Purge the packet free list to free up memory.
JS_METHOD(FlushPacketFreelist);
std::list<BaseObjectPtr<BaseObject>> packet_freelist;
std::unordered_map<Endpoint*, BaseObjectPtr<BaseObject>> listening_endpoints;
bool in_ngtcp2_callback_scope = false;
+48 -40
View File
@@ -1,7 +1,6 @@
#if HAVE_OPENSSL && HAVE_QUIC
#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include "endpoint.h"
#include <aliased_struct-inl.h>
#include <async_wrap-inl.h>
#include <debug_utils-inl.h>
@@ -12,7 +11,6 @@
#include <node_external_reference.h>
#include <node_process-inl.h>
#include <node_sockaddr-inl.h>
#include <req_wrap-inl.h>
#include <util-inl.h>
#include <uv.h>
#include <v8.h>
@@ -20,6 +18,7 @@
#include "application.h"
#include "bindingdata.h"
#include "defs.h"
#include "endpoint.h"
#include "http3.h"
#include "ncrypto.h"
@@ -464,35 +463,27 @@ SocketAddress Endpoint::UDP::local_address() const {
return SocketAddress::FromSockName(impl_->handle_);
}
int Endpoint::UDP::Send(const BaseObjectPtr<Packet>& packet) {
int Endpoint::UDP::Send(Packet::Ptr packet) {
DCHECK(packet);
DCHECK(!packet->IsDispatched());
if (is_closed_or_closing()) return UV_EBADF;
uv_buf_t buf = *packet;
// We don't use the default implementation of Dispatch because the packet
// itself is going to be reset and added to a freelist to be reused. The
// default implementation of Dispatch will cause the packet to be deleted,
// which we don't want.
packet->ClearWeak();
packet->Dispatched();
int err = uv_udp_send(packet->req(),
// Detach from the Ptr — libuv takes ownership until the callback fires.
Packet* raw = packet.release();
uv_buf_t buf = *raw;
int err = uv_udp_send(raw->req(),
&impl_->handle_,
&buf,
1,
packet->destination().data(),
uv_udp_send_cb{[](uv_udp_send_t* req, int status) {
auto ptr = BaseObjectPtr<Packet>(static_cast<Packet*>(
ReqWrap<uv_udp_send_t>::from_req(req)));
ptr->env()->DecreaseWaitingRequestCounter();
ptr->Done(status);
}});
raw->destination().data(),
[](uv_udp_send_t* req, int status) {
Packet* p = Packet::FromReq(req);
p->listener()->PacketDone(status);
ArenaPool<Packet>::Release(p);
});
if (err < 0) {
// The packet failed.
packet->Done(err);
packet->MakeWeak();
} else {
packet->env()->IncreaseWaitingRequestCounter();
// Send failed — release the packet back to the pool immediately.
ArenaPool<Packet>::Release(raw);
}
return err;
}
@@ -599,6 +590,8 @@ Endpoint::Endpoint(Environment* env,
stats_(env->isolate()),
state_(env->isolate()),
options_(options),
packet_pool_(kDefaultMaxPacketLength,
ArenaPool<Packet>::kDefaultSlotsPerBlock),
udp_(this),
addrLRU_(options_.address_lru_size) {
MakeWeak();
@@ -614,6 +607,18 @@ Endpoint::Endpoint(Environment* env,
env, object, env->stats_string(), stats_.GetArrayBuffer());
}
Packet::Ptr Endpoint::CreatePacket(const SocketAddress& destination,
size_t length,
const char* diagnostic_label) {
auto ptr = packet_pool_.AcquireExtra(static_cast<Packet::Listener*>(this),
destination);
if (ptr) {
ptr->Truncate(std::min(length, ptr->capacity()));
ptr->set_diagnostic_label(diagnostic_label);
}
return ptr;
}
SocketAddress Endpoint::local_address() const {
DCHECK(!is_closed() && !is_closing());
return udp_.local_address();
@@ -724,32 +729,37 @@ void Endpoint::DisassociateStatelessResetToken(
}
}
void Endpoint::Send(const BaseObjectPtr<Packet>& packet) {
void Endpoint::Send(Packet::Ptr packet) {
#ifdef DEBUG
// When diagnostic packet loss is enabled, the packet will be randomly
// dropped. This can happen to any type of packet. We use this only in
// testing to test various reliability issues.
if (is_diagnostic_packet_loss(options_.tx_loss)) [[unlikely]] {
packet->Done();
// Simulating tx packet loss
// Simulating tx packet loss. Ptr destructor releases the packet.
return;
}
#endif // DEBUG
if (is_closed() || is_closing() || packet->length() == 0) {
packet->CancelPacket();
// Ptr destructor releases the packet back to the pool.
return;
}
Debug(this, "Sending %s", packet->ToString());
size_t packet_length = packet->length();
state_->pending_callbacks++;
int err = udp_.Send(packet);
env()->IncreaseWaitingRequestCounter();
int err = udp_.Send(std::move(packet));
if (err != 0) {
Debug(this, "Sending packet failed with error %d", err);
packet->Done(err);
// The packet was already released in UDP::Send on error.
state_->pending_callbacks--;
env()->DecreaseWaitingRequestCounter();
Destroy(CloseContext::SEND_FAILURE, err);
return;
}
STAT_INCREMENT_N(Stats, bytes_sent, packet->length());
STAT_INCREMENT_N(Stats, bytes_sent, packet_length);
STAT_INCREMENT(Stats, packets_sent);
}
@@ -767,11 +777,10 @@ void Endpoint::SendRetry(const PathDescriptor& options) {
auto info = addrLRU_.Upsert(options.remote_address);
if (++(info->retry_count) <= options_.max_retries) {
auto packet =
Packet::CreateRetryPacket(env(), this, options, options_.token_secret);
Packet::CreateRetryPacket(*this, options, options_.token_secret);
if (packet) {
STAT_INCREMENT(Stats, retry_count);
Send(std::move(packet));
packet.reset();
}
// If creating the retry is unsuccessful, we just drop things on the floor.
@@ -789,11 +798,10 @@ void Endpoint::SendVersionNegotiation(const PathDescriptor& options) {
// reset packets. If the packet is sent, then we'll at least increment the
// version_negotiation_count statistic so that application code can keep an
// eye on it.
auto packet = Packet::CreateVersionNegotiationPacket(env(), this, options);
auto packet = Packet::CreateVersionNegotiationPacket(*this, options);
if (packet) {
STAT_INCREMENT(Stats, version_negotiation_count);
Send(std::move(packet));
packet.reset();
}
// If creating the packet is unsuccessful, we just drop things on the floor.
@@ -823,13 +831,12 @@ bool Endpoint::SendStatelessReset(const PathDescriptor& options,
if (exceeds_limits()) return false;
auto packet = Packet::CreateStatelessResetPacket(
env(), this, options, options_.reset_token_secret, source_len);
*this, options, options_.reset_token_secret, source_len);
if (packet) {
addrLRU_.Upsert(options.remote_address)->reset_count++;
STAT_INCREMENT(Stats, stateless_reset_count);
Send(std::move(packet));
packet.reset();
return true;
}
return false;
@@ -843,12 +850,11 @@ void Endpoint::SendImmediateConnectionClose(const PathDescriptor& options,
reason);
// While it is possible for a malicious peer to cause us to create a large
// number of these, generating them is fairly trivial.
auto packet = Packet::CreateImmediateConnectionClosePacket(
env(), this, options, reason);
auto packet =
Packet::CreateImmediateConnectionClosePacket(*this, options, reason);
if (packet) {
STAT_INCREMENT(Stats, immediate_close_count);
Send(std::move(packet));
packet.reset();
}
}
@@ -1571,6 +1577,7 @@ void Endpoint::PacketDone(int status) {
// At this point we should be waiting on at least one packet.
DCHECK_GE(state_->pending_callbacks, 1);
state_->pending_callbacks--;
env()->DecreaseWaitingRequestCounter();
// Can we go ahead and close now?
if (state_->closing == 1) MaybeDestroy();
}
@@ -1597,6 +1604,7 @@ bool Endpoint::is_listening() const {
void Endpoint::MemoryInfo(MemoryTracker* tracker) const {
tracker->TrackField("options", options_);
tracker->TrackField("packet_pool", packet_pool_);
tracker->TrackField("udp", udp_);
if (server_state_.has_value()) {
tracker->TrackField("server_options", server_state_->options);
+11 -2
View File
@@ -10,6 +10,7 @@
#include <v8.h>
#include <algorithm>
#include <optional>
#include "arena.h"
#include "bindingdata.h"
#include "packet.h"
#include "session.h"
@@ -189,7 +190,14 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
Session* session);
void DisassociateStatelessResetToken(const StatelessResetToken& token);
void Send(const BaseObjectPtr<Packet>& packet);
void Send(Packet::Ptr packet);
// Acquire a Packet from the pool. length sets the initial working
// size (must be <= pool capacity). The slot is always allocated at
// full capacity to avoid fragmentation.
Packet::Ptr CreatePacket(const SocketAddress& destination,
size_t length = kDefaultMaxPacketLength,
const char* diagnostic_label = nullptr);
// Generates and sends a retry packet. This is terminal for the connection.
// Retry packets are used to force explicit path validation by issuing a token
@@ -255,7 +263,7 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
int Start();
void Stop();
void Close();
int Send(const BaseObjectPtr<Packet>& packet);
int Send(Packet::Ptr packet);
// Returns the local UDP socket address to which we are bound,
// or fail with an assert if we are not bound.
@@ -363,6 +371,7 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
AliasedStruct<Stats> stats_;
AliasedStruct<State> state_;
const Options options_;
ArenaPool<Packet> packet_pool_;
UDP udp_;
struct ServerState {
+65 -249
View File
@@ -1,34 +1,25 @@
#if HAVE_OPENSSL && HAVE_QUIC
#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include "packet.h"
#include <base_object-inl.h>
#include <crypto/crypto_util.h>
#include <env-inl.h>
#include <ngtcp2/ngtcp2.h>
#include <ngtcp2/ngtcp2_crypto.h>
#include <node_sockaddr-inl.h>
#include <req_wrap-inl.h>
#include <uv.h>
#include <v8.h>
#include <cstring>
#include <string>
#include "bindingdata.h"
#include "cid.h"
#include "defs.h"
#include "endpoint.h"
#include "ncrypto.h"
#include "packet.h"
#include "tokens.h"
namespace node {
using v8::Local;
using v8::Object;
namespace quic {
namespace node::quic {
namespace {
static constexpr size_t kRandlen = NGTCP2_MIN_STATELESS_RESET_RANDLEN * 5;
static constexpr size_t kMinStatelessResetLen = 41;
static constexpr size_t kMaxFreeList = 100;
} // namespace
std::string PathDescriptor::ToString() const {
@@ -44,180 +35,39 @@ std::string PathDescriptor::ToString() const {
return res;
}
struct Packet::Data final : public MemoryRetainer {
MaybeStackBuffer<uint8_t, kDefaultMaxPacketLength> data_;
// ============================================================================
// Packet
// The diagnostic_label_ is used only as a debugging tool when
// logging debug information about the packet. It identifies
// the purpose of the packet.
const std::string diagnostic_label_;
void MemoryInfo(MemoryTracker* tracker) const override {
tracker->TrackFieldWithSize("data", data_.length());
}
SET_MEMORY_INFO_NAME(Data)
SET_SELF_SIZE(Data)
Data(size_t length, std::string_view diagnostic_label)
: diagnostic_label_(diagnostic_label) {
data_.AllocateSufficientStorage(length);
}
size_t length() const { return data_.length(); }
operator uv_buf_t() {
return uv_buf_init(reinterpret_cast<char*>(data_.out()), data_.length());
}
operator ngtcp2_vec() { return ngtcp2_vec{data_.out(), data_.length()}; }
std::string ToString() const {
return diagnostic_label_ + ", " + std::to_string(length());
}
};
const SocketAddress& Packet::destination() const {
return destination_;
}
size_t Packet::length() const {
return data_ ? data_->length() : 0;
}
Packet::operator uv_buf_t() const {
return !data_ ? uv_buf_init(nullptr, 0) : *data_;
}
Packet::operator ngtcp2_vec() const {
return !data_ ? ngtcp2_vec{nullptr, 0} : *data_;
}
void Packet::Truncate(size_t len) {
DCHECK(data_);
DCHECK_LE(len, data_->length());
data_->data_.SetLength(len);
}
JS_CONSTRUCTOR_IMPL(Packet, packet_constructor_template, {
JS_ILLEGAL_CONSTRUCTOR();
JS_INHERIT(ReqWrap<uv_udp_send_t>);
JS_CLASS(packetwrap);
})
BaseObjectPtr<Packet> Packet::Create(Environment* env,
Listener* listener,
const SocketAddress& destination,
size_t length,
const char* diagnostic_label) {
if (BindingData::Get(env).packet_freelist.empty()) {
JS_NEW_INSTANCE_OR_RETURN(env, obj, {});
return MakeBaseObject<Packet>(
env, listener, obj, destination, length, diagnostic_label);
}
return FromFreeList(env,
std::make_shared<Data>(length, diagnostic_label),
listener,
destination);
}
BaseObjectPtr<Packet> Packet::Clone() const {
// Cloning is copy-free. Our data_ is a shared_ptr so we can just
// share it with the cloned packet.
auto& binding = BindingData::Get(env());
if (binding.packet_freelist.empty()) {
JS_NEW_INSTANCE_OR_RETURN(env(), obj, {});
return MakeBaseObject<Packet>(env(), listener_, obj, destination_, data_);
}
return FromFreeList(env(), data_, listener_, destination_);
}
BaseObjectPtr<Packet> Packet::FromFreeList(Environment* env,
std::shared_ptr<Data> data,
Listener* listener,
const SocketAddress& destination) {
auto& binding = BindingData::Get(env);
if (binding.packet_freelist.empty()) return {};
auto obj = binding.packet_freelist.front();
binding.packet_freelist.pop_front();
CHECK(obj);
CHECK_EQ(env, obj->env());
auto packet = BaseObjectPtr<Packet>(static_cast<Packet*>(obj.get()));
Debug(packet.get(), "Reusing packet from freelist");
packet->data_ = std::move(data);
packet->destination_ = destination;
packet->listener_ = listener;
return packet;
}
Packet::Packet(Environment* env,
Packet::Packet(uint8_t* data,
size_t capacity,
Listener* listener,
Local<Object> object,
const SocketAddress& destination,
std::shared_ptr<Data> data)
: ReqWrap<uv_udp_send_t>(env, object, PROVIDER_QUIC_PACKET),
const SocketAddress& destination)
: data_(data),
capacity_(capacity),
length_(capacity),
listener_(listener),
destination_(destination),
data_(std::move(data)) {
ClearWeak();
Debug(this, "Created a new packet");
}
Packet::Packet(Environment* env,
Listener* listener,
Local<Object> object,
const SocketAddress& destination,
size_t length,
const char* diagnostic_label)
: Packet(env,
listener,
object,
destination,
std::make_shared<Data>(length, diagnostic_label)) {}
void Packet::Done(int status) {
Debug(this, "Packet is done with status %d", status);
BaseObjectPtr<Packet> self(this);
self->MakeWeak();
if (listener_ != nullptr && IsDispatched()) {
listener_->PacketDone(status);
}
// As a performance optimization, we add this packet to a freelist
// rather than deleting it but only if the freelist isn't too
// big, we don't want to accumulate these things forever.
auto& binding = BindingData::Get(env());
if (binding.packet_freelist.size() >= kMaxFreeList) {
Debug(this, "Freelist full, destroying packet");
data_.reset();
return;
}
Debug(this, "Returning packet to freelist");
listener_ = nullptr;
data_.reset();
Reset();
binding.packet_freelist.push_back(std::move(self));
}
Packet::operator bool() const {
return data_ != nullptr;
}
req_{} {}
std::string Packet::ToString() const {
if (!data_) return "Packet (<empty>)";
return "Packet (" + data_->ToString() + ")";
std::string res = "Packet(";
#ifdef DEBUG
if (diagnostic_label_) {
res += diagnostic_label_;
res += ", ";
}
#endif
res += std::to_string(length_);
res += ")";
return res;
}
void Packet::MemoryInfo(MemoryTracker* tracker) const {
tracker->TrackField("destination", destination_);
if (data_) tracker->TrackField("data", data_);
}
// ============================================================================
// Static factory methods
BaseObjectPtr<Packet> Packet::CreateRetryPacket(
Environment* env,
Listener* listener,
const PathDescriptor& path_descriptor,
const TokenSecret& token_secret) {
Packet::Ptr Packet::CreateRetryPacket(Endpoint& endpoint,
const PathDescriptor& path_descriptor,
const TokenSecret& token_secret) {
auto& random = CID::Factory::random();
CID cid = random.Generate();
RetryToken token(path_descriptor.version,
@@ -225,7 +75,7 @@ BaseObjectPtr<Packet> Packet::CreateRetryPacket(
cid,
path_descriptor.dcid,
token_secret);
if (!token) return {};
if (!token) return Ptr();
const ngtcp2_vec& vec = token;
@@ -233,7 +83,7 @@ BaseObjectPtr<Packet> Packet::CreateRetryPacket(
vec.len + (2 * NGTCP2_MAX_CIDLEN) + path_descriptor.scid.length() + 8;
auto packet =
Create(env, listener, path_descriptor.remote_address, pktlen, "retry");
endpoint.CreatePacket(path_descriptor.remote_address, pktlen, "retry");
if (!packet) return packet;
ngtcp2_vec dest = *packet;
@@ -246,108 +96,80 @@ BaseObjectPtr<Packet> Packet::CreateRetryPacket(
path_descriptor.dcid,
vec.base,
vec.len);
if (nwrite <= 0) {
packet->CancelPacket();
return {};
}
if (nwrite <= 0) return Ptr();
packet->Truncate(static_cast<size_t>(nwrite));
return packet;
}
BaseObjectPtr<Packet> Packet::CreateConnectionClosePacket(
Environment* env,
Listener* listener,
Packet::Ptr Packet::CreateConnectionClosePacket(
Endpoint& endpoint,
const SocketAddress& destination,
ngtcp2_conn* conn,
const QuicError& error) {
auto packet = Create(
env, listener, destination, kDefaultMaxPacketLength, "connection close");
auto packet = endpoint.CreatePacket(
destination, kDefaultMaxPacketLength, "connection close");
if (!packet) return packet;
ngtcp2_vec vec = *packet;
ssize_t nwrite = ngtcp2_conn_write_connection_close(
conn, nullptr, nullptr, vec.base, vec.len, error, uv_hrtime());
if (nwrite < 0) {
packet->CancelPacket();
return {};
}
if (nwrite < 0) return Ptr();
packet->Truncate(static_cast<size_t>(nwrite));
return packet;
}
BaseObjectPtr<Packet> Packet::CreateImmediateConnectionClosePacket(
Environment* env,
Listener* listener,
Packet::Ptr Packet::CreateImmediateConnectionClosePacket(
Endpoint& endpoint,
const PathDescriptor& path_descriptor,
const QuicError& reason) {
auto packet = Create(env,
listener,
path_descriptor.remote_address,
kDefaultMaxPacketLength,
"immediate connection close (endpoint)");
auto packet = endpoint.CreatePacket(path_descriptor.remote_address,
kDefaultMaxPacketLength,
"immediate connection close (endpoint)");
if (!packet) return packet;
ngtcp2_vec vec = *packet;
ssize_t nwrite = ngtcp2_crypto_write_connection_close(
vec.base,
vec.len,
path_descriptor.version,
path_descriptor.dcid,
path_descriptor.scid,
reason.code(),
// We do not bother sending a reason string here, even if
// there is one in the QuicError
nullptr,
0);
if (nwrite <= 0) {
packet->CancelPacket();
return {};
}
ssize_t nwrite = ngtcp2_crypto_write_connection_close(vec.base,
vec.len,
path_descriptor.version,
path_descriptor.dcid,
path_descriptor.scid,
reason.code(),
nullptr,
0);
if (nwrite <= 0) return Ptr();
packet->Truncate(static_cast<size_t>(nwrite));
return packet;
}
BaseObjectPtr<Packet> Packet::CreateStatelessResetPacket(
Environment* env,
Listener* listener,
Packet::Ptr Packet::CreateStatelessResetPacket(
Endpoint& endpoint,
const PathDescriptor& path_descriptor,
const TokenSecret& token_secret,
size_t source_len) {
// Per the QUIC spec, a stateless reset token must be strictly smaller than
// the packet that triggered it. This is one of the mechanisms to prevent
// infinite looping exchange of stateless tokens with the peer. An endpoint
// should never send a stateless reset token smaller than 41 bytes per the
// QUIC spec. The reason is that packets less than 41 bytes may allow an
// observer to reliably determine that it's a stateless reset.
// the packet that triggered it.
size_t pktlen = source_len - 1;
if (pktlen < kMinStatelessResetLen) return {};
if (pktlen < kMinStatelessResetLen) return Ptr();
StatelessResetToken token(token_secret, path_descriptor.dcid);
uint8_t random[kRandlen];
CHECK(ncrypto::CSPRNG(random, kRandlen));
auto packet = Create(env,
listener,
path_descriptor.remote_address,
kDefaultMaxPacketLength,
"stateless reset");
auto packet = endpoint.CreatePacket(path_descriptor.remote_address,
kDefaultMaxPacketLength,
"stateless reset");
if (!packet) return packet;
ngtcp2_vec vec = *packet;
ssize_t nwrite = ngtcp2_pkt_write_stateless_reset(
vec.base, pktlen, token, random, kRandlen);
if (nwrite <= static_cast<ssize_t>(kMinStatelessResetLen)) {
packet->CancelPacket();
return {};
}
if (nwrite <= static_cast<ssize_t>(kMinStatelessResetLen)) return Ptr();
packet->Truncate(static_cast<size_t>(nwrite));
return packet;
}
BaseObjectPtr<Packet> Packet::CreateVersionNegotiationPacket(
Environment* env,
Listener* listener,
const PathDescriptor& path_descriptor) {
Packet::Ptr Packet::CreateVersionNegotiationPacket(
Endpoint& endpoint, const PathDescriptor& path_descriptor) {
const auto generateReservedVersion = [&] {
socklen_t addrlen = path_descriptor.remote_address.length();
uint32_t h = 0x811C9DC5u;
@@ -375,11 +197,9 @@ BaseObjectPtr<Packet> Packet::CreateVersionNegotiationPacket(
size_t pktlen = path_descriptor.dcid.length() +
path_descriptor.scid.length() + (sizeof(sv)) + 7;
auto packet = Create(env,
listener,
path_descriptor.remote_address,
kDefaultMaxPacketLength,
"version negotiation");
auto packet = endpoint.CreatePacket(path_descriptor.remote_address,
kDefaultMaxPacketLength,
"version negotiation");
if (!packet) return packet;
ngtcp2_vec vec = *packet;
@@ -393,16 +213,12 @@ BaseObjectPtr<Packet> Packet::CreateVersionNegotiationPacket(
path_descriptor.scid.length(),
sv,
arraysize(sv));
if (nwrite <= 0) {
packet->CancelPacket();
return {};
}
if (nwrite <= 0) return Ptr();
packet->Truncate(static_cast<size_t>(nwrite));
return packet;
}
} // namespace quic
} // namespace node
} // namespace node::quic
#endif // OPENSSL_NO_QUIC
#endif // HAVE_OPENSSL && HAVE_QUIC
+87 -89
View File
@@ -2,16 +2,11 @@
#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
#include <base_object.h>
#include <env.h>
#include <ngtcp2/ngtcp2.h>
#include <node_external_reference.h>
#include <node_sockaddr.h>
#include <req_wrap.h>
#include <uv.h>
#include <v8.h>
#include <string>
#include "bindingdata.h"
#include "arena.h"
#include "cid.h"
#include "data.h"
#include "defs.h"
@@ -19,6 +14,8 @@
namespace node::quic {
class Endpoint;
struct PathDescriptor final {
uint32_t version;
const CID& dcid;
@@ -28,127 +25,128 @@ struct PathDescriptor final {
std::string ToString() const;
};
// A Packet encapsulates serialized outbound QUIC data. Packets must never be
// larger than the path MTU. The default QUIC packet maximum length is 1200
// bytes, which we assume by default. The packet storage will be stack allocated
// up to this size.
// A Packet encapsulates serialized outbound QUIC data.
//
// Packets are maintained in a freelist held by the BindingData instance. When
// using Create() to create a Packet, we'll check to see if there is a free
// packet in the freelist and use it instead of starting fresh with a new
// packet. The freelist can store at most kMaxFreeList packets. This is a
// performance optimization to avoid excessive allocation churn when creating
// lots of packets since each one is ReqWrap and has a fair amount of associated
// overhead. However, we don't want to accumulate too many of these in the
// freelist either, so we cap the size.
// Packets are allocated from an ArenaPool owned by the Endpoint, which
// allocates fixed-size slots from contiguous memory blocks. This avoids
// heap fragmentation and the overhead of per-packet V8 object allocation.
//
// Packets are always encrypted so their content should be considered opaque
// to us. We leave it entirely up to ngtcp2 how to encode QUIC frames into
// the packet.
class Packet final : public ReqWrap<uv_udp_send_t> {
private:
struct Data;
// Each Packet contains:
// - A uv_udp_send_t for the libuv send operation
// - A destination socket address
// - A data buffer (trailing memory in the arena slot)
//
// Packets are always encrypted; their content is opaque. We leave it
// entirely up to ngtcp2 how to encode QUIC frames into the packet.
//
// Member layout is ordered so that fields touched on the hot path
// (data_, capacity_, length_, listener_) share the first cache line.
// The uv_udp_send_t (320 bytes, only touched by libuv) is placed last.
class Packet final {
public:
JS_CONSTRUCTOR(Packet);
using Ptr = ArenaPool<Packet>::Ptr;
class Listener {
public:
virtual void PacketDone(int status) = 0;
};
// Do not use the Packet constructors directly to create them. These are
// public only to support MakeBaseObject. Use the Create, or Create variants
// to create or acquire packet instances.
Packet(Environment* env,
// Constructor takes the trailing data buffer and capacity from the
// arena slot, plus the listener and destination. The data/capacity
// are injected by ArenaPool::AcquireExtra().
Packet(uint8_t* data,
size_t capacity,
Listener* listener,
v8::Local<v8::Object> object,
const SocketAddress& destination,
size_t length,
const char* diagnostic_label = "<unknown>");
Packet(Environment* env,
Listener* listener,
v8::Local<v8::Object> object,
const SocketAddress& destination,
std::shared_ptr<Data> data);
const SocketAddress& destination);
DISALLOW_COPY_AND_MOVE(Packet)
const SocketAddress& destination() const;
size_t length() const;
operator uv_buf_t() const;
operator ngtcp2_vec() const;
operator bool() const;
// --- Inline accessors (hot path) ---
// Modify the size of the packet after ngtcp2 has written
// to it. len must be <= length(). We call this after we've
// asked ngtcp2 to encode frames into the packet and ngtcp2
// tells us how many of the packets bytes were used.
void Truncate(size_t len);
uint8_t* data() { return data_; }
const uint8_t* data() const { return data_; }
size_t length() const { return length_; }
size_t capacity() const { return capacity_; }
const SocketAddress& destination() const { return destination_; }
Listener* listener() const { return listener_; }
uv_udp_send_t* req() { return &req_; }
// Create (or acquire from the freelist) a Packet with the given
// destination and length. The diagnostic_label is used to help
// identify the packet purpose in debugging output.
static BaseObjectPtr<Packet> Create(
Environment* env,
Listener* listener,
const SocketAddress& destination,
size_t length = kDefaultMaxPacketLength,
const char* diagnostic_label = "<unknown>");
operator uv_buf_t() const {
return uv_buf_init(reinterpret_cast<char*>(data_), length_);
}
operator ngtcp2_vec() const { return ngtcp2_vec{data_, length_}; }
BaseObjectPtr<Packet> Clone() const;
// Modify the logical size of the packet after ngtcp2 has written
// to it. len must be <= capacity().
void Truncate(size_t len) {
DCHECK_LE(len, capacity_);
length_ = len;
}
void MemoryInfo(MemoryTracker* tracker) const override;
SET_MEMORY_INFO_NAME(Packet)
SET_SELF_SIZE(Packet)
// Recover Packet* from a uv_udp_send_t* in the libuv send callback.
static Packet* FromReq(uv_udp_send_t* req) {
return ContainerOf(&Packet::req_, req);
}
std::string ToString() const;
// --- Static factory methods ---
// These create fully-formed packets for specific QUIC operations.
// They acquire from the endpoint's packet pool and return Ptr.
// An empty Ptr indicates failure.
static BaseObjectPtr<Packet> CreateRetryPacket(
Environment* env,
Listener* listener,
[[nodiscard]] static Ptr CreateRetryPacket(
Endpoint& endpoint,
const PathDescriptor& path_descriptor,
const TokenSecret& token_secret);
static BaseObjectPtr<Packet> CreateConnectionClosePacket(
Environment* env,
Listener* listener,
[[nodiscard]] static Ptr CreateConnectionClosePacket(
Endpoint& endpoint,
const SocketAddress& destination,
ngtcp2_conn* conn,
const QuicError& error);
static BaseObjectPtr<Packet> CreateImmediateConnectionClosePacket(
Environment* env,
Listener* listener,
[[nodiscard]] static Ptr CreateImmediateConnectionClosePacket(
Endpoint& endpoint,
const PathDescriptor& path_descriptor,
const QuicError& reason);
static BaseObjectPtr<Packet> CreateStatelessResetPacket(
Environment* env,
Listener* listener,
[[nodiscard]] static Ptr CreateStatelessResetPacket(
Endpoint& endpoint,
const PathDescriptor& path_descriptor,
const TokenSecret& token_secret,
size_t source_len);
static BaseObjectPtr<Packet> CreateVersionNegotiationPacket(
Environment* env,
Listener* listener,
const PathDescriptor& path_descriptor);
[[nodiscard]] static Ptr CreateVersionNegotiationPacket(
Endpoint& endpoint, const PathDescriptor& path_descriptor);
// Called when the packet is done being sent.
void Done(int status = 0);
inline void CancelPacket() { Done(UV_ECANCELED); }
// --- Diagnostic label: zero cost in release builds ---
#ifdef DEBUG
void set_diagnostic_label(const char* label) {
diagnostic_label_ = label;
}
const char* diagnostic_label() const {
return diagnostic_label_;
}
#else
void set_diagnostic_label(const char*) {}
#endif
std::string ToString() const;
private:
static BaseObjectPtr<Packet> FromFreeList(Environment* env,
std::shared_ptr<Data> data,
Listener* listener,
const SocketAddress& destination);
// Hot fields first — all on cache line 0 during the fill loop.
uint8_t* data_;
size_t capacity_;
size_t length_;
Listener* listener_;
// Touched at send time.
SocketAddress destination_;
std::shared_ptr<Data> data_;
// Only touched by libuv during uv_udp_send and in the send callback.
uv_udp_send_t req_;
#ifdef DEBUG
const char* diagnostic_label_ = nullptr;
#endif
};
} // namespace node::quic
+22 -33
View File
@@ -1,7 +1,6 @@
#if HAVE_OPENSSL && HAVE_QUIC
#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include "session.h"
#include <aliased_struct-inl.h>
#include <async_wrap-inl.h>
#include <crypto/crypto_util.h>
@@ -29,6 +28,7 @@
#include "ncrypto.h"
#include "packet.h"
#include "preferredaddress.h"
#include "session.h"
#include "sessionticket.h"
#include "streams.h"
#include "tlscontext.h"
@@ -1719,7 +1719,7 @@ bool Session::Receive(Store&& store,
return false;
}
void Session::Send(const BaseObjectPtr<Packet>& packet) {
void Session::Send(Packet::Ptr packet) {
// Sending a Packet is generally best effort. If we're not in a state
// where we can send a packet, it's ok to drop it on the floor. The
// packet loss mechanisms will cause the packet data to be resent later
@@ -1731,21 +1731,21 @@ void Session::Send(const BaseObjectPtr<Packet>& packet) {
DCHECK(!is_in_draining_period());
if (!can_send_packets()) [[unlikely]] {
return packet->CancelPacket();
// Ptr destructor releases the packet back to the pool.
return;
}
Debug(this, "Session is sending %s", packet->ToString());
auto& stats_ = impl_->stats_;
STAT_INCREMENT_N(Stats, bytes_sent, packet->length());
endpoint().Send(packet);
endpoint().Send(std::move(packet));
}
void Session::Send(const BaseObjectPtr<Packet>& packet,
const PathStorage& path) {
void Session::Send(Packet::Ptr packet, const PathStorage& path) {
DCHECK(!is_destroyed());
DCHECK(!is_in_draining_period());
UpdatePath(path);
Send(packet);
Send(std::move(packet));
}
datagram_id Session::SendDatagram(Store&& data) {
@@ -1778,7 +1778,7 @@ datagram_id Session::SendDatagram(Store&& data) {
return 0;
}
BaseObjectPtr<Packet> packet;
Packet::Ptr packet;
uint8_t* pos = nullptr;
int accepted = 0;
ngtcp2_vec vec = data;
@@ -1804,11 +1804,10 @@ datagram_id Session::SendDatagram(Store&& data) {
// datagram. It's entirely up to ngtcp2 whether to include the datagram
// in the packet on each call to ngtcp2_conn_writev_datagram.
if (!packet) {
packet = Packet::Create(env(),
endpoint(),
impl_->remote_address_,
ngtcp2_conn_get_max_tx_udp_payload_size(*this),
"datagram");
packet = endpoint().CreatePacket(
impl_->remote_address_,
ngtcp2_conn_get_max_tx_udp_payload_size(*this),
"datagram");
// Typically sending datagrams is best effort, but if we cannot create
// the packet, then we handle it as a fatal error as that indicates
// something else is likely very wrong.
@@ -1817,7 +1816,7 @@ datagram_id Session::SendDatagram(Store&& data) {
Close(CloseMethod::SILENT);
return 0;
}
pos = ngtcp2_vec(*packet).base;
pos = packet->data();
}
ssize_t nwrite = ngtcp2_conn_writev_datagram(*this,
@@ -1840,7 +1839,6 @@ datagram_id Session::SendDatagram(Store&& data) {
// not fit. Since datagrams are best effort, we are going to abandon
// the attempt and just return.
DCHECK_EQ(accepted, 0);
packet->CancelPacket();
return 0;
}
case NGTCP2_ERR_WRITE_MORE: {
@@ -1853,14 +1851,12 @@ datagram_id Session::SendDatagram(Store&& data) {
// The remote endpoint does not want to accept datagrams. That's ok,
// just return 0.
DCHECK_EQ(accepted, 0);
packet->CancelPacket();
return 0;
}
case NGTCP2_ERR_INVALID_ARGUMENT: {
// The datagram is too large. That should have been caught above but
// that's ok. We'll just abandon the attempt and return.
DCHECK_EQ(accepted, 0);
packet->CancelPacket();
return 0;
}
case NGTCP2_ERR_PKT_NUM_EXHAUSTED: {
@@ -1896,7 +1892,6 @@ datagram_id Session::SendDatagram(Store&& data) {
break;
}
}
packet->CancelPacket();
SetLastError(QuicError::ForTransport(nwrite));
Close(CloseMethod::SILENT);
return 0;
@@ -1906,8 +1901,8 @@ datagram_id Session::SendDatagram(Store&& data) {
// Note that this doesn't mean that the packet actually contains the
// datagram! We'll check that next by checking the accepted value.
packet->Truncate(nwrite);
Send(packet);
packet.reset();
Send(std::move(packet));
// packet is now empty; next loop iteration creates a new one.
if (accepted) {
// Yay! The datagram was accepted into the packet we just sent and we can
@@ -2292,12 +2287,9 @@ void Session::SendConnectionClose() {
if (is_server()) {
if (auto packet = Packet::CreateConnectionClosePacket(
env(),
endpoint(),
impl_->remote_address_,
*this,
impl_->last_error_)) [[likely]] {
return Send(packet);
endpoint(), impl_->remote_address_, *this, impl_->last_error_))
[[likely]] {
return Send(std::move(packet));
}
// If we are unable to create a connection close packet then
@@ -2309,11 +2301,9 @@ void Session::SendConnectionClose() {
return ErrorAndSilentClose();
}
auto packet = Packet::Create(env(),
endpoint(),
impl_->remote_address_,
kDefaultMaxPacketLength,
"immediate connection close (client)");
auto packet = endpoint().CreatePacket(impl_->remote_address_,
kDefaultMaxPacketLength,
"immediate connection close (client)");
if (!packet) [[unlikely]] {
return ErrorAndSilentClose();
}
@@ -2329,12 +2319,11 @@ void Session::SendConnectionClose() {
uv_hrtime());
if (nwrite < 0) [[unlikely]] {
packet->CancelPacket();
return ErrorAndSilentClose();
}
packet->Truncate(nwrite);
return Send(packet);
return Send(std::move(packet));
}
void Session::OnTimeout() {
+2 -2
View File
@@ -328,8 +328,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
const SocketAddress& local_address,
const SocketAddress& remote_address);
void Send(const BaseObjectPtr<Packet>& packet);
void Send(const BaseObjectPtr<Packet>& packet, const PathStorage& path);
void Send(Packet::Ptr packet);
void Send(Packet::Ptr packet, const PathStorage& path);
datagram_id SendDatagram(Store&& data);
// A non-const variation to allow certain modifications.
+232
View File
@@ -0,0 +1,232 @@
#if HAVE_OPENSSL && HAVE_QUIC
#include <gtest/gtest.h>
#include <quic/arena.h>
#include <cstdint>
#include <utility>
#include <vector>
namespace node::quic {
namespace {
// A simple test type with non-trivial constructor/destructor to verify
// that ArenaPool properly manages object lifecycles.
struct TestObj {
int value;
bool* destroyed; // pointer to external flag set on destruction
TestObj(int v, bool* d) : value(v), destroyed(d) {}
~TestObj() {
if (destroyed) *destroyed = true;
}
};
// A test type that accepts extra data (like Packet).
struct TestObjWithExtra {
uint8_t* data;
size_t capacity;
int tag;
TestObjWithExtra(uint8_t* d, size_t cap, int t)
: data(d), capacity(cap), tag(t) {}
};
TEST(QuicArenaPool, BasicAcquireRelease) {
ArenaPool<TestObj> pool(0, 4);
EXPECT_EQ(pool.total_slots(), 0u);
EXPECT_EQ(pool.in_use_count(), 0u);
bool destroyed = false;
{
auto ptr = pool.Acquire(42, &destroyed);
ASSERT_TRUE(ptr);
EXPECT_EQ(ptr->value, 42);
EXPECT_EQ(pool.total_slots(), 4); // first block allocated
EXPECT_EQ(pool.in_use_count(), 1);
EXPECT_EQ(pool.block_count(), 1);
EXPECT_FALSE(destroyed);
}
// Ptr went out of scope — destructor called, slot released
EXPECT_TRUE(destroyed);
EXPECT_EQ(pool.in_use_count(), 0);
}
TEST(QuicArenaPool, GrowsOnDemand) {
ArenaPool<TestObj> pool(0, 2); // 2 slots per block
bool d1 = false, d2 = false, d3 = false;
auto p1 = pool.Acquire(1, &d1);
auto p2 = pool.Acquire(2, &d2);
EXPECT_EQ(pool.block_count(), 1);
EXPECT_EQ(pool.total_slots(), 2);
// This should trigger a second block
auto p3 = pool.Acquire(3, &d3);
EXPECT_EQ(pool.block_count(), 2);
EXPECT_EQ(pool.total_slots(), 4);
EXPECT_EQ(pool.in_use_count(), 3);
EXPECT_EQ(p1->value, 1);
EXPECT_EQ(p2->value, 2);
EXPECT_EQ(p3->value, 3);
}
TEST(QuicArenaPool, PtrMoveSemantics) {
ArenaPool<TestObj> pool(0, 4);
bool destroyed = false;
auto p1 = pool.Acquire(10, &destroyed);
ASSERT_TRUE(p1);
// Move construction
auto p2 = std::move(p1);
EXPECT_FALSE(p1); // NOLINT — testing moved-from state
ASSERT_TRUE(p2);
EXPECT_EQ(p2->value, 10);
EXPECT_FALSE(destroyed);
// Move assignment
bool d2 = false;
auto p3 = pool.Acquire(20, &d2);
p3 = std::move(p2);
EXPECT_TRUE(d2); // p3's old object destroyed
EXPECT_FALSE(p2); // NOLINT
ASSERT_TRUE(p3);
EXPECT_EQ(p3->value, 10);
EXPECT_FALSE(destroyed);
p3.reset();
EXPECT_TRUE(destroyed);
}
TEST(QuicArenaPool, PtrRelease) {
ArenaPool<TestObj> pool(0, 4);
bool destroyed = false;
auto ptr = pool.Acquire(99, &destroyed);
TestObj* raw = ptr.release();
EXPECT_FALSE(ptr); // Ptr is now empty
EXPECT_FALSE(destroyed); // Object not yet destroyed
// Verify the raw pointer is valid
EXPECT_EQ(raw->value, 99);
EXPECT_EQ(pool.in_use_count(), 1);
// Release via static method — this calls ~TestObj and returns slot
ArenaPool<TestObj>::Release(raw);
EXPECT_TRUE(destroyed);
EXPECT_EQ(pool.in_use_count(), 0);
}
TEST(QuicArenaPool, MaybeGCFreesEmptyBlocks) {
ArenaPool<TestObj> pool(0, 2);
// Fill two blocks
bool d[4] = {};
auto p1 = pool.Acquire(1, &d[0]);
auto p2 = pool.Acquire(2, &d[1]);
auto p3 = pool.Acquire(3, &d[2]);
auto p4 = pool.Acquire(4, &d[3]);
EXPECT_EQ(pool.block_count(), 2u);
EXPECT_EQ(pool.total_slots(), 4u);
EXPECT_EQ(pool.in_use_count(), 4u);
// Release all four. With in_use == 0 and 2 blocks, GC should
// free one block (keeping at least one).
p1.reset();
p2.reset();
p3.reset();
p4.reset();
EXPECT_EQ(pool.in_use_count(), 0u);
// GC should have freed one block (triggered when in_use < total/2)
EXPECT_EQ(pool.block_count(), 1u);
EXPECT_EQ(pool.total_slots(), 2u);
}
TEST(QuicArenaPool, FlushKeepsAtLeastOneBlock) {
ArenaPool<TestObj> pool(0, 2);
bool d = false;
auto p = pool.Acquire(1, &d);
p.reset();
// One block, all slots free.
EXPECT_EQ(pool.block_count(), 1);
pool.Flush();
// Should keep at least one block even though it's fully free.
EXPECT_EQ(pool.block_count(), 1);
}
TEST(QuicArenaPool, AcquireExtraWiresExtraData) {
static constexpr size_t kExtraBytes = 256;
ArenaPool<TestObjWithExtra> pool(kExtraBytes, 4);
auto ptr = pool.AcquireExtra(42);
ASSERT_TRUE(ptr);
EXPECT_EQ(ptr->tag, 42);
EXPECT_EQ(ptr->capacity, kExtraBytes);
EXPECT_NE(ptr->data, nullptr);
// The extra data pointer from Ptr should match what the constructor got.
EXPECT_EQ(ptr.extra_data(), ptr->data);
EXPECT_EQ(ptr.extra_bytes(), kExtraBytes);
// Verify we can write to the extra data region.
std::memset(ptr->data, 0xAB, kExtraBytes);
EXPECT_EQ(ptr->data[0], 0xAB);
EXPECT_EQ(ptr->data[kExtraBytes - 1], 0xAB);
}
TEST(QuicArenaPool, SlotReuse) {
ArenaPool<TestObj> pool(0, 2);
bool d1 = false, d2 = false;
// Acquire and release
{
auto p = pool.Acquire(1, &d1);
EXPECT_EQ(pool.total_slots(), 2);
}
EXPECT_TRUE(d1);
// Acquire again — should reuse the slot without growing
auto p2 = pool.Acquire(2, &d2);
EXPECT_EQ(pool.total_slots(), 2); // no growth
EXPECT_EQ(pool.block_count(), 1);
EXPECT_EQ(p2->value, 2);
}
TEST(QuicArenaPool, SlotSizeIncludesExtraBytes) {
static constexpr size_t kExtra = 1200;
ArenaPool<TestObjWithExtra> pool(kExtra, 4);
// Slot size should accommodate header + TestObjWithExtra + 1200 extra bytes
EXPECT_GE(pool.slot_size(), sizeof(TestObjWithExtra) + kExtra);
EXPECT_EQ(pool.extra_bytes(), kExtra);
}
TEST(QuicArenaPool, MultipleAcquireReleaseChurn) {
ArenaPool<TestObj> pool(0, 4);
// Simulate a burst of acquires and releases
for (int round = 0; round < 10; round++) {
std::vector<ArenaPool<TestObj>::Ptr> ptrs;
bool destroyed[8] = {};
for (int i = 0; i < 8; i++) {
ptrs.push_back(pool.Acquire(i, &destroyed[i]));
}
EXPECT_EQ(pool.in_use_count(), 8);
// Release all
ptrs.clear();
EXPECT_EQ(pool.in_use_count(), 0);
for (int i = 0; i < 8; i++) {
EXPECT_TRUE(destroyed[i]);
}
}
// Pool may have grown but should be stable
EXPECT_EQ(pool.in_use_count(), 0);
}
} // namespace
} // namespace node::quic
#endif // HAVE_OPENSSL && HAVE_QUIC