perf(runtime/spawn): collect output using op_read_all (#16596)

**This patch**
```
benchmark      time (avg)             (min … max)       p75       p99      p995
------------------------------------------------- -----------------------------
echo deno   23.99 ms/iter   (22.51 ms … 33.61 ms)  23.97 ms  33.61 ms  33.61 ms
cat 16kb    24.27 ms/iter    (22.5 ms … 35.21 ms)   24.2 ms  35.21 ms  35.21 ms
cat 1mb     25.88 ms/iter   (25.04 ms … 30.28 ms)  26.12 ms  30.28 ms  30.28 ms
cat 15mb    38.41 ms/iter       (35.7 ms … 50 ms)  38.31 ms     50 ms     50 ms
```

**main**
```
benchmark      time (avg)             (min … max)       p75       p99      p995
------------------------------------------------- -----------------------------
echo deno   35.66 ms/iter   (34.53 ms … 41.84 ms)  35.79 ms  41.84 ms  41.84 ms
cat 16kb    35.99 ms/iter   (34.52 ms … 44.94 ms)  36.05 ms  44.94 ms  44.94 ms
cat 1mb     38.68 ms/iter   (36.67 ms … 50.44 ms)  37.95 ms  50.44 ms  50.44 ms
cat 15mb     48.4 ms/iter   (46.19 ms … 58.41 ms)  49.16 ms  58.41 ms  58.41 ms
```

cli/bench/spawn.js

@@ -0,0 +1,11 @@
+// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
+
+Deno.bench("echo deno", async () => {
+  await Deno.spawn("echo", { args: ["deno"] });
+});
+
+Deno.bench("cat 128kb", async () => {
+  await Deno.spawn("cat", {
+    args: ["./cli/bench/testdata/128k.bin"],
+  });
+});

ext/web/06_streams.js

@@ -730,6 +730,7 @@
     const stream = webidl.createBranded(ReadableStream);
     stream[promiseIdSymbol] = undefined;
     stream[_isUnref] = false;
+    stream[_resourceBackingUnrefable] = { rid, autoClose: true };
     const underlyingSource = {
       type: "bytes",
       async pull(controller) {
@@ -767,8 +768,14 @@
     return stream;
   }
 
+  function readableStreamIsUnrefable(stream) {
+    return _isUnref in stream;
+  }
+
   function readableStreamForRidUnrefableRef(stream) {
-    if (!(_isUnref in stream)) throw new TypeError("Not an unrefable stream");
+    if (!readableStreamIsUnrefable(stream)) {
+      throw new TypeError("Not an unrefable stream");
+    }
     stream[_isUnref] = false;
     if (stream[promiseIdSymbol] !== undefined) {
       core.refOp(stream[promiseIdSymbol]);
@@ -776,7 +783,9 @@
   }
 
   function readableStreamForRidUnrefableUnref(stream) {
-    if (!(_isUnref in stream)) throw new TypeError("Not an unrefable stream");
+    if (!readableStreamIsUnrefable(stream)) {
+      throw new TypeError("Not an unrefable stream");
+    }
     stream[_isUnref] = true;
     if (stream[promiseIdSymbol] !== undefined) {
       core.unrefOp(stream[promiseIdSymbol]);
@@ -787,15 +796,25 @@
     return stream[_resourceBacking];
   }
 
+  function getReadableStreamResourceBackingUnrefable(stream) {
+    return stream[_resourceBackingUnrefable];
+  }
+
   async function readableStreamCollectIntoUint8Array(stream) {
-    const resourceBacking = getReadableStreamResourceBacking(stream);
+    const resourceBacking = getReadableStreamResourceBacking(stream) ||
+      getReadableStreamResourceBackingUnrefable(stream);
     const reader = acquireReadableStreamDefaultReader(stream);
 
     if (resourceBacking) {
       // fast path, read whole body in a single op call
       try {
         readableStreamDisturb(stream);
-        const buf = await core.opAsync("op_read_all", resourceBacking.rid);
+        const promise = core.opAsync("op_read_all", resourceBacking.rid);
+        if (readableStreamIsUnrefable(stream)) {
+          const promiseId = stream[promiseIdSymbol] = promise[promiseIdSymbol];
+          if (stream[_isUnref]) core.unrefOp(promiseId);
+        }
+        const buf = await promise;
         readableStreamThrowIfErrored(stream);
         readableStreamClose(stream);
         return buf;
@@ -4585,6 +4604,9 @@
   }
 
   const _resourceBacking = Symbol("[[resourceBacking]]");
+  // This distinction exists to prevent unrefable streams being used in
+  // regular fast streams that are unaware of refability
+  const _resourceBackingUnrefable = Symbol("[[resourceBackingUnrefable]]");
   /** @template R */
   class ReadableStream {
     /** @type {ReadableStreamDefaultController | ReadableByteStreamController} */

runtime/js/40_spawn.js

@@ -12,12 +12,12 @@
     ObjectEntries,
     String,
     TypeError,
-    Uint8Array,
     PromisePrototypeThen,
     SafePromiseAll,
     SymbolFor,
   } = window.__bootstrap.primordials;
   const {
+    readableStreamCollectIntoUint8Array,
     readableStreamForRidUnrefable,
     readableStreamForRidUnrefableRef,
     readableStreamForRidUnrefableUnref,
@@ -64,26 +64,12 @@
     };
   }
 
-  async function collectOutput(readableStream) {
+  function collectOutput(readableStream) {
     if (!(readableStream instanceof ReadableStream)) {
       return null;
     }
 
-    const bufs = [];
+    return readableStreamCollectIntoUint8Array(readableStream);
-    let size = 0;
-    for await (const chunk of readableStream) {
-      bufs.push(chunk);
-      size += chunk.byteLength;
-    }
-
-    const buffer = new Uint8Array(size);
-    let offset = 0;
-    for (const chunk of bufs) {
-      buffer.set(chunk, offset);
-      offset += chunk.byteLength;
-    }
-
-    return buffer;
   }
 
   class Child {