net: allow reading data into a static buffer

Author: mscdex

benchmark/net/net-s2c.js

@@ -5,48 +5,70 @@ const common = require('../common.js');
 const PORT = common.PORT;
 
 const bench = common.createBenchmark(main, {
-  len: [64, 102400, 1024 * 1024 * 16],
+  sendchunklen: [256, 32 * 1024, 128 * 1024, 16 * 1024 * 1024],
   type: ['utf', 'asc', 'buf'],
+  recvbuflen: [0, 64 * 1024, 1024 * 1024],
   dur: [5]
 });
 
 var chunk;
 var encoding;
+var recvbuf;
+var received = 0;
+
+function main({ dur, sendchunklen, type, recvbuflen }) {
+  if (isFinite(recvbuflen) && recvbuflen > 0)
+    recvbuf = Buffer.alloc(recvbuflen);
 
-function main({ dur, len, type }) {
   switch (type) {
     case 'buf':
-      chunk = Buffer.alloc(len, 'x');
+      chunk = Buffer.alloc(sendchunklen, 'x');
       break;
     case 'utf':
       encoding = 'utf8';
-      chunk = 'ü'.repeat(len / 2);
+      chunk = 'ü'.repeat(sendchunklen / 2);
       break;
     case 'asc':
       encoding = 'ascii';
-      chunk = 'x'.repeat(len);
+      chunk = 'x'.repeat(sendchunklen);
       break;
     default:
       throw new Error(`invalid type: ${type}`);
   }
 
   const reader = new Reader();
-  const writer = new Writer();
+  var writer;
+  var socketOpts;
+  if (recvbuf === undefined) {
+    writer = new Writer();
+    socketOpts = { port: PORT };
+  } else {
+    socketOpts = {
+      port: PORT,
+      onread: {
+        buffer: recvbuf,
+        callback: function(nread, buf) {
+          received += nread;
+        }
+      }
+    };
+  }
 
   // the actual benchmark.
   const server = net.createServer(function(socket) {
     reader.pipe(socket);
   });
 
   server.listen(PORT, function() {
-    const socket = net.connect(PORT);
+    const socket = net.connect(socketOpts);
     socket.on('connect', function() {
       bench.start();
 
-      socket.pipe(writer);
+      if (recvbuf === undefined)
+        socket.pipe(writer);
 
       setTimeout(function() {
-        const bytes = writer.received;
+        const bytes = received;
         const gbits = (bytes * 8) / (1024 * 1024 * 1024);
         bench.end(gbits);
         process.exit(0);
@@ -58,12 +80,11 @@ function main({ dur, len, type }) {
 const net = require('net');
 
 function Writer() {
-  this.received = 0;
   this.writable = true;
 }
 
 Writer.prototype.write = function(chunk, encoding, cb) {
-  this.received += chunk.length;
+  received += chunk.length;
 
   if (typeof encoding === 'function')
     encoding();

lib/internal/stream_base_commons.js

@@ -17,6 +17,8 @@ const { owner_symbol } = require('internal/async_hooks').symbols;
 const kMaybeDestroy = Symbol('kMaybeDestroy');
 const kUpdateTimer = Symbol('kUpdateTimer');
 const kAfterAsyncWrite = Symbol('kAfterAsyncWrite');
+const kBuffer = Symbol('kBuffer');
+const kBufferCb = Symbol('kBufferCb');
 
 function handleWriteReq(req, data, encoding) {
   const { handle } = req;
@@ -140,9 +142,16 @@ function onStreamRead(arrayBuffer) {
   stream[kUpdateTimer]();
 
   if (nread > 0 && !stream.destroyed) {
-    const offset = streamBaseState[kArrayBufferOffset];
-    const buf = new FastBuffer(arrayBuffer, offset, nread);
-    if (!stream.push(buf)) {
+    var ret;
+    const userBuf = stream[kBuffer];
+    if (userBuf) {
+      ret = (stream[kBufferCb](nread, userBuf) !== false);
+    } else {
+      const offset = streamBaseState[kArrayBufferOffset];
+      const buf = new FastBuffer(arrayBuffer, offset, nread);
+      ret = stream.push(buf);
+    }
+    if (!ret) {
       handle.reading = false;
       if (!stream.destroyed) {
         const err = handle.readStop();
@@ -186,4 +195,6 @@ module.exports = {
   kAfterAsyncWrite,
   kMaybeDestroy,
   kUpdateTimer,
+  kBuffer,
+  kBufferCb,
 };

lib/net.js

@@ -63,7 +63,9 @@ const {
   writeGeneric,
   onStreamRead,
   kAfterAsyncWrite,
-  kUpdateTimer
+  kUpdateTimer,
+  kBuffer,
+  kBufferCb
 } = require('internal/stream_base_commons');
 const {
   codes: {
@@ -101,18 +103,20 @@ function getFlags(ipv6Only) {
   return ipv6Only === true ? TCPConstants.UV_TCP_IPV6ONLY : 0;
 }
 
-function createHandle(fd, is_server) {
+function createHandle(fd, is_server, buf) {
   validateInt32(fd, 'fd', 0);
   const type = TTYWrap.guessHandleType(fd);
   if (type === 'PIPE') {
     return new Pipe(
-      is_server ? PipeConstants.SERVER : PipeConstants.SOCKET
+      is_server ? PipeConstants.SERVER : PipeConstants.SOCKET,
+      buf
     );
   }
 
   if (type === 'TCP') {
     return new TCP(
-      is_server ? TCPConstants.SERVER : TCPConstants.SOCKET
+      is_server ? TCPConstants.SERVER : TCPConstants.SOCKET,
+      buf
     );
   }
 
@@ -241,6 +245,8 @@ function Socket(options) {
   this._host = null;
   this[kLastWriteQueueSize] = 0;
   this[kTimeout] = null;
+  this[kBuffer] = null;
+  this[kBufferCb] = null;
 
   if (typeof options === 'number')
     options = { fd: options }; // Legacy interface.
@@ -265,40 +271,50 @@ function Socket(options) {
   if (options.handle) {
     this._handle = options.handle; // private
     this[async_id_symbol] = getNewAsyncId(this._handle);
-  } else if (options.fd !== undefined) {
-    const { fd } = options;
-    let err;
-
-    // createHandle will throw ERR_INVALID_FD_TYPE if `fd` is not
-    // a valid `PIPE` or `TCP` descriptor
-    this._handle = createHandle(fd, false);
-
-    err = this._handle.open(fd);
+  } else {
+    const onread = options.onread;
+    if (onread !== null && typeof onread === 'object' &&
+        Buffer.isBuffer(onread.buffer) &&
+        typeof onread.callback === 'function') {
+      this[kBuffer] = onread.buffer;
+      this[kBufferCb] = onread.callback;
+    }
+    if (options.fd !== undefined) {
+      const { fd } = options;
+      let err;
 
-    // While difficult to fabricate, in some architectures
-    // `open` may return an error code for valid file descriptors
-    // which cannot be opened. This is difficult to test as most
-    // un-openable fds will throw on `createHandle`
-    if (err)
-      throw errnoException(err, 'open');
+      // createHandle will throw ERR_INVALID_FD_TYPE if `fd` is not
+      // a valid `PIPE` or `TCP` descriptor
+      this._handle = createHandle(fd, false);
 
-    this[async_id_symbol] = this._handle.getAsyncId();
+      err = this._handle.open(fd);
 
-    if ((fd === 1 || fd === 2) &&
-        (this._handle instanceof Pipe) &&
-        process.platform === 'win32') {
-      // Make stdout and stderr blocking on Windows
-      err = this._handle.setBlocking(true);
+      // While difficult to fabricate, in some architectures
+      // `open` may return an error code for valid file descriptors
+      // which cannot be opened. This is difficult to test as most
+      // un-openable fds will throw on `createHandle`
       if (err)
-        throw errnoException(err, 'setBlocking');
-
-      this._writev = null;
-      this._write = makeSyncWrite(fd);
-      // makeSyncWrite adjusts this value like the original handle would, so
-      // we need to let it do that by turning it into a writable, own property.
-      Object.defineProperty(this._handle, 'bytesWritten', {
-        value: 0, writable: true
-      });
+        throw errnoException(err, 'open');
+
+      this[async_id_symbol] = this._handle.getAsyncId();
+
+      if ((fd === 1 || fd === 2) &&
+          (this._handle instanceof Pipe) &&
+          process.platform === 'win32') {
+        // Make stdout and stderr blocking on Windows
+        err = this._handle.setBlocking(true);
+        if (err)
+          throw errnoException(err, 'setBlocking');
+
+        this._writev = null;
+        this._write = makeSyncWrite(fd);
+        // makeSyncWrite adjusts this value like the original handle would, so
+        // we need to let it do that by turning it into a writable, own
+        // property.
+        Object.defineProperty(this._handle, 'bytesWritten', {
+          value: 0, writable: true
+        });
+      }
     }
   }
 
@@ -888,8 +904,8 @@ Socket.prototype.connect = function(...args) {
 
   if (!this._handle) {
     this._handle = pipe ?
-      new Pipe(PipeConstants.SOCKET) :
-      new TCP(TCPConstants.SOCKET);
+      new Pipe(PipeConstants.SOCKET, this[kBuffer]) :
+      new TCP(TCPConstants.SOCKET, this[kBuffer]);
     initSocketHandle(this);
   }
 

src/connection_wrap.cc

@@ -29,6 +29,17 @@ ConnectionWrap<WrapType, UVType>::ConnectionWrap(Environment* env,
                       reinterpret_cast<uv_stream_t*>(&handle_),
                       provider) {}
 
+template <typename WrapType, typename UVType>
+ConnectionWrap<WrapType, UVType>::ConnectionWrap(Environment* env,
+                                                 Local<Object> object,
+                                                 ProviderType provider,
+                                                 uv_buf_t buf)
+    : LibuvStreamWrap(env,
+                      object,
+                      reinterpret_cast<uv_stream_t*>(&handle_),
+                      provider,
+                      buf) {}
+
 
 template <typename WrapType, typename UVType>
 void ConnectionWrap<WrapType, UVType>::OnConnection(uv_stream_t* handle,
@@ -116,11 +127,23 @@ template ConnectionWrap<PipeWrap, uv_pipe_t>::ConnectionWrap(
     Local<Object> object,
     ProviderType provider);
 
+template ConnectionWrap<PipeWrap, uv_pipe_t>::ConnectionWrap(
+    Environment* env,
+    Local<Object> object,
+    ProviderType provider,
+    uv_buf_t buf);
+
 template ConnectionWrap<TCPWrap, uv_tcp_t>::ConnectionWrap(
     Environment* env,
     Local<Object> object,
     ProviderType provider);
 
+template ConnectionWrap<TCPWrap, uv_tcp_t>::ConnectionWrap(
+    Environment* env,
+    Local<Object> object,
+    ProviderType provider,
+    uv_buf_t buf);
+
 template void ConnectionWrap<PipeWrap, uv_pipe_t>::OnConnection(
     uv_stream_t* handle, int status);
 

src/connection_wrap.h

@@ -19,6 +19,10 @@ class ConnectionWrap : public LibuvStreamWrap {
   ConnectionWrap(Environment* env,
                  v8::Local<v8::Object> object,
                  ProviderType provider);
+  ConnectionWrap(Environment* env,
+                 v8::Local<v8::Object> object,
+                 ProviderType provider,
+                 uv_buf_t buf);
 
   UVType handle_;
 };

src/pipe_wrap.cc

@@ -149,7 +149,14 @@ void PipeWrap::New(const FunctionCallbackInfo<Value>& args) {
       UNREACHABLE();
   }
 
-  new PipeWrap(env, args.This(), provider, ipc);
+  if (args.Length() > 1 && Buffer::HasInstance(args[1])) {
+    uv_buf_t buf;
+    buf.base = Buffer::Data(args[1]);
+    buf.len = Buffer::Length(args[1]);
+    new PipeWrap(env, args.This(), provider, ipc, buf);
+  } else {
+    new PipeWrap(env, args.This(), provider, ipc);
+  }
 }
 
 
@@ -163,6 +170,17 @@ PipeWrap::PipeWrap(Environment* env,
                    // Suggestion: uv_pipe_init() returns void.
 }
 
+PipeWrap::PipeWrap(Environment* env,
+                   Local<Object> object,
+                   ProviderType provider,
+                   bool ipc,
+                   uv_buf_t buf)
+    : ConnectionWrap(env, object, provider, buf) {
+  int r = uv_pipe_init(env->event_loop(), &handle_, ipc);
+  CHECK_EQ(r, 0);  // How do we proxy this error up to javascript?
+                   // Suggestion: uv_pipe_init() returns void.
+}
+
 
 void PipeWrap::Bind(const FunctionCallbackInfo<Value>& args) {
   PipeWrap* wrap;

src/pipe_wrap.h

@@ -55,6 +55,11 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
            v8::Local<v8::Object> object,
            ProviderType provider,
            bool ipc);
+  PipeWrap(Environment* env,
+           v8::Local<v8::Object> object,
+           ProviderType provider,
+           bool ipc,
+           uv_buf_t buf);
 
   static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Bind(const v8::FunctionCallbackInfo<v8::Value>& args);

src/stream_base-inl.h

@@ -150,13 +150,25 @@ inline void StreamResource::EmitWantsWrite(size_t suggested_size) {
 }
 
 inline StreamBase::StreamBase(Environment* env) : env_(env) {
+  buf_.base = nullptr;
+  buf_.len = 0;
+  PushStreamListener(&default_listener_);
+}
+
+inline StreamBase::StreamBase(Environment* env, uv_buf_t buf)
+    : env_(env),
+      buf_(buf) {
   PushStreamListener(&default_listener_);
 }
 
 inline Environment* StreamBase::stream_env() const {
   return env_;
 }
 
+inline uv_buf_t StreamBase::stream_buf() const {
+  return buf_;
+}
+
 inline int StreamBase::Shutdown(v8::Local<v8::Object> req_wrap_obj) {
   Environment* env = stream_env();