stream: reduce allocations in StreamBase reads and writes

benchmark/net/tcp-raw-c2s.js

@@ -23,7 +23,11 @@ function main({ dur, len, type }) {
     TCPConnectWrap,
     constants: TCPConstants,
   } = common.binding('tcp_wrap');
-  const { WriteWrap } = common.binding('stream_wrap');
+  const {
+    WriteWrap,
+    kReadBytesOrError,
+    streamBaseState,
+  } = common.binding('stream_wrap');
   const PORT = common.PORT;
 
   const serverHandle = new TCP(TCPConstants.SERVER);
@@ -55,9 +59,7 @@ function main({ dur, len, type }) {
       if (!buffer)
         fail('read');
 
-      // Don't slice the buffer. The point of this is to isolate, not
-      // simulate real traffic.
-      bytes += buffer.byteLength;
+      bytes += streamBaseState[kReadBytesOrError];
     };
 
     clientHandle.readStart();

benchmark/net/tcp-raw-pipe.js

@@ -23,7 +23,12 @@ function main({ dur, len, type }) {
     TCPConnectWrap,
     constants: TCPConstants,
   } = common.binding('tcp_wrap');
-  const { WriteWrap } = common.binding('stream_wrap');
+  const {
+    WriteWrap,
+    kReadBytesOrError,
+    kArrayBufferOffset,
+    streamBaseState,
+  } = common.binding('stream_wrap');
   const PORT = common.PORT;
 
   function fail(err, syscall) {
@@ -50,9 +55,11 @@ function main({ dur, len, type }) {
       if (!buffer)
         fail('read');
 
+      const nread = streamBaseState[kReadBytesOrError];
+      const offset = streamBaseState[kArrayBufferOffset];
       const writeReq = new WriteWrap();
       writeReq.async = false;
-      err = clientHandle.writeBuffer(writeReq, Buffer.from(buffer));
+      err = clientHandle.writeBuffer(writeReq, Buffer.from(buffer, offset, nread));
 
       if (err)
         fail(err, 'write');
@@ -94,7 +101,7 @@ function main({ dur, len, type }) {
     if (!buffer)
       fail('read');
 
-    bytes += buffer.byteLength;
+    bytes += streamBaseState[kReadBytesOrError];
   };
 
   connectReq.oncomplete = function(err) {

benchmark/net/tcp-raw-s2c.js

@@ -23,7 +23,11 @@ function main({ dur, len, type }) {
     TCPConnectWrap,
     constants: TCPConstants,
   } = common.binding('tcp_wrap');
-  const { WriteWrap } = common.binding('stream_wrap');
+  const {
+    WriteWrap,
+    kReadBytesOrError,
+    streamBaseState,
+  } = common.binding('stream_wrap');
   const PORT = common.PORT;
 
   const serverHandle = new TCP(TCPConstants.SERVER);
@@ -116,9 +120,7 @@ function main({ dur, len, type }) {
         if (!buffer)
           fail('read');
 
-        // Don't slice the buffer. The point of this is to isolate, not
-        // simulate real traffic.
-        bytes += buffer.byteLength;
+        bytes += streamBaseState[kReadBytesOrError];
       };
 
       clientHandle.readStart();

lib/internal/stream_base_commons.js

@@ -8,11 +8,10 @@ const {
 const { Buffer } = require('buffer');
 const { FastBuffer } = require('internal/buffer');
 const {
-  WriteWrap,
   kReadBytesOrError,
   kArrayBufferOffset,
   kBytesWritten,
-  kLastWriteWasAsync,
+  kLastWriteErr,
   streamBaseState,
 } = internalBinding('stream_wrap');
 const { UV_EOF } = internalBinding('uv');
@@ -43,41 +42,6 @@ const kBuffer = Symbol('kBuffer');
 const kBufferGen = Symbol('kBufferGen');
 const kBufferCb = Symbol('kBufferCb');
 
-function handleWriteReq(req, data, encoding) {
-  const { handle } = req;
-
-  switch (encoding) {
-    case 'buffer':
-    {
-      const ret = handle.writeBuffer(req, data);
-      if (streamBaseState[kLastWriteWasAsync])
-        req.buffer = data;
-      return ret;
-    }
-    case 'latin1':
-    case 'binary':
-      return handle.writeLatin1String(req, data);
-    case 'utf8':
-    case 'utf-8':
-      return handle.writeUtf8String(req, data);
-    case 'ascii':
-      return handle.writeAsciiString(req, data);
-    case 'ucs2':
-    case 'ucs-2':
-    case 'utf16le':
-    case 'utf-16le':
-      return handle.writeUcs2String(req, data);
-    default:
-    {
-      const buffer = Buffer.from(data, encoding);
-      const ret = handle.writeBuffer(req, buffer);
-      if (streamBaseState[kLastWriteWasAsync])
-        req.buffer = buffer;
-      return ret;
-    }
-  }
-}
-
 function onWriteComplete(status) {
   debug('onWriteComplete', status, this.error);
 
@@ -105,21 +69,8 @@ function onWriteComplete(status) {
     this.callback(null);
 }
 
-function createWriteWrap(handle, callback) {
-  const req = new WriteWrap();
-
-  req.handle = handle;
-  req.oncomplete = onWriteComplete;
-  req.async = false;
-  req.bytes = 0;
-  req.buffer = null;
-  req.callback = callback;
-
-  return req;
-}
-
 function writevGeneric(self, data, cb) {
-  const req = createWriteWrap(self[kHandle], cb);
+  const handle = self[kHandle];
   const allBuffers = data.allBuffers;
   let chunks;
   if (allBuffers) {
@@ -134,33 +85,87 @@ function writevGeneric(self, data, cb) {
       chunks[i * 2 + 1] = entry.encoding;
     }
   }
-  const err = req.handle.writev(req, chunks, allBuffers);
-
-  // Retain chunks
-  if (err === 0) req._chunks = chunks;
+  const ret = handle.writev(null, chunks, allBuffers);
 
-  afterWriteDispatched(req, err, cb);
-  return req;
+  return afterWriteDispatched(handle, ret, chunks, cb);
 }
 
 function writeGeneric(self, data, encoding, cb) {
-  const req = createWriteWrap(self[kHandle], cb);
-  const err = handleWriteReq(req, data, encoding);
+  const handle = self[kHandle];
+  let ret;
+  let buffer = null;
 
-  afterWriteDispatched(req, err, cb);
-  return req;
+  switch (encoding) {
+    case 'buffer':
+      buffer = data;
+      ret = handle.writeBuffer(null, data);
+      break;
+    case 'latin1':
+    case 'binary':
+      ret = handle.writeLatin1String(null, data);
+      break;
+    case 'utf8':
+    case 'utf-8':
+      ret = handle.writeUtf8String(null, data);
+      break;
+    case 'ascii':
+      ret = handle.writeAsciiString(null, data);
+      break;
+    case 'ucs2':
+    case 'ucs-2':
+    case 'utf16le':
+    case 'utf-16le':
+      ret = handle.writeUcs2String(null, data);
+      break;
+    default:
+      buffer = Buffer.from(data, encoding);
+      ret = handle.writeBuffer(null, buffer);
+      break;
+  }
+
+  return afterWriteDispatched(handle, ret, buffer, cb);
 }
 
-function afterWriteDispatched(req, err, cb) {
-  req.bytes = streamBaseState[kBytesWritten];
-  req.async = !!streamBaseState[kLastWriteWasAsync];
+// `ret` is either a numeric error code (when the write - or its failure -
+// completed synchronously and no write request was created) or the WriteWrap
+// object of a dispatched write.
+function afterWriteDispatched(handle, ret, buffer, cb) {
+  const bytes = streamBaseState[kBytesWritten];
+
+  if (typeof ret === 'number') {
+    // The write (or its failure) completed synchronously.
+    if (ret !== 0)
+      cb(new ErrnoException(ret, 'write'));
+    else if (typeof cb === 'function')
+      cb();
+    return { async: false, bytes };
+  }
 
-  if (err !== 0)
-    return cb(new ErrnoException(err, 'write', req.error));
+  const req = ret;
+  const err = streamBaseState[kLastWriteErr];
+  if (err !== 0) {
+    cb(new ErrnoException(err, 'write', req.error));
+    return { async: false, bytes };
+  }
 
-  if (!req.async && typeof req.callback === 'function') {
-    req.callback();
+  req.handle = handle;
+  req.oncomplete = onWriteComplete;
+  req.callback = cb;
+  req.async = true;
+  req.bytes = bytes;
+  // Retain the data (or chunks) being written until the write completes.
+  req.buffer = buffer;
+
+  // If the write completed synchronously inside the write call, before
+  // `oncomplete` could be attached, the completion status has been recorded;
+  // deliver it now.
+  const status = req.writeStatus;
+  if (status !== null) {
+    req.writeStatus = null;
+    req.oncomplete(status);
   }
+
+  return req;
 }
 
 function onStreamRead(arrayBuffer) {

src/env.cc

@@ -767,6 +767,105 @@ std::unique_ptr<BackingStore> Environment::release_managed_buffer(
   return bs;
 }
 
+uv_buf_t StreamReadSlab::Allocate(Isolate* isolate, size_t suggested) {
+  DCHECK_GT(suggested, 0);
+  // Reads always get the full `suggested` size: handing out a smaller
+  // remainder would shrink the read() buffer and fragment large reads into
+  // more system calls, which costs more than the slab tail it saves.
+  size_t remaining = current_.bs ? current_.size() - current_.offset : 0;
+  if (remaining < suggested) {
+    // Retire the current slab; it stays alive through `retired_` if reads
+    // are still pending on it, or through JS views over it otherwise.
+    if (current_.bs && current_.pending > 0)
+      retired_.push_back(std::move(current_));
+    current_ = Slab();
+    std::unique_ptr<BackingStore> bs = ArrayBuffer::NewBackingStore(
+        isolate,
+        std::max(kSlabSize, suggested),
+        BackingStoreInitializationMode::kUninitialized);
+    current_.bs = std::move(bs);
+  }
+  char* base = current_.data() + current_.offset;
+  current_.offset += suggested;
+  current_.last_base = base;
+  current_.last_end = current_.offset;
+  current_.pending++;
+  return uv_buf_init(base, suggested);
+}
+
+StreamReadSlab::Slab* StreamReadSlab::FindSlab(const char* base) {
+  if (current_.Contains(base)) return &current_;
+  for (Slab& slab : retired_)
+    if (slab.Contains(base)) return &slab;
+  return nullptr;
+}
+
+void StreamReadSlab::CompleteReservation(Slab* slab,
+                                         const uv_buf_t& buf,
+                                         size_t used) {
+  DCHECK_GT(slab->pending, 0);
+  slab->pending--;
+  if (buf.base == slab->last_base && slab->offset == slab->last_end) {
+    // This was the most recent reservation and nothing was reserved after
+    // it: rewind the unused remainder so it can be reserved again.
+    slab->offset = (buf.base - slab->data()) + used;
+    slab->last_end = slab->offset;
+  }
+  if (slab != &current_ && slab->pending == 0) {
+    for (auto it = retired_.begin(); it != retired_.end(); ++it) {
+      if (&*it == slab) {
+        retired_.erase(it);
+        break;
+      }
+    }
+  }
+}
+
+bool StreamReadSlab::Commit(Isolate* isolate,
+                            const uv_buf_t& buf,
+                            size_t nread,
+                            Local<ArrayBuffer>* ab,
+                            size_t* offset) {
+  Slab* slab = FindSlab(buf.base);
+  if (slab == nullptr) return false;
+  DCHECK_LE(nread, buf.len);
+
+  // A partial read would leave the rest of its reservation as waste once
+  // the slab retires - memory that counts towards V8's external memory and
+  // drives up GC frequency. If most of the reservation would be wasted,
+  // give the read a right-sized copy instead (as if it had never been read
+  // into the slab) and return its reservation in full. Reads that (mostly)
+  // fill their reservation get a zero-copy view into the slab.
+  if (nread < buf.len - buf.len / 4 && buf.base == slab->last_base &&
+      slab->offset == slab->last_end) {
+    std::unique_ptr<BackingStore> bs = ArrayBuffer::NewBackingStore(
+        isolate, nread, BackingStoreInitializationMode::kUninitialized);
+    memcpy(bs->Data(), buf.base, nread);
+    *ab = ArrayBuffer::New(isolate, std::move(bs));
+    *offset = 0;
+    CompleteReservation(slab, buf, 0);
+    return true;
+  }
+
+  *offset = buf.base - slab->data();
+  if (slab->ab.IsEmpty()) {
+    *ab = ArrayBuffer::New(isolate, slab->bs);
+    slab->ab.Reset(isolate, *ab);
+  } else {
+    *ab = slab->ab.Get(isolate);
+  }
+  CompleteReservation(slab, buf, nread);
+  return true;
+}
+
+bool StreamReadSlab::Release(const uv_buf_t& buf) {
+  if (buf.base == nullptr) return true;
+  Slab* slab = FindSlab(buf.base);
+  if (slab == nullptr) return false;
+  CompleteReservation(slab, buf, 0);
+  return true;
+}
+
 std::string Environment::GetExecPath(const std::vector<std::string>& argv) {
   char exec_path_buf[2 * PATH_MAX];
   size_t exec_path_len = sizeof(exec_path_buf);