Merge the koffi allocation-leak fix into the retention line
This brings fix batch 3c (per-runtime out-buffers, freed on the last release, plus a steady-state RSS test in its own child process) onto the branch carrying fix batches 3a and 3b and retention. The merge was automatic. Checked by hand: release() frees the reused buffers only after the reference count reaches zero and the pump has exited. retain() (from 3a) is called only while the provider already holds a reference, so it cannot resurrect a runtime that is closing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
58a6fc5c06
3 changed files with 177 additions and 60 deletions
|
|
@ -18,7 +18,7 @@
|
|||
"build": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json",
|
||||
"prepack": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json",
|
||||
"typecheck": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json --pretty false",
|
||||
"test": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --test dist-test/test/*.test.js"
|
||||
"test": "node ../../node_modules/typescript/bin/tsc -p tsconfig.json && node ../../node_modules/typescript/bin/tsc -p tsconfig.test.json && node --expose-gc --test dist-test/test/*.test.js"
|
||||
},
|
||||
"keywords": [
|
||||
"cultmesh",
|
||||
|
|
|
|||
|
|
@ -88,35 +88,26 @@ interface NativeEventStruct {
|
|||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
type KoffiLib = any;
|
||||
|
||||
/** `nextEventAsync`'s signature, typed explicitly because `promisify` cannot
|
||||
* infer one from koffi's untyped (`any`) async export. */
|
||||
type NextEventAsyncFn = (
|
||||
runtime: CultMeshQuicNativeHandle,
|
||||
timeoutMs: number,
|
||||
outEvent: unknown,
|
||||
payload: Uint8Array | null,
|
||||
payloadCapacity: number,
|
||||
outRequired: unknown,
|
||||
) => Promise<number>;
|
||||
|
||||
/**
|
||||
* The loaded native library: koffi itself, the declared struct, and the
|
||||
* twelve raw v2 exports. Untyped (koffi is `any`); the runtime class methods
|
||||
* below are the type surface for these calls, so wrapping them a second time
|
||||
* in a typed interface would only restate what the methods already say.
|
||||
* in a typed interface would only restate what the methods already say. The
|
||||
* type is inferred from `buildLibrary`'s return value instead of restated.
|
||||
*/
|
||||
interface LoadedNativeLibrary {
|
||||
koffi: KoffiLib;
|
||||
event: KoffiLib;
|
||||
runtimeOpenFn: KoffiLib;
|
||||
runtimeCloseFn: KoffiLib;
|
||||
listenerOpenFn: KoffiLib;
|
||||
listenerCloseFn: KoffiLib;
|
||||
connectionOpenFn: KoffiLib;
|
||||
connectionCertificateCompleteFn: KoffiLib;
|
||||
connectionShutdownFn: KoffiLib;
|
||||
streamOpenFn: KoffiLib;
|
||||
streamSendFrameFn: KoffiLib;
|
||||
streamShutdownFn: KoffiLib;
|
||||
nextEventAsync: (
|
||||
runtime: CultMeshQuicNativeHandle,
|
||||
timeoutMs: number,
|
||||
outEvent: unknown,
|
||||
payload: Uint8Array | null,
|
||||
payloadCapacity: number,
|
||||
outRequired: unknown,
|
||||
) => Promise<number>;
|
||||
lastErrorFn: KoffiLib;
|
||||
}
|
||||
// eslint-disable-next-line @typescript-eslint/no-use-before-define
|
||||
type LoadedNativeLibrary = ReturnType<typeof buildLibrary>;
|
||||
|
||||
let libraryCache: LoadedNativeLibrary | undefined;
|
||||
|
||||
|
|
@ -140,10 +131,8 @@ function resolveNativeDir(): string {
|
|||
return join(__dirname, "..", "native", platformArchDir());
|
||||
}
|
||||
|
||||
/** Lazily loads koffi, resolves the binary directory, and declares the ABI. */
|
||||
function loadBindings(): LoadedNativeLibrary {
|
||||
if (libraryCache) return libraryCache;
|
||||
|
||||
/** Loads koffi, resolves the binary directory, and declares the ABI. */
|
||||
function buildLibrary() {
|
||||
// Untyped on purpose: koffi's own type declarations are ESM-flavored and
|
||||
// fight a CommonJS `typeof import(...)` reference under Node16 resolution.
|
||||
// eslint-disable-next-line @typescript-eslint/no-var-requires, @typescript-eslint/no-explicit-any
|
||||
|
|
@ -222,9 +211,9 @@ function loadBindings(): LoadedNativeLibrary {
|
|||
|
||||
// eslint-disable-next-line @typescript-eslint/no-var-requires
|
||||
const { promisify } = require("node:util") as typeof import("node:util");
|
||||
const nextEventAsync = promisify(nextEventFn.async) as LoadedNativeLibrary["nextEventAsync"];
|
||||
const nextEventAsync = promisify(nextEventFn.async) as NextEventAsyncFn;
|
||||
|
||||
libraryCache = {
|
||||
return {
|
||||
koffi,
|
||||
event,
|
||||
runtimeOpenFn,
|
||||
|
|
@ -240,6 +229,12 @@ function loadBindings(): LoadedNativeLibrary {
|
|||
nextEventAsync,
|
||||
lastErrorFn,
|
||||
};
|
||||
}
|
||||
|
||||
/** Lazily builds and caches the loaded native library. */
|
||||
function loadBindings(): LoadedNativeLibrary {
|
||||
if (libraryCache) return libraryCache;
|
||||
libraryCache = buildLibrary();
|
||||
return libraryCache;
|
||||
}
|
||||
|
||||
|
|
@ -275,9 +270,21 @@ export class CultMeshQuicNativeRuntime {
|
|||
private closed = false;
|
||||
private pumpLoop: Promise<void>;
|
||||
|
||||
/**
|
||||
* `nextEvent`'s out-parameter buffers, allocated once and reused for every
|
||||
* call instead of per call: the pump is single-threaded with at most one
|
||||
* `nextEvent` in flight (`nextEventAsync` is only ever awaited, never
|
||||
* fired concurrently), so decode-then-reuse is safe. Freed once in
|
||||
* `release()` after the pump loop has exited.
|
||||
*/
|
||||
private readonly outEventBuffer: unknown;
|
||||
private readonly outRequiredBuffer: unknown;
|
||||
|
||||
private constructor(handle: CultMeshQuicNativeHandle, lib: LoadedNativeLibrary) {
|
||||
this.handle = handle;
|
||||
this.lib = lib;
|
||||
this.outEventBuffer = lib.koffi.alloc(lib.event, 1);
|
||||
this.outRequiredBuffer = lib.koffi.alloc("int32_t", 1);
|
||||
this.pumpLoop = this.pump();
|
||||
}
|
||||
|
||||
|
|
@ -287,9 +294,11 @@ export class CultMeshQuicNativeRuntime {
|
|||
const out = lib.koffi.alloc("void *", 1);
|
||||
const status = lib.runtimeOpenFn(null, out);
|
||||
if (status !== 0) {
|
||||
lib.koffi.free(out);
|
||||
throw new Error(`cultmesh_quic_runtime_open failed with status ${status}.`);
|
||||
}
|
||||
const handle = toBigInt64(lib.koffi.decode(out, "void *"));
|
||||
lib.koffi.free(out);
|
||||
CultMeshQuicNativeRuntime.shared = new CultMeshQuicNativeRuntime(handle, lib);
|
||||
}
|
||||
CultMeshQuicNativeRuntime.refCountValue += 1;
|
||||
|
|
@ -328,6 +337,8 @@ export class CultMeshQuicNativeRuntime {
|
|||
CultMeshQuicNativeRuntime.shared = undefined;
|
||||
await this.pumpLoop;
|
||||
this.lib.runtimeCloseFn(this.handle);
|
||||
this.lib.koffi.free(this.outEventBuffer);
|
||||
this.lib.koffi.free(this.outRequiredBuffer);
|
||||
}
|
||||
|
||||
onListenerEvent(listenerId: bigint, listener: NativeEventListener): void {
|
||||
|
|
@ -362,23 +373,28 @@ export class CultMeshQuicNativeRuntime {
|
|||
): { listenerId: bigint; boundPort: number } {
|
||||
const outListenerId = this.lib.koffi.alloc("uint64_t", 1);
|
||||
const outBoundPort = this.lib.koffi.alloc("uint16_t", 1);
|
||||
const status = this.lib.listenerOpenFn(
|
||||
this.handle,
|
||||
host,
|
||||
port,
|
||||
pkcs12,
|
||||
pkcs12.length,
|
||||
password,
|
||||
outListenerId,
|
||||
outBoundPort,
|
||||
);
|
||||
if (status !== 0) {
|
||||
throw new Error(`cultmesh_quic_listener_open failed: ${readLastError(this.lib, this.handle)}`);
|
||||
try {
|
||||
const status = this.lib.listenerOpenFn(
|
||||
this.handle,
|
||||
host,
|
||||
port,
|
||||
pkcs12,
|
||||
pkcs12.length,
|
||||
password,
|
||||
outListenerId,
|
||||
outBoundPort,
|
||||
);
|
||||
if (status !== 0) {
|
||||
throw new Error(`cultmesh_quic_listener_open failed: ${readLastError(this.lib, this.handle)}`);
|
||||
}
|
||||
return {
|
||||
listenerId: toBigInt64(this.lib.koffi.decode(outListenerId, "uint64_t")),
|
||||
boundPort: this.lib.koffi.decode(outBoundPort, "uint16_t") as number,
|
||||
};
|
||||
} finally {
|
||||
this.lib.koffi.free(outListenerId);
|
||||
this.lib.koffi.free(outBoundPort);
|
||||
}
|
||||
return {
|
||||
listenerId: toBigInt64(this.lib.koffi.decode(outListenerId, "uint64_t")),
|
||||
boundPort: this.lib.koffi.decode(outBoundPort, "uint16_t") as number,
|
||||
};
|
||||
}
|
||||
|
||||
listenerClose(listenerId: bigint): void {
|
||||
|
|
@ -388,11 +404,15 @@ export class CultMeshQuicNativeRuntime {
|
|||
|
||||
connectionOpen(host: string, port: number): bigint {
|
||||
const out = this.lib.koffi.alloc("uint64_t", 1);
|
||||
const status = this.lib.connectionOpenFn(this.handle, host, port, out);
|
||||
if (status !== 0) {
|
||||
throw new Error(`cultmesh_quic_connection_open failed: ${readLastError(this.lib, this.handle)}`);
|
||||
try {
|
||||
const status = this.lib.connectionOpenFn(this.handle, host, port, out);
|
||||
if (status !== 0) {
|
||||
throw new Error(`cultmesh_quic_connection_open failed: ${readLastError(this.lib, this.handle)}`);
|
||||
}
|
||||
return toBigInt64(this.lib.koffi.decode(out, "uint64_t"));
|
||||
} finally {
|
||||
this.lib.koffi.free(out);
|
||||
}
|
||||
return toBigInt64(this.lib.koffi.decode(out, "uint64_t"));
|
||||
}
|
||||
|
||||
connectionCertificateComplete(connectionId: bigint, accept: boolean): void {
|
||||
|
|
@ -410,11 +430,15 @@ export class CultMeshQuicNativeRuntime {
|
|||
|
||||
streamOpen(connectionId: bigint, kind: number): bigint {
|
||||
const out = this.lib.koffi.alloc("uint64_t", 1);
|
||||
const status = this.lib.streamOpenFn(this.handle, connectionId, kind, out);
|
||||
if (status !== 0) {
|
||||
throw new Error(`cultmesh_quic_stream_open failed: ${readLastError(this.lib, this.handle)}`);
|
||||
try {
|
||||
const status = this.lib.streamOpenFn(this.handle, connectionId, kind, out);
|
||||
if (status !== 0) {
|
||||
throw new Error(`cultmesh_quic_stream_open failed: ${readLastError(this.lib, this.handle)}`);
|
||||
}
|
||||
return toBigInt64(this.lib.koffi.decode(out, "uint64_t"));
|
||||
} finally {
|
||||
this.lib.koffi.free(out);
|
||||
}
|
||||
return toBigInt64(this.lib.koffi.decode(out, "uint64_t"));
|
||||
}
|
||||
|
||||
streamSendFrame(streamId: bigint, encodedFrame: Uint8Array, fin: boolean): void {
|
||||
|
|
@ -497,10 +521,21 @@ export class CultMeshQuicNativeRuntime {
|
|||
payload: Uint8Array | null,
|
||||
payloadCapacity: number,
|
||||
): Promise<{ status: number; event: NativeEventStruct | null; required: number }> {
|
||||
const outEvent = this.lib.koffi.alloc(this.lib.event, 1);
|
||||
const outRequired = this.lib.koffi.alloc("int32_t", 1);
|
||||
const status = await this.lib.nextEventAsync(this.handle, timeoutMs, outEvent, payload, payloadCapacity, outRequired);
|
||||
const decoded = status === 1 ? (this.lib.koffi.decode(outEvent, this.lib.event) as NativeEventStruct) : null;
|
||||
// Reuses the buffers allocated once in the constructor instead of
|
||||
// allocating fresh ones per call: the pump awaits each `nextEventAsync`
|
||||
// before issuing the next one, so at most one call is ever in flight
|
||||
// against them, and both are fully decoded here before the buffers can
|
||||
// be reused by the pump's next call.
|
||||
const status = await this.lib.nextEventAsync(
|
||||
this.handle,
|
||||
timeoutMs,
|
||||
this.outEventBuffer,
|
||||
payload,
|
||||
payloadCapacity,
|
||||
this.outRequiredBuffer,
|
||||
);
|
||||
const decoded =
|
||||
status === 1 ? (this.lib.koffi.decode(this.outEventBuffer, this.lib.event) as NativeEventStruct) : null;
|
||||
return {
|
||||
status,
|
||||
event: decoded && {
|
||||
|
|
@ -510,7 +545,7 @@ export class CultMeshQuicNativeRuntime {
|
|||
stream_id: toBigInt64(decoded.stream_id),
|
||||
code: toBigInt64(decoded.code),
|
||||
},
|
||||
required: this.lib.koffi.decode(outRequired, "int32_t") as number,
|
||||
required: this.lib.koffi.decode(this.outRequiredBuffer, "int32_t") as number,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@
|
|||
// `packages/cultmesh-ts/native/<platform>` tree once a later cut lands it.
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import test from "node:test";
|
||||
|
|
@ -304,3 +305,84 @@ test("negative grep: no koffi callbacks, .async only on nextEvent", () => {
|
|||
}
|
||||
assert.doesNotMatch(source, /koffi\.register/, "no koffi callbacks");
|
||||
});
|
||||
|
||||
// The measurement below runs in its own `node --expose-gc` child process
|
||||
// instead of in this test's own process: by the time this test would run,
|
||||
// ~150 earlier tests in the suite have already grown and partly freed the
|
||||
// glibc heap, so a real per-call native leak can land in already-mapped free
|
||||
// space and never show up as RSS growth in *this* process (confirmed by hand:
|
||||
// the in-process version of this test passed even with `nextEvent` mutated
|
||||
// back to allocating fresh koffi buffers every call). A freshly spawned
|
||||
// process has no such history, matching the ~8 MiB (fixed) vs ~30 MiB
|
||||
// (pre-fix) split measured in isolation.
|
||||
const MEMORY_PROBE_SCRIPT = `
|
||||
const { CultMeshQuicNativeRuntime } = require(process.argv[1]);
|
||||
(async () => {
|
||||
const runtime = await CultMeshQuicNativeRuntime.open();
|
||||
const nextEvent = runtime.nextEvent.bind(runtime);
|
||||
try {
|
||||
for (let i = 0; i < 2000; i += 1) await nextEvent(0, null, 0);
|
||||
global.gc();
|
||||
const before = process.memoryUsage().rss;
|
||||
const ITERATIONS = 200000;
|
||||
for (let i = 0; i < ITERATIONS; i += 1) await nextEvent(0, null, 0);
|
||||
global.gc();
|
||||
const after = process.memoryUsage().rss;
|
||||
process.stdout.write(JSON.stringify({ before, after, growthBytes: after - before, ITERATIONS }));
|
||||
} finally {
|
||||
await runtime.release();
|
||||
}
|
||||
})().catch((error) => {
|
||||
console.error(error);
|
||||
process.exit(1);
|
||||
});
|
||||
`;
|
||||
|
||||
test(
|
||||
"memory: sustained nextEvent calls do not leak koffi-allocated out-parameter buffers",
|
||||
{ timeout: 60_000 },
|
||||
(t) => {
|
||||
if (!nativeBridgeAvailable()) {
|
||||
t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge.");
|
||||
return;
|
||||
}
|
||||
|
||||
// `nextEvent` is private; the compiled module path is resolved the same
|
||||
// way this file's own `import` resolves it, and handed to the child by
|
||||
// argv so the child needs no path logic of its own.
|
||||
const modulePath = require.resolve("../src/realtime-quic-native");
|
||||
const result = spawnSync(process.execPath, ["--expose-gc", "-e", MEMORY_PROBE_SCRIPT, modulePath], {
|
||||
encoding: "utf8",
|
||||
env: process.env,
|
||||
});
|
||||
if (result.status !== 0 || typeof result.stdout !== "string" || result.stdout.trim().length === 0) {
|
||||
if (/--expose-gc/.test(result.stderr ?? "")) {
|
||||
t.skip("this Node build does not support --expose-gc.");
|
||||
return;
|
||||
}
|
||||
assert.fail(
|
||||
`memory probe child process failed (status ${String(result.status)}):\n${result.stderr ?? "(no stderr)"}`,
|
||||
);
|
||||
}
|
||||
|
||||
const { before, after, growthBytes, ITERATIONS } = JSON.parse(result.stdout.trim()) as {
|
||||
before: number;
|
||||
after: number;
|
||||
growthBytes: number;
|
||||
ITERATIONS: number;
|
||||
};
|
||||
// Measured on Yggdrasil, same build and host, in a dedicated process:
|
||||
// fixed code (0 `koffi.alloc` calls in the loop): ~8.0 MiB RSS growth.
|
||||
// pre-fix code (400k `koffi.alloc` calls, never freed): ~29.9 MiB.
|
||||
// 18 MiB sits well above the fixed baseline and well below the leaking
|
||||
// one, so the bound stays robust to ordinary run-to-run noise while
|
||||
// still catching the regression.
|
||||
const BOUND_BYTES = 18 * 1024 * 1024;
|
||||
assert.ok(
|
||||
growthBytes < BOUND_BYTES,
|
||||
`RSS grew by ${growthBytes} bytes (${(growthBytes / 1024 / 1024).toFixed(2)} MiB, from ${before} to ${after}) ` +
|
||||
`over ${ITERATIONS} nextEvent calls; expected it to stay under ${BOUND_BYTES / 1024 / 1024} MiB once ` +
|
||||
"out-parameter buffers are reused instead of allocated per call.",
|
||||
);
|
||||
},
|
||||
);
|
||||
|
|
|
|||
Loading…
Reference in a new issue