net: move createPipe onto net

Author: kingces95

doc/api/child_process.md

@@ -1055,8 +1055,8 @@ 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). Pipe endpoints returned by [`pipe.createPipe()`][] may be passed
-   here. A readable pipe endpoint returned by [`pipe.createPipe()`][] must not
+   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.
@@ -1445,7 +1445,7 @@ 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
+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.
@@ -2382,7 +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
+[`net.createPipe()`]: net.md#netcreatepipe
 [`options.detached`]: #optionsdetached
 [`process.disconnect()`]: process.md#processdisconnect
 [`process.env`]: process.md#processenv

doc/api/index.md

@@ -44,7 +44,6 @@
 * [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/net.md

@@ -2138,6 +2138,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 +2348,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 +2362,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 +2395,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

doc/api/pipe.md

@@ -77,7 +77,7 @@ const {
 } = internalBinding('uv');
 
 const { SocketListSend, SocketListReceive } = SocketList;
-const { kReaderOfPair, kWriterOfPair } = require('internal/pipe');
+const { kReaderOfPair, kWriterOfPair } = require('internal/net');
 const kLeasedTo = Symbol('kLeasedTo');
 const kStreamLeaseInUseMessage =
   'Stream is already in use by a child process';

lib/internal/child_process.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,

lib/internal/net.js

@@ -47,12 +47,14 @@ let debug = require('internal/util/debuglog').debuglog('net', (fn) => {
   debug = fn;
 });
 const {
+  kReaderOfPair,
   kReinitializeHandle,
   kSetNoDelay,
   kSetKeepAlive,
   kSetKeepAliveInitialDelay,
   kSetKeepAliveInterval,
   kSetKeepAliveCount,
+  kWriterOfPair,
   isIP,
   isIPv4,
   isIPv6,
@@ -80,6 +82,7 @@ const {
 const {
   Pipe,
   PipeConnectWrap,
+  pairPipes,
   constants: PipeConstants,
 } = internalBinding('pipe_wrap');
 const {
@@ -211,6 +214,29 @@ function createHandle(fd, is_server) {
   throw new ERR_INVALID_FD_TYPE(type);
 }
 
+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 };
+}
+
 
 function getNewAsyncId(handle) {
   return (!handle || typeof handle.getAsyncId !== 'function') ?
@@ -2863,6 +2889,7 @@ module.exports = {
   BoundSocket,
   connect,
   createConnection: connect,
+  createPipe,
   createServer,
   isIP: isIP,
   isIPv4: isIPv4,