stream: cache chunk data between 1st and 2nd pass in Writev

Signed-off-by: Daijiro Wachi <daijiro.wachi@gmail.com>

Author: watilde

benchmark/streams/writable-writev-string.js

@@ -0,0 +1,43 @@
+'use strict';
+
+const common = require('../common.js');
+const { Writable } = require('stream');
+
+// Benchmarks StreamBase::Writev with string chunks, exercising the chunk
+// cache that avoids redundant V8 array accesses, ToString, and ParseEncoding
+// calls between the sizing pass and the write pass.
+const bench = common.createBenchmark(main, {
+  n: [1e4],
+  chunks: [4, 16, 64],
+  encoding: ['utf8', 'latin1'],
+  type: ['string', 'buffer', 'mixed'],
+});
+
+function main({ n, chunks, encoding, type }) {
+  const str = 'Hello, benchmark! '.repeat(4);
+  const buf = Buffer.from(str, encoding);
+
+  const wr = new Writable({
+    writev(chunks, cb) { cb(); },
+    write(chunk, enc, cb) { cb(); },
+  });
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    wr.cork();
+    for (let j = 0; j < chunks; j++) {
+      if (type === 'buffer') {
+        wr.write(buf);
+      } else if (type === 'string') {
+        wr.write(str, encoding);
+      } else if (j % 2 === 0) {
+        // Alternate buffer and string to hit the mixed (non-all_buffers) path.
+        wr.write(buf);
+      } else {
+        wr.write(str, encoding);
+      }
+    }
+    wr.uncork();
+  }
+  bench.end(n);
+}

src/stream_base.cc

@@ -201,12 +201,24 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
   size_t offset;
 
   if (!all_buffers) {
+    // Cache per-chunk data from the first pass so the second pass avoids
+    // redundant V8 array accesses, ToString conversions, and ParseEncoding
+    // calls. Local<> handles remain valid for the duration of this scope.
+    struct CachedChunk {
+      Local<Value> value;
+      Local<String> string;  // empty for Buffer chunks
+      enum encoding enc;
+    };
+    MaybeStackBuffer<CachedChunk, 16> chunk_cache(count);
+
     // Determine storage size first
     for (size_t i = 0; i < count; i++) {
       Local<Value> chunk;
       if (!chunks->Get(context, i * 2).ToLocal(&chunk))
         return -1;
 
+      chunk_cache[i].value = chunk;
+
       if (Buffer::HasInstance(chunk))
         continue;
         // Buffer chunk, no additional storage required
@@ -219,6 +231,8 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
       if (!chunks->Get(context, i * 2 + 1).ToLocal(&next_chunk))
         return -1;
       enum encoding encoding = ParseEncoding(isolate, next_chunk);
+      chunk_cache[i].string = string;
+      chunk_cache[i].enc = encoding;
       size_t chunk_size;
       if ((encoding == UTF8 &&
              string->Length() > 65535 &&
@@ -230,35 +244,23 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
       storage_size += chunk_size;
     }
 
-    if (storage_size > INT_MAX)
-      return UV_ENOBUFS;
-  } else {
-    for (size_t i = 0; i < count; i++) {
-      Local<Value> chunk;
-      if (!chunks->Get(context, i).ToLocal(&chunk))
-        return -1;
-      bufs[i].base = Buffer::Data(chunk);
-      bufs[i].len = Buffer::Length(chunk);
-    }
-  }
+    if (storage_size > INT_MAX) return UV_ENOBUFS;
 
-  std::unique_ptr<BackingStore> bs;
-  if (storage_size > 0) {
-    bs = ArrayBuffer::NewBackingStore(
-        isolate, storage_size, BackingStoreInitializationMode::kUninitialized);
-  }
+    std::unique_ptr<BackingStore> bs;
+    if (storage_size > 0) {
+      bs = ArrayBuffer::NewBackingStore(
+          isolate,
+          storage_size,
+          BackingStoreInitializationMode::kUninitialized);
+    }
 
-  offset = 0;
-  if (!all_buffers) {
+    offset = 0;
     for (size_t i = 0; i < count; i++) {
-      Local<Value> chunk;
-      if (!chunks->Get(context, i * 2).ToLocal(&chunk))
-        return -1;
-
-      // Write buffer
-      if (Buffer::HasInstance(chunk)) {
-        bufs[i].base = Buffer::Data(chunk);
-        bufs[i].len = Buffer::Length(chunk);
+      // string.IsEmpty() signals a Buffer chunk; enc is uninitialised in
+      // that case so we must not read it.
+      if (chunk_cache[i].string.IsEmpty()) {
+        bufs[i].base = Buffer::Data(chunk_cache[i].value);
+        bufs[i].len = Buffer::Length(chunk_cache[i].value);
         continue;
       }
 
@@ -268,28 +270,32 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
           static_cast<char*>(bs ? bs->Data() : nullptr) + offset;
       size_t str_size = (bs ? bs->ByteLength() : 0) - offset;
 
-      Local<String> string;
-      if (!chunk->ToString(context).ToLocal(&string))
-        return -1;
-      Local<Value> next_chunk;
-      if (!chunks->Get(context, i * 2 + 1).ToLocal(&next_chunk))
-        return -1;
-      enum encoding encoding = ParseEncoding(isolate, next_chunk);
       str_size = StringBytes::Write(isolate,
                                     str_storage,
                                     str_size,
-                                    string,
-                                    encoding);
+                                    chunk_cache[i].string,
+                                    chunk_cache[i].enc);
       bufs[i].base = str_storage;
       bufs[i].len = str_size;
       offset += str_size;
     }
+
+    StreamWriteResult res = Write(*bufs, count, nullptr, req_wrap_obj);
+    SetWriteResult(res);
+    if (res.wrap != nullptr && storage_size > 0)
+      res.wrap->SetBackingStore(std::move(bs));
+    return res.err;
+  } else {
+    for (size_t i = 0; i < count; i++) {
+      Local<Value> chunk;
+      if (!chunks->Get(context, i).ToLocal(&chunk)) return -1;
+      bufs[i].base = Buffer::Data(chunk);
+      bufs[i].len = Buffer::Length(chunk);
+    }
   }
 
   StreamWriteResult res = Write(*bufs, count, nullptr, req_wrap_obj);
   SetWriteResult(res);
-  if (res.wrap != nullptr && storage_size > 0)
-    res.wrap->SetBackingStore(std::move(bs));
   return res.err;
 }
 

test/parallel/test-stream-writev-string-chunks.js

@@ -0,0 +1,111 @@
+'use strict';
+
+// Tests that StreamBase::Writev correctly handles string chunks by verifying
+// data integrity when string chunks, buffer chunks, and mixed chunks are
+// written via cork/uncork. This exercises the chunk-cache path (all_buffers=false)
+// introduced to avoid redundant V8 array accesses between the sizing and write
+// passes.
+
+const common = require('../common');
+const assert = require('assert');
+const { Writable } = require('stream');
+
+function collectWritev(decodeStrings) {
+  const received = [];
+  const w = new Writable({
+    decodeStrings,
+    writev(chunks, cb) {
+      for (const { chunk, encoding } of chunks)
+        received.push({ chunk, encoding });
+      cb();
+    },
+    write(chunk, encoding, cb) {
+      received.push({ chunk, encoding });
+      cb();
+    },
+  });
+  return { w, received };
+}
+
+// String chunks only — various encodings, decodeStrings: false so we can
+// inspect the raw string/encoding pairs the stream layer received.
+{
+  const { w, received } = collectWritev(false);
+  w.cork();
+  w.write('hello', 'utf8');
+  w.write('world', 'latin1');
+  w.write('cafebabe', 'hex');
+  w.uncork();
+  w.end();
+
+  w.on('finish', common.mustCall(() => {
+    assert.strictEqual(received[0].chunk, 'hello');
+    assert.strictEqual(received[0].encoding, 'utf8');
+    assert.strictEqual(received[1].chunk, 'world');
+    assert.strictEqual(received[1].encoding, 'latin1');
+    assert.strictEqual(received[2].chunk, 'cafebabe');
+    assert.strictEqual(received[2].encoding, 'hex');
+  }));
+}
+
+// Mixed buffer + string chunks — the non-all_buffers branch in Writev.
+{
+  const { w, received } = collectWritev(false);
+  const buf = Buffer.from('buffered');
+  w.cork();
+  w.write(buf);
+  w.write('stringy', 'utf8');
+  w.write(buf);
+  w.uncork();
+  w.end();
+
+  w.on('finish', common.mustCall(() => {
+    assert(Buffer.isBuffer(received[0].chunk));
+    assert.strictEqual(received[0].chunk.toString(), 'buffered');
+    assert.strictEqual(received[1].chunk, 'stringy');
+    assert.strictEqual(received[1].encoding, 'utf8');
+    assert(Buffer.isBuffer(received[2].chunk));
+    assert.strictEqual(received[2].chunk.toString(), 'buffered');
+  }));
+}
+
+// More than 16 chunks so MaybeStackBuffer<CachedChunk, 16> must heap-allocate.
+// Verify that every string arrives correctly after the realloc.
+{
+  const COUNT = 20;
+  const chunks = [];
+  const { w, received } = collectWritev(false);
+  w.cork();
+  for (let i = 0; i < COUNT; i++) {
+    const s = `chunk-${i}`;
+    chunks.push(s);
+    w.write(s, 'utf8');
+  }
+  w.uncork();
+  w.end();
+
+  w.on('finish', common.mustCall(() => {
+    assert.strictEqual(received.length, COUNT);
+    for (let i = 0; i < COUNT; i++) {
+      assert.strictEqual(received[i].chunk, chunks[i]);
+      assert.strictEqual(received[i].encoding, 'utf8');
+    }
+  }));
+}
+
+// decodeStrings: true — all strings should arrive as Buffers with correct data.
+{
+  const { w, received } = collectWritev(true);
+  w.cork();
+  w.write('café', 'utf8');
+  w.write(Buffer.from('raw'));
+  w.uncork();
+  w.end();
+
+  w.on('finish', common.mustCall(() => {
+    assert(Buffer.isBuffer(received[0].chunk));
+    assert.strictEqual(received[0].chunk.toString('utf8'), 'café');
+    assert(Buffer.isBuffer(received[1].chunk));
+    assert.strictEqual(received[1].chunk.toString(), 'raw');
+  }));
+}