Make node streams faster (#1502)

* Make node streams faster

* Fix for macOS, improve performance, handle ref and unref

Co-authored-by: Jarred Sumner <709451+Jarred-Sumner@users.noreply.github.com>

3 files changed, 112 insertions, 113 deletions

diff --git a/test/bun.js/child_process-node.test.js b/test/bun.js/child_process-node.test.js
index 0f04bb2cd..908dc2a36 100644
--- a/test/bun.js/child_process-node.test.js
+++ b/test/bun.js/child_process-node.test.js
@@ -12,7 +12,7 @@ import { tmpdir } from "node:os";
 
 const debug = process.env.DEBUG ? console.log : () => {};
 
-const platformTmpDir = tmpdir();
+const platformTmpDir = require("fs").realpathSync(tmpdir());
 
 // Copyright Joyent, Inc. and other Node contributors.
 //
@@ -135,7 +135,7 @@ describe("ChildProcess.spawn", () => {
     child.spawn({
       file: "node",
       // file: process.execPath,
-      args: ["--interactive"],
+      args: ["node", "--interactive"],
       cwd: process.cwd(),
       stdio: ["ignore", "ignore", "ignore", "ipc"],
     });
@@ -179,8 +179,6 @@ describe("ChildProcess spawn bad stdio", () => {
 
       this.stdout.destroy();
       this.stderr.destroy();
-      this.stdout = null;
-      this.stderr = null;
 
       return err;
     };
@@ -256,7 +254,6 @@ describe("child_process cwd", () => {
     let data = "";
     child.stdout.on("data", (chunk) => {
       data += chunk;
-      console.trace("here");
     });
 
     // TODO: Test exit events
@@ -367,6 +364,8 @@ describe("child_process cwd", () => {
 
 describe("child_process default options", () => {
   it("should use process.env as default env", (done) => {
+    globalThis.process.env.TMPDIR = platformTmpDir;
+
     let child = spawn("printenv", [], {});
     let response = "";
 
@@ -389,106 +388,106 @@ describe("child_process default options", () => {
   });
 });
 
-// describe("child_process double pipe", () => {
-//   it("should allow two pipes to be used at once", (done) => {
-//     const { mustCallAtLeast, mustCall } = createCallCheckCtx(done);
-//     let grep, sed, echo;
-//     grep = spawn("grep", ["o"]);
-//     sed = spawn("sed", ["s/o/O/"]);
-//     echo = spawn("echo", ["hello\nnode\nand\nworld\n"]);
-
-//     // pipe echo | grep
-//     echo.stdout.on(
-//       "data",
-//       mustCallAtLeast((data) => {
-//         debug(`grep stdin write ${data.length}`);
-//         if (!grep.stdin.write(data)) {
-//           echo.stdout.pause();
-//         }
-//       }),
-//     );
-
-//     // TODO(Derrick): We don't implement the full API for this yet,
-//     // So stdin has no 'drain' event.
-//     // // TODO(@jasnell): This does not appear to ever be
-//     // // emitted. It's not clear if it is necessary.
-//     // grep.stdin.on("drain", (data) => {
-//     //   echo.stdout.resume();
-//     // });
-
-//     // Propagate end from echo to grep
-//     echo.stdout.on(
-//       "end",
-//       mustCall(() => {
-//         debug("echo stdout end");
-//         grep.stdin.end();
-//       }),
-//     );
-
-//     echo.on(
-//       "exit",
-//       mustCall(() => {
-//         debug("echo exit");
-//       }),
-//     );
-
-//     grep.on(
-//       "exit",
-//       mustCall(() => {
-//         debug("grep exit");
-//       }),
-//     );
-
-//     sed.on(
-//       "exit",
-//       mustCall(() => {
-//         debug("sed exit");
-//       }),
-//     );
-
-//     // pipe grep | sed
-//     grep.stdout.on(
-//       "data",
-//       mustCallAtLeast((data) => {
-//         debug(`grep stdout ${data.length}`);
-//         if (!sed.stdin.write(data)) {
-//           grep.stdout.pause();
-//         }
-//       }),
-//     );
-
-//     // // TODO(@jasnell): This does not appear to ever be
-//     // // emitted. It's not clear if it is necessary.
-//     // sed.stdin.on("drain", (data) => {
-//     //   grep.stdout.resume();
-//     // });
-
-//     // Propagate end from grep to sed
-//     grep.stdout.on(
-//       "end",
-//       mustCall((code) => {
-//         debug("grep stdout end");
-//         sed.stdin.end();
-//       }),
-//     );
-
-//     let result = "";
-
-//     // print sed's output
-//     sed.stdout.on(
-//       "data",
-//       mustCallAtLeast((data) => {
-//         result += data.toString("utf8");
-//         debug(data);
-//       }),
-//     );
-
-//     sed.stdout.on(
-//       "end",
-//       mustCall(() => {
-//         debug("result: " + result);
-//         strictEqual(result, `hellO\nnOde\nwOrld\n`);
-//       }),
-//     );
-//   });
-// });
+describe("child_process double pipe", () => {
+  it("should allow two pipes to be used at once", (done) => {
+    const { mustCallAtLeast, mustCall } = createCallCheckCtx(done);
+    let grep, sed, echo;
+    grep = spawn("grep", ["o"]);
+    sed = spawn("sed", ["s/o/O/"]);
+    echo = spawn("echo", ["hello\nnode\nand\nworld\n"]);
+
+    // pipe echo | grep
+    echo.stdout.on(
+      "data",
+      mustCallAtLeast((data) => {
+        debug(`grep stdin write ${data.length}`);
+        if (!grep.stdin.write(data)) {
+          echo.stdout.pause();
+        }
+      }),
+    );
+
+    // TODO(Derrick): We don't implement the full API for this yet,
+    // So stdin has no 'drain' event.
+    // // TODO(@jasnell): This does not appear to ever be
+    // // emitted. It's not clear if it is necessary.
+    // grep.stdin.on("drain", (data) => {
+    //   echo.stdout.resume();
+    // });
+
+    // Propagate end from echo to grep
+    echo.stdout.on(
+      "end",
+      mustCall(() => {
+        debug("echo stdout end");
+        grep.stdin.end();
+      }),
+    );
+
+    echo.on(
+      "exit",
+      mustCall(() => {
+        debug("echo exit");
+      }),
+    );
+
+    grep.on(
+      "exit",
+      mustCall(() => {
+        debug("grep exit");
+      }),
+    );
+
+    sed.on(
+      "exit",
+      mustCall(() => {
+        debug("sed exit");
+      }),
+    );
+
+    // pipe grep | sed
+    grep.stdout.on(
+      "data",
+      mustCallAtLeast((data) => {
+        debug(`grep stdout ${data.length}`);
+        if (!sed.stdin.write(data)) {
+          grep.stdout.pause();
+        }
+      }),
+    );
+
+    // // TODO(@jasnell): This does not appear to ever be
+    // // emitted. It's not clear if it is necessary.
+    sed.stdin.on("drain", (data) => {
+      grep.stdout.resume();
+    });
+
+    // Propagate end from grep to sed
+    grep.stdout.on(
+      "end",
+      mustCall((code) => {
+        debug("grep stdout end");
+        sed.stdin.end();
+      }),
+    );
+
+    let result = "";
+
+    // print sed's output
+    sed.stdout.on(
+      "data",
+      mustCallAtLeast((data) => {
+        result += data.toString("utf8");
+        debug(data);
+      }),
+    );
+
+    sed.stdout.on(
+      "end",
+      mustCall(() => {
+        debug("result: " + result);
+        strictEqual(result, `hellO\nnOde\nwOrld\n`);
+      }),
+    );
+  });
+});
diff --git a/test/bun.js/child_process.test.ts b/test/bun.js/child_process.test.ts
index c862ff1b4..95a589401 100644
--- a/test/bun.js/child_process.test.ts
+++ b/test/bun.js/child_process.test.ts
@@ -13,7 +13,7 @@ import { tmpdir } from "node:os";
 
 const debug = process.env.DEBUG ? console.log : () => {};
 
-const platformTmpDir = tmpdir();
+const platformTmpDir = require("fs").realpathSync(tmpdir());
 
 // Semver regex: https://gist.github.com/jhorsman/62eeea161a13b80e39f5249281e17c39?permalink_comment_id=2896416#gistcomment-2896416
 // Not 100% accurate, but good enough for this test
@@ -82,12 +82,12 @@ describe("spawn()", () => {
     expect(SEMVER_REGEX.test(result.trim())).toBe(true);
   });
 
-  it("should allow stdout to be read via .read() API", async () => {
+  it("should allow stdout to be read via .read() API", async (done) => {
     const child = spawn("bun", ["-v"]);
     const result: string = await new Promise((resolve) => {
       let finalData = "";
       child.stdout.on("error", (e) => {
-        console.error(e);
+        done(e);
       });
       child.stdout.on("readable", () => {
         let data;
@@ -99,6 +99,7 @@ describe("spawn()", () => {
       });
     });
     expect(SEMVER_REGEX.test(result.trim())).toBe(true);
+    done();
   });
 
   it("should accept stdio option with 'ignore' for no stdio fds", async () => {
diff --git a/test/bun.js/spawn.test.ts b/test/bun.js/spawn.test.ts
index 5be0b1ddc..418178e2b 100644
--- a/test/bun.js/spawn.test.ts
+++ b/test/bun.js/spawn.test.ts
@@ -294,9 +294,8 @@ for (let [gcTick, label] of [
                 var output = "";
                 var reader = process.stdout!.getReader();
                 var done = false;
-                var value;
                 while (!done) {
-                  ({ value, done } = await reader.read());
+                  var { value, done } = await reader.read();
                   if (value) output += new TextDecoder().decode(value);
                 }