diff --git a/binding.gyp b/binding.gyp index c35d8e66..8b918472 100644 --- a/binding.gyp +++ b/binding.gyp @@ -13,6 +13,7 @@ "target_name": "dd_pprof", "sources": [ "bindings/profilers/heap.cc", + "bindings/profilers/near-oom.cc", "bindings/profilers/wall.cc", "bindings/per-isolate-data.cc", "bindings/thread-cpu-clock.cc", @@ -40,6 +41,7 @@ "target_name": "test_dd_pprof", "sources": [ "bindings/profilers/heap.cc", + "bindings/profilers/near-oom.cc", "bindings/profilers/wall.cc", "bindings/per-isolate-data.cc", "bindings/thread-cpu-clock.cc", diff --git a/bindings/otel-thread-ctx.cc b/bindings/otel-thread-ctx.cc index ddee80cf..b7c7dfd4 100644 --- a/bindings/otel-thread-ctx.cc +++ b/bindings/otel-thread-ctx.cc @@ -23,15 +23,16 @@ // Node.js writer for the OTEP-4947 Thread Local Context Record, adapted for // the Node.js asynchronous context model. The record is wrapped in a JS -// object (CtxWrap) and stored in an AsyncLocalStorage instance; an -// out-of-process reader discovers it by walking the V8 isolate's +// object and stored in an AsyncLocalStorage instance; an out-of-process +// reader discovers it by walking the V8 isolate's // ContinuationPreservedEmbedderData to the AsyncContextFrame (a JS Map), -// looking up the ALS instance as the key, reading the resulting CtxWrap, -// and finally the record it owns. +// looking up the ALS instance as the key, and reading the record pointer +// out of the resulting object's internal field. That field points +// straight at the record. The record is preceded in memory by an instance +// of CtxWrap used for internal bookkeeping. #include "otel-thread-ctx.hh" -#include "defer.hh" #include "internal-field.hh" #include @@ -42,9 +43,9 @@ #include #include +#include #include -#include -#include +#include #include // Single thread-local read from outside the process via TLSDESC. It @@ -62,10 +63,10 @@ // - the (per-isolate) tagged address of the `undefined` singleton // (`undefined_addr`). After looking up the value for our ALS key in // the ACF map, the reader can compare against this to skip the -// JSObject / internal-field-0 dereference when no CtxWrap is +// JSObject / internal-field-0 dereference when no ThreadContext is // currently attached; without it, a reader walking through undefined // would have to rely on structural validation of the bytes at -// undefined+wrapped_object_offset to detect the absence. +// undefined+js_object_record_offset to detect the absence. // // Layout is part of the reader ABI: see the README "Discovery contract" // section and the static_asserts below. @@ -131,7 +132,7 @@ struct OtelThreadCtxRecord { uint8_t trace_id[16]; // offset 0 uint8_t span_id[8]; // offset 16 uint8_t valid; // offset 24 - uint8_t reserved; // offset 25 + uint8_t trace_flags; // offset 25 uint16_t attrs_data_size; // offset 26 uint8_t attrs_data[]; // offset 28; length is attrs_data_size }; @@ -140,23 +141,19 @@ static_assert(sizeof(OtelThreadCtxRecord) == 28, static_assert(offsetof(OtelThreadCtxRecord, trace_id) == 0, "trace_id offset"); static_assert(offsetof(OtelThreadCtxRecord, span_id) == 16, "span_id offset"); static_assert(offsetof(OtelThreadCtxRecord, valid) == 24, "valid offset"); -static_assert(offsetof(OtelThreadCtxRecord, reserved) == 25, "reserved offset"); +static_assert(offsetof(OtelThreadCtxRecord, trace_flags) == 25, + "trace_flags offset"); static_assert(offsetof(OtelThreadCtxRecord, attrs_data_size) == 26, "attrs_data_size offset"); static_assert(offsetof(OtelThreadCtxRecord, attrs_data) == 28, "attrs_data offset"); -struct OtelThreadCtxRecordDeleter { - void operator()(OtelThreadCtxRecord* p) const noexcept { free(p); } -}; -using OwnedRecord = - std::unique_ptr; - -// Floor on the attrs_data capacity of a freshly allocated record. Sized so -// the total allocation is one 64-byte cache line — matching the OTEP-4947 -// "frugal writer" guidance ("a frugal writer may aim to keep the entire -// record under 64 bytes") — and giving small records some slack so the -// first few appends (if any) can be in-place. +// Floor on the attrs_data capacity of a freshly allocated record. Matching the +// OTEP-4947 "frugal writer" guidance ("a frugal writer may aim to keep the +// entire record under 64 bytes") — and giving small records some slack so the +// first few appends (if any) can be in-place. The CtxWrap fields preceding +// the record in the same block are writer bookkeeping the reader never +// sees, so they don't count against that budget. constexpr size_t MIN_INITIAL_CAPACITY = 64 - sizeof(OtelThreadCtxRecord); // Upper bound on the attribute payload. Sized so the total record (28-byte @@ -167,39 +164,29 @@ constexpr size_t MIN_INITIAL_CAPACITY = 64 - sizeof(OtelThreadCtxRecord); // as best-effort. constexpr size_t MAX_ATTRS_DATA_SIZE = 640 - sizeof(OtelThreadCtxRecord); -// Wraps a heap-allocated OtelThreadCtxRecord. Lifetime is managed by V8 -// GC: when no JS code (or AsyncLocalStorage entry) holds a reference, the -// record is freed. -// -// Layout note for the reader: `record_` is private to C++ but its byte -// position within CtxWrap is part of the reader contract. It is the first -// field of the class, at offset zero. `capacity_` and -// `truncated_` sit after `record_` purely for the writer's own -// bookkeeping — the reader never touches them. -// Deliberately not a node::ObjectWrap. That base registers a per-instance -// environment cleanup hook in its constructor and calls -// RemoveEnvironmentCleanupHook from its destructor, which CHECKs that an -// Environment is current: -// -// node[107]: void node::RemoveEnvironmentCleanupHook(...) hooks.cc:142 -// Assertion failed: (env) != nullptr -// 3: otel_thread_ctx_nodejs::CtxWrap::~CtxWrap() -// -// A CtxWrap is owned by a weak V8 handle, so V8 chooses when it dies, and -// weak callbacks run during isolate teardown with no context entered — -// Environment::GetCurrent(isolate) returns null on `!isolate->InContext()` -// alone — so the CHECK fires and aborts. Reproducible today by creating a few -// thousand ThreadContexts and exiting normally; see the regression test. +// Wraps an OTEP-4947 record. The record isn't a separate allocation: a +// CtxWrap is one contiguous block whose head is the CtxWrap object itself +// and whose tail — starting at `RECORD_OFFSET`, immediately past the +// object's own fields — holds the record header followed by `capacity_` +// bytes of attrs_data. Lifetime is managed by V8 GC: when no JS code (or +// AsyncLocalStorage entry) holds a reference, the whole block is freed. // -// Note the CHECK is guarding something real, so this must not be worked -// around by skipping the removal: the Environment may well still be alive, -// and leaving a hook behind whose arg is a freed pointer turns an abort into -// a use-after-free at Drain(). The fix is to never register the per-instance -// hook, and to provide the teardown deletion it was giving us (see -// g_live_ctx_wraps below). +// Layout note for the reader: the holder JSObject's internal field 0 points +// at the record itself, not at the CtxWrap, so a reader that has walked to +// the holder is a single dereference away from the record and never has to +// know that CtxWrap exists at all. C++ code goes the other way with +// FromRecord(), which just subtracts RECORD_OFFSET. +// Deliberately not a node::ObjectWrap, for two independent reasons. First, it +// has a known bug in interaction with GC when numerous instances are created +// and can abort the process during isolate teardown; instances live at +// shutdown are deleted using DrainLive instead. Second, node::ObjectWrap owns +// internal field 0 — its Wrap() stores the ObjectWrap pointer there and its +// Unwrap() reads it back — so deriving from it would force that slot to hold a +// CtxWrap pointer, and every reader would be stuck with the extra hop through +// the wrapper that the layout above exists to avoid. Not deriving from it is +// what frees the slot for the record pointer. class CtxWrap { public: - ~CtxWrap(); static void Init(Local exports); CtxWrap(const CtxWrap&) = delete; @@ -212,6 +199,7 @@ class CtxWrap { static void DebugBytes(const FunctionCallbackInfo& args); static void Append(const FunctionCallbackInfo& args); static void Invalidate(const FunctionCallbackInfo& args); + static void SetTraceFlags(const FunctionCallbackInfo& args); static void IsTruncated(const FunctionCallbackInfo& args); // Encode the JS array at `attrs_val` into `out` as packed (key, len, value) @@ -227,28 +215,48 @@ class CtxWrap { std::vector* out, bool* out_truncated); - CtxWrap(OtelThreadCtxRecord* record, size_t capacity, bool truncated); + // Splice `appended` onto the end of the record: in place if it fits in the + // current allocation's slack, otherwise by moving the whole wrap to a larger + // one and repointing `holder`'s internal field at the new record. In that + // case `this` is destroyed before returning, so the caller must not touch it + // afterwards. Returns false if the allocation fails. + bool AppendEncoded(Local holder, + const std::vector& appended); + + explicit CtxWrap(size_t capacity); + ~CtxWrap(); - // Attach to the holder JSObject: store `this` in internal field 0 and take - // a weak handle on the holder, so V8 deletes us once it collects it. + // Allocate one zero-initialised block big enough for the object plus a + // record with `capacity` bytes of attrs_data, and construct the CtxWrap + // at its head. Returns nullptr if the allocation fails. CtxWraps are + // never created with plain `new` — the trailing record wouldn't be there. + static CtxWrap* Create(size_t capacity); + // Destroy and free a CtxWrap obtained from Create(). Runs the destructor + // and then `free`, which is what the `calloc` in Create() pairs with; + // `delete` would reach for `operator delete` instead, and the mismatch is + // undefined behaviour that a replaced global allocator or ASan will + // actually trip over. Deleting `operator delete` below makes reaching for + // it a compile error. + static void Destroy(CtxWrap* self); + static void operator delete(void*) = delete; + + // The record living in the tail of this object's own allocation. + OtelThreadCtxRecord* record(); + // The CtxWrap that precedes the `record`. Inverse of record(). + static CtxWrap* FromRecord(void* record); + + // Attach to the holder JSObject: store our record pointer in internal + // field 0 (the reader's entry point) and take a weak handle on the + // holder, so V8 destroys us once it collects it. void Wrap(Local holder); static CtxWrap* Unwrap(Local holder); static void WeakCallback(const v8::WeakCallbackInfo& data); + static void DrainLive(void* arg); - // The fields are kept in one access section because C++ leaves - // the relative layout of fields in different access controls - // implementation-defined. `record_` must come first — its offset - // within CtxWrap is part of the reader contract (see the - // static_assert below) — and is therefore `public`. The bookkeeping - // fields after it would normally be private, but the access change - // would let a conforming compiler reorder them in front of `record_`; - // exposing them publicly keeps everything in one ordering-stable - // block. Readers never touch them. - public: - OtelThreadCtxRecord* record_; - // attrs_data capacity in bytes of the record_ allocation. The total - // allocation is `sizeof(OtelThreadCtxRecord) + capacity_`. Always - // `record_->attrs_data_size <= capacity_ <= MAX_ATTRS_DATA_SIZE`. + // attrs_data capacity in bytes of the record in this object's tail. The + // total allocation is `RECORD_OFFSET + sizeof(OtelThreadCtxRecord) + + // capacity_`. Always `record()->attrs_data_size <= capacity_ <= + // MAX_ATTRS_DATA_SIZE`. size_t capacity_; // Set to true (once, never cleared) if at any point in this record's // lifetime — during New() or any subsequent Append() — at least one @@ -259,8 +267,8 @@ class CtxWrap { // attribute value, which can execute user JS (e.g. a custom // `toString`) that in turn calls `appendAttributes` on the same // ThreadContext. A reentrant Append would mutate attrs_data_size out - // from under the outer call's `current_used` snapshot, causing the - // outer memcpy to overwrite the reentrant call's bytes and the outer + // from under the outer call's snapshot of it, causing the outer memcpy + // to overwrite the reentrant call's bytes and the outer // attrs_data_size write to shrink the record. We reject the reentrant // call instead. New() doesn't need the guard because a freshly constructed // CtxWrap isn't observable to JS until New() returns. @@ -276,15 +284,37 @@ class CtxWrap { v8::Global handle_; }; -// Pin the offset of `record_` — the field the reader walks to from the -// JSObject's internal field 0. With no base class it is simply the first -// member, so the offset is zero and the published -// `threadlocal.native_wrap_fields_offset` is computed from this. -static_assert(std::is_standard_layout::value, - "CtxWrap must stay standard-layout: the reader contract depends " - "on offsetof(record_) being well-defined"); -static_assert(offsetof(CtxWrap, record_) == 0, - "record_ must be the first field of CtxWrap"); +// Byte offset of the record within a CtxWrap allocation: the record starts +// immediately after the object's own fields. Both record() and FromRecord() +// are defined in terms of it. +constexpr size_t RECORD_OFFSET = sizeof(CtxWrap); +static_assert(RECORD_OFFSET % alignof(OtelThreadCtxRecord) == 0, + "record must land on its natural alignment"); +// Most likely subsumed in the previous assert, but still call out directly +// that record must be 2-aligned as that's a requirement for storing it +// with v8::Object::SetAlignedPointerInInternalField(). +static_assert(RECORD_OFFSET % 2 == 0, "record must land on 2 boundary"); + +inline OtelThreadCtxRecord* CtxWrap::record() { + return reinterpret_cast( + reinterpret_cast(this) + RECORD_OFFSET); +} + +inline CtxWrap* CtxWrap::FromRecord(void* record) { + return reinterpret_cast(static_cast(record) - + RECORD_OFFSET); +} + +CtxWrap* CtxWrap::Create(size_t capacity) { + void* mem = calloc(1, RECORD_OFFSET + sizeof(OtelThreadCtxRecord) + capacity); + if (mem == nullptr) return nullptr; + return new (mem) CtxWrap(capacity); +} + +void CtxWrap::Destroy(CtxWrap* self) { + self->~CtxWrap(); + free(self); +} // Head of the live-CtxWrap list for this thread. Node pins each isolate to a // thread, and CtxWraps are only ever constructed and destroyed on their own @@ -293,13 +323,10 @@ static_assert(offsetof(CtxWrap, record_) == 0, // `otel_thread_ctx_nodejs_v1` above is thread-local for the same reason. thread_local CtxWrap* g_live_ctx_wraps = nullptr; -// Delete every CtxWrap V8 has not collected yet. This is the teardown deletion -// that node::ObjectWrap's per-instance cleanup hook used to provide; without it -// the records would simply leak at exit. Registered once per isolate from -// Init(), which runs at module initialisation with a context entered, so -// AddEnvironmentCleanupHook's own CHECK is satisfied, and never removed — it -// fires exactly once, at teardown, while the Environment is still alive. -void DrainLiveCtxWraps(void* arg) { +// Destroy every CtxWrap V8 has not collected yet. Without it the records would +// simply leak at exit. Registered once per isolate from Init(). It fires +// exactly once, at teardown, while the Environment is still alive. +void CtxWrap::DrainLive(void* arg) { auto* isolate = static_cast(arg); // We must allocate our own HandleScope here as node::FreeEnvironment wraps // RunCleanup in a SealHandleScope, so handle_.Get() below has to allocate @@ -312,8 +339,8 @@ void DrainLiveCtxWraps(void* arg) { CtxWrap* next = p->next_; p->pprev_ = nullptr; p->next_ = nullptr; - // Clear the holder's internal field (containing p as pointer value), so - // nothing can reach a dangling CtxWrap through it including the + // Clear the holder's internal field (containing p's record pointer), so + // nothing can reach a dangling record through it including the // out-of-process reader, which walks this slot. Being on the live list // means V8 has not collected the holder, so the handle is safe to read // here; the WeakCallback path cannot do this and does not need to, @@ -321,7 +348,8 @@ void DrainLiveCtxWraps(void* arg) { if (!p->handle_.IsEmpty()) { SetAlignedPointerInInternalField(p->handle_.Get(isolate), 0, nullptr); } - delete p; + std::atomic_signal_fence(std::memory_order_acq_rel); + Destroy(p); p = next; } g_live_ctx_wraps = nullptr; @@ -335,16 +363,15 @@ CtxWrap::~CtxWrap() { *pprev_ = next_; if (next_ != nullptr) next_->pprev_ = pprev_; } - free(record_); } void CtxWrap::WeakCallback(const v8::WeakCallbackInfo& data) { - delete data.GetParameter(); + Destroy(data.GetParameter()); } void CtxWrap::Wrap(Local holder) { Isolate* isolate = Isolate::GetCurrent(); - SetAlignedPointerInInternalField(holder, 0, this); + SetAlignedPointerInInternalField(holder, 0, record()); handle_.Reset(isolate, holder); handle_.SetWeak(this, &WeakCallback, v8::WeakCallbackType::kParameter); next_ = g_live_ctx_wraps; @@ -355,13 +382,14 @@ void CtxWrap::Wrap(Local holder) { CtxWrap* CtxWrap::Unwrap(Local holder) { if (holder->InternalFieldCount() < 1) return nullptr; - return static_cast(GetAlignedPointerFromInternalField(*holder, 0)); + void* record = GetAlignedPointerFromInternalField(*holder, 0); + if (record == nullptr) return nullptr; + return FromRecord(record); } -CtxWrap::CtxWrap(OtelThreadCtxRecord* record, size_t capacity, bool truncated) - : record_(record), - capacity_(capacity), - truncated_(truncated), +CtxWrap::CtxWrap(size_t capacity) + : capacity_(capacity), + truncated_(false), encoding_(false), pprev_(nullptr), next_(nullptr) {} @@ -380,6 +408,35 @@ bool CopyBytes(Local value, size_t expected_bytes, uint8_t* out) { return true; } +// Read a trace-flags argument: the W3C trace-flags byte that accompanies the +// trace and span ids. Absent, undefined or null means zero, which is what +// OTEP-4947 prescribes when no flags are known. Every value in 0..255 is +// accepted rather than masked to the currently defined bits: W3C requires +// unknown flag bits to be propagated, so they are not ours to drop. +bool ToTraceFlags(Isolate* isolate, Local value, uint8_t* out) { + if (value.IsEmpty() || value->IsUndefined() || value->IsNull()) { + *out = 0; + return true; + } + if (!value->IsNumber()) { + isolate->ThrowError("traceFlags must be an integer in 0..255"); + return false; + } + // NaN fails this comparison too, so the cast below is always in range. + const double d = value.As()->Value(); + if (!(d >= 0 && d <= 255)) { + isolate->ThrowError("traceFlags must be an integer in 0..255"); + return false; + } + const uint8_t byte = static_cast(d); + if (static_cast(byte) != d) { + isolate->ThrowError("traceFlags must be an integer in 0..255"); + return false; + } + *out = byte; + return true; +} + // Encode the JS array `attrs_val` (positional, index N = uint8 key N) into // `*out` as packed `(key:u8, len:u8, value:u8[len])` entries. // `existing_size` is the number of bytes already in any pre-existing @@ -473,10 +530,10 @@ void CtxWrap::New(const FunctionCallbackInfo& args) { isolate->ThrowError("ThreadContext must be called with `new`"); return; } - if (args.Length() < 2 || args.Length() > 3) { + if (args.Length() < 2 || args.Length() > 4) { isolate->ThrowError( - "ThreadContext expects 2 or 3 arguments: traceId, spanId, " - "attributes?"); + "ThreadContext expects 2 to 4 arguments: traceId, spanId, " + "traceFlags?, attributes?"); return; } @@ -492,6 +549,8 @@ void CtxWrap::New(const FunctionCallbackInfo& args) { isolate->ThrowError("spanId must be an 8-byte Uint8Array"); return; } + uint8_t trace_flags = 0; + if (!ToTraceFlags(isolate, args[2], &trace_flags)) return; // Encode attributes into a transient buffer first so we can size the // record allocation correctly. The 612-byte attrs_data cap mirrors the @@ -501,7 +560,7 @@ void CtxWrap::New(const FunctionCallbackInfo& args) { // truncated flag below. std::vector attrs_buf; bool truncated = false; - if (!EncodeAttrs(isolate, context, args[2], 0, &attrs_buf, &truncated)) { + if (!EncodeAttrs(isolate, context, args[3], 0, &attrs_buf, &truncated)) { return; } @@ -512,14 +571,16 @@ void CtxWrap::New(const FunctionCallbackInfo& args) { // doesn't change the geometric-growth amortized cost of subsequent // appends). size_t capacity = std::max(attrs_buf.size(), MIN_INITIAL_CAPACITY); - const size_t total = sizeof(OtelThreadCtxRecord) + capacity; - OwnedRecord record(static_cast(calloc(1, total))); - if (!record) { + CtxWrap* self = CtxWrap::Create(capacity); + if (self == nullptr) { isolate->ThrowError("allocation failed"); return; } + self->truncated_ = truncated; + OtelThreadCtxRecord* record = self->record(); memcpy(record->trace_id, trace_id, sizeof(trace_id)); memcpy(record->span_id, span_id, sizeof(span_id)); + record->trace_flags = trace_flags; record->attrs_data_size = static_cast(attrs_buf.size()); if (!attrs_buf.empty()) { memcpy(record->attrs_data, attrs_buf.data(), attrs_buf.size()); @@ -533,15 +594,14 @@ void CtxWrap::New(const FunctionCallbackInfo& args) { std::atomic_signal_fence(std::memory_order_release); *reinterpret_cast(&record->valid) = 1; - CtxWrap* self = new CtxWrap(record.release(), capacity, truncated); + // Only now does the record become reachable — Wrap() is what publishes + // its address into the holder's internal field. self->Wrap(args.This()); args.GetReturnValue().Set(args.This()); } -// Append entries to the active record. Either modifies the record in place -// (if the appended bytes fit in the current allocation's slack) or -// reallocates to a larger one (geometrically), keeping invariant -// `record_->attrs_data_size <= capacity_`. +// `appendAttributes(attributes)`: validate and encode the attributes, then +// hand the encoded bytes to AppendEncoded to splice onto the active record. void CtxWrap::Append(const FunctionCallbackInfo& args) { Isolate* isolate = args.GetIsolate(); Local context = isolate->GetCurrentContext(); @@ -556,27 +616,31 @@ void CtxWrap::Append(const FunctionCallbackInfo& args) { return; } - // Reject reentrant Append on the same wrap. EncodeAttrs' `ToString` - // below can execute user JS, and if that JS calls `appendAttributes` - // on this same ThreadContext, the reentrant call would grow - // attrs_data_size out from under the outer call's `current_used` - // snapshot, causing the outer memcpy to overwrite the reentrant call's - // bytes and the outer attrs_data_size write to shrink the record. + // Reject reentrant Append on the same wrap. EncodeAttrs' element getters + // and `ToString` below can execute user JS, and if that JS calls + // `appendAttributes` on this same ThreadContext, the reentrant call would + // grow attrs_data_size out from under the outer call's snapshot of it, + // causing the outer memcpy to overwrite the reentrant call's bytes and + // the outer attrs_data_size write to shrink the record. if (self->encoding_) { isolate->ThrowError( "reentrant appendAttributes on the same ThreadContext is not allowed"); return; } - self->encoding_ = true; - defer { - self->encoding_ = false; - }; - const size_t current_used = self->record_->attrs_data_size; std::vector appended; bool truncated = false; - if (!EncodeAttrs( - isolate, context, args[0], current_used, &appended, &truncated)) { + self->encoding_ = true; + const bool encoded = EncodeAttrs(isolate, + context, + args[0], + self->record()->attrs_data_size, + &appended, + &truncated); + // EncodeAttrs was the only thing that can run user JS, so the + // reentrancy guard had to span only it. + self->encoding_ = false; + if (!encoded) { return; } if (truncated) self->truncated_ = true; @@ -586,10 +650,23 @@ void CtxWrap::Append(const FunctionCallbackInfo& args) { // already at the cap. if (appended.empty()) return; + if (!self->AppendEncoded(args.This(), appended)) { + isolate->ThrowError("allocation failed"); + } +} + +// Splice `appended` onto the end of the record: in place if the bytes fit in +// the current allocation's slack, otherwise by moving the whole wrap to a +// larger allocation (grown geometrically), keeping invariant +// `record()->attrs_data_size <= capacity_`. See the declaration for the +// `this`-is-destroyed caveat on the growing path. +bool CtxWrap::AppendEncoded(Local holder, + const std::vector& appended) { + const size_t current_used = record()->attrs_data_size; const size_t new_used = current_used + appended.size(); // EncodeAttrs already enforced the cap; new_used <= MAX_ATTRS_DATA_SIZE. - if (new_used <= self->capacity_) { + if (new_used <= capacity_) { // In-place: write the new entries past the current attrs_data_size, // then bump attrs_data_size with a release fence + volatile store so // the content writes are visible before the size store from the @@ -601,55 +678,50 @@ void CtxWrap::Append(const FunctionCallbackInfo& args) { // mid-append sees either the old size (old extent, ignores the // half-written tail) or the new size (full new extent, all bytes // written). Either is consistent. - memcpy(&self->record_->attrs_data[current_used], - appended.data(), - appended.size()); + memcpy( + &record()->attrs_data[current_used], appended.data(), appended.size()); std::atomic_signal_fence(std::memory_order_release); - *reinterpret_cast(&self->record_->attrs_data_size) = + *reinterpret_cast(&record()->attrs_data_size) = static_cast(new_used); - return; + return true; } - // Doesn't fit. Reallocate with geometric growth with cap. + // Doesn't fit. Reallocate with geometric growth with cap. The record lives + // inside the CtxWrap allocation, so growing it means moving the CtxWrap + // too: build a replacement, hand it the same holder object, and retire the + // old one. size_t new_cap = - std::min(std::max(self->capacity_ * 2, new_used), MAX_ATTRS_DATA_SIZE); + std::min(std::max(capacity_ * 2, new_used), MAX_ATTRS_DATA_SIZE); - const size_t total = sizeof(OtelThreadCtxRecord) + new_cap; - OwnedRecord new_rec(static_cast(calloc(1, total))); - if (!new_rec) { - isolate->ThrowError("allocation failed"); - return; - } - // Capture before the copy: the point of the assert below is that the memcpy - // carried the header across intact, not that the record is valid. It used to - // assert `valid == 1`, which invalidate() legitimately makes false — and - // since NDEBUG is not defined for this addon, that aborted release builds - // too, not just debug ones. - const uint8_t src_valid = self->record_->valid; + CtxWrap* new_self = CtxWrap::Create(new_cap); + if (new_self == nullptr) return false; + new_self->truncated_ = truncated_; // Copy the existing record (header + already-written attrs_data). memcpy( - new_rec.get(), self->record_, sizeof(OtelThreadCtxRecord) + current_used); + new_self->record(), record(), sizeof(OtelThreadCtxRecord) + current_used); // Append the new entries and update attrs_data_size. - memcpy(&new_rec->attrs_data[current_used], appended.data(), appended.size()); - new_rec->attrs_data_size = static_cast(new_used); - // The copy should've carried the source record's header across verbatim, - // whatever its validity was. - assert(new_rec->valid == src_valid); - - // Publish: the pointer swap is the atomic boundary the reader sees. The - // first fence keeps the new_rec content writes ordered before the pointer - // store from the compiler's perspective. The second fence prevents free() - // from being hoisted above the pointer swap — without it, a reader stopped - // between a reordered free() and the not-yet-completed swap would follow - // self->record_ into freed memory. OTEP signal-handler semantics (the - // writer is stopped during reads) take care of CPU-side ordering and make - // immediate freeing of the old record safe. + memcpy(&new_self->record()->attrs_data[current_used], + appended.data(), + appended.size()); + new_self->record()->attrs_data_size = static_cast(new_used); + + // Publish: the internal-field store inside Wrap() is the atomic boundary + // the reader sees. The first fence keeps the new_self content writes + // ordered before that store from the compiler's perspective. The second + // fence prevents the free() inside Destroy() from being hoisted above it — + // without it, a reader stopped between a reordered free() and the + // not-yet-completed store would follow the internal field into freed + // memory. OTEP signal-handler semantics (the writer is stopped during + // reads) take care of CPU-side ordering and make immediate freeing of the + // old block safe. std::atomic_signal_fence(std::memory_order_release); - OtelThreadCtxRecord* old_rec = self->record_; - self->record_ = new_rec.release(); - self->capacity_ = new_cap; + new_self->Wrap(holder); std::atomic_signal_fence(std::memory_order_acq_rel); - free(old_rec); + // Destroying the old wrap runs ~Global on its handle_, which resets the + // weak handle and so cancels the WeakCallback V8 would otherwise fire on + // the freed block. + CtxWrap::Destroy(this); + return true; } // Mark this record's `valid` byte as 0 in place. Every async-context @@ -668,7 +740,30 @@ void CtxWrap::Invalidate(const FunctionCallbackInfo& args) { return; } std::atomic_signal_fence(std::memory_order_release); - *reinterpret_cast(&self->record_->valid) = 0; + *reinterpret_cast(&self->record()->valid) = 0; +} + +// Overwrite the record's trace-flags byte in place. The W3C flags are not +// always known when a context is built: an SDK whose sampling decision is +// deferred only learns the sampled bit later, and a decision already made can +// still be overridden. So this mirrors invalidate() — one byte, a fence and a +// volatile store, visible at once to every frame sharing the record. +void CtxWrap::SetTraceFlags(const FunctionCallbackInfo& args) { + Isolate* isolate = args.GetIsolate(); + CtxWrap* self = CtxWrap::Unwrap(args.This()); + if (!self) { + isolate->ThrowError("not a ThreadContext"); + return; + } + if (args.Length() != 1) { + isolate->ThrowError("setTraceFlags expects 1 argument: traceFlags"); + return; + } + uint8_t trace_flags = 0; + if (!ToTraceFlags(isolate, args[0], &trace_flags)) return; + std::atomic_signal_fence(std::memory_order_release); + *reinterpret_cast(&self->record()->trace_flags) = + trace_flags; } // Returns true if any attribute was ever dropped from this wrapper's @@ -695,15 +790,15 @@ void CtxWrap::DebugBytes(const FunctionCallbackInfo& args) { return; } const size_t total = - sizeof(OtelThreadCtxRecord) + self->record_->attrs_data_size; + sizeof(OtelThreadCtxRecord) + self->record()->attrs_data_size; Local buf = v8::ArrayBuffer::New(isolate, total); - memcpy(buf->GetBackingStore()->Data(), self->record_, total); + memcpy(buf->GetBackingStore()->Data(), self->record(), total); args.GetReturnValue().Set(Uint8Array::New(buf, 0, total)); } void CtxWrap::Init(Local exports) { Isolate* isolate = Isolate::GetCurrent(); - node::AddEnvironmentCleanupHook(isolate, DrainLiveCtxWraps, isolate); + node::AddEnvironmentCleanupHook(isolate, DrainLive, isolate); Local context = isolate->GetCurrentContext(); Local tpl = FunctionTemplate::New(isolate, New); @@ -719,6 +814,9 @@ void CtxWrap::Init(Local exports) { tpl->PrototypeTemplate()->Set( String::NewFromUtf8Literal(isolate, "invalidate"), FunctionTemplate::New(isolate, Invalidate)); + tpl->PrototypeTemplate()->Set( + String::NewFromUtf8Literal(isolate, "setTraceFlags"), + FunctionTemplate::New(isolate, SetTraceFlags)); tpl->PrototypeTemplate()->Set( String::NewFromUtf8Literal(isolate, "isTruncated"), FunctionTemplate::New(isolate, IsTruncated)); @@ -738,7 +836,12 @@ void CtxWrap::Init(Local exports) { // this as a per-isolate cleanup hook the first time StoreAls is called // keeps the handle safely scoped to the isolate. void ResetDiscoveryStruct(void* /*arg*/) { - otel_thread_ctx_nodejs_v1.cped_slot = nullptr; + // Clear cped_slot with volatile + signal fence so a reader that stops this + // thread mid-teardown either sees a null cped_slot and bails, or sees the + // struct still whole. + *reinterpret_cast( + &otel_thread_ctx_nodejs_v1.cped_slot) = nullptr; + std::atomic_signal_fence(std::memory_order_release); otel_thread_ctx_nodejs_v1.als_handle.Reset(); otel_thread_ctx_nodejs_v1.als_identity_hash = 0; otel_thread_ctx_nodejs_v1.undefined_addr = 0; @@ -754,25 +857,23 @@ void StoreAls(const FunctionCallbackInfo& args) { otel_thread_ctx_nodejs_v1.als_identity_hash = obj->GetIdentityHash(); otel_thread_ctx_nodejs_v1.als_handle = Global(isolate, obj); #if NODE_MAJOR_VERSION >= 22 - otel_thread_ctx_nodejs_v1.cped_slot = - reinterpret_cast( - reinterpret_cast(isolate) + - v8::internal::Internals::kContinuationPreservedEmbedderDataOffset); + v8::internal::Address* slot = reinterpret_cast( + reinterpret_cast(isolate) + + v8::internal::Internals::kContinuationPreservedEmbedderDataOffset); #else // Node < 22 lacks ContinuationPreservedEmbedderData entirely (and the // associated V8 internal offset). The TS layer refuses to install the // hook on these versions via isAsyncContextFrameActive, so StoreAls is - // never called from JS — this null assignment is just here so the - // addon compiles on the older Node versions the package supports. - otel_thread_ctx_nodejs_v1.cped_slot = nullptr; + // never called from JS — this null assignment is here so the addon + // compiles on the older Node versions the package supports and also + // cped_slot == nullptr serves as the reader gate. + v8::internal::Address* slot = nullptr; #endif - // `undefined_addr == 0` doubles as the "not yet initialized on this - // isolate" flag: it starts at zero (thread-local zero-init), any real - // V8 undefined singleton address is non-zero, and ResetDiscoveryStruct - // clears it back to zero — so a subsequent StoreAls (e.g. isolate - // tear-down then re-init on the same thread) re-registers the cleanup - // hook. Register BEFORE the write so the flag transition is the last - // observable step. + // `undefined_addr == 0` marks "no cleanup hook registered for this thread + // yet": it starts at zero (thread-local zero-init) and ResetDiscoveryStruct + // clears it back to zero, so an isolate re-initialized on this thread + // registers a fresh hook. This is bookkeeping for us, not the reader's + // gate, we use cped_slot for that. if (otel_thread_ctx_nodejs_v1.undefined_addr == 0) { node::AddEnvironmentCleanupHook(isolate, ResetDiscoveryStruct, nullptr); } @@ -781,6 +882,14 @@ void StoreAls(const FunctionCallbackInfo& args) { // address is fine — no Global<> tracking needed. otel_thread_ctx_nodejs_v1.undefined_addr = reinterpret_cast(*v8::Undefined(isolate)); + + // Write `cped_slot` last with signal fence + volatile. It is what a reader + // tests before it dereferences anything, so publishing it after every other + // field means a reader either sees null and bails or sees a fully populated + // struct. + std::atomic_signal_fence(std::memory_order_release); + *reinterpret_cast( + &otel_thread_ctx_nodejs_v1.cped_slot) = slot; } // Without a function that explicitly reads the TLS variable, on x86 the @@ -794,11 +903,13 @@ void GetStoredAlsHash(const FunctionCallbackInfo& args) { // V8 layout constants captured at addon-compile time from the same V8 // headers Node bundles. Published via the discovery contract so an -// out-of-process reader can decode our wrapper / V8's internal hashmap +// out-of-process reader can decode V8's JSObject / internal hashmap // layout without doing its own V8-internal-symbol lookups for the -// pointer-compression / sandbox state. +// pointer-compression / sandbox state. Note that nothing published here +// describes our own wrapper: internal field 0 points straight at the +// record, so the reader needs no offset of ours to reach it. #if NODE_MAJOR_VERSION >= 22 -constexpr int WRAPPED_OBJECT_OFFSET = +constexpr int JS_OBJECT_RECORD_OFFSET = v8::internal::Internals::kJSObjectHeaderSize + v8::internal::Internals::kEmbedderDataSlotExternalPointerOffset; #else @@ -807,18 +918,10 @@ constexpr int WRAPPED_OBJECT_OFFSET = // either — see StoreAls), so this value is published only to keep the // addon's exported surface consistent across Node majors. A would-be // reader cannot reach a live record through it. -constexpr int WRAPPED_OBJECT_OFFSET = 0; +constexpr int JS_OBJECT_RECORD_OFFSET = 0; #endif constexpr int TAGGED_SIZE = v8::internal::kApiTaggedSize; -// Given a pointer to a CtxWrap — reached from the JSObject's V8 -// wrapped-object slot — add this offset to arrive at `record_`. CtxWrap has -// no base class, so `record_` is its first member and the offset is zero; -// computing it with offsetof keeps the published value correct if the layout -// ever changes again. -constexpr int NATIVE_WRAP_FIELDS_OFFSET = - static_cast(offsetof(CtxWrap, record_)); - // V8 JSMap layout: kTableOffset within the JSMap object holds a tagged // pointer to the backing OrderedHashMap table. Not exposed in V8's // public headers; kept in sync with @@ -850,11 +953,10 @@ void OtelThreadCtx::Init(Local exports) { .FromJust(); }; publish_int("otelThreadCtxJsMapTableOffset", JS_MAP_TABLE_OFFSET); - publish_int("otelThreadCtxNativeWrapFieldsOffset", NATIVE_WRAP_FIELDS_OFFSET); publish_int("otelThreadCtxOrderedHashMapHeaderSize", ORDERED_HASH_MAP_HEADER_SIZE); publish_int("otelThreadCtxTaggedSize", TAGGED_SIZE); - publish_int("otelThreadCtxWrappedObjectOffset", WRAPPED_OBJECT_OFFSET); + publish_int("otelThreadCtxJsObjectRecordOffset", JS_OBJECT_RECORD_OFFSET); } } // namespace dd diff --git a/bindings/per-isolate-data.cc b/bindings/per-isolate-data.cc index 5898fce2..21f668fd 100644 --- a/bindings/per-isolate-data.cc +++ b/bindings/per-isolate-data.cc @@ -14,6 +14,7 @@ * limitations under the License. */ +#include #include #include #include @@ -64,4 +65,22 @@ std::shared_ptr& PerIsolateData::GetHeapProfilerState() { return heap_profiler_state; } +#if DD_V8_HAS_DICTIONARY_TEMPLATE +v8::Local PerIsolateData::GetDictionaryTemplate( + v8::Isolate* isolate, + DictionaryTemplateId id, + v8::MemorySpan names) { + const size_t index = static_cast(id); + auto& tmpl = dictionary_templates[index]; + auto& arity = dictionary_template_arities[index]; + if (tmpl.IsEmpty()) { + arity = names.size(); + tmpl.Reset(v8::DictionaryTemplate::New(isolate, names)); + } else { + assert(arity == names.size()); + } + return Nan::New(tmpl); +} +#endif + } // namespace dd diff --git a/bindings/per-isolate-data.hh b/bindings/per-isolate-data.hh index 618aa8a9..4221d1ac 100644 --- a/bindings/per-isolate-data.hh +++ b/bindings/per-isolate-data.hh @@ -18,19 +18,45 @@ #include #include +#include #include +#include +#include #include +#include + +// v8::DictionaryTemplate landed in V8 12.3, i.e. Node.js >= 22. +#if V8_MAJOR_VERSION > 12 || (V8_MAJOR_VERSION == 12 && V8_MINOR_VERSION >= 3) +#define DD_V8_HAS_DICTIONARY_TEMPLATE 1 +#else +#define DD_V8_HAS_DICTIONARY_TEMPLATE 0 +#endif namespace dd { struct HeapProfilerState; +#if DD_V8_HAS_DICTIONARY_TEMPLATE +enum class DictionaryTemplateId : size_t { + kWallSampleContext, + kCount, +}; + +constexpr size_t kDictionaryTemplateCount = + static_cast(DictionaryTemplateId::kCount); +#endif + class PerIsolateData { private: Nan::Global wall_profiler_constructor; Nan::Global allocation_node_constructor; Nan::Global time_profile_node_constructor; std::shared_ptr heap_profiler_state; +#if DD_V8_HAS_DICTIONARY_TEMPLATE + std::array, kDictionaryTemplateCount> + dictionary_templates; + std::array dictionary_template_arities{}; +#endif PerIsolateData() {} @@ -41,6 +67,12 @@ class PerIsolateData { Nan::Global& AllocationNodeConstructor(); Nan::Global& TimeProfileNodeConstructor(); std::shared_ptr& GetHeapProfilerState(); +#if DD_V8_HAS_DICTIONARY_TEMPLATE + v8::Local GetDictionaryTemplate( + v8::Isolate* isolate, + DictionaryTemplateId id, + v8::MemorySpan names); +#endif }; } // namespace dd diff --git a/bindings/profilers/heap.cc b/bindings/profilers/heap.cc index 6106625d..3b87fad1 100644 --- a/bindings/profilers/heap.cc +++ b/bindings/profilers/heap.cc @@ -18,15 +18,13 @@ #include "allocation-profile-node.hh" #include "allocation-profile.hh" -#include "defer.hh" +#include "near-oom.hh" #include "per-isolate-data.hh" #include "translate-heap-profile.hh" -#include #include #include #include -#include #include #include @@ -51,441 +49,6 @@ static void HeapProfilerCleanupHook(void* data) { } } -static size_t NearHeapLimit(void* data, - size_t current_heap_limit, - size_t initial_heap_limit); -static void InterruptCallback(v8::Isolate* isolate, void* data); -static void AsyncCallback(uv_async_t* handle); - -enum CallbackMode { - kNoCallback = 0, - kAsyncCallback = 1, - kInterruptCallback = 2, -}; - -struct HeapProfilerState { - explicit HeapProfilerState(v8::Isolate* isolate) : isolate(isolate) {} - - ~HeapProfilerState() { - // Uninstall first. By the time we run, the shared_ptr in PerIsolateData is - // already empty (that is what destroyed us), so NearHeapLimit would find no - // state to work with; anything below that can trigger a GC must not be able - // to reach it. - UninstallNearHeapLimitCallback(); - - auto profiler = isolate->GetHeapProfiler(); - if (profiler) { - profiler->StopSamplingHeapProfiler(); - } - - if (async) { - // defer deletion of async when uv_close callback is invoked - uv_close(reinterpret_cast(async), [](uv_handle_t* handle) { - delete reinterpret_cast(handle); - }); - async = nullptr; - } - } - - void UninstallNearHeapLimitCallback() { - if (isolate && callbackInstalled) { - isolate->RemoveNearHeapLimitCallback(&NearHeapLimit, 0); - callbackInstalled = false; - } - } - - void InstallNearHeapLimitCallback() { - if (callbackInstalled) { - return; - } - if (isolate) { - isolate->AddNearHeapLimitCallback(&NearHeapLimit, nullptr); - // Restore the original heap limit once live old-generation usage falls - // below 90% of the original limit. The threshold controls when V8 - // restores the limit, not the restored limit size. - constexpr double kHeapLimitRestoreThreshold = 0.90; - isolate->AutomaticallyRestoreInitialHeapLimit(kHeapLimitRestoreThreshold); - callbackInstalled = true; - } - } - - void RegisterAsyncCallback() { - if (async) { - return; - } - // async is dynamically allocated so that its lifetime can be different - // from the one of HeapProfilerState since uv_close is asynchronous - async = new uv_async_t(); - uv_async_init(Nan::GetCurrentEventLoop(), async, AsyncCallback); - uv_unref(reinterpret_cast(async)); - } - - void OnNewProfile() { - profile.reset(); - // Only (re)install the NearHeapLimit callback when OOM monitoring is - // configured. Otherwise a plain start()+profile() flow would silently - // register a callback that the user never asked for. - if (max_heap_extension_count > 0) { - InstallNearHeapLimitCallback(); - } - } - - v8::Isolate* isolate = nullptr; - uint32_t heap_extension_size = 0; - uint32_t max_heap_extension_count = 0; - uint32_t current_heap_extension_count = 0; - uv_async_t* async = nullptr; - std::shared_ptr profile; - std::vector export_command; - bool allocations = false; - bool dumpProfileOnStderr = false; - Nan::Callback callback; - uint32_t callbackMode = 0; - bool isMainThread = true; - bool callbackInstalled = false; - bool insideCallback = false; -}; - -static void dumpAllocationProfile(FILE* file, - Node* node, - std::string& cur_stack) { - auto initial_len = cur_stack.size(); - char buf[256]; - - snprintf(buf, - sizeof(buf), - "%s%s:%s:%d", - cur_stack.empty() ? "" : ";", - node->script_name.empty() ? "_" : node->script_name.c_str(), - node->name.empty() ? "(anonymous)" : node->name.c_str(), - node->line_number); - cur_stack += buf; - for (auto& allocation : node->allocations) { - fprintf(file, - "%s %u %zu\n", - cur_stack.c_str(), - allocation.count, - allocation.count * allocation.size); - } - for (auto& child : node->children) { - dumpAllocationProfile(file, child.get(), cur_stack); - } - cur_stack.resize(initial_len); -} - -static void dumpAllocationProfile(FILE* file, Node* node) { - std::string stack; - dumpAllocationProfile(file, node, stack); -} - -static void dumpAllocationProfileAsJSON(FILE* file, Node* node) { - fprintf( - file, - R"({"name":"%s","scriptName":"%s","scriptId":%d,"lineNumber":%d,"columnNumber":%d,"children":[)", - node->name.c_str(), - node->script_name.c_str(), - node->script_id, - node->line_number, - node->column_number); - - bool first = true; - for (auto& child : node->children) { - if (!first) { - fputs(",", file); - } else { - first = false; - } - dumpAllocationProfileAsJSON(file, child.get()); - } - fprintf(file, R"(],"allocations":[)"); - first = true; - for (auto& allocation : node->allocations) { - fprintf(file, - R"(%s{"sizeBytes":%zu,"count":%d})", - first ? "" : ",", - allocation.size, - allocation.count); - first = false; - } - fputs("]}", file); -} - -static void OnExit(uv_process_t* req, int64_t, int) { - if (req->data) { - uv_timer_stop(reinterpret_cast(req->data)); - } - uv_close((uv_handle_t*)req, nullptr); -} - -static void CloseLoop(uv_loop_t& loop) { - uv_run(&loop, UV_RUN_DEFAULT); - uv_walk( - &loop, - [](uv_handle_t* handle, void* arg) { - if (!uv_is_closing(handle)) { - uv_close(handle, nullptr); - } - }, - nullptr); - int r; - do { - r = uv_run(&loop, UV_RUN_ONCE); - } while (r != 0); - - if (uv_loop_close(&loop)) { - fprintf(stderr, "Failed to close event loop\n"); - } -} - -static int CreateTempFile(uv_loop_t& loop, std::string& filepath) { - char buf[PATH_MAX]; - size_t sz = sizeof(buf); - int r; - if ((r = uv_os_tmpdir(buf, &sz)) != 0) { - fprintf(stderr, "Failed to retrieve temp directory: %s\n", uv_strerror(r)); - return -1; - } - -#if defined(__linux__) || defined(__APPLE__) - filepath = std::string{buf, sz} + "/heap_profile_XXXXXX"; - int fd = mkstemp(&filepath[0]); - if (fd < 0) { - fprintf(stderr, - "Failed to create temp file %s : %s\n", - filepath.c_str(), - strerror(errno)); - return -1; - } - return fd; -#else - // Use custom implementation of mkstemp() for Windows - // uv_fs_mkstemp() is not used because it fails unexpectedly on Windows - // (fail fast exception is raised when trying to write to the returned file - // descriptor) - const int max_tries = 3; - for (int i = 0; i < max_tries; ++i) { - filepath = std::string{buf, sz} + "/heap_profile_" + - std::to_string( - std::chrono::system_clock::now().time_since_epoch().count()); - uv_fs_t fs_req{}; - int fd = uv_fs_open(&loop, - &fs_req, - filepath.c_str(), - UV_FS_O_CREAT | UV_FS_O_EXCL | UV_FS_O_WRONLY, - 0600, - nullptr); - uv_fs_req_cleanup(&fs_req); - if (fd >= 0) { - return r; - } - if (fd != UV_EEXIST) { - fprintf(stderr, "Failed to create temp file: %s\n", uv_strerror(fd)); - return -1; - } - } - return -1; -#endif -} - -static void ExportProfile(HeapProfilerState& state) { - const int64_t timeoutMs = 15000; - uv_loop_t loop; - int r; - - if ((r = uv_loop_init(&loop)) != 0) { - fprintf(stderr, "Failed to init new event loop: %s\n", uv_strerror(r)); - return; - } - - defer { - CloseLoop(loop); - }; - - std::string filepath; - int fd; - if ((fd = CreateTempFile(loop, filepath)) < 0) { - return; - } - FILE* file = fdopen(fd, "w"); - dumpAllocationProfileAsJSON(file, state.profile.get()); - fclose(file); - std::vector args; - for (auto& arg : state.export_command) { - args.push_back(const_cast(arg.data())); - } - args.push_back(&filepath[0]); - args.push_back(nullptr); - uv_process_options_t options = {}; - options.flags = UV_PROCESS_DETACHED; - options.file = args[0]; - options.args = args.data(); - options.exit_cb = &OnExit; - uv_stdio_container_t child_stdio[3]; - child_stdio[0].flags = UV_IGNORE; - child_stdio[1].flags = UV_INHERIT_FD; - child_stdio[1].data.fd = 2; - child_stdio[2].flags = UV_INHERIT_FD; - child_stdio[2].data.fd = 2; - options.stdio = child_stdio; - options.stdio_count = 3; - uv_process_t child_req; - uv_timer_t timer; - timer.data = &child_req; - child_req.data = &timer; - - fprintf(stderr, "Spawning export process:"); - for (auto arg : args) { - fprintf(stderr, " %s", arg ? arg : "\n"); - } - if ((r = uv_spawn(&loop, &child_req, &options))) { - fprintf(stderr, "Failed to spawn export process: %s\n", uv_strerror(r)); - return; - } - if ((r = uv_timer_init(&loop, &timer)) != 0) { - fprintf(stderr, "Failed to init timer: %s\n", uv_strerror(r)); - return; - } - if ((r = uv_timer_start( - &timer, - [](uv_timer_t* handle) { - uv_process_kill(reinterpret_cast(handle->data), - SIGKILL); - }, - timeoutMs, - 0))) { - fprintf(stderr, "Failed to start timer: %s\n", uv_strerror(r)); - return; - } - uv_run(&loop, UV_RUN_DEFAULT); - - // Delete temp file - uv_fs_t fs_req{}; - uv_fs_unlink(&loop, &fs_req, filepath.c_str(), nullptr); - uv_fs_req_cleanup(&fs_req); -} - -size_t NearHeapLimit(void* data, - size_t current_heap_limit, - size_t initial_heap_limit) { - auto isolate = v8::Isolate::GetCurrent(); - auto state = PerIsolateData::For(isolate)->GetHeapProfilerState(); - - if (!state) { - // StopSamplingHeapProfiler uninstalls us before dropping the state, so - // normally this cannot happen. The gap is the other destruction path: a - // shared_ptr copy taken by an in-flight NearHeapLimit or InterruptCallback - // can outlive the per-isolate slot — the OOM JS callback calling - // process.exit() erases PerIsolateData while InterruptCallback still holds - // a reference, so ~HeapProfilerState never runs to uninstall us. Decline - // and let V8 do its normal OOM handling. - // - // Deliberately no RemoveNearHeapLimitCallback here: the state that tracked - // the installation is already unreachable, so callbackInstalled cannot be - // cleared, and the only way to get here is a process on its way out. - return current_heap_limit; - } - - if (state->insideCallback) { - // Reentrant call detected, try to increase heap limit a bit so that - // previous callback can proceed - const uint32_t default_heap_extension_size = 10 * 1024 * 1024; - auto extension_size = state->heap_extension_size - ? state->heap_extension_size - : default_heap_extension_size; - return current_heap_limit + extension_size; - } - state->insideCallback = true; - defer { - state->insideCallback = false; - }; - - ++state->current_heap_extension_count; - fprintf(stderr, - "NearHeapLimit(count=%d): current_heap_limit=%zu, " - "initial_heap_limit=%zu\n", - state->current_heap_extension_count, - current_heap_limit, - initial_heap_limit); - - auto n = isolate->NumberOfTrackedHeapObjectTypes(); - v8::HeapObjectStatistics stats; - - for (size_t i = 0; i < n; ++i) { - if (isolate->GetHeapObjectStatisticsAtLastGC(&stats, i) && - stats.object_count() > 0) { - fprintf(stderr, - "HeapObjectStats: type=%s, subtype=%s, size=%zu, count=%zu\n", - stats.object_type(), - stats.object_sub_type(), - stats.object_size(), - stats.object_count()); - } - } - // GetAllocationProfile returns null when V8's sampling heap profiler isn't - // running, and that can happen while this callback is still installed: - // HeapProfilerCleanupHook stops V8's sampler without touching our state, so - // between that hook and the isolate actually going away we stay registered - // with nothing to sample. The heap-limit bookkeeping below still has to run, - // so skip only the profile-dependent work. - std::unique_ptr profile{ - isolate->GetHeapProfiler()->GetAllocationProfile()}; - if (profile) { - state->profile = TranslateAllocationProfileToCpp(profile->GetRootNode()); - if (state->dumpProfileOnStderr) { - dumpAllocationProfile(stderr, state->profile.get()); - } - - if (!state->export_command.empty()) { - ExportProfile(*state); - } - - if (!state->callback.IsEmpty()) { - if (state->callbackMode & kInterruptCallback) { - isolate->RequestInterrupt(InterruptCallback, nullptr); - } - if (state->callbackMode & kAsyncCallback) { - uv_async_send(state->async); - } - } else { - state->profile.reset(); - } - } else { - // Drop any profile retained from an earlier invocation: it is stale, and - // nothing below is going to consume or replace it. - state->profile.reset(); - fprintf(stderr, - "NearHeapLimit: heap profiler is not enabled, no allocation " - "profile to report\n"); - } - - if (!state->isMainThread) { - // In worker thread, OOM is not fatal to the whole process and will only - // terminate the worker. - // This is done by a callback registered by node, that's why we remove our - // callback and then call LowMemoryNotification() here to trigger another - // garbage collection, which will eventually call the callback registered by - // node. - state->UninstallNearHeapLimitCallback(); - isolate->LowMemoryNotification(); - // use the same value as node plus 1 - constexpr size_t kExtraHeapAllowance = 16 * 1024 * 1024; - return current_heap_limit + kExtraHeapAllowance + 1; - } - - size_t new_heap_limit = - current_heap_limit + - ((state->current_heap_extension_count <= state->max_heap_extension_count) - ? state->heap_extension_size - : 0); - if (state->current_heap_extension_count >= state->max_heap_extension_count) { - // On Node 14, NearLimitCallback is sometimes called many times, without the - // process aborting, even when returned limit is not increased. Disable - // callback until next call to GetAllocationProfile() - state->UninstallNearHeapLimitCallback(); - } - return new_heap_limit; -} - NAN_METHOD(HeapProfiler::StartSamplingHeapProfiler) { auto isolate = info.GetIsolate(); @@ -645,73 +208,6 @@ NAN_METHOD(HeapProfiler::MapAllocationProfile) { } } -NAN_METHOD(HeapProfiler::MonitorOutOfMemory) { - if (info.Length() != 7) { - return Nan::ThrowTypeError("MonitorOOMCondition must have 7 arguments."); - } - if (!info[0]->IsUint32()) { - return Nan::ThrowTypeError("Heap limit extension size must be a uint32."); - } - if (!info[1]->IsUint32()) { - return Nan::ThrowTypeError( - "Max heap limit extension count must be a uint32."); - } - if (!info[2]->IsBoolean()) { - return Nan::ThrowTypeError("DumpHeapProfileOnStdErr must be a boolean."); - } - if (!info[3]->IsArray()) { - return Nan::ThrowTypeError("Export command must be a string array."); - } - if (!info[4]->IsNullOrUndefined() && !info[4]->IsFunction()) { - return Nan::ThrowTypeError("Callback name must be a function."); - } - if (!info[5]->IsUint32()) { - return Nan::ThrowTypeError("CallbackMode must be a uint32."); - } - if (!info[6]->IsBoolean()) { - return Nan::ThrowTypeError("IsMainThread must be a boolean."); - } - - auto isolate = v8::Isolate::GetCurrent(); - - // Reuse existing state if present so sample_interval/allocations set by - // StartSamplingHeapProfiler survive. Only OOM-owned fields are reset below. - // callbackInstalled is intentionally left alone — - // InstallNearHeapLimitCallback below is idempotent. - auto& state = PerIsolateData::For(isolate)->GetHeapProfilerState(); - if (!state) { - state = std::make_shared(isolate); - } - - state->current_heap_extension_count = 0; - state->profile.reset(); - state->export_command.clear(); - state->callback.Reset(); - - state->heap_extension_size = info[0].As()->Value(); - state->max_heap_extension_count = info[1].As()->Value(); - state->dumpProfileOnStderr = info[2].As()->Value(); - state->callbackMode = info[5].As()->Value(); - state->isMainThread = info[6].As()->Value(); - state->InstallNearHeapLimitCallback(); - if (!info[4]->IsNullOrUndefined() && state->callbackMode != kNoCallback) { - state->callback.Reset(Nan::To(info[4]).ToLocalChecked()); - } - - auto commands = info[3].As(); - for (uint32_t i = 0; i < commands->Length(); ++i) { - auto value = Nan::Get(commands, i).ToLocalChecked(); - if (value->IsString()) { - Nan::Utf8String arg{value}; - state->export_command.emplace_back(*arg, arg.length()); - } - } - - if (!state->callback.IsEmpty() && (state->callbackMode & kAsyncCallback)) { - state->RegisterAsyncCallback(); - } -} - NAN_MODULE_INIT(HeapProfiler::Init) { v8::Local heapProfiler = Nan::New(); Nan::SetMethod( @@ -726,24 +222,4 @@ NAN_MODULE_INIT(HeapProfiler::Init) { heapProfiler); } -void InterruptCallback(v8::Isolate* isolate, void* data) { - v8::HandleScope scope(isolate); - auto state = PerIsolateData::For(isolate)->GetHeapProfilerState(); - // The interrupt is requested from NearHeapLimit but runs later, so - // StopSamplingHeapProfiler() may have dropped the state in between. - if (!state || !state->profile) { - return; - } - v8::Local argv[1] = { - dd::TranslateAllocationProfile(state->profile.get())}; - Nan::AsyncResource resource("NearHeapLimit"); - state->callback.Call(1, argv, &resource); - // Release the retained native profile once the callback has been invoked. - state->profile.reset(); -} - -void AsyncCallback(uv_async_t* handle) { - InterruptCallback(v8::Isolate::GetCurrent(), nullptr); -} - } // namespace dd diff --git a/bindings/profilers/near-oom.cc b/bindings/profilers/near-oom.cc new file mode 100644 index 00000000..e03094e1 --- /dev/null +++ b/bindings/profilers/near-oom.cc @@ -0,0 +1,495 @@ +/* + * Copyright 2026 Datadog, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "near-oom.hh" + +#include "defer.hh" +#include "heap.hh" +#include "per-isolate-data.hh" +#include "translate-heap-profile.hh" + +#include +#include +#include +#include +#include + +#include +#include + +namespace dd { + +static void dumpAllocationProfile(FILE* file, + Node* node, + std::string& cur_stack) { + auto initial_len = cur_stack.size(); + char buf[256]; + + snprintf(buf, + sizeof(buf), + "%s%s:%s:%d", + cur_stack.empty() ? "" : ";", + node->script_name.empty() ? "_" : node->script_name.c_str(), + node->name.empty() ? "(anonymous)" : node->name.c_str(), + node->line_number); + cur_stack += buf; + for (auto& allocation : node->allocations) { + fprintf(file, + "%s %u %zu\n", + cur_stack.c_str(), + allocation.count, + allocation.count * allocation.size); + } + for (auto& child : node->children) { + dumpAllocationProfile(file, child.get(), cur_stack); + } + cur_stack.resize(initial_len); +} + +static void dumpAllocationProfile(FILE* file, Node* node) { + std::string stack; + dumpAllocationProfile(file, node, stack); +} + +static void dumpAllocationProfileAsJSON(FILE* file, Node* node) { + fprintf( + file, + R"({"name":"%s","scriptName":"%s","scriptId":%d,"lineNumber":%d,"columnNumber":%d,"children":[)", + node->name.c_str(), + node->script_name.c_str(), + node->script_id, + node->line_number, + node->column_number); + + bool first = true; + for (auto& child : node->children) { + if (!first) { + fputs(",", file); + } else { + first = false; + } + dumpAllocationProfileAsJSON(file, child.get()); + } + fprintf(file, R"(],"allocations":[)"); + first = true; + for (auto& allocation : node->allocations) { + fprintf(file, + R"(%s{"sizeBytes":%zu,"count":%d})", + first ? "" : ",", + allocation.size, + allocation.count); + first = false; + } + fputs("]}", file); +} + +static void OnExit(uv_process_t* req, int64_t, int) { + if (req->data) { + uv_timer_stop(reinterpret_cast(req->data)); + } + uv_close((uv_handle_t*)req, nullptr); +} + +static void CloseLoop(uv_loop_t& loop) { + uv_run(&loop, UV_RUN_DEFAULT); + uv_walk( + &loop, + [](uv_handle_t* handle, void* arg) { + if (!uv_is_closing(handle)) { + uv_close(handle, nullptr); + } + }, + nullptr); + int r; + do { + r = uv_run(&loop, UV_RUN_ONCE); + } while (r != 0); + + if (uv_loop_close(&loop)) { + fprintf(stderr, "Failed to close event loop\n"); + } +} + +static int CreateTempFile(uv_loop_t& loop, std::string& filepath) { + char buf[PATH_MAX]; + size_t sz = sizeof(buf); + int r; + if ((r = uv_os_tmpdir(buf, &sz)) != 0) { + fprintf(stderr, "Failed to retrieve temp directory: %s\n", uv_strerror(r)); + return -1; + } + +#if defined(__linux__) || defined(__APPLE__) + filepath = std::string{buf, sz} + "/heap_profile_XXXXXX"; + int fd = mkstemp(&filepath[0]); + if (fd < 0) { + fprintf(stderr, + "Failed to create temp file %s : %s\n", + filepath.c_str(), + strerror(errno)); + return -1; + } + return fd; +#else + // Use custom implementation of mkstemp() for Windows + // uv_fs_mkstemp() is not used because it fails unexpectedly on Windows + // (fail fast exception is raised when trying to write to the returned file + // descriptor) + const int max_tries = 3; + for (int i = 0; i < max_tries; ++i) { + filepath = std::string{buf, sz} + "/heap_profile_" + + std::to_string( + std::chrono::system_clock::now().time_since_epoch().count()); + uv_fs_t fs_req{}; + int fd = uv_fs_open(&loop, + &fs_req, + filepath.c_str(), + UV_FS_O_CREAT | UV_FS_O_EXCL | UV_FS_O_WRONLY, + 0600, + nullptr); + uv_fs_req_cleanup(&fs_req); + if (fd >= 0) { + return r; + } + if (fd != UV_EEXIST) { + fprintf(stderr, "Failed to create temp file: %s\n", uv_strerror(fd)); + return -1; + } + } + return -1; +#endif +} + +static void ExportProfile(HeapProfilerState& state) { + const int64_t timeoutMs = 15000; + uv_loop_t loop; + int r; + + if ((r = uv_loop_init(&loop)) != 0) { + fprintf(stderr, "Failed to init new event loop: %s\n", uv_strerror(r)); + return; + } + + defer { + CloseLoop(loop); + }; + + std::string filepath; + int fd; + if ((fd = CreateTempFile(loop, filepath)) < 0) { + return; + } + FILE* file = fdopen(fd, "w"); + dumpAllocationProfileAsJSON(file, state.profile.get()); + fclose(file); + std::vector args; + for (auto& arg : state.export_command) { + args.push_back(const_cast(arg.data())); + } + args.push_back(&filepath[0]); + args.push_back(nullptr); + uv_process_options_t options = {}; + options.flags = UV_PROCESS_DETACHED; + options.file = args[0]; + options.args = args.data(); + options.exit_cb = &OnExit; + uv_stdio_container_t child_stdio[3]; + child_stdio[0].flags = UV_IGNORE; + child_stdio[1].flags = UV_INHERIT_FD; + child_stdio[1].data.fd = 2; + child_stdio[2].flags = UV_INHERIT_FD; + child_stdio[2].data.fd = 2; + options.stdio = child_stdio; + options.stdio_count = 3; + uv_process_t child_req; + uv_timer_t timer; + timer.data = &child_req; + child_req.data = &timer; + + fprintf(stderr, "Spawning export process:"); + for (auto arg : args) { + fprintf(stderr, " %s", arg ? arg : "\n"); + } + if ((r = uv_spawn(&loop, &child_req, &options))) { + fprintf(stderr, "Failed to spawn export process: %s\n", uv_strerror(r)); + return; + } + if ((r = uv_timer_init(&loop, &timer)) != 0) { + fprintf(stderr, "Failed to init timer: %s\n", uv_strerror(r)); + return; + } + if ((r = uv_timer_start( + &timer, + [](uv_timer_t* handle) { + uv_process_kill(reinterpret_cast(handle->data), + SIGKILL); + }, + timeoutMs, + 0))) { + fprintf(stderr, "Failed to start timer: %s\n", uv_strerror(r)); + return; + } + uv_run(&loop, UV_RUN_DEFAULT); + + // Delete temp file + uv_fs_t fs_req{}; + uv_fs_unlink(&loop, &fs_req, filepath.c_str(), nullptr); + uv_fs_req_cleanup(&fs_req); +} + +// V8 only raises the limit when the returned value is strictly greater than +// current_heap_limit, and clamps it to its own allocator maximum, so +// saturating is enough to stay well-defined in the extreme case. +static size_t ExtendedHeapLimit(size_t current_heap_limit, size_t extension) { + return extension > std::numeric_limits::max() - current_heap_limit + ? std::numeric_limits::max() + : current_heap_limit + extension; +} + +size_t NearHeapLimit(void* data, + size_t current_heap_limit, + size_t initial_heap_limit) { + auto isolate = v8::Isolate::GetCurrent(); + auto state = PerIsolateData::For(isolate)->GetHeapProfilerState(); + + if (!state) { + // StopSamplingHeapProfiler uninstalls us before dropping the state, so + // normally this cannot happen. The gap is the other destruction path: a + // shared_ptr copy taken by an in-flight NearHeapLimit or InterruptCallback + // can outlive the per-isolate slot — the OOM JS callback calling + // process.exit() erases PerIsolateData while InterruptCallback still holds + // a reference, so ~HeapProfilerState never runs to uninstall us. Decline + // and let V8 do its normal OOM handling. + // + // Deliberately no RemoveNearHeapLimitCallback here: the state that tracked + // the installation is already unreachable, so callbackInstalled cannot be + // cleared, and the only way to get here is a process on its way out. + return current_heap_limit; + } + + size_t extension = state->heap_extension_size; + if (state->automatic_heap_extension) { + if (!state->automatic_heap_extension_size.has_value()) { + // Grant at most one young generation, as Node.js does for its near-OOM + // heap snapshot callback. In current V8, heap_size_limit() is the + // old-generation limit this callback was handed plus the maximum + // young-generation size, so the delta is that young generation. It is + // fixed for the isolate, so sample it once and reuse it. + v8::HeapStatistics heap_statistics; + isolate->GetHeapStatistics(&heap_statistics); + const size_t total_heap_limit = heap_statistics.heap_size_limit(); + // Only cache a usable sample: a degenerate one must not disable + // automatic sizing for the rest of the isolate's lifetime. + if (total_heap_limit > current_heap_limit) { + state->automatic_heap_extension_size = + total_heap_limit - current_heap_limit; + } + } + extension = state->automatic_heap_extension_size.value_or(0); + } + + if (state->insideCallback) { + // Reentrant call: GetAllocationProfile() allocated its way back into us. + // The in-progress capture still needs room to finish, so rescue it even + // when the caller asked for no top-level extension at all. + constexpr size_t kReentrantRescueExtension = 10 * 1024 * 1024; + return ExtendedHeapLimit( + current_heap_limit, + extension != 0 ? extension : kReentrantRescueExtension); + } + state->insideCallback = true; + defer { + state->insideCallback = false; + }; + + ++state->current_heap_extension_count; + fprintf(stderr, + "NearHeapLimit(count=%d): current_heap_limit=%zu, " + "initial_heap_limit=%zu\n", + state->current_heap_extension_count, + current_heap_limit, + initial_heap_limit); + + auto n = isolate->NumberOfTrackedHeapObjectTypes(); + v8::HeapObjectStatistics stats; + + for (size_t i = 0; i < n; ++i) { + if (isolate->GetHeapObjectStatisticsAtLastGC(&stats, i) && + stats.object_count() > 0) { + fprintf(stderr, + "HeapObjectStats: type=%s, subtype=%s, size=%zu, count=%zu\n", + stats.object_type(), + stats.object_sub_type(), + stats.object_size(), + stats.object_count()); + } + } + // GetAllocationProfile returns null when V8's sampling heap profiler isn't + // running, and that can happen while this callback is still installed: + // HeapProfilerCleanupHook stops V8's sampler without touching our state, so + // between that hook and the isolate actually going away we stay registered + // with nothing to sample. The heap-limit bookkeeping below still has to run, + // so skip only the profile-dependent work. + std::unique_ptr profile{ + isolate->GetHeapProfiler()->GetAllocationProfile()}; + if (profile) { + state->profile = TranslateAllocationProfileToCpp(profile->GetRootNode()); + if (state->dumpProfileOnStderr) { + dumpAllocationProfile(stderr, state->profile.get()); + } + + if (!state->export_command.empty()) { + ExportProfile(*state); + } + + if (!state->callback.IsEmpty()) { + if (state->callbackMode & kInterruptCallback) { + isolate->RequestInterrupt(InterruptCallback, nullptr); + } + if (state->callbackMode & kAsyncCallback) { + uv_async_send(state->async); + } + } else { + state->profile.reset(); + } + } else { + // Drop any profile retained from an earlier invocation: it is stale, and + // nothing below is going to consume or replace it. + state->profile.reset(); + fprintf(stderr, + "NearHeapLimit: heap profiler is not enabled, no allocation " + "profile to report\n"); + } + + if (!state->isMainThread) { + // In worker thread, OOM is not fatal to the whole process and will only + // terminate the worker. + // This is done by a callback registered by node, that's why we remove our + // callback and then call LowMemoryNotification() here to trigger another + // garbage collection, which will eventually call the callback registered by + // node. + state->UninstallNearHeapLimitCallback(); + isolate->LowMemoryNotification(); + // use the same value as node plus 1 + constexpr size_t kExtraHeapAllowance = 16 * 1024 * 1024; + return current_heap_limit + kExtraHeapAllowance + 1; + } + + if (state->current_heap_extension_count >= state->max_heap_extension_count) { + // On Node 14, NearLimitCallback is sometimes called many times, without the + // process aborting, even when returned limit is not increased. Disable + // callback until next call to GetAllocationProfile() + state->UninstallNearHeapLimitCallback(); + } + return state->current_heap_extension_count <= state->max_heap_extension_count + ? ExtendedHeapLimit(current_heap_limit, extension) + : current_heap_limit; +} + +NAN_METHOD(HeapProfiler::MonitorOutOfMemory) { + if (info.Length() != 8) { + return Nan::ThrowTypeError("MonitorOOMCondition must have 8 arguments."); + } + if (!info[0]->IsUint32()) { + return Nan::ThrowTypeError("Heap limit extension size must be a uint32."); + } + if (!info[1]->IsUint32()) { + return Nan::ThrowTypeError( + "Max heap limit extension count must be a uint32."); + } + if (!info[2]->IsBoolean()) { + return Nan::ThrowTypeError("DumpHeapProfileOnStdErr must be a boolean."); + } + if (!info[3]->IsArray()) { + return Nan::ThrowTypeError("Export command must be a string array."); + } + if (!info[4]->IsNullOrUndefined() && !info[4]->IsFunction()) { + return Nan::ThrowTypeError("Callback name must be a function."); + } + if (!info[5]->IsUint32()) { + return Nan::ThrowTypeError("CallbackMode must be a uint32."); + } + if (!info[6]->IsBoolean()) { + return Nan::ThrowTypeError("IsMainThread must be a boolean."); + } + if (!info[7]->IsBoolean()) { + return Nan::ThrowTypeError( + "AutomaticHeapLimitExtension must be a boolean."); + } + + auto isolate = v8::Isolate::GetCurrent(); + + // Reuse existing state if present so sample_interval/allocations set by + // StartSamplingHeapProfiler survive. Only OOM-owned fields are reset below. + // callbackInstalled is intentionally left alone — + // InstallNearHeapLimitCallback below is idempotent. + auto& state = PerIsolateData::For(isolate)->GetHeapProfilerState(); + if (!state) { + state = std::make_shared(isolate); + } + + state->current_heap_extension_count = 0; + state->automatic_heap_extension_size.reset(); + state->profile.reset(); + state->export_command.clear(); + state->callback.Reset(); + + state->heap_extension_size = info[0].As()->Value(); + state->max_heap_extension_count = info[1].As()->Value(); + state->dumpProfileOnStderr = info[2].As()->Value(); + state->callbackMode = info[5].As()->Value(); + state->isMainThread = info[6].As()->Value(); + state->automatic_heap_extension = info[7].As()->Value(); + state->InstallNearHeapLimitCallback(); + if (!info[4]->IsNullOrUndefined() && state->callbackMode != kNoCallback) { + state->callback.Reset(Nan::To(info[4]).ToLocalChecked()); + } + + auto commands = info[3].As(); + for (uint32_t i = 0; i < commands->Length(); ++i) { + auto value = Nan::Get(commands, i).ToLocalChecked(); + if (value->IsString()) { + Nan::Utf8String arg{value}; + state->export_command.emplace_back(*arg, arg.length()); + } + } + + if (!state->callback.IsEmpty() && (state->callbackMode & kAsyncCallback)) { + state->RegisterAsyncCallback(); + } +} + +void InterruptCallback(v8::Isolate* isolate, void* data) { + v8::HandleScope scope(isolate); + auto state = PerIsolateData::For(isolate)->GetHeapProfilerState(); + // The interrupt is requested from NearHeapLimit but runs later, so + // StopSamplingHeapProfiler() may have dropped the state in between. + if (!state || !state->profile) { + return; + } + v8::Local argv[1] = { + dd::TranslateAllocationProfile(state->profile.get())}; + Nan::AsyncResource resource("NearHeapLimit"); + state->callback.Call(1, argv, &resource); + // Release the retained native profile once the callback has been invoked. + state->profile.reset(); +} + +void AsyncCallback(uv_async_t* handle) { + InterruptCallback(v8::Isolate::GetCurrent(), nullptr); +} + +} // namespace dd diff --git a/bindings/profilers/near-oom.hh b/bindings/profilers/near-oom.hh new file mode 100644 index 00000000..ce9008c2 --- /dev/null +++ b/bindings/profilers/near-oom.hh @@ -0,0 +1,133 @@ +/* + * Copyright 2026 Datadog, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "translate-heap-profile.hh" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace dd { + +size_t NearHeapLimit(void* data, + size_t current_heap_limit, + size_t initial_heap_limit); +void InterruptCallback(v8::Isolate* isolate, void* data); +void AsyncCallback(uv_async_t* handle); + +enum CallbackMode { + kNoCallback = 0, + kAsyncCallback = 1, + kInterruptCallback = 2, +}; + +struct HeapProfilerState { + explicit HeapProfilerState(v8::Isolate* isolate) : isolate(isolate) {} + + ~HeapProfilerState() { + // Uninstall first. By the time we run, the shared_ptr in PerIsolateData is + // already empty (that is what destroyed us), so NearHeapLimit would find no + // state to work with; anything below that can trigger a GC must not be able + // to reach it. + UninstallNearHeapLimitCallback(); + + auto profiler = isolate->GetHeapProfiler(); + if (profiler) { + profiler->StopSamplingHeapProfiler(); + } + + if (async) { + // defer deletion of async when uv_close callback is invoked + uv_close(reinterpret_cast(async), [](uv_handle_t* handle) { + delete reinterpret_cast(handle); + }); + async = nullptr; + } + } + + void UninstallNearHeapLimitCallback() { + if (isolate && callbackInstalled) { + isolate->RemoveNearHeapLimitCallback(&NearHeapLimit, 0); + callbackInstalled = false; + } + } + + void InstallNearHeapLimitCallback() { + if (callbackInstalled) { + return; + } + if (isolate) { + isolate->AddNearHeapLimitCallback(&NearHeapLimit, nullptr); + // Restore the original heap limit once live old-generation usage falls + // below 90% of the original limit. The threshold controls when V8 + // restores the limit, not the restored limit size. + constexpr double kHeapLimitRestoreThreshold = 0.90; + isolate->AutomaticallyRestoreInitialHeapLimit(kHeapLimitRestoreThreshold); + callbackInstalled = true; + } + } + + void RegisterAsyncCallback() { + if (async) { + return; + } + // async is dynamically allocated so that its lifetime can be different + // from the one of HeapProfilerState since uv_close is asynchronous + async = new uv_async_t(); + uv_async_init(Nan::GetCurrentEventLoop(), async, AsyncCallback); + uv_unref(reinterpret_cast(async)); + } + + void OnNewProfile() { + profile.reset(); + // Only (re)install the NearHeapLimit callback when OOM monitoring is + // configured. Otherwise a plain start()+profile() flow would silently + // register a callback that the user never asked for. + if (max_heap_extension_count > 0) { + InstallNearHeapLimitCallback(); + } + } + + v8::Isolate* isolate = nullptr; + uint32_t heap_extension_size = 0; + // When true, heap_extension_size is ignored in favour of one maximum-sized + // young generation, sampled once into automatic_heap_extension_size. + bool automatic_heap_extension = false; + std::optional automatic_heap_extension_size; + uint32_t max_heap_extension_count = 0; + uint32_t current_heap_extension_count = 0; + uv_async_t* async = nullptr; + std::shared_ptr profile; + std::vector export_command; + bool allocations = false; + bool dumpProfileOnStderr = false; + Nan::Callback callback; + uint32_t callbackMode = 0; + bool isMainThread = true; + bool callbackInstalled = false; + bool insideCallback = false; +}; + +} // namespace dd diff --git a/bindings/profilers/wall.cc b/bindings/profilers/wall.cc index 0187af47..32e8ef7a 100644 --- a/bindings/profilers/wall.cc +++ b/bindings/profilers/wall.cc @@ -20,9 +20,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -33,6 +35,12 @@ #include "translate-time-profile.hh" #include "wall.hh" +// #if on an undefined macro is 0, which would silently drop the fast path. +#ifndef DD_V8_HAS_DICTIONARY_TEMPLATE +#error \ + "DD_V8_HAS_DICTIONARY_TEMPLATE undefined; per-isolate-data.hh not included" +#endif + #ifndef _WIN32 #define DD_WALL_USE_SIGPROF true @@ -493,6 +501,14 @@ void WallProfiler::Cleanup(Isolate* isolate) { } } +// NewInstance matches values to names by position; generating both from this +// one list keeps them in sync. +#define DD_SAMPLE_CONTEXT_FIELDS \ + X(timestamp) \ + X(cpuTime) \ + X(context) \ + X(asyncId) + ContextsByNode WallProfiler::GetContextsByNode(CpuProfile* profile, ContextBuffer& contexts, int64_t startCpuTime) { @@ -511,10 +527,30 @@ ContextsByNode WallProfiler::GetContextsByNode(CpuProfile* profile, // iteration index int deltaIdx = 0; - auto contextKey = String::NewFromUtf8Literal(isolate, "context"); - auto timestampKey = String::NewFromUtf8Literal(isolate, "timestamp"); - auto cpuTimeKey = String::NewFromUtf8Literal(isolate, "cpuTime"); - auto asyncIdKey = String::NewFromUtf8Literal(isolate, "asyncId"); + Local undefined = Undefined(isolate); +#if DD_V8_HAS_DICTIONARY_TEMPLATE +#define X(name) #name, + static constexpr std::string_view kNames[] = {DD_SAMPLE_CONTEXT_FIELDS}; +#undef X + auto tmpl = PerIsolateData::For(isolate)->GetDictionaryTemplate( + isolate, DictionaryTemplateId::kWallSampleContext, kNames); + + auto newSampleContext = [&](auto& values) { + return tmpl->NewInstance(v8Context, values); + }; +#else +#define X(name) String::NewFromUtf8Literal(isolate, #name), + Local keys[] = {DD_SAMPLE_CONTEXT_FIELDS}; +#undef X + + auto newSampleContext = [&](auto& values) { + auto object = Object::New(isolate); + for (size_t i = 0; i < std::size(values); i++) { + object->Set(v8Context, keys[i], values[i].ToLocalChecked()).Check(); + } + return object; + }; +#endif auto V8toEpochOffset = GetV8ToEpochOffset(); auto lastCpuTime = startCpuTime; @@ -565,44 +601,36 @@ ContextsByNode WallProfiler::GetContextsByNode(CpuProfile* profile, array = it->second.contexts; ++it->second.hitcount; } - // Conforms to TimeProfileNodeContext defined in v8-types.ts - Local timedContext = Object::New(isolate); - timedContext - ->Set(v8Context, - timestampKey, - BigInt::New(isolate, sampleTimestamp + V8toEpochOffset)) - .Check(); + Local timestamp = + BigInt::New(isolate, sampleTimestamp + V8toEpochOffset); + Local cpuTime = undefined; + Local context = undefined; + Local asyncId = undefined; + auto* function_name = sample->GetFunctionNameStr(); // If current sample is program, reports its cpu time to the next sample if (strcmp(function_name, "(program)") != 0) { if (collectCpuTime_) { - timedContext - ->Set( - v8Context, - cpuTimeKey, - Number::New(isolate, sampleContext.cpu_time - lastCpuTime)) - .Check(); + cpuTime = + Number::New(isolate, sampleContext.cpu_time - lastCpuTime); lastCpuTime = sampleContext.cpu_time; } // If current sample is neither program nor idle, associate a sampling // context and async ID if (strcmp(function_name, "(idle)") != 0) { if (sampleContext.context) { - timedContext - ->Set(v8Context, - contextKey, - sampleContext.context.get()->Get(isolate)) - .Check(); + context = sampleContext.context.get()->Get(isolate); } if (collectAsyncId_) { - timedContext - ->Set(v8Context, - asyncIdKey, - Number::New(isolate, sampleContext.async_id)) - .Check(); + asyncId = Number::New(isolate, sampleContext.async_id); } } } + +#define X(name) name, + MaybeLocal values[] = {DD_SAMPLE_CONTEXT_FIELDS}; +#undef X + auto timedContext = newSampleContext(values); array->Set(v8Context, array->Length(), timedContext).Check(); // Sample context was consumed, fetch the next one @@ -614,6 +642,7 @@ ContextsByNode WallProfiler::GetContextsByNode(CpuProfile* profile, return contextsByNode; } +#undef DD_SAMPLE_CONTEXT_FIELDS void GCPrologueCallback(Isolate* isolate, GCType type, diff --git a/package-lock.json b/package-lock.json index 972bf4cc..100a76d9 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@datadog/pprof", - "version": "5.18.1", + "version": "5.19.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@datadog/pprof", - "version": "5.18.1", + "version": "5.19.0", "license": "Apache-2.0", "dependencies": { "node-gyp-build": "^4.8.4", @@ -15,7 +15,7 @@ }, "devDependencies": { "@types/mocha": "^10.0.1", - "@types/node": "26.2.0", + "@types/node": "26.4.1", "@types/semver": "^7.8.0", "@types/sinon": "^22.0.0", "@types/tmp": "^0.2.3", @@ -962,9 +962,9 @@ "license": "MIT" }, "node_modules/@types/node": { - "version": "26.2.0", - "resolved": "https://registry.npmjs.org/@types/node/-/node-26.2.0.tgz", - "integrity": "sha512-5IviulTZeRNp2vAJ514cc/HUlY5nZ9fCbq9DMyC52BrhFZACo3nI0R7qBxhQmo/d27NFe96ur/b7Wwxklda+kg==", + "version": "26.4.1", + "resolved": "https://registry.npmjs.org/@types/node/-/node-26.4.1.tgz", + "integrity": "sha512-k97ENvZWtvA6yqz5/FS6a7duDgOPEeOQOc2iKS/nY6mX6qJUKtLnWzQS+Xj6tXweyj6ZcTAK2Qecetnvi9nCLA==", "dev": true, "license": "MIT", "dependencies": { diff --git a/package.json b/package.json index 85062239..5f2a3395 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@datadog/pprof", - "version": "5.18.1", + "version": "5.19.0", "description": "pprof support for Node.js", "repository": { "type": "git", @@ -43,7 +43,7 @@ }, "devDependencies": { "@types/mocha": "^10.0.1", - "@types/node": "26.2.0", + "@types/node": "26.4.1", "@types/semver": "^7.8.0", "@types/sinon": "^22.0.0", "@types/tmp": "^0.2.3", diff --git a/ts/src/heap-profiler-bindings.ts b/ts/src/heap-profiler-bindings.ts index 9ca439c6..cc5d0d4e 100644 --- a/ts/src/heap-profiler-bindings.ts +++ b/ts/src/heap-profiler-bindings.ts @@ -64,6 +64,7 @@ export function monitorOutOfMemory( callback: NearHeapLimitCallback | undefined, callbackMode: number, isMainThread: boolean, + automaticHeapLimitExtension: boolean, ) { profiler.heapProfiler.monitorOutOfMemory( heapLimitExtensionSize, @@ -73,5 +74,6 @@ export function monitorOutOfMemory( callback, callbackMode, isMainThread, + automaticHeapLimitExtension, ); } diff --git a/ts/src/heap-profiler.ts b/ts/src/heap-profiler.ts index 1d00d422..e520d45e 100644 --- a/ts/src/heap-profiler.ts +++ b/ts/src/heap-profiler.ts @@ -242,25 +242,39 @@ export const CallbackMode = { Both: 3, }; +/** + * How much the heap limit is raised when v8 signals it is near the limit. + * + * A number is an exact byte count, and 0 means "grant no extension and let v8 + * run its normal OOM handling". `'auto'` instead sizes the extension to one + * maximum young generation - the same budget Node.js grants its own near-OOM + * heap snapshot callback - which is what v8 actually needs to finish one more + * GC while the profile is captured. + */ +export type HeapLimitExtensionSize = number | 'auto'; + /** * Add monitoring for v8 heap, heap profiler must already be started. * When an out of heap memory event occurs: - * - an extension of heap memory of |heapLimitExtensionSize| bytes is - * requested to v8. This extension can occur |maxHeapLimitExtensionCount| - * number of times. If the extension amount is not enough to satisfy - * memory allocation that triggers GC and OOM, process will abort. + * - the heap limit is extended by |heapLimitExtensionSize| so a profile can + * be captured before the process dies. If the extension amount is not + * enough to satisfy the memory allocation that triggers GC and OOM, the + * process will abort, so prefer 'auto' over a hand-picked constant. This + * top-level extension can occur |maxHeapLimitExtensionCount| times. + * Reentrant rescue extensions used to finish an in-progress capture are + * additional and are not included in that count. * - heap profile is dumped as folded stacks on stderr if * |dumpHeapProfileOnSdterr| is true * - heap profile is dumped in temporary file and a new process is spawned * with |exportCommand| arguments and profile path appended at the end. - * - |callback| is called. Callback can be invoked only if - * heapLimitExtensionSize is enough for the process to continue. Invocation - * will be done by a RequestInterrupt if |callbackMode| is Interrupt or Both, - * this might be unsafe since Isolate should not be reentered - * from RequestInterrupt, but this allows to interrupt synchronous code. - * Otherwise the callback is scheduled to be called asynchronously. + * - |callback| is called. Callback can be invoked only if the extension is + * enough for the process to continue. Invocation will be done by a + * RequestInterrupt if |callbackMode| is Interrupt or Both, this might be + * unsafe since Isolate should not be reentered from RequestInterrupt, but + * this allows to interrupt synchronous code. Otherwise the callback is + * scheduled to be called asynchronously. * @param heapLimitExtensionSize - amount of bytes heap should be expanded - * with upon OOM + * with upon OOM, or 'auto' to size it to one maximum young generation * @param maxHeapLimitExtensionCount - maximum number of times heap size * extension can occur * @param dumpHeapProfileOnSdterr - dump heap profile on stderr upon OOM @@ -270,7 +284,7 @@ export const CallbackMode = { * @param callbackMode */ export function monitorOutOfMemory( - heapLimitExtensionSize: number, + heapLimitExtensionSize: HeapLimitExtensionSize, maxHeapLimitExtensionCount: number, dumpHeapProfileOnSdterr: boolean, exportCommand?: Array, @@ -288,13 +302,15 @@ export function monitorOutOfMemory( callback(convertProfile(profile)); }; } + const automatic = heapLimitExtensionSize === 'auto'; monitorOutOfMemoryImported( - heapLimitExtensionSize, + automatic ? 0 : heapLimitExtensionSize, maxHeapLimitExtensionCount, dumpHeapProfileOnSdterr, exportCommand || [], newCallback, typeof callbackMode !== 'undefined' ? callbackMode : CallbackMode.Async, isMainThread, + automatic, ); } diff --git a/ts/src/index.ts b/ts/src/index.ts index b4d35efe..10d2052b 100644 --- a/ts/src/index.ts +++ b/ts/src/index.ts @@ -26,6 +26,8 @@ export { LabelSet, } from './v8-types'; +export {HeapLimitExtensionSize} from './heap-profiler'; + export {encode, encodeSync} from './profile-encoder'; export {SourceMapper} from './sourcemapper/sourcemapper'; export {setLogger} from './logger'; diff --git a/ts/src/otel-thread-ctx.ts b/ts/src/otel-thread-ctx.ts index 6ee061ed..d2a4c1e7 100644 --- a/ts/src/otel-thread-ctx.ts +++ b/ts/src/otel-thread-ctx.ts @@ -48,9 +48,8 @@ import { export interface ProcessContextAttributes { readonly 'threadlocal.schema_version': 'nodejs_v1_dev'; readonly 'threadlocal.attribute_key_map': readonly string[]; - readonly 'threadlocal.wrapped_object_offset': number; + readonly 'threadlocal.js_object_record_offset': number; readonly 'threadlocal.tagged_size': number; - readonly 'threadlocal.native_wrap_fields_offset': number; readonly 'threadlocal.js_map_table_offset': number; readonly 'threadlocal.ordered_hash_map_header_size': number; } @@ -63,9 +62,9 @@ export interface ProcessContextAttributes { * * `appendAttributes` mutates the context's record in place. Because every * async-context frame that holds the same `ThreadContext` reference observes - * the same native record buffer, an append is visible across all those - * frames even when the reallocate path runs (the context's internal - * pointer is updated, the JS object is not replaced). + * the same native record, an append is visible across all those frames even + * when the reallocate path runs (the record is re-published on the same JS + * object, which is never replaced). */ export interface ThreadContext { appendAttributes( @@ -85,6 +84,20 @@ export interface ThreadContext { */ invalidate(): void; + /** + * Overwrite this context's W3C trace-flags byte in place, for the case + * where the flags are not yet known when the context is built — an SDK + * whose sampling decision is deferred learns the sampled bit later, and a + * decision already taken can still be overridden. Like {@link invalidate}, + * the write is seen at once by every async-context frame holding this + * context, because they all share one record. + * + * Must be an integer in 0..255. Bits beyond those W3C currently defines are + * stored as given rather than masked off, since W3C requires unknown flag + * bits to be propagated. + */ + setTraceFlags(traceFlags: number): void; + isTruncated(): boolean; /** Debug-only: returns the on-the-wire record bytes. Not stable. */ debugBytes(): Uint8Array; @@ -119,6 +132,7 @@ export interface ThreadContextCtor { new ( traceId: Uint8Array, spanId: Uint8Array, + traceFlags?: number, attributes?: Array, ): ThreadContext; readonly prototype: ThreadContext; @@ -128,9 +142,8 @@ interface Addon { threadContext: ThreadContextCtor; otelThreadCtxStoreAls(als: AsyncLocalStorage): void; otelThreadCtxGetStoredAlsHash(): number; - otelThreadCtxWrappedObjectOffset: number; + otelThreadCtxJsObjectRecordOffset: number; otelThreadCtxTaggedSize: number; - otelThreadCtxNativeWrapFieldsOffset: number; otelThreadCtxJsMapTableOffset: number; otelThreadCtxOrderedHashMapHeaderSize: number; } @@ -142,9 +155,8 @@ const SCHEMA_VERSION = 'nodejs_v1_dev'; // (no V8 pointer compression, no sandbox); the reader is Linux-only per // the OTEP anyway, so the fallbacks just keep processContextAttributes // consistent in shape. -let WRAPPED_OBJECT_OFFSET = 24; +let JS_OBJECT_RECORD_OFFSET = 0x18; let TAGGED_SIZE = 8; -let NATIVE_WRAP_FIELDS_OFFSET = 0; let JS_MAP_TABLE_OFFSET = 0x18; let ORDERED_HASH_MAP_HEADER_SIZE = 0x10; @@ -171,9 +183,8 @@ if (process.platform === 'linux') { // eslint-disable-next-line @typescript-eslint/no-require-imports const findBinding = require('node-gyp-build'); const addon: Addon = findBinding(join(__dirname, '..', '..')); - WRAPPED_OBJECT_OFFSET = addon.otelThreadCtxWrappedObjectOffset; + JS_OBJECT_RECORD_OFFSET = addon.otelThreadCtxJsObjectRecordOffset; TAGGED_SIZE = addon.otelThreadCtxTaggedSize; - NATIVE_WRAP_FIELDS_OFFSET = addon.otelThreadCtxNativeWrapFieldsOffset; JS_MAP_TABLE_OFFSET = addon.otelThreadCtxJsMapTableOffset; ORDERED_HASH_MAP_HEADER_SIZE = addon.otelThreadCtxOrderedHashMapHeaderSize; @@ -234,6 +245,7 @@ if (process.platform === 'linux') { class NoopThreadContext implements ThreadContext { appendAttributes(): void {} invalidate(): void {} + setTraceFlags(): void {} isTruncated(): boolean { return false; } @@ -286,9 +298,8 @@ export function getProcessContextAttributes( return Object.freeze({ 'threadlocal.schema_version': SCHEMA_VERSION, 'threadlocal.attribute_key_map': Object.freeze(keys.slice()), - 'threadlocal.wrapped_object_offset': WRAPPED_OBJECT_OFFSET, + 'threadlocal.js_object_record_offset': JS_OBJECT_RECORD_OFFSET, 'threadlocal.tagged_size': TAGGED_SIZE, - 'threadlocal.native_wrap_fields_offset': NATIVE_WRAP_FIELDS_OFFSET, 'threadlocal.js_map_table_offset': JS_MAP_TABLE_OFFSET, 'threadlocal.ordered_hash_map_header_size': ORDERED_HASH_MAP_HEADER_SIZE, }) as ProcessContextAttributes; diff --git a/ts/test/oom-heap-limit-extension.ts b/ts/test/oom-heap-limit-extension.ts new file mode 100644 index 00000000..0e5320a9 --- /dev/null +++ b/ts/test/oom-heap-limit-extension.ts @@ -0,0 +1,65 @@ +/* + * Copyright 2026 Datadog, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +'use strict'; + +import * as v8 from 'v8'; + +import {heap, HeapLimitExtensionSize} from '../src/index'; + +const MB = 1024 * 1024; +const CHUNK_SIZE = 4 * MB; +const MAX_CHUNKS = 64; +const heapLimitExtensionSize: HeapLimitExtensionSize = + process.argv[2] === 'auto' ? 'auto' : Number(process.argv[2] || 0); + +function heapLimit() { + return v8.getHeapStatistics().heap_size_limit; +} + +heap.start(MB, 64); +heap.monitorOutOfMemory(heapLimitExtensionSize, 1, false); + +const initialLimit = heapLimit(); +const retained: number[][] = []; +let chunks = 0; + +// Report every heap limit the process observes so the parent can tell whether +// a top-level extension was granted even if v8 aborts us mid-leak. A near-heap +// limit event is not necessarily fatal - v8 may free enough and carry on - so +// the limit is the only reliable signal here, not survival. +console.log(`limit ${initialLimit}`); + +function leak() { + const limit = heapLimit(); + if (limit !== initialLimit) { + console.log(`limit ${limit}`); + process.exit(0); + } + if (chunks >= MAX_CHUNKS) { + process.exit(0); + } + + const chunk = new Array(CHUNK_SIZE / 8); + for (let i = 0; i < chunk.length; i++) { + chunk[i] = i + 0.1; + } + retained.push(chunk); + chunks++; + setTimeout(leak, 5); +} + +leak(); diff --git a/ts/test/oom-restore-heap-limit.ts b/ts/test/oom-restore-heap-limit.ts index c053a03c..6432bac6 100644 --- a/ts/test/oom-restore-heap-limit.ts +++ b/ts/test/oom-restore-heap-limit.ts @@ -18,11 +18,13 @@ import * as v8 from 'v8'; -import {heap} from '../src/index'; +import {heap, HeapLimitExtensionSize} from '../src/index'; const MB = 1024 * 1024; const LIMIT_TOLERANCE = 16 * MB; const CHUNK_SIZE = 4 * MB; +const heapLimitExtensionSize: HeapLimitExtensionSize = + process.argv[2] === 'auto' ? 'auto' : Number(process.argv[2] || 0); const gc = (global as typeof globalThis & {gc?: () => void}).gc; function heapLimit() { @@ -44,7 +46,7 @@ async function main() { heap.start(MB, 64); try { - heap.monitorOutOfMemory(64 * MB, 1, false); + heap.monitorOutOfMemory(heapLimitExtensionSize, 1, false); const initialLimit = heapLimit(); const retained: number[][] = []; diff --git a/ts/test/otel-ctx-teardown.ts b/ts/test/otel-ctx-teardown.ts index 1780b73f..50a0f58c 100644 --- a/ts/test/otel-ctx-teardown.ts +++ b/ts/test/otel-ctx-teardown.ts @@ -46,7 +46,7 @@ function id(n: number, len: number): Uint8Array { const retained: unknown[] = []; for (let i = 0; i < N; i++) { - const ctx = new otelThreadCtx.ThreadContext(id(i, 16), id(i, 8), [ + const ctx = new otelThreadCtx.ThreadContext(id(i, 16), id(i, 8), 0, [ 'k', String(i), ]); diff --git a/ts/test/test-heap-profiler.ts b/ts/test/test-heap-profiler.ts index 20b62b80..26bcca70 100644 --- a/ts/test/test-heap-profiler.ts +++ b/ts/test/test-heap-profiler.ts @@ -377,10 +377,8 @@ describe('foreign heap sampler', () => { }); describe('OOMMonitoring', () => { - it('should restore heap limit after v8 recovers from OOM', async function () { - this.timeout(30000); - - const proc = fork(path.join(__dirname, 'oom-restore-heap-limit.js'), { + async function runOomFixture(script: string, heapLimitExtensionSize: string) { + const proc = fork(path.join(__dirname, script), [heapLimitExtensionSize], { execArgv: ['--expose-gc', '--max-old-space-size=64'], silent: true, }); @@ -393,18 +391,88 @@ describe('OOMMonitoring', () => { output += chunk; }); - await new Promise((resolve, reject) => { - proc.on('error', reject); - proc.on('exit', code => { - if (code === 0) { - resolve(); - } else { - reject( - new Error(`oom-restore-heap-limit exited with ${code}\n${output}`), - ); - } - }); - }); + return new Promise<{code: number | null; output: string}>( + (resolve, reject) => { + proc.on('error', reject); + proc.on('exit', code => { + resolve({code, output}); + }); + }, + ); + } + + async function assertHeapLimitIsRestored(heapLimitExtensionSize: string) { + const {code, output} = await runOomFixture( + 'oom-restore-heap-limit.js', + heapLimitExtensionSize, + ); + assert.strictEqual( + code, + 0, + `oom-restore-heap-limit exited with ${code}\n${output}`, + ); + } + + it('should restore an automatic heap limit extension', async function () { + this.timeout(30000); + await assertHeapLimitIsRestored('auto'); + }); + + it('should restore a configured heap limit extension', async function () { + this.timeout(30000); + await assertHeapLimitIsRestored(String(64 * 1024 * 1024)); + }); + + // The fixture runs under --max-old-space-size=64, and v8 reports + // heap_size_limit as the old generation limit plus one maximum young + // generation, so the young generation the automatic mode should grant is + // recoverable from the first limit the fixture reports. + const MAX_OLD_SPACE = 64 * 1024 * 1024; + + async function grantedHeapLimitExtension(heapLimitExtensionSize: string) { + const {output} = await runOomFixture( + 'oom-heap-limit-extension.js', + heapLimitExtensionSize, + ); + const limits = [...output.matchAll(/^limit (\d+)$/gm)].map(match => + Number(match[1]), + ); + assert.ok( + output.includes('NearHeapLimit(count='), + `the near heap limit callback never ran\n${output}`, + ); + assert.ok(limits.length > 0, `no heap limit was reported\n${output}`); + return { + granted: Math.max(...limits) - limits[0], + youngGeneration: limits[0] - MAX_OLD_SPACE, + output, + }; + } + + it('should grant exactly one young generation when set to auto', async function () { + this.timeout(30000); + const {granted, youngGeneration, output} = + await grantedHeapLimitExtension('auto'); + assert.strictEqual( + granted, + youngGeneration, + `expected one young generation of headroom\n${output}`, + ); + }); + + // A size of 0 must keep meaning "grant no top-level extension" so that + // upgrading does not silently start extending the heap of callers already + // passing 0. Only the reentrant rescue grant that lets an in-progress + // capture finish may raise the limit, and it is far below a young + // generation. + it('should not grant a top-level extension when the size is 0', async function () { + this.timeout(30000); + const {granted, youngGeneration, output} = + await grantedHeapLimitExtension('0'); + assert.ok( + granted < youngGeneration, + `expected no young-generation extension, got ${granted}\n${output}`, + ); }); it('should call external process upon OOM', async function () { diff --git a/ts/test/test-otel-thread-ctx.ts b/ts/test/test-otel-thread-ctx.ts index e28a6f03..372a5fff 100644 --- a/ts/test/test-otel-thread-ctx.ts +++ b/ts/test/test-otel-thread-ctx.ts @@ -44,13 +44,24 @@ import { interface PosOpts { traceId: Uint8Array; spanId: Uint8Array; + traceFlags?: number; attributes?: Array; } function tcRun(fn: () => T, opts: PosOpts): T { - return new ThreadContext(opts.traceId, opts.spanId, opts.attributes).run(fn); + return new ThreadContext( + opts.traceId, + opts.spanId, + opts.traceFlags, + opts.attributes, + ).run(fn); } function tcEnter(opts: PosOpts): void { - new ThreadContext(opts.traceId, opts.spanId, opts.attributes).enter(); + new ThreadContext( + opts.traceId, + opts.spanId, + opts.traceFlags, + opts.attributes, + ).enter(); } function tcAppend( attributes: Array | undefined, @@ -83,7 +94,7 @@ interface Header { traceId: Uint8Array; spanId: Uint8Array; valid: number; - reserved: number; + traceFlags: number; attrsDataSize: number; } @@ -102,7 +113,7 @@ function decodeHeader(bytes: Uint8Array): Header { traceId: bytes.slice(0, 16), spanId: bytes.slice(16, 24), valid: bytes[24], - reserved: bytes[25], + traceFlags: bytes[25], attrsDataSize, }; } @@ -191,7 +202,7 @@ function captureBytes(opts: { strictAssert.deepEqual(hdr.traceId, TRACE_ID_BYTES); strictAssert.deepEqual(hdr.spanId, SPAN_ID_BYTES); strictAssert.equal(hdr.valid, 1); - strictAssert.equal(hdr.reserved, 0); + strictAssert.equal(hdr.traceFlags, 0); strictAssert.equal(hdr.attrsDataSize, 0); }); @@ -726,6 +737,98 @@ function captureBytes(opts: { }); }); + describe('traceFlags', () => { + it('defaults to 0 when not supplied', () => { + const bytes = tcRun(() => _currentRecordBytes()!, { + traceId: TRACE_ID_BYTES, + spanId: SPAN_ID_BYTES, + }); + strictAssert.equal(decodeHeader(bytes).traceFlags, 0); + }); + + it('writes the byte given at construction', () => { + const bytes = tcRun(() => _currentRecordBytes()!, { + traceId: TRACE_ID_BYTES, + spanId: SPAN_ID_BYTES, + traceFlags: 0x01, + }); + strictAssert.equal(decodeHeader(bytes).traceFlags, 0x01); + }); + + it('keeps bits W3C has not defined rather than masking them off', () => { + const bytes = tcRun(() => _currentRecordBytes()!, { + traceId: TRACE_ID_BYTES, + spanId: SPAN_ID_BYTES, + traceFlags: 0xff, + }); + strictAssert.equal(decodeHeader(bytes).traceFlags, 0xff); + }); + + it('coexists with attributes in the fourth argument', () => { + const bytes = tcRun(() => _currentRecordBytes()!, { + traceId: TRACE_ID_BYTES, + spanId: SPAN_ID_BYTES, + traceFlags: 0x03, + attributes: ['v0'], + }); + const hdr = decodeHeader(bytes); + strictAssert.equal(hdr.traceFlags, 0x03); + strictAssert.deepEqual(decodeAttrs(bytes), ['v0']); + }); + + it('rejects non-integers and out-of-range values', () => { + for (const bad of [-1, 256, 1.5, NaN, '1' as unknown as number]) { + strictAssert.throws( + () => new ThreadContext(TRACE_ID_BYTES, SPAN_ID_BYTES, bad), + /traceFlags must be an integer in 0\.\.255/, + `expected ${String(bad)} to be rejected`, + ); + } + }); + + it('setTraceFlags overwrites in place, visible on the shared record', () => { + // The deferred-sampling case: the record is built before the sampled + // bit is known, and every frame holding this context must see the + // update, not just the one that made it. + const ctx = new ThreadContext(TRACE_ID_BYTES, SPAN_ID_BYTES); + ctx.run(() => { + strictAssert.equal( + decodeHeader(_currentRecordBytes()!).traceFlags, + 0, + ); + ctx.setTraceFlags(0x01); + strictAssert.equal( + decodeHeader(_currentRecordBytes()!).traceFlags, + 0x01, + ); + }); + }); + + it('survives an append that reallocates the record', () => { + const ctx = new ThreadContext(TRACE_ID_BYTES, SPAN_ID_BYTES, 0x01); + ctx.run(() => { + // Overflow the initial 36-byte slack so the wrap has to move. + ctx.appendAttributes([undefined, 'x'.repeat(200)]); + ctx.appendAttributes([undefined, undefined, 'y'.repeat(200)]); + const hdr = decodeHeader(_currentRecordBytes()!); + strictAssert.equal(hdr.traceFlags, 0x01); + strictAssert.equal(hdr.valid, 1); + }); + }); + + it('setTraceFlags still reaches the record after a reallocation', () => { + const ctx = new ThreadContext(TRACE_ID_BYTES, SPAN_ID_BYTES); + ctx.run(() => { + ctx.appendAttributes([undefined, 'x'.repeat(300)]); + ctx.setTraceFlags(0x03); + strictAssert.equal( + decodeHeader(_currentRecordBytes()!).traceFlags, + 0x03, + ); + }); + }); + }); + describe('invalidate', () => { it('flips the record valid byte to 0 in place', () => { // Verified through the shared record: same ThreadContext reference @@ -837,9 +940,8 @@ function captureBytes(opts: { const pca = getProcessContextAttributes(keys); strictAssert.equal(pca['threadlocal.schema_version'], 'nodejs_v1_dev'); strictAssert.deepEqual(pca['threadlocal.attribute_key_map'], keys); - strictAssert.equal(pca['threadlocal.wrapped_object_offset'], 24); + strictAssert.equal(pca['threadlocal.js_object_record_offset'], 0x18); strictAssert.equal(pca['threadlocal.tagged_size'], 8); - strictAssert.equal(pca['threadlocal.native_wrap_fields_offset'], 0); strictAssert.equal(pca['threadlocal.js_map_table_offset'], 0x18); strictAssert.equal( pca['threadlocal.ordered_hash_map_header_size'], @@ -848,11 +950,10 @@ function captureBytes(opts: { strictAssert.deepEqual(Object.keys(pca).sort(), [ 'threadlocal.attribute_key_map', 'threadlocal.js_map_table_offset', - 'threadlocal.native_wrap_fields_offset', + 'threadlocal.js_object_record_offset', 'threadlocal.ordered_hash_map_header_size', 'threadlocal.schema_version', 'threadlocal.tagged_size', - 'threadlocal.wrapped_object_offset', ]); });