http2: refactor read mechanism

Refactor the read mechanism to completely avoid copying.

Instead of copying individual `DATA` frame contents into buffers,
create `ArrayBuffer` instances for all socket reads and emit
slices of those `ArrayBuffer`s to JS.

PR-URL: https://github.com/nodejs/node/pull/18030
Reviewed-By: James M Snell <jasnell@gmail.com>

4 files changed, 69 insertions, 0 deletions

diff --git a/test/common/README.md b/test/common/README.md
index b50d8ca88e..3e3e5543b4 100644
--- a/test/common/README.md
+++ b/test/common/README.md
@@ -268,6 +268,17 @@ fail.
 
 If `fn` is not provided, an empty function will be used.
 
+### mustCallAsync([fn][, exact])
+* `fn` [<Function>]
+* `exact` [<Number>] default = 1
+* return [<Function>]
+
+The same as `mustCall()`, except that it is also checked that the Promise
+returned by the function is fulfilled for each invocation of the function.
+
+The return value of the wrapped function is the return value of the original
+function, if necessary wrapped as a promise.
+
 ### mustCallAtLeast([fn][, minimum])
 * `fn` [<Function>] default = () => {}
 * `minimum` [<Number>] default = 1
diff --git a/test/common/index.js b/test/common/index.js
index 418abef1b3..0cc48cf74c 100644
--- a/test/common/index.js
+++ b/test/common/index.js
@@ -501,6 +501,12 @@ exports.mustCallAtLeast = function(fn, minimum) {
   return _mustCallInner(fn, minimum, 'minimum');
 };
 
+exports.mustCallAsync = function(fn, exact) {
+  return exports.mustCall((...args) => {
+    return Promise.resolve(fn(...args)).then(exports.mustCall((val) => val));
+  }, exact);
+};
+
 function _mustCallInner(fn, criteria = 1, field) {
   if (process._exiting)
     throw new Error('Cannot use common.mustCall*() in process exit handler');
diff --git a/test/parallel/test-http2-backpressure.js b/test/parallel/test-http2-backpressure.js
new file mode 100644
index 0000000000..9b69dddbfd
--- /dev/null
+++ b/test/parallel/test-http2-backpressure.js
@@ -0,0 +1,49 @@
+'use strict';
+
+// Verifies that a full HTTP2 pipeline handles backpressure.
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const makeDuplexPair = require('../common/duplexpair');
+
+common.crashOnUnhandledRejection();
+
+{
+  let req;
+  const server = http2.createServer();
+  server.on('stream', common.mustCallAsync(async (stream, headers) => {
+    stream.respond({
+      'content-type': 'text/html',
+      ':status': 200
+    });
+    req._readableState.highWaterMark = 20;
+    stream._writableState.highWaterMark = 20;
+    assert.strictEqual(stream.write('A'.repeat(5)), true);
+    assert.strictEqual(stream.write('A'.repeat(40)), false);
+    assert.strictEqual(await event(req, 'data'), 'A'.repeat(5));
+    assert.strictEqual(await event(req, 'data'), 'A'.repeat(40));
+    await event(stream, 'drain');
+    assert.strictEqual(stream.write('A'.repeat(5)), true);
+    assert.strictEqual(stream.write('A'.repeat(40)), false);
+  }));
+
+  const { clientSide, serverSide } = makeDuplexPair();
+  server.emit('connection', serverSide);
+
+  const client = http2.connect('http://localhost:80', {
+    createConnection: common.mustCall(() => clientSide)
+  });
+
+  req = client.request({ ':path': '/' });
+  req.setEncoding('utf8');
+  req.end();
+}
+
+function event(ee, eventName) {
+  return new Promise((resolve) => {
+    ee.once(eventName, common.mustCall(resolve));
+  });
+}
diff --git a/test/parallel/test-http2-misbehaving-flow-control-paused.js b/test/parallel/test-http2-misbehaving-flow-control-paused.js
index 0b7299d5ac..d69e0fd802 100644
--- a/test/parallel/test-http2-misbehaving-flow-control-paused.js
+++ b/test/parallel/test-http2-misbehaving-flow-control-paused.js
@@ -56,6 +56,9 @@ let client;
 
 const server = h2.createServer({ settings: { initialWindowSize: 36 } });
 server.on('stream', (stream) => {
+  // Set the high water mark to zero, since otherwise we still accept
+  // reads from the source stream (if we can consume them).
+  stream._readableState.highWaterMark = 0;
   stream.pause();
   stream.on('error', common.expectsError({
     code: 'ERR_HTTP2_STREAM_ERROR',