net: add createPipe()

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

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)

doc/api/pipe.md

@@ -0,0 +1,111 @@
+# Pipe
+
+<!--introduced_in=REPLACEME-->
+
+> Stability: 1.1 - Active development
+
+<!-- source_link=lib/pipe.js -->
+
+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()`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* 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

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,

lib/internal/pipe.js

@@ -0,0 +1,10 @@
+'use strict';
+
+const {
+  Symbol,
+} = primordials;
+
+module.exports = {
+  kReaderOfPair: Symbol('kReaderOfPair'),
+  kWriterOfPair: Symbol('kWriterOfPair'),
+};

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,
+};

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")                                               \