WIP: fastpipe

Author: addaleax

lib/internal/http2/core.js

@@ -117,7 +117,8 @@ const {
   kUpdateTimer,
   kHandle,
   kSession,
-  setStreamTimeout
+  setStreamTimeout,
+  installFastpipe
 } = require('internal/stream_base_commons');
 const { kTimeout } = require('internal/timers');
 const { isArrayBufferView } = require('internal/util/types');
@@ -2119,6 +2120,8 @@ class Http2Stream extends Duplex {
   }
 }
 
+installFastpipe(Http2Stream);
+
 function processHeaders(oldHeaders) {
   assertIsObject(oldHeaders, 'headers');
   const headers = Object.create(null);

lib/internal/stream_base_commons.js

@@ -25,6 +25,9 @@ const {
 } = require('internal/timers');
 const { isUint8Array } = require('internal/util/types');
 const { clearTimeout } = require('timers');
+const {
+  installFastpipe
+} = require('internal/stream_base_fastpipe');
 
 const kMaybeDestroy = Symbol('kMaybeDestroy');
 const kUpdateTimer = Symbol('kUpdateTimer');
@@ -263,4 +266,5 @@ module.exports = {
   kBuffer,
   kBufferCb,
   kBufferGen
+  installFastpipe
 };

lib/internal/stream_base_fastpipe.js

@@ -0,0 +1,135 @@
+'use strict';
+const { Readable, Writable } = require('stream');
+const { owner_symbol } = require('internal/async_hooks').symbols;
+
+const kFastpipeSupported = Symbol('kFastpipeSupported');
+const kFastpipeWritableBailoutMethods = {};
+const kFastpipeWritableBailoutMethodsOriginals = {};
+const kFastpipeReadableBailoutMethods = {};
+const kFastpipeReadableBailoutMethodsOriginals = {};
+
+for (const name of ['cork', 'uncork', 'write', 'destroy']) {
+  const original = Writable.prototype[name];
+  kFastpipeWritableBailoutMethodsOriginals[name] = original;
+  kFastpipeWritableBailoutMethods[name] = function(...args) {
+    bailoutFromWritable(this);
+    return original.call(this, ...args);
+  };
+}
+
+for (const name of [
+  'pause', 'resume', 'setEncoding', 'unshift', 'push', 'destroy',
+  'pipe', 'unpipe'
+]) {
+  const original = Readable.prototype[name];
+  kFastpipeReadableBailoutMethodsOriginals[name] = original;
+  kFastpipeReadableBailoutMethods[name] = function(...args) {
+    bailoutFromReadable(this);
+    return original.call(this, ...args);
+  };
+}
+
+const OriginalPipe = Readable.prototype.pipe;
+
+function installFastpipe(StreamClass) {
+  StreamClass.prototype[kFastpipeSupported] = true;
+  StreamClass.prototype.pipe = function(dest, opts) {
+    // If one of the streams is already part of a pipe, we first need to unpipe
+    // those streams. This will call the original `.pipe()` function, which
+    // adds `data` listeners, which prevents a new fastpipe from being formed
+    // later here.
+    if (dest._handle && dest._handle.pipeSource)
+      bailoutFromWritable(dest);
+
+    if (!dest[kFastpipeSupported] ||
+        this.listenerCount('data') > 0 ||
+        this.listenerCount('readable') > 0 ||
+        dest.listenerCount('drain') > 0 ||
+        this._readableState.decoder ||
+        this._readableState.objectMode ||
+        (dest._writableState && dest._writableState.objectMode)) {
+      return OriginalPipe.call(this, dest, opts);
+    }
+
+    if (this.connecting) {
+      OriginalPipe.call(this, dest, opts);
+      this.once('connect', () => {
+        this.unpipe(dest);
+        this.pipe(dest, opts);
+      });
+      return this;
+    }
+
+    if (this._readableState.length > 0) {
+      const data = this.read();
+      dest.write(data);
+    }
+
+    if (dest._writableState &&
+        (dest._writableState.corked || dest._writableState.writing)) {
+      OriginalPipe.call(this, dest, opts);
+      dest.once('drain', () => {
+        this.unpipe(dest);
+        this.pipe(dest, opts);
+      });
+      return this;
+    }
+
+    Object.assign(dest, kFastpipeWritableBailoutMethods);
+    dest.on('newListener', writableNewListener);
+
+    Object.assign(this, kFastpipeReadableBailoutMethods);
+    this.on('newListener', readableNewListener);
+
+    const { internalBinding } = require('internal/bootstrap/loaders');
+    const { StreamPipe } = internalBinding('stream_pipe');
+    const pipe = new StreamPipe(this._handle._externalStream,
+                                dest._handle._externalStream);
+    pipe.onunpipe = onunpipe;
+    pipe.start();
+    return this;
+  };
+}
+
+function onunpipe() {
+  const self = this.source[owner_symbol];
+  bailoutFromReadable(self);
+}
+
+function writableNewListener(name) {
+  if (name === 'drain')
+    bailoutFromWritable(this);
+}
+
+function readableNewListener(name) {
+  if (name === 'data' || name === 'readable')
+    bailoutFromReadable(this);
+}
+
+function bailoutFromReadable(self) {
+  if (!self._handle)
+    return;
+  const pipe = self._handle.pipeTarget;
+  const dest = pipe.sink[owner_symbol];
+  self.removeListener('newListener', readableNewListener);
+  Object.assign(self, kFastpipeReadableBailoutMethodsOriginals);
+  dest.removeListener('newListener', writableNewListener);
+  Object.assign(dest, kFastpipeWritableBailoutMethodsOriginals);
+  pipe.unpipe();
+  pipe.onunpipe = function() {};
+  if (pipe.pendingWrites() > 0) {
+    pipe.oncomplete = function() {
+      OriginalPipe.call(self, dest);
+    };
+  } else {
+    OriginalPipe.call(self, dest);
+  }
+}
+
+function bailoutFromWritable(dest) {
+  bailoutFromReadable(dest._handle.pipeSource.source[owner_symbol]);
+}
+
+module.exports = {
+  installFastpipe
+};

lib/net.js

@@ -71,6 +71,7 @@ const {
   kBuffer,
   kBufferCb,
   kBufferGen
+  installFastpipe
 } = require('internal/stream_base_commons');
 const {
   codes: {
@@ -779,6 +780,8 @@ Socket.prototype._write = function(data, encoding, cb) {
   this._writeGeneric(false, data, encoding, cb);
 };
 
+installFastpipe(Socket);
+
 
 // Legacy alias. Having this is probably being overly cautious, but it doesn't
 // really hurt anyone either. This can probably be removed safely if desired.

node.gyp

@@ -195,6 +195,7 @@
       'lib/internal/v8_prof_processor.js',
       'lib/internal/validators.js',
       'lib/internal/stream_base_commons.js',
+      'lib/internal/stream_base_fastpipe.js',
       'lib/internal/vm/source_text_module.js',
       'lib/internal/worker.js',
       'lib/internal/worker/io.js',