stream: allocate stream read buffers from a slab

Read buffers for streams that emit their data to JS were allocated per
read: a 64KB backing store, tracked in a map, and then - since reads
rarely fill the whole buffer - reallocated to the right size and copied.

Allocate read buffers from a 64KB slab instead. Reads reserve the
suggested size from the slab and JS receives a view over the slab's
ArrayBuffer at the read's offset, using the offset mechanism that
onStreamRead already supports. Unused reservation space is rewound when
a read returns less than was reserved, so small reads (e.g. TLS records)
share a slab.

This removes the per-read allocations, the map bookkeeping and the
resize copy.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

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();

src/env.cc

@@ -816,6 +816,105 @@ void Environment::recycle_managed_buffer(std::unique_ptr<BackingStore> bs) {
     managed_buffer_cache_ = std::move(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);

src/env.h

@@ -277,6 +277,63 @@ struct ContextInfo {
 
 class EnabledDebugList;
 
+// A bump allocator for stream read buffers. Reads reserve a chunk of the
+// current slab and, once completed, are handed to JS as a view (ArrayBuffer +
+// offset) over the slab, so that no per-read allocation or copy is needed.
+// Unused reservation space is rewound when a read returns fewer bytes than
+// were reserved. Slabs with reads still pending when a new slab is started
+// (possible when multiple reads are in flight, e.g. on Windows) are kept
+// alive in `retired_` until those reads complete.
+class StreamReadSlab {
+ public:
+  // Sized to match the read buffer size that libuv suggests for stream
+  // reads: a slab typically serves a single large read (still avoiding the
+  // copy that right-sizing the buffer would need), or many small ones.
+  // Larger slabs amortize allocations further, but stay alive (pinned by
+  // chunk views) long enough to be promoted to V8's old generation, where
+  // their external memory is only reclaimed by major GCs.
+  static constexpr size_t kSlabSize = 64 * 1024;
+
+  // Reserve `suggested` bytes. Starts a new slab if the current one does
+  // not have enough space left.
+  uv_buf_t Allocate(v8::Isolate* isolate, size_t suggested);
+  // Commit a completed read of `nread` bytes into the buffer previously
+  // returned by Allocate(), rewinding the unused remainder of the
+  // reservation if possible. Returns the slab's ArrayBuffer and the offset
+  // of `buf.base` within it. Returns false if the buffer was not allocated
+  // from this slab (e.g. reads rerouted from another stream listener).
+  bool Commit(v8::Isolate* isolate,
+              const uv_buf_t& buf,
+              size_t nread,
+              v8::Local<v8::ArrayBuffer>* ab,
+              size_t* offset);
+  // Return an unused reservation (failed or empty read). Returns false if
+  // the buffer was not allocated from this slab.
+  bool Release(const uv_buf_t& buf);
+
+ private:
+  struct Slab {
+    std::shared_ptr<v8::BackingStore> bs;
+    v8::Global<v8::ArrayBuffer> ab;
+    size_t offset = 0;          // Bump pointer.
+    size_t pending = 0;         // Reservations not yet committed/released.
+    char* last_base = nullptr;  // Most recent reservation...
+    size_t last_end = 0;        // ...and the bump pointer after it.
+
+    char* data() const { return static_cast<char*>(bs->Data()); }
+    size_t size() const { return bs->ByteLength(); }
+    bool Contains(const char* p) const {
+      return bs && p >= data() && p < data() + size();
+    }
+  };
+
+  Slab* FindSlab(const char* base);
+  void CompleteReservation(Slab* slab, const uv_buf_t& buf, size_t used);
+
+  Slab current_;
+  std::vector<Slab> retired_;
+};
+
 namespace per_process {
 extern std::shared_ptr<KVStore> system_environment;
 }
@@ -1066,6 +1123,10 @@ class Environment final : public MemoryRetainer {
   // Only buffers that were not exposed externally may be recycled.
   void recycle_managed_buffer(std::unique_ptr<v8::BackingStore> bs);
 
+  StreamReadSlab& stream_read_slab() {
+    return stream_read_slab_;
+  }
+
   void AddUnmanagedFd(int fd);
   void RemoveUnmanagedFd(int fd);
 
@@ -1283,6 +1344,9 @@ class Environment final : public MemoryRetainer {
       released_allocated_buffers_;
   std::unique_ptr<v8::BackingStore> managed_buffer_cache_;
 
+  // Used by EmitToJSStreamListener to allocate stream read buffers.
+  StreamReadSlab stream_read_slab_;
+
   v8::CpuProfiler* cpu_profiler_ = nullptr;
   std::vector<v8::ProfilerId> pending_profiles_;
 };

src/stream_base.cc

@@ -684,7 +684,7 @@ void StreamResource::ClearError() {
 uv_buf_t EmitToJSStreamListener::OnStreamAlloc(size_t suggested_size) {
   CHECK_NOT_NULL(stream_);
   Environment* env = static_cast<StreamBase*>(stream_)->stream_env();
-  return env->allocate_managed_buffer(suggested_size);
+  return env->stream_read_slab().Allocate(env->isolate(), suggested_size);
 }
 
 void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
@@ -694,25 +694,34 @@ void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
   Isolate* isolate = env->isolate();
   HandleScope handle_scope(isolate);
   Context::Scope context_scope(env->context());
-  std::unique_ptr<BackingStore> bs = env->release_managed_buffer(buf_);
 
   if (nread <= 0)  {
-    env->recycle_managed_buffer(std::move(bs));
+    if (!env->stream_read_slab().Release(buf_))
+      env->recycle_managed_buffer(env->release_managed_buffer(buf_));
     if (nread < 0)
       stream->CallJSOnreadMethod(nread, Local<ArrayBuffer>());
     return;
   }
 
-  CHECK_LE(static_cast<size_t>(nread), bs->ByteLength());
-  if (static_cast<size_t>(nread) != bs->ByteLength()) {
-    std::unique_ptr<BackingStore> old_bs = std::move(bs);
-    bs = ArrayBuffer::NewBackingStore(
-        isolate, nread, BackingStoreInitializationMode::kUninitialized);
-    memcpy(bs->Data(), old_bs->Data(), nread);
-    env->recycle_managed_buffer(std::move(old_bs));
+  CHECK_LE(static_cast<size_t>(nread), buf_.len);
+  size_t offset = 0;
+  Local<ArrayBuffer> ab;
+  if (!env->stream_read_slab().Commit(isolate, buf_, nread, &ab, &offset)) {
+    // The buffer was allocated through allocate_managed_buffer() by another
+    // stream listener (e.g. StreamPipe's) whose read was rerouted here after
+    // that listener was removed.
+    std::unique_ptr<BackingStore> bs = env->release_managed_buffer(buf_);
+    CHECK_LE(static_cast<size_t>(nread), bs->ByteLength());
+    if (static_cast<size_t>(nread) != bs->ByteLength()) {
+      std::unique_ptr<BackingStore> old_bs = std::move(bs);
+      bs = ArrayBuffer::NewBackingStore(
+          isolate, nread, BackingStoreInitializationMode::kUninitialized);
+      memcpy(bs->Data(), old_bs->Data(), nread);
+      env->recycle_managed_buffer(std::move(old_bs));
+    }
+    ab = ArrayBuffer::New(isolate, std::move(bs));
   }
-
-  stream->CallJSOnreadMethod(nread, ArrayBuffer::New(isolate, std::move(bs)));
+  stream->CallJSOnreadMethod(nread, ab, offset);
 }