Something went wrong. Try again.
Offload functions to worker threads with shared memory primitives for Node.js.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586import type { Server } from 'node:net';import { channel } from '../channel.ts';import { shared } from '../shared/shared.ts';import { int32atomic } from '../shared/descriptors.ts';import type { WorkerHandle } from '../runner.ts';import { PushChannel } from './push-channel.ts';import { leastConns } from './strategies.ts';import type { ListenArgs, ListenOptions, ServerThread, ServerThreads, ServerThreadsOptions } from './types.ts';
const DEFAULT_HIGH_WATER_MARK = 64;const DEFAULT_DRAIN_TIMEOUT = 30_000;
export function serverThreads( workers: readonly WorkerHandle[], server: Server, opts: ServerThreadsOptions = {},): ServerThreads { const balance = opts.balance ?? leastConns(); const hwm = opts.highWaterMark ?? DEFAULT_HIGH_WATER_MARK; const listenOpts: Required<ListenOptions> = { drainTimeout: opts.listen?.drainTimeout ?? DEFAULT_DRAIN_TIMEOUT, };
const n = workers.length; if (n === 0) throw new Error('serverThreads: workers array must be non-empty');
// Per-worker fd push channels (each is an AsyncIterable<number>). const pushChannels = Array.from({ length: n }, () => new PushChannel<number>()); // Wrap each PushChannel in moroutine's channel() so it is recognized by // moroutine's arg-preparation layer and piped cross-thread via MessagePort. const channels = pushChannels.map((pc) => channel<number>(pc, { highWaterMark: hwm })); // Shared atomic counter slots, one per worker. const counters = shared(Array.from({ length: n }, () => int32atomic));
const onConnection = (socket: any) => { const origFd: number = socket._handle?.fd; if (typeof origFd !== 'number') { // Windows or unexpected handle type — drop hard. socket.destroy(); return; } // Snapshot per-worker state, pick target. const threads: ServerThread[] = new Array(n); for (let i = 0; i < n; i++) threads[i] = { worker: workers[i], conns: counters.elements[i].load() }; const selected = balance.select(threads); const idx = workers.indexOf(selected.worker);
// Detach the socket from libuv without closing the fd. readStop() // deregisters the read watcher; no-oping close() prevents destroy() // from closing the underlying fd. The worker will open it fresh. const handle = socket._handle; handle.readStop(); handle.close = (cb: any) => { if (cb) cb(); }; socket._handle = null; socket.destroy();
counters.elements[idx].add(1); try { pushChannels[idx].send(origFd); } catch { counters.elements[idx].sub(1); } };
server.on('connection', onConnection);
let disposed = false; const dispose = () => { if (disposed) return; disposed = true; server.off('connection', onConnection); server.off('close', onClose); for (const pc of pushChannels) pc.close(); }; const onClose = () => dispose(); server.once('close', onClose);
// Build the indexable/iterable/disposable result. const entries = workers.map((w, i) => [w, [channels[i], counters.elements[i], listenOpts] as ListenArgs] as const); const result: readonly (readonly [WorkerHandle, ListenArgs])[] = entries; Object.defineProperty(result, Symbol.dispose, { value: dispose, enumerable: false }); return result as unknown as ServerThreads;}