src: improve StreamBase read throughput

Improve performance by providing JS with the raw ingridients
for the read data, i.e. an `ArrayBuffer` + offset + length
fields, instead of creating `Buffer` instances in C++ land.

PR-URL: https://github.com/nodejs/node/pull/23797
Reviewed-By: Tiancheng "Timothy" Gu <timothygu99@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

1 files changed, 25 insertions, 8 deletions

diff --git a/src/stream_base.cc b/src/stream_base.cc
index c6cce9c2d0..57713d5eaf 100644
--- a/src/stream_base.cc
+++ b/src/stream_base.cc
@@ -17,6 +17,7 @@
 namespace node {
 
 using v8::Array;
+using v8::ArrayBuffer;
 using v8::Boolean;
 using v8::Context;
 using v8::FunctionCallbackInfo;
@@ -303,17 +304,29 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
 }
 
 
-void StreamBase::CallJSOnreadMethod(ssize_t nread, Local<Object> buf) {
+void StreamBase::CallJSOnreadMethod(ssize_t nread,
+                                    Local<ArrayBuffer> ab,
+                                    size_t offset) {
   Environment* env = env_;
 
+#ifdef DEBUG
+  CHECK_EQ(static_cast<int32_t>(nread), nread);
+  CHECK_EQ(static_cast<int32_t>(offset), offset);
+
+  if (ab.IsEmpty()) {
+    CHECK_EQ(offset, 0);
+    CHECK_LE(nread, 0);
+  } else {
+    CHECK_GE(nread, 0);
+  }
+#endif
+  env->stream_base_state()[kReadBytesOrError] = nread;
+  env->stream_base_state()[kArrayBufferOffset] = offset;
+
   Local<Value> argv[] = {
-    Integer::New(env->isolate(), nread),
-    buf
+    ab.IsEmpty() ? Undefined(env->isolate()).As<Value>() : ab.As<Value>()
   };
 
-  if (argv[1].IsEmpty())
-    argv[1] = Undefined(env->isolate());
-
   AsyncWrap* wrap = GetAsyncWrap();
   CHECK_NOT_NULL(wrap);
   wrap->MakeCallback(env->onread_string(), arraysize(argv), argv);
@@ -366,14 +379,18 @@ void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   if (nread <= 0)  {
     free(buf.base);
     if (nread < 0)
-      stream->CallJSOnreadMethod(nread, Local<Object>());
+      stream->CallJSOnreadMethod(nread, Local<ArrayBuffer>());
     return;
   }
 
   CHECK_LE(static_cast<size_t>(nread), buf.len);
   char* base = Realloc(buf.base, nread);
 
-  Local<Object> obj = Buffer::New(env, base, nread).ToLocalChecked();
+  Local<ArrayBuffer> obj = ArrayBuffer::New(
+      env->isolate(),
+      base,
+      nread,
+      v8::ArrayBufferCreationMode::kInternalized);  // Transfer ownership to V8.
   stream->CallJSOnreadMethod(nread, obj);
 }