Rewrite internal Web Streams to use less memory

src/js/builtins/ProcessObjectInternals.ts

@@ -104,22 +104,7 @@ export function getStdinStream(fd, isTTY: boolean, fdType: BunProcessStdinFdType
 
         // Releasing the lock is not possible as there are active reads
         // we will instead pretend we are unref'd, and release the lock once the reads are finished.
-        shouldUnref = true;
-
-        // unref the native part of the stream
-        try {
-          $getByIdDirectPrivate(
-            $getByIdDirectPrivate(native, "readableStreamController"),
-            "underlyingByteSource",
-          ).$resume(false);
-        } catch (e) {
-          if (IS_BUN_DEVELOPMENT) {
-            // we assume this isn't possible, but because we aren't sure
-            // we will ignore if error during release, but make a big deal in debug
-            console.error(e);
-            $assert(!"reachable");
-          }
-        }
+        source?.updateRef?.(false);
       }
     } else if (source) {
       source.updateRef(false);
@@ -183,25 +168,11 @@ export function getStdinStream(fd, isTTY: boolean, fdType: BunProcessStdinFdType
   async function internalRead(stream) {
     $debug("internalRead();");
     try {
-      var done: boolean, value: Uint8Array[];
       $assert(reader);
-      const pendingRead = reader.readMany();
+      const { done, value } = await reader.read();
 
-      if ($isPromise(pendingRead)) {
-        ({ done, value } = await pendingRead);
-      } else {
-        $debug("readMany() did not return a promise");
-        ({ done, value } = pendingRead);
-      }
-
-      if (!done) {
-        stream.push(value[0]);
-
-        // shouldn't actually happen, but just in case
-        const length = value.length;
-        for (let i = 1; i < length; i++) {
-          stream.push(value[i]);
-        }
+      if (value) {
+        stream.push(value);
 
         if (shouldUnref) unref();
       } else {

src/js/builtins/ReadableStreamDefaultReader.ts

@@ -51,19 +51,23 @@ export function readMany(this: ReadableStreamDefaultReader): ReadableStreamDefau
 
   const state = $getByIdDirectPrivate(stream, "state");
   stream.$disturbed = true;
-  if (state === $streamClosed) return { value: [], size: 0, done: true };
-  else if (state === $streamErrored) {
+  if (state === $streamErrored) {
     throw $getByIdDirectPrivate(stream, "storedError");
   }
 
   var controller = $getByIdDirectPrivate(stream, "readableStreamController");
-  var queue = $getByIdDirectPrivate(controller, "queue");
-  if (!queue) {
+  if (controller) {
+    var queue = $getByIdDirectPrivate(controller, "queue");
+  }
+
+  if (!queue && state !== $streamClosed) {
     // This is a ReadableStream direct controller implemented in JS
     // It hasn't been started yet.
     return controller.$pull(controller).$then(function ({ done, value }) {
-      return done ? { done: true, value: [], size: 0 } : { value: [value], size: 1, done: false };
+      return done ? { done: true, value: value ? [value] : [], size: 0 } : { value: [value], size: 1, done: false };
     });
+  } else if (!queue) {
+    return { done: true, value: [], size: 0 };
   }
 
   const content = queue.content;
@@ -98,27 +102,31 @@ export function readMany(this: ReadableStreamDefaultReader): ReadableStreamDefau
         $putByValDirect(outValues, i, values[i].value);
       }
     }
-    $resetQueue($getByIdDirectPrivate(controller, "queue"));
 
-    if ($getByIdDirectPrivate(controller, "closeRequested")) {
-      $readableStreamCloseIfPossible($getByIdDirectPrivate(controller, "controlledReadableStream"));
-    } else if ($isReadableStreamDefaultController(controller)) {
-      $readableStreamDefaultControllerCallPullIfNeeded(controller);
-    } else if ($isReadableByteStreamController(controller)) {
-      $readableByteStreamControllerCallPullIfNeeded(controller);
+    if (state !== $streamClosed) {
+      if ($getByIdDirectPrivate(controller, "closeRequested")) {
+        $readableStreamCloseIfPossible($getByIdDirectPrivate(controller, "controlledReadableStream"));
+      } else if ($isReadableStreamDefaultController(controller)) {
+        $readableStreamDefaultControllerCallPullIfNeeded(controller);
+      } else if ($isReadableByteStreamController(controller)) {
+        $readableByteStreamControllerCallPullIfNeeded(controller);
+      }
     }
+    $resetQueue($getByIdDirectPrivate(controller, "queue"));
 
     return { value: outValues, size, done: false };
   }
 
   var onPullMany = result => {
+    const resultValue = result.value;
+
     if (result.done) {
-      return { value: [], size: 0, done: true };
+      return { value: resultValue ? [resultValue] : [], size: 0, done: true };
     }
     var controller = $getByIdDirectPrivate(stream, "readableStreamController");
 
     var queue = $getByIdDirectPrivate(controller, "queue");
-    var value = [result.value].concat(queue.content.toArray(false));
+    var value = [resultValue].concat(queue.content.toArray(false));
     var length = value.length;
 
     if ($isReadableByteStreamController(controller)) {
@@ -136,8 +144,6 @@ export function readMany(this: ReadableStreamDefaultReader): ReadableStreamDefau
     }
 
     var size = queue.size;
-    $resetQueue(queue);
-
     if ($getByIdDirectPrivate(controller, "closeRequested")) {
       $readableStreamCloseIfPossible($getByIdDirectPrivate(controller, "controlledReadableStream"));
     } else if ($isReadableStreamDefaultController(controller)) {
@@ -146,12 +152,18 @@ export function readMany(this: ReadableStreamDefaultReader): ReadableStreamDefau
       $readableByteStreamControllerCallPullIfNeeded(controller);
     }
 
+    $resetQueue($getByIdDirectPrivate(controller, "queue"));
+
     return { value: value, size: size, done: false };
   };
 
+  if (state === $streamClosed) {
+    return { value: [], size: 0, done: true };
+  }
+
   var pullResult = controller.$pull(controller);
   if (pullResult && $isPromise(pullResult)) {
-    return pullResult.$then(onPullMany) as any;
+    return pullResult.then(onPullMany) as any;
   }
 
   return onPullMany(pullResult);