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 [`net.createPipe()`][] may be passed
+   here. A readable pipe endpoint returned by [`net.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 [`net.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
+[`net.createPipe()`]: net.md#netcreatepipe
 [`options.detached`]: #optionsdetached
 [`process.disconnect()`]: process.md#processdisconnect
 [`process.env`]: process.md#processenv

doc/api/net.md

@@ -8,9 +8,10 @@
 
 <!-- source_link=lib/net.js -->
 
-The `node:net` module provides an asynchronous network API for creating stream-based
-TCP or [IPC][] servers ([`net.createServer()`][]) and clients
-([`net.createConnection()`][]).
+The `node:net` module provides an asynchronous network API for creating
+stream-based TCP or [IPC][] servers ([`net.createServer()`][]) and clients
+([`net.createConnection()`][]), and operating system pipe pairs
+([`net.createPipe()`][]).
 
 It can be accessed using:
 
@@ -2138,6 +2139,90 @@ Use `nc` to connect to a Unix domain socket server:
 nc -U /tmp/echo.sock
 ```
 
+## `net.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 `net.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 [`net.createPipe()`][] are released from their lease and may be
+passed to another [`child_process.spawn()`][] call. Endpoints created by
+[`net.createPipe()`][] are not supported by synchronous child process APIs such
+as [`child_process.spawnSync()`][].
+
+A `readable` endpoint created by [`net.createPipe()`][] must not be flowing
+when it is passed to [`child_process.spawn()`][]. The child process
+[`'close'` event][child-process-close] 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:net');
+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:net';
+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
+});
+```
+
 ## `net.getDefaultAutoSelectFamily()`
 
 <!-- YAML
@@ -2264,6 +2349,8 @@ net.isIPv6('fhqwhgads'); // returns false
 [`ERR_SOCKET_HANDLE_ADOPTED`]: errors.md#err_socket_handle_adopted
 [`EventEmitter`]: events.md#class-eventemitter
 [`child_process.fork()`]: child_process.md#child_processforkmodulepath-args-options
+[`child_process.spawn()`]: child_process.md#child_processspawncommand-args-options
+[`child_process.spawnSync()`]: child_process.md#child_processspawnsynccommand-args-options
 [`dns.lookup()`]: dns.md#dnslookuphostname-options-callback
 [`dns.lookup()` hints]: dns.md#supported-getaddrinfo-flags
 [`net.Server`]: #class-netserver
@@ -2276,6 +2363,7 @@ net.isIPv6('fhqwhgads'); // returns false
 [`net.createConnection(options)`]: #netcreateconnectionoptions-connectlistener
 [`net.createConnection(path)`]: #netcreateconnectionpath-connectlistener
 [`net.createConnection(port, host)`]: #netcreateconnectionport-host-connectlistener
+[`net.createPipe()`]: #netcreatepipe
 [`net.createServer()`]: #netcreateserveroptions-connectionlistener
 [`net.getDefaultAutoSelectFamily()`]: #netgetdefaultautoselectfamily
 [`net.getDefaultAutoSelectFamilyAttemptTimeout()`]: #netgetdefaultautoselectfamilyattempttimeout
@@ -2308,13 +2396,15 @@ net.isIPv6('fhqwhgads'); // returns false
 [`socket.setTimeout()`]: #socketsettimeouttimeout-callback
 [`socket.setTimeout(timeout)`]: #socketsettimeouttimeout-callback
 [`stream.getDefaultHighWaterMark()`]: stream.md#streamgetdefaulthighwatermarkobjectmode
+[`stdio`]: child_process.md#optionsstdio
 [`worker_threads`]: worker_threads.md
 [`writable.destroy()`]: stream.md#writabledestroyerror
 [`writable.destroyed`]: stream.md#writabledestroyed
 [`writable.end()`]: stream.md#writableendchunk-encoding-callback
 [`writable.writableLength`]: stream.md#writablewritablelength
 [dot-decimal notation]: https://en.wikipedia.org/wiki/Dot-decimal_notation
 [half-closed]: https://tools.ietf.org/html/rfc1122
+[child-process-close]: child_process.md#event-close
 [stream_writable_write]: stream.md#writablewritechunk-encoding-callback
 [unspecified IPv4 address]: https://en.wikipedia.org/wiki/0.0.0.0
 [unspecified IPv6 address]: https://en.wikipedia.org/wiki/IPv6_address#Unspecified_address

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/net');
+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/net.js

@@ -99,12 +99,14 @@ function isLoopback(host) {
 }
 
 module.exports = {
+  kReaderOfPair: Symbol('kReaderOfPair'),
   kReinitializeHandle: Symbol('kReinitializeHandle'),
   kSetNoDelay: Symbol('kSetNoDelay'),
   kSetKeepAlive: Symbol('kSetKeepAlive'),
   kSetKeepAliveInitialDelay: Symbol('kSetKeepAliveInitialDelay'),
   kSetKeepAliveInterval: Symbol('kSetKeepAliveInterval'),
   kSetKeepAliveCount: Symbol('kSetKeepAliveCount'),
+  kWriterOfPair: Symbol('kWriterOfPair'),
   isIP,
   isIPv4,
   isIPv6,