Something went wrong. Try again.
Offload functions to worker threads with shared memory primitives for Node.js.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126import type { MessagePort, Transferable } from 'node:worker_threads';import { serializeArg, deserializeArg } from './shared/reconstruct.ts';import { collectTransferables, extractTransferables } from './transfer.ts';import { shared } from './shared/shared.ts';import { int32atomic } from './shared/descriptors.ts';import type { Int32Atomic } from './shared/int32-atomic.ts';import type { SharedStruct } from './shared/shared-struct.ts';
export const CANCEL = 1;export const DEFAULT_HIGH_WATER = 16;
/** Inflight count + state for one producer/consumer pair. */export type PipeFlags = SharedStruct<{ inflight: Int32Atomic; state: Int32Atomic }>;
export function newPipeFlags(): PipeFlags { return shared({ inflight: int32atomic, state: int32atomic });}
/** * Everything a producer and consumer need to agree on for one stream * endpoint: the MessagePort, the per-endpoint backpressure `flags`, an * optional `readySignal` used by multi-consumer channels, and the * `highWater` cap. One object captures it all. * * Cross postMessage via {@link serializeStreamHandle} + {@link deserializeStreamHandle}. */export interface StreamHandle { port: MessagePort; flags: PipeFlags; readySignal?: Int32Atomic; highWater: number;}
/** Transport form — safe to drop into a dispatch message; SABs pass by reference. */export interface SerializedStreamHandle { __stream__: true; port: MessagePort; flags: unknown; readySignal?: unknown; highWater: number;}
export function serializeStreamHandle(h: StreamHandle): SerializedStreamHandle { return { __stream__: true, port: h.port, flags: serializeArg(h.flags), readySignal: h.readySignal ? serializeArg(h.readySignal) : undefined, highWater: h.highWater, };}
export function deserializeStreamHandle(obj: SerializedStreamHandle): StreamHandle { return { port: obj.port, flags: deserializeArg(obj.flags) as PipeFlags, readySignal: obj.readySignal ? (deserializeArg(obj.readySignal) as Int32Atomic) : undefined, highWater: obj.highWater, };}
export function isSerializedStreamHandle(arg: unknown): arg is SerializedStreamHandle { return typeof arg === 'object' && arg !== null && (arg as any).__stream__ === true;}
export interface PipeOptions { /** * When true, call extractTransferables on each value to pull out any * `transfer(buf)` markers into the transfer list. Set on the parent side * where user code may wrap values; off on the worker side. */ extractTransfers?: boolean;}
/** * Atomics-based pipe. Pumps values from `source` onto `handle.port`, * parking on `flags.inflight` when it reaches `handle.highWater`. */export async function pipeIterable<T>( source: AsyncIterable<T>, handle: StreamHandle, opts: PipeOptions = {},): Promise<void> { const extract = opts.extractTransfers ?? false; const { port, flags, highWater } = handle; const inflight = flags.fields.inflight; const state = flags.fields.state;
try { for await (const value of source) { if (state.load() === CANCEL) return;
let serialized: unknown; const transferList: Transferable[] = []; if (extract) { const extracted = extractTransferables([value]); serialized = serializeArg(extracted.args[0]); transferList.push(...extracted.transfer); collectTransferables(extracted.args[0], transferList); } else { serialized = serializeArg(value); collectTransferables(value, transferList); } port.postMessage({ value: serialized, done: false }, transferList);
const newInflight = inflight.add(1) + 1; if (newInflight >= highWater) { while (true) { if (state.load() === CANCEL) return; const current = inflight.load(); if (current < highWater) break; await inflight.waitAsync(current); } } } if (state.load() !== CANCEL) port.postMessage({ done: true }); } catch (err) { if (state.load() !== CANCEL) { port.postMessage({ done: true, error: err instanceof Error ? err : new Error(String(err)) }); } } try { port.close(); } catch {}}