From 4d06c46ecea5dce5043ba31935c3eab7b8a5fd20 Mon Sep 17 00:00:00 2001 From: Chris King Date: Tue, 4 Aug 2026 22:28:13 -1000 Subject: [PATCH] net: add parent-owned pipe endpoints Signed-off-by: Chris King --- doc/api/child_process.md | 11 +- doc/api/index.md | 1 + doc/api/pipe.md | 111 ++++++++ lib/internal/child_process.js | 87 +++++++ lib/internal/pipe.js | 10 + lib/pipe.js | 41 +++ src/env_properties.h | 1 + src/pipe_wrap.cc | 46 ++++ src/pipe_wrap.h | 1 + src/process_wrap.cc | 3 +- test/common/pipe.js | 60 +++++ .../test-child-process-leased-pipe-bash.js | 213 ++++++++++++++++ .../test-child-process-leased-pipe-errors.js | 237 ++++++++++++++++++ .../test-child-process-leased-pipe-unit.js | 177 +++++++++++++ .../test-child-process-leased-pipe.js | 76 ++++++ .../test-child-process-validate-stdio.js | 40 +++ test/parallel/test-pipe-create-pipe-docs.js | 77 ++++++ .../parallel/test-pipe-create-pipe-finally.js | 45 ++++ test/parallel/test-pipe-create-pipe.js | 23 ++ 19 files changed, 1258 insertions(+), 2 deletions(-) create mode 100644 doc/api/pipe.md create mode 100644 lib/internal/pipe.js create mode 100644 lib/pipe.js create mode 100644 test/common/pipe.js create mode 100644 test/parallel/test-child-process-leased-pipe-bash.js create mode 100644 test/parallel/test-child-process-leased-pipe-errors.js create mode 100644 test/parallel/test-child-process-leased-pipe-unit.js create mode 100644 test/parallel/test-child-process-leased-pipe.js create mode 100644 test/parallel/test-pipe-create-pipe-docs.js create mode 100644 test/parallel/test-pipe-create-pipe-finally.js create mode 100644 test/parallel/test-pipe-create-pipe.js diff --git a/doc/api/child_process.md b/doc/api/child_process.md index e90759b16d3f..6ffb3135da41 100644 --- a/doc/api/child_process.md +++ b/doc/api/child_process.md @@ -1055,7 +1055,9 @@ pipes between the parent and child. The value is one of the following: file descriptor is duplicated in the child process to the fd that corresponds to the index in the `stdio` array. The stream must have an underlying descriptor (file streams do not start until the `'open'` event has - occurred). + occurred). Pipe endpoints returned by [`pipe.createPipe()`][] may be passed + here. A readable pipe endpoint returned by [`pipe.createPipe()`][] must not + be flowing when it is passed here. **NOTE:** While it is technically possible to pass `stdin` as a writable or `stdout`/`stderr` as readable, it is not recommended. Readable and writable streams are designed with distinct behaviors, and using @@ -1441,6 +1443,12 @@ streams of a child process have been closed. This is distinct from the [`'exit'`][] event, since multiple processes might share the same stdio streams. The `'close'` event will always emit after [`'exit'`][] was already emitted, or [`'error'`][] if the child process failed to spawn. +Readable stdio streams created by Node.js are resumed after the child process +exits so they can be fully consumed and closed before the `'close'` event is +emitted. Endpoints created by [`pipe.createPipe()`][] are an exception to this +rule and are not resumed by the child process. Their stream lifecycle remains +owned by the parent process, and consequently the child process `'close'` event +does not wait for such streams to close. If the process exited, `code` is the final exit code of the process, otherwise `null`. If the process terminated due to receipt of a signal, `signal` is the @@ -2374,6 +2382,7 @@ or [`child_process.fork()`][]. [`maxBuffer` and Unicode]: #maxbuffer-and-unicode [`net.Server`]: net.md#class-netserver [`net.Socket`]: net.md#class-netsocket +[`pipe.createPipe()`]: pipe.md#pipecreatepipe [`options.detached`]: #optionsdetached [`process.disconnect()`]: process.md#processdisconnect [`process.env`]: process.md#processenv diff --git a/doc/api/index.md b/doc/api/index.md index 24c38d0f3a70..fa9834f73270 100644 --- a/doc/api/index.md +++ b/doc/api/index.md @@ -44,6 +44,7 @@ * [Net](net.md) * [OS](os.md) * [Path](path.md) +* [Pipe](pipe.md) * [Performance hooks](perf_hooks.md) * [Permissions](permissions.md) * [Process](process.md) diff --git a/doc/api/pipe.md b/doc/api/pipe.md new file mode 100644 index 000000000000..7bfbf8d6cfa0 --- /dev/null +++ b/doc/api/pipe.md @@ -0,0 +1,111 @@ +# Pipe + + + +> Stability: 1.1 - Active development + + + +The `node:pipe` module provides APIs for creating operating system pipes. + +It can be accessed using: + +```mjs +import pipe from 'node:pipe'; +``` + +```cjs +const pipe = require('node:pipe'); +``` + +## `pipe.createPipe()` + + + +* Returns: {Object} + * `readable` {net.Socket} The readable end of the pipe. + * `writable` {net.Socket} The writable end of the pipe. + +The `pipe.createPipe()` method creates an operating system pipe pair. The +returned `readable` and `writable` streams are owned by the current process and +may be passed to [`child_process.spawn()`][] using the [`stdio`][] option. + +When a `readable` endpoint is passed as child stdin or as another child fd, the +child leases a readable handle. When a `writable` endpoint is passed as child +stdout, stderr, or another child fd, the child leases a writable handle. A +`readable` endpoint may not be passed as child stdout or stderr, and a +`writable` endpoint may not be passed as child stdin. An endpoint may be leased +to only one child process at a time. After the child process exits, endpoints +created by [`pipe.createPipe()`][] are released from their lease and may be +passed to another [`child_process.spawn()`][] call. +Endpoints created by [`pipe.createPipe()`][] are not supported by synchronous +child process APIs such as [`child_process.spawnSync()`][]. + +A `readable` endpoint created by [`pipe.createPipe()`][] must not be flowing +when it is passed to [`child_process.spawn()`][]. The child process +[`'close'`][] event does not wait for such an endpoint to close and does not +resume it after the child process exits. + +The current process is responsible for the endpoint streams. Use normal stream +idioms such as `end()` to finish writing and stream consumption to drain a +readable endpoint. Use `resume()` when an unread readable endpoint should be +drained without observing its data, and use `destroy()` when an endpoint is no +longer needed without being naturally ended or drained. + +```cjs +const { spawn } = require('node:child_process'); +const { createPipe } = require('node:pipe'); +const { text } = require('node:stream/consumers'); + +const { readable, writable } = createPipe(); +const child = spawn(process.execPath, ['-e', ` + const fs = require('node:fs'); + const buffer = Buffer.alloc(1); + const count = fs.readSync(0, buffer, 0, 1, null); + fs.writeSync(1, buffer.subarray(0, count)); +`], { + stdio: [readable, 'pipe', 'inherit'], +}); + +const output = text(child.stdout); +writable.end('abc'); + +child.on('close', async () => { + console.log(await output); // Prints: a + console.log(await text(readable)); // Prints: bc +}); +``` + +```mjs +import { spawn } from 'node:child_process'; +import { createPipe } from 'node:pipe'; +import { text } from 'node:stream/consumers'; + +const { readable, writable } = createPipe(); +const child = spawn(process.execPath, ['-e', ` + const fs = require('node:fs'); + const buffer = Buffer.alloc(1); + const count = fs.readSync(0, buffer, 0, 1, null); + fs.writeSync(1, buffer.subarray(0, count)); +`], { + stdio: [readable, 'pipe', 'inherit'], +}); + +const output = text(child.stdout); +writable.end('abc'); + +child.on('close', async () => { + console.log(await output); // Prints: a + console.log(await text(readable)); // Prints: bc +}); +``` + +[`'close'`]: child_process.md#event-close +[`child_process.spawn()`]: child_process.md#child_processspawncommand-args-options +[`child_process.spawnSync()`]: child_process.md#child_processspawnsynccommand-args-options +[`net.Socket`]: net.md#class-netsocket +[`stdio`]: child_process.md#optionsstdio +[`writable.destroy()`]: stream.md#writabledestroyerror +[`writable.end()`]: stream.md#writableendchunk-encoding-callback diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js index a12b2954db81..9a8ae98d3362 100644 --- a/lib/internal/child_process.js +++ b/lib/internal/child_process.js @@ -2,6 +2,7 @@ const { ArrayIsArray, + ArrayPrototypeFilter, ArrayPrototypePush, ArrayPrototypeReduce, ArrayPrototypeSlice, @@ -21,6 +22,7 @@ const { ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE, ERR_INVALID_HANDLE_TYPE, + ERR_INVALID_STATE, ERR_INVALID_SYNC_FORK_INPUT, ERR_IPC_CHANNEL_CLOSED, ERR_IPC_DISCONNECTED, @@ -75,6 +77,15 @@ const { } = internalBinding('uv'); const { SocketListSend, SocketListReceive } = SocketList; +const { kReaderOfPair, kWriterOfPair } = require('internal/pipe'); +const kLeasedTo = Symbol('kLeasedTo'); +const kStreamLeaseInUseMessage = + 'Stream is already in use by a child process'; +const kReadableStreamLeaseFlowingMessage = + 'Readable pipe must not be flowing'; +const kSyncLeasedStdioMessage = + 'cannot be used with spawnSync() because parent-owned pipe streams are ' + + 'only supported by spawn()'; // Lazy loaded for startup performance and to allow monkey patching of // internalBinding('http_parser').HTTPParser. @@ -278,6 +289,8 @@ function ChildProcess() { this.stdin.destroy(); } + releaseStreamLeases(this, this._leasedStreams); + this._handle.close(); this._handle = null; @@ -351,6 +364,50 @@ function closePendingHandle(target) { target._pendingMessage = null; } +function releaseStreamLeases(target, entries) { + if (entries === undefined) return; + + for (let i = 0; i < entries.length; i++) { + if (entries[i].type !== 'leased') continue; + + const stream = entries[i].stream; + + assert(stream !== undefined); + + if (stream[kLeasedTo] === target) + stream[kLeasedTo] = undefined; + } + + target._leasedStreams = undefined; +} + + +function acquireStreamLeases(target, stdio) { + assert(stdio !== undefined); + + for (let i = 0; i < stdio.length; i++) { + if (stdio[i].type !== 'leased') continue; + + const stream = stdio[i].stream; + + assert(stream !== undefined); + + if (stream[kLeasedTo]) { + releaseStreamLeases(target, stdio); + throw new ERR_INVALID_STATE(kStreamLeaseInUseMessage); + } + + if (stream[kReaderOfPair] && stream.readableFlowing === true) { + releaseStreamLeases(target, stdio); + throw new ERR_INVALID_STATE(kReadableStreamLeaseFlowingMessage); + } + + stream[kLeasedTo] = target; + } + + return ArrayPrototypeFilter(stdio, (stream) => stream.type === 'leased'); +} + ChildProcess.prototype.spawn = function spawn(options) { let i = 0; @@ -405,6 +462,8 @@ ChildProcess.prototype.spawn = function spawn(options) { if (options.windowsVerbatimArguments) spawnFlags |= processConstants.kProcessFlagWindowsVerbatimArguments; + this._leasedStreams = acquireStreamLeases(this, stdio); + const err = this._handle.spawn( options.file, options.args, @@ -422,6 +481,8 @@ ChildProcess.prototype.spawn = function spawn(options) { err === UV_EMFILE || err === UV_ENFILE || err === UV_ENOENT) { + releaseStreamLeases(this, this._leasedStreams); + if (childProcessSpawn.hasSubscribers) { childProcessSpawn.error.publish({ process: this, @@ -446,6 +507,7 @@ ChildProcess.prototype.spawn = function spawn(options) { this._handle.close(); this._handle = null; + releaseStreamLeases(this, this._leasedStreams); if (childProcessSpawn.hasSubscribers) { childProcessSpawn.error.publish({ @@ -468,6 +530,7 @@ ChildProcess.prototype.spawn = function spawn(options) { for (i = 0; i < stdio.length; i++) { const stream = stdio[i]; if (stream.type === 'ignore') continue; + if (stream.type === 'leased') continue; if (stream.ipc) { this._closesNeeded++; @@ -1077,6 +1140,28 @@ function getValidStdio(stdio, sync) { type: 'fd', fd: typeof stdio === 'number' ? stdio : stdio.fd, }); + } else if (stdio[kReaderOfPair] || stdio[kWriterOfPair]) { + if (sync) { + cleanup(); + throw new ERR_INVALID_ARG_VALUE('stdio', stdio, + kSyncLeasedStdioMessage); + } + + if (stdio.readable && !stdio.writable && (i === 1 || i === 2)) { + cleanup(); + throw new ERR_INVALID_ARG_VALUE('stdio', stdio); + } + + if (stdio.writable && !stdio.readable && i === 0) { + cleanup(); + throw new ERR_INVALID_ARG_VALUE('stdio', stdio); + } + + ArrayPrototypePush(acc, { + type: 'leased', + handle: stdio._handle, + stream: stdio, + }); } else if (getHandleWrapType(stdio) || getHandleWrapType(stdio.handle) || getHandleWrapType(stdio._handle)) { const handle = getHandleWrapType(stdio) ? @@ -1152,6 +1237,8 @@ function spawnSync(options) { module.exports = { ChildProcess, kChannelHandle, + kReaderOfPair, + kWriterOfPair, setupChannel, getValidStdio, stdioStringToArray, diff --git a/lib/internal/pipe.js b/lib/internal/pipe.js new file mode 100644 index 000000000000..35a790092c2d --- /dev/null +++ b/lib/internal/pipe.js @@ -0,0 +1,10 @@ +'use strict'; + +const { + Symbol, +} = primordials; + +module.exports = { + kReaderOfPair: Symbol('kReaderOfPair'), + kWriterOfPair: Symbol('kWriterOfPair'), +}; diff --git a/lib/pipe.js b/lib/pipe.js new file mode 100644 index 000000000000..e0068d3ab94a --- /dev/null +++ b/lib/pipe.js @@ -0,0 +1,41 @@ +'use strict'; + +const { Socket } = require('net'); + +const { + kReaderOfPair, + kWriterOfPair, +} = require('internal/pipe'); + +const { + Pipe, + pairPipes, + constants: PipeConstants, +} = internalBinding('pipe_wrap'); + +function createPipe() { + const readHandle = new Pipe(PipeConstants.SOCKET); + const writeHandle = new Pipe(PipeConstants.SOCKET); + pairPipes(readHandle, writeHandle); + + const readable = new Socket({ + handle: readHandle, + pauseOnCreate: true, + readable: true, + writable: false, + }); + const writable = new Socket({ + handle: writeHandle, + readable: false, + writable: true, + }); + + readable[kReaderOfPair] = true; + writable[kWriterOfPair] = true; + + return { readable, writable }; +} + +module.exports = { + createPipe, +}; diff --git a/src/env_properties.h b/src/env_properties.h index e3179287dce4..d3d392818513 100644 --- a/src/env_properties.h +++ b/src/env_properties.h @@ -290,6 +290,7 @@ V(options_string, "options") \ V(original_string, "original") \ V(output_string, "output") \ + V(leased_string, "leased") \ V(overlapped_string, "overlapped") \ V(parse_error_string, "Parse Error") \ V(password_string, "password") \ diff --git a/src/pipe_wrap.cc b/src/pipe_wrap.cc index 48bd22e301d8..6faa3a8855f7 100644 --- a/src/pipe_wrap.cc +++ b/src/pipe_wrap.cc @@ -89,6 +89,7 @@ void PipeWrap::Initialize(Local target, SetConstructorFunction(context, target, "Pipe", t); env->set_pipe_constructor_template(t); + SetMethod(context, target, "pairPipes", Pair); // Create FunctionTemplate for PipeConnectWrap. auto cwt = AsyncWrap::MakeLazilyInitializedJSTemplate(env); @@ -106,6 +107,7 @@ void PipeWrap::Initialize(Local target, void PipeWrap::RegisterExternalReferences(ExternalReferenceRegistry* registry) { registry->Register(New); + registry->Register(Pair); registry->Register(Bind); registry->Register(Listen); registry->Register(Connect); @@ -149,6 +151,50 @@ void PipeWrap::New(const FunctionCallbackInfo& args) { new PipeWrap(env, args.This(), provider, ipc); } +void PipeWrap::Pair(const FunctionCallbackInfo& args) { + Environment* env = Environment::GetCurrent(args); + uv_fs_t req; + uv_file fds[2]; + PipeWrap* read_wrap; + PipeWrap* write_wrap; + int err; + + CHECK(args[0]->IsObject()); + CHECK(args[1]->IsObject()); + ASSIGN_OR_RETURN_UNWRAP(&read_wrap, args[0].As()); + ASSIGN_OR_RETURN_UNWRAP(&write_wrap, args[1].As()); + + err = uv_pipe(fds, UV_NONBLOCK_PIPE, UV_NONBLOCK_PIPE); + if (err) { + env->ThrowUVException(err, "uv_pipe"); + return; + } + + err = uv_pipe_open(&read_wrap->handle_, fds[0]); + if (err) + goto error_close_fds; + read_wrap->set_fd(fds[0]); + + err = uv_pipe_open(&write_wrap->handle_, fds[1]); + if (err) + goto error_close_read_wrap; + write_wrap->set_fd(fds[1]); + return; + +error_close_read_wrap: + read_wrap->Close(); + goto error_close_write_fd; + +error_close_fds: + uv_fs_close(nullptr, &req, fds[0], nullptr); + uv_fs_req_cleanup(&req); + +error_close_write_fd: + uv_fs_close(nullptr, &req, fds[1], nullptr); + uv_fs_req_cleanup(&req); + env->ThrowUVException(err, "uv_pipe_open"); +} + PipeWrap::PipeWrap(Environment* env, Local object, ProviderType provider, diff --git a/src/pipe_wrap.h b/src/pipe_wrap.h index c0722b63d853..ef46be47ce29 100644 --- a/src/pipe_wrap.h +++ b/src/pipe_wrap.h @@ -60,6 +60,7 @@ class PipeWrap : public ConnectionWrap { bool ipc); static void New(const v8::FunctionCallbackInfo& args); + static void Pair(const v8::FunctionCallbackInfo& args); static void Bind(const v8::FunctionCallbackInfo& args); static void Listen(const v8::FunctionCallbackInfo& args); static void Connect(const v8::FunctionCallbackInfo& args); diff --git a/src/process_wrap.cc b/src/process_wrap.cc index 21ccb2a9989b..cd16493a2253 100644 --- a/src/process_wrap.cc +++ b/src/process_wrap.cc @@ -170,7 +170,8 @@ class ProcessWrap : public HandleWrap { if (!StreamForWrap(env, stdio).To(&(*options_stdio)[i].data.stream)) { return Nothing(); } - } else if (type->StrictEquals(env->wrap_string())) { + } else if (type->StrictEquals(env->wrap_string()) || + type->StrictEquals(env->leased_string())) { (*options_stdio)[i].flags = UV_INHERIT_STREAM; if (!StreamForWrap(env, stdio).To(&(*options_stdio)[i].data.stream)) { return Nothing(); diff --git a/test/common/pipe.js b/test/common/pipe.js new file mode 100644 index 000000000000..01dd1a55fc3f --- /dev/null +++ b/test/common/pipe.js @@ -0,0 +1,60 @@ +'use strict'; + +const common = require('./'); +const { once } = require('node:events'); +const { createPipe } = require('node:pipe'); +const { test } = require('node:test'); + +const kPipeCloseTimeout = common.platformTimeout(10_000); + +function isClosed(stream) { + return stream.closed || stream.destroyed; +} + +function waitForClose(stream) { + if (isClosed(stream)) + return Promise.resolve(); + + return once(stream, 'close'); +} + +async function assertPipeCloses(readable, writable) { + const timeout = new Promise((_, reject) => { + const timer = setTimeout(() => { + reject(new Error('pipe endpoints did not close')); + }, kPipeCloseTimeout); + timer.unref(); + }); + + await Promise.race([ + Promise.all([ + waitForClose(readable), + waitForClose(writable), + ]), + timeout, + ]); +} + +async function withCreatePipe(fn) { + const pipe = createPipe(); + + try { + const result = await fn(pipe); + await assertPipeCloses(pipe.readable, pipe.writable); + return result; + } finally { + pipe.readable.destroy(); + pipe.writable.destroy(); + } +} + +function testCreatePipe(name, fn) { + test(name, (t) => withCreatePipe(({ readable, writable }) => { + return fn(readable, writable, t); + })); +} + +module.exports = { + testCreatePipe, + withCreatePipe, +}; diff --git a/test/parallel/test-child-process-leased-pipe-bash.js b/test/parallel/test-child-process-leased-pipe-bash.js new file mode 100644 index 000000000000..dd506ab30da9 --- /dev/null +++ b/test/parallel/test-child-process-leased-pipe-bash.js @@ -0,0 +1,213 @@ +'use strict'; +const { isWindows } = require('../common'); +const assert = require('node:assert'); +const { spawn } = require('node:child_process'); +const { once } = require('node:events'); +const { createPipe } = require('node:pipe'); +const { test } = require('node:test'); +const { text } = require('node:stream/consumers'); + +// This test sketches how bash idioms could be expressed from JavaScript with +// process-owned pipe endpoints. The helpers between this comment and the test +// are small stubs that make the example executable; the test body is the part +// intended to demonstrate the user-facing syntax. + +function waitForClose(child) { + return once(child, 'close').then(([code, signal]) => { + assert.strictEqual(code, 0); + assert.strictEqual(signal, null); + }); +} + +function commandFromTemplate(strings, values) { + const args = []; + for (let i = 0; i < strings.length; i++) { + const words = strings[i].trim().split(/\s+/).filter(Boolean); + Array.prototype.push.apply(args, words); + if (i < values.length) + args.push(String(values[i])); + } + + const cmd = args.shift(); + return { cmd, args }; +} + +const curlStub = ` + const lines = [ + 'event: log\\n', + 'data: alpha\\n', + '\\n', + 'event: log\\n', + 'data: beta\\n', + '\\n', + 'event: end\\n', + '\\n', + ]; + + function writeLine() { + const line = lines.shift(); + if (line == null) + return; + + process.stdout.write(line); + setTimeout(writeLine, 20); + } + + writeLine(); +`; + +function startCommand(command, { stdin = 'ignore', stdout = 'pipe' } = {}) { + const cmd = command.cmd === 'curl' ? process.execPath : command.cmd; + const args = command.cmd === 'curl' ? ['-e', curlStub] : command.args; + + return spawn(cmd, args, { + stdio: [stdin, stdout, 'inherit'], + }); +} + +function createSubprocess(child, stdout) { + const close = waitForClose(child); + const read = stdout == null ? null : text(stdout); + + return { + async close() { + await close; + }, + + async read() { + assert.notStrictEqual(read, null); + const result = await read; + await close; + return result; + }, + + async readLine() { + const result = await this.read(); + return result.split(/\r?\n/, 1)[0]; + }, + }; +} + +function runCommand(command, options = {}) { + const capture = options.stdout == null; + const child = startCommand(command, { + ...options, + stdout: capture ? 'pipe' : options.stdout, + }); + + return createSubprocess(child, capture ? child.stdout : null); +} + +function runPipeline(commands, options = {}) { + const capture = options.stdout == null; + const stdout = capture ? createPipe() : null; + let stdin = options.stdin; + const children = []; + const pipes = []; + + for (let i = 0; i < commands.length; i++) { + const last = i === commands.length - 1; + const pipe = last ? null : createPipe(); + const child = startCommand(commands[i], { + stdin, + stdout: last ? (capture ? stdout.writable : options.stdout) : + pipe.writable, + }); + + children.push(child); + if (pipe != null) + pipes.push(pipe); + stdin = pipe?.readable; + } + + for (const pipe of pipes) + pipe.writable.end(); + if (capture) + stdout.writable.end(); + + const close = Promise.all(children.map(waitForClose)); + const read = capture ? text(stdout.readable) : null; + + return { + async close() { + await close; + }, + + async read() { + assert.notStrictEqual(read, null); + const result = await read; + await close; + return result; + }, + + async readLine() { + const result = await this.read(); + return result.split(/\r?\n/, 1)[0]; + }, + }; +} + +function $(strings, ...values) { + if (Array.isArray(strings.raw)) { + return command(strings, values); + } + + const commands = [strings, ...values].map((command) => command.command); + return (options) => runPipeline(commands, options); +} + +function command(strings, values) { + const command = commandFromTemplate(strings, values); + const fn = (options) => runCommand(command, options); + fn.command = command; + return fn; +} + +$.writer = function writer(command) { + const pipe = createPipe(); + const child = startCommand(command.command, { + stdout: pipe.writable, + }); + + child.once('close', () => pipe.writable.end()); + pipe.close = () => Promise.all([ + waitForClose(child), + text(pipe.readable), + ]); + + return pipe.readable; +}; + +if (!isWindows) test('bashjs-style pipeline leases readable endpoint', + async () => { + const events = $.writer( + $`curl -N ${'https://example.com/events'}` + ); + const logs = []; + + while (true) { + // Read only the event type line, leaving the rest of the event in the + // stream for the selected handler. + const type = await $`sed -n ${'1{s/^event: //p;q}'}`({ + stdin: events, + }).readLine(); + + switch (type) { + case 'end': + return assert.deepStrictEqual(logs, ['alpha', 'beta']); + + case 'log': { + // Read through the blank line that terminates this event, select + // data lines, and strip the "data: " prefix. + const log = await $( + $`sed -n ${'/^$/q; /^data: /p'}`, + $`cut -d ${' '} ${'-f2-'}` + )({ + stdin: events, + }).readLine(); + logs.push(log); + break; + } + } + } + }); diff --git a/test/parallel/test-child-process-leased-pipe-errors.js b/test/parallel/test-child-process-leased-pipe-errors.js new file mode 100644 index 000000000000..57e5b80ee122 --- /dev/null +++ b/test/parallel/test-child-process-leased-pipe-errors.js @@ -0,0 +1,237 @@ +'use strict'; +const assert = require('node:assert'); +const { spawn, spawnSync } = require('node:child_process'); +const { once } = require('node:events'); +const { test } = require('node:test'); +const { text } = require('node:stream/consumers'); +const { + testCreatePipe, + withCreatePipe, +} = require('../common/pipe'); + +const holdOpen = ` + process.send('ready'); + process.on('message', (message) => { + if (message === 'close') process.exit(0); + }); +`; + +const holdStdinOpen = ` + process.stdin.resume(); + process.send('ready'); +`; + +function trySpawn(...args) { + try { + return { child: spawn(...args) }; + } catch (error) { + return { error }; + } +} + +testCreatePipe('spawn failure leaves parent-owned endpoints usable', + async (readable, writable) => { + const child = spawn('program-that-had-better-not-exist', [], { + stdio: [readable, 'ignore', 'ignore'], + }); + + const close = new Promise((resolve) => child.on('close', resolve)); + const [[err]] = await Promise.all([ + once(child, 'error'), + close, + ]); + assert.strictEqual(err.code, 'ENOENT'); + assert.strictEqual(readable.destroyed, false); + assert.strictEqual(writable.destroyed, false); + + const output = text(readable); + writable.end('abc'); + assert.strictEqual(await output, 'abc'); + }); + +testCreatePipe('spawnSync rejects parent-owned pipe streams', + (readable, writable) => { + assert.throws(() => { + spawnSync(process.execPath, ['-e', ''], { + stdio: [readable, 'ignore', 'ignore'], + }); + }, { + code: 'ERR_INVALID_ARG_VALUE', + message: /parent-owned pipe streams are only supported by spawn\(\)/, + }); + + readable.resume(); + writable.end(); + }); + +testCreatePipe('writable endpoint is rejected as child stdin', + (readable, writable) => { + assert.throws(() => { + spawn(process.execPath, ['-e', ''], { + stdio: [writable, 'ignore', 'ignore'], + }); + }, { + code: 'ERR_INVALID_ARG_VALUE', + }); + + readable.resume(); + writable.end(); + }); + +testCreatePipe('readable endpoint is rejected as child stdout', + (readable, writable) => { + assert.throws(() => { + spawn(process.execPath, ['-e', ''], { + stdio: ['ignore', readable, 'ignore'], + }); + }, { + code: 'ERR_INVALID_ARG_VALUE', + }); + + readable.resume(); + writable.end(); + }); + +testCreatePipe('readable endpoint is rejected as child stderr', + (readable, writable) => { + assert.throws(() => { + spawn(process.execPath, ['-e', ''], { + stdio: ['ignore', 'ignore', readable], + }); + }, { + code: 'ERR_INVALID_ARG_VALUE', + }); + + readable.resume(); + writable.end(); + }); + +testCreatePipe('readable cannot be leased by two children concurrently', + async (readable, writable) => { + const child = spawn(process.execPath, ['-e', holdStdinOpen], { + stdio: [readable, 'ignore', 'inherit', 'ipc'], + }); + await once(child, 'message'); + + const result = trySpawn(process.execPath, ['-e', holdStdinOpen], { + stdio: [readable, 'ignore', 'inherit', 'ipc'], + }); + + const childClose = once(child, 'close'); + const resultClose = result.child != null ? once(result.child, 'close') : + null; + if (result.child != null) + result.child.kill(); + + writable.end(); + await childClose; + if (resultClose != null) + await resultClose; + readable.resume(); + + assert.strictEqual(result.error?.code, 'ERR_INVALID_STATE'); + }); + +testCreatePipe('readable cannot be leased twice by the same child', + (readable, writable) => { + assert.throws(() => { + spawn(process.execPath, ['-e', ''], { + stdio: [readable, 'ignore', 'inherit', readable], + }); + }, { + code: 'ERR_INVALID_STATE', + }); + + readable.resume(); + writable.end(); + }); + +testCreatePipe('writable cannot be leased by two children concurrently', + async (readable, writable) => { + const child = spawn(process.execPath, ['-e', holdOpen], { + stdio: ['ignore', writable, 'inherit', 'ipc'], + }); + await once(child, 'message'); + + const result = trySpawn(process.execPath, ['-e', holdOpen], { + stdio: ['ignore', writable, 'inherit', 'ipc'], + }); + + const childClose = once(child, 'close'); + const resultClose = result.child != null ? once(result.child, 'close') : + null; + if (result.child != null) + result.child.kill(); + + child.send('close'); + await childClose; + if (resultClose != null) + await resultClose; + readable.resume(); + writable.end(); + + assert.strictEqual(result.error?.code, 'ERR_INVALID_STATE'); + }); + +testCreatePipe('writable cannot be leased twice by the same child', + (readable, writable) => { + assert.throws(() => { + spawn(process.execPath, ['-e', ''], { + stdio: ['ignore', writable, 'inherit', writable], + }); + }, { + code: 'ERR_INVALID_STATE', + }); + + readable.resume(); + writable.end(); + }); + +test('failed lease releases earlier streams from the same attempt', + async () => { + await withCreatePipe(async (busy) => { + await withCreatePipe(async (free) => { + const childA = spawn(process.execPath, ['-e', holdStdinOpen], { + stdio: [busy.readable, 'ignore', 'inherit', 'ipc'], + }); + await once(childA, 'message'); + + assert.throws(() => { + spawn(process.execPath, ['-e', holdStdinOpen], { + stdio: [free.readable, 'ignore', 'inherit', busy.readable], + }); + }, { + code: 'ERR_INVALID_STATE', + }); + + const childC = spawn(process.execPath, ['-e', holdStdinOpen], { + stdio: [free.readable, 'ignore', 'inherit', 'ipc'], + }); + await once(childC, 'message'); + + const childAClose = once(childA, 'close'); + const childCClose = once(childC, 'close'); + busy.writable.end(); + free.writable.end(); + await childAClose; + await childCClose; + busy.readable.resume(); + free.readable.resume(); + }); + }); + }); + +testCreatePipe('flowing readable cannot be leased', (readable, writable) => { + readable.resume(); + + assert.throws(() => { + spawn(process.execPath, ['-e', holdStdinOpen], { + stdio: [readable, 'ignore', 'inherit'], + }); + }, { + code: 'ERR_INVALID_STATE', + message: /Readable pipe must not be flowing/, + }); + + writable.end(); +}); diff --git a/test/parallel/test-child-process-leased-pipe-unit.js b/test/parallel/test-child-process-leased-pipe-unit.js new file mode 100644 index 000000000000..1ecf91147ca8 --- /dev/null +++ b/test/parallel/test-child-process-leased-pipe-unit.js @@ -0,0 +1,177 @@ +'use strict'; +const assert = require('node:assert'); +const { spawn } = require('node:child_process'); +const { once } = require('node:events'); +const { createPipe } = require('node:pipe'); +const { test } = require('node:test'); +const { text } = require('node:stream/consumers'); + +const pipeStdinToStdout = ` + process.stdin.pipe(process.stdout); +`; + +const pipeFd3ToStdout = ` + const fs = require('node:fs'); + fs.createReadStream(null, { fd: 3 }).pipe(process.stdout); +`; + +const writeStdout = ` + process.stdout.write('hello stdout\\n'); +`; + +const writeStderr = ` + process.stderr.write('hello stderr\\n'); +`; + +const writeFd3 = ` + const fs = require('node:fs'); + fs.writeSync(3, 'hello fd3\\n'); +`; + +async function waitForClose(child) { + const [code, signal] = await once(child, 'close'); + assert.strictEqual(code, 0); + assert.strictEqual(signal, null); +} + +test('parent-owned streams do not appear as stdin', () => { + const { readable, writable } = createPipe(); + + const child = spawn(process.execPath, ['-e', pipeStdinToStdout], { + stdio: [readable, 'pipe', 'inherit'], + }); + assert.strictEqual(child.stdin, null); + + child.kill(); + readable.destroy(); + writable.destroy(); +}); + +test('parent-owned streams do not appear as stdout', () => { + const { readable, writable } = createPipe(); + + const child = spawn(process.execPath, ['-e', pipeStdinToStdout], { + stdio: ['pipe', writable, 'inherit'], + }); + assert.strictEqual(child.stdout, null); + + child.kill(); + readable.destroy(); + writable.destroy(); +}); + +test('parent-owned streams do not appear as stderr', () => { + const { readable, writable } = createPipe(); + + const child = spawn(process.execPath, ['-e', pipeStdinToStdout], { + stdio: ['pipe', 'pipe', writable], + }); + assert.strictEqual(child.stderr, null); + + child.kill(); + readable.destroy(); + writable.destroy(); +}); + +test('child stdin leases readable end and pipes to stdout', async () => { + const { readable, writable } = createPipe(); + + const child = spawn(process.execPath, ['-e', pipeStdinToStdout], { + stdio: [readable, 'pipe', 'inherit'], + }); + const output = text(child.stdout); + + writable.end('abc'); + await waitForClose(child); + assert.strictEqual(await output, 'abc'); +}); + +test('fd 3 leases readable end and pipes to stdout', async () => { + const { readable, writable } = createPipe(); + + const child = spawn(process.execPath, ['-e', pipeFd3ToStdout], { + stdio: ['ignore', 'pipe', 'inherit', readable], + }); + const output = text(child.stdout); + + writable.end('abc'); + await waitForClose(child); + assert.strictEqual(await output, 'abc'); +}); + +test('child stdout leases writable end and parent reads output', async () => { + const { readable, writable } = createPipe(); + const child = spawn(process.execPath, ['-e', writeStdout], { + stdio: ['ignore', writable, 'inherit'], + }); + + writable.destroy(); + await waitForClose(child); + + assert.strictEqual(await text(readable), 'hello stdout\n'); +}); + +test('child stderr leases writable end and parent reads output', async () => { + const { readable, writable } = createPipe(); + const child = spawn(process.execPath, ['-e', writeStderr], { + stdio: ['ignore', 'ignore', writable], + }); + + writable.destroy(); + + assert.strictEqual(await text(readable), 'hello stderr\n'); + await waitForClose(child); +}); + +test('fd 3 leases writable end and parent reads output', async () => { + const { readable, writable } = createPipe(); + const child = spawn(process.execPath, ['-e', writeFd3], { + stdio: ['ignore', 'ignore', 'inherit', writable], + }); + + writable.destroy(); + + assert.strictEqual(await text(readable), 'hello fd3\n'); + await waitForClose(child); +}); + +test('parent-owned readable does not flow after child exits', + async () => { + const { readable, writable } = createPipe(); + let flowed = ''; + readable.on('data', (chunk) => { + flowed += chunk; + }); + readable.pause(); + + const child = spawn(process.execPath, ['-e', writeStdout], { + stdio: ['ignore', writable, 'inherit'], + }); + + writable.destroy(); + await waitForClose(child); + + assert.strictEqual(flowed, ''); + assert.strictEqual(readable.readableFlowing, false); + assert.strictEqual(await text(readable), 'hello stdout\n'); + }); + +test('parent-owned pipes must be explicitly closed in a pipeline', async () => { + const { readable, writable } = createPipe(); + const producer = spawn(process.execPath, ['-e', "process.stdout.write('a')"], { + stdio: ['ignore', writable, 'inherit'], + }); + const consumer = spawn(process.execPath, ['-e', pipeStdinToStdout], { + stdio: [readable, 'pipe', 'inherit'], + }); + const output = text(consumer.stdout); + + await waitForClose(producer); + + // The parent owns the pipe endpoints. Close the writable endpoint so the + // consumer can observe EOF and close. This is a degenerate case. A 'pipe' + // should be used instead of a parent-owned pipe to form a pipeline. + writable.destroy(); + await waitForClose(consumer); + assert.strictEqual(await output, 'a'); +}); diff --git a/test/parallel/test-child-process-leased-pipe.js b/test/parallel/test-child-process-leased-pipe.js new file mode 100644 index 000000000000..5a82f43e96d4 --- /dev/null +++ b/test/parallel/test-child-process-leased-pipe.js @@ -0,0 +1,76 @@ +'use strict'; +const assert = require('node:assert'); +const { spawn } = require('node:child_process'); +const { once } = require('node:events'); +const { createPipe } = require('node:pipe'); +const { test } = require('node:test'); +const { text } = require('node:stream/consumers'); + +const readOneByteFromStdin = ` + const fs = require('node:fs'); + const buffer = Buffer.alloc(1); + const count = fs.readSync(0, buffer, 0, 1, null); + fs.writeSync(1, buffer.subarray(0, count)); +`; + +const writeArgToStdout = ` + process.stdout.write(process.argv[1]); +`; + +async function waitForClose(child) { + const [code, signal] = await once(child, 'close'); + assert.strictEqual(code, 0); + assert.strictEqual(signal, null); +} + +test('parent can lease readable to children sequentially then reclaim', + async () => { + const { readable, writable } = createPipe(); + + writable.write('abc'); + + // child A consumes 'a' + const childA = spawn(process.execPath, ['-e', readOneByteFromStdin], { + stdio: [readable, 'pipe', 'inherit'], + }); + const outputA = text(childA.stdout); + + await waitForClose(childA); + assert.strictEqual(await outputA, 'a'); + + // child B consumes 'b' + const childB = spawn(process.execPath, ['-e', readOneByteFromStdin], { + stdio: [readable, 'pipe', 'inherit'], + }); + const outputB = text(childB.stdout); + + await waitForClose(childB); + assert.strictEqual(await outputB, 'b'); + + // parent consumes 'c' + writable.end(); + assert.strictEqual(await text(readable), 'c'); + }); + +test('parent can lease writable to children sequentially then write', + async () => { + const { readable, writable } = createPipe(); + + // child A writes 'a' + const childA = spawn(process.execPath, ['-e', writeArgToStdout, 'a'], { + stdio: ['ignore', writable, 'inherit'], + }); + await waitForClose(childA); + + // child B writes 'b' + const childB = spawn(process.execPath, ['-e', writeArgToStdout, 'b'], { + stdio: ['ignore', writable, 'inherit'], + }); + await waitForClose(childB); + + // parent writes 'c' + const output = text(readable); + writable.end('c'); + assert.strictEqual(await output, 'abc'); + }); + diff --git a/test/parallel/test-child-process-validate-stdio.js b/test/parallel/test-child-process-validate-stdio.js index 5ba6f0fd123c..042aaecfe403 100644 --- a/test/parallel/test-child-process-validate-stdio.js +++ b/test/parallel/test-child-process-validate-stdio.js @@ -3,6 +3,8 @@ const common = require('../common'); const assert = require('assert'); +const { createPipe } = require('node:pipe'); +const { PassThrough } = require('stream'); const getValidStdio = require('internal/child_process').getValidStdio; const expectedError = { code: 'ERR_INVALID_ARG_VALUE', name: 'TypeError' }; @@ -43,6 +45,44 @@ assert.throws(() => getValidStdio(stdio2, true), assert.throws(() => getValidStdio(stdio), expectedError); } +// Pure JavaScript streams do not have an OS handle that can be passed to a +// child process as stdio. +{ + const stdio = [new PassThrough()]; + assert.throws(() => getValidStdio(stdio), expectedError); +} + +// Parent-owned pipe streams are normalized to their own stdio type because +// they lend only the OS handle to the child. +{ + const { readable, writable } = createPipe(); + const result = getValidStdio([readable, writable, 'ignore']); + + assert.strictEqual(result.stdio[0].type, 'leased'); + assert.strictEqual(result.stdio[0].stream, readable); + assert.strictEqual(result.stdio[0].handle, readable._handle); + assert.strictEqual(result.stdio[1].type, 'leased'); + assert.strictEqual(result.stdio[1].stream, writable); + assert.strictEqual(result.stdio[1].handle, writable._handle); + + readable.destroy(); + writable.destroy(); +} + +// Parent-owned pipe streams cannot be used with spawnSync() because the sync +// process runner only owns stdio pipes it creates internally. +{ + const { readable, writable } = createPipe(); + + assert.throws(() => getValidStdio([readable, 'ignore', 'ignore'], true), { + code: 'ERR_INVALID_ARG_VALUE', + message: /parent-owned pipe streams are only supported by spawn\(\)/, + }); + + readable.destroy(); + writable.destroy(); +} + const { isMainThread } = require('worker_threads'); if (isMainThread) { diff --git a/test/parallel/test-pipe-create-pipe-docs.js b/test/parallel/test-pipe-create-pipe-docs.js new file mode 100644 index 000000000000..07536f7ac27f --- /dev/null +++ b/test/parallel/test-pipe-create-pipe-docs.js @@ -0,0 +1,77 @@ +'use strict'; +const assert = require('node:assert'); +const { spawn } = require('node:child_process'); +const { once } = require('node:events'); +const { test } = require('node:test'); +const { text } = require('node:stream/consumers'); + +// Keep these examples in sync with the pipe.createPipe() examples in +// doc/api/pipe.md. +const createPipeCjsExample = ` + const { spawn } = require('node:child_process'); + const { createPipe } = require('node:pipe'); + const { text } = require('node:stream/consumers'); + + const { readable, writable } = createPipe(); + const child = spawn(process.execPath, ['-e', \` + const fs = require('node:fs'); + const buffer = Buffer.alloc(1); + const count = fs.readSync(0, buffer, 0, 1, null); + fs.writeSync(1, buffer.subarray(0, count)); + \`], { + stdio: [readable, 'pipe', 'inherit'], + }); + + const output = text(child.stdout); + writable.end('abc'); + + child.on('close', async () => { + console.log(await output); + console.log(await text(readable)); + }); +`; + +const createPipeMjsExample = ` + import { spawn } from 'node:child_process'; + import { createPipe } from 'node:pipe'; + import { text } from 'node:stream/consumers'; + + const { readable, writable } = createPipe(); + const child = spawn(process.execPath, ['-e', \` + const fs = require('node:fs'); + const buffer = Buffer.alloc(1); + const count = fs.readSync(0, buffer, 0, 1, null); + fs.writeSync(1, buffer.subarray(0, count)); + \`], { + stdio: [readable, 'pipe', 'inherit'], + }); + + const output = text(child.stdout); + writable.end('abc'); + + child.on('close', async () => { + console.log(await output); + console.log(await text(readable)); + }); +`; + +async function waitForClose(child) { + const [code, signal] = await once(child, 'close'); + assert.strictEqual(code, 0); + assert.strictEqual(signal, null); +} + +async function runExample(args, code) { + const child = spawn(process.execPath, [...args, code], { + stdio: ['ignore', 'pipe', 'inherit'], + }); + const output = text(child.stdout); + + await waitForClose(child); + assert.strictEqual(await output, 'a\nbc\n'); +} + +test('pipe.createPipe documentation examples', async () => { + await runExample(['-e'], createPipeCjsExample); + await runExample(['--input-type=module', '-e'], createPipeMjsExample); +}); diff --git a/test/parallel/test-pipe-create-pipe-finally.js b/test/parallel/test-pipe-create-pipe-finally.js new file mode 100644 index 000000000000..2c194a4452b5 --- /dev/null +++ b/test/parallel/test-pipe-create-pipe-finally.js @@ -0,0 +1,45 @@ +'use strict'; +const assert = require('node:assert'); +const { once } = require('node:events'); +const { text } = require('node:stream/consumers'); +const { testCreatePipe } = require('../common/pipe'); + +async function raceFinishBeforeRead(readable, writable) { + const finish = once(writable, 'finish').then(() => 'finish'); + const tick = new Promise((resolve) => setImmediate(resolve, 'pending')); + writable.end('abc'); + const result = await Promise.race([finish, tick]); + + readable.resume(); + await text(readable); + await finish; + return result; +} + +if (process.platform !== 'win32') { + testCreatePipe('pipe writer can finish before pipe reader consumes', + async (readable, writable) => { + assert.strictEqual( + await raceFinishBeforeRead(readable, writable), + 'finish'); + }); +} + +if (process.platform === 'win32') { + testCreatePipe('pipe writer finish waits until pipe reader consumes', + async (readable, writable) => { + assert.strictEqual( + await raceFinishBeforeRead(readable, writable), + 'pending'); + }); +} + +// A possible hack to unify this behavior would be to bypass shutdown for +// parent-owned pipe writers would be to add the following to +// Socket.prototype._final: +// +// if (this[kWriterOfPair]) { +// debug('_final: pipe pair writable, close handle'); +// process.nextTick(() => this.destroy()); +// return cb(); +// } diff --git a/test/parallel/test-pipe-create-pipe.js b/test/parallel/test-pipe-create-pipe.js new file mode 100644 index 000000000000..847a8099bb93 --- /dev/null +++ b/test/parallel/test-pipe-create-pipe.js @@ -0,0 +1,23 @@ +'use strict'; +const assert = require('node:assert'); +const { testCreatePipe } = require('../common/pipe'); + +testCreatePipe('createPipe returns directional OS-backed streams', + (readable, writable) => { + assert.strictEqual(readable.readable, true); + assert.strictEqual(readable.writable, false); + assert.strictEqual(writable.readable, false); + assert.strictEqual(writable.writable, true); + + readable.resume(); + writable.end(); + }); + +testCreatePipe('readable starts paused so the parent does not pre-consume bytes', + (readable, writable) => { + assert.strictEqual(readable.readableFlowing, false); + + readable.resume(); + writable.end(); + }); +