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>

7 files changed, 51 insertions, 13 deletions

diff --git a/src/env-inl.h b/src/env-inl.h
index 6ace0bf825..9d369d492c 100644
--- a/src/env-inl.h
+++ b/src/env-inl.h
@@ -446,6 +446,11 @@ Environment::trace_category_state() {
   return trace_category_state_;
 }
 
+inline AliasedBuffer<int32_t, v8::Int32Array>&
+Environment::stream_base_state() {
+  return stream_base_state_;
+}
+
 inline uint32_t Environment::get_next_module_id() {
   return module_id_counter_++;
 }
diff --git a/src/env.cc b/src/env.cc
index f77bcbf169..9d8fd967a4 100644
--- a/src/env.cc
+++ b/src/env.cc
@@ -158,6 +158,7 @@ Environment::Environment(IsolateData* isolate_data,
       makecallback_cntr_(0),
       should_abort_on_uncaught_toggle_(isolate_, 1),
       trace_category_state_(isolate_, kTraceCategoryCount),
+      stream_base_state_(isolate_, StreamBase::kNumStreamBaseStateFields),
       http_parser_buffer_(nullptr),
       fs_stats_field_array_(isolate_, kFsStatsFieldsLength * 2),
       fs_stats_field_bigint_array_(isolate_, kFsStatsFieldsLength * 2),
diff --git a/src/env.h b/src/env.h
index 3daa48f9cb..a85058f895 100644
--- a/src/env.h
+++ b/src/env.h
@@ -668,6 +668,7 @@ class Environment {
   should_abort_on_uncaught_toggle();
 
   inline AliasedBuffer<uint8_t, v8::Uint8Array>& trace_category_state();
+  inline AliasedBuffer<int32_t, v8::Int32Array>& stream_base_state();
 
   // The necessary API for async_hooks.
   inline double new_async_id();
@@ -951,6 +952,8 @@ class Environment {
   AliasedBuffer<uint8_t, v8::Uint8Array> trace_category_state_;
   std::unique_ptr<TrackingTraceStateObserver> trace_state_observer_;
 
+  AliasedBuffer<int32_t, v8::Int32Array> stream_base_state_;
+
   std::unique_ptr<performance::performance_state> performance_state_;
   std::unordered_map<std::string, uint64_t> performance_marks_;
 
diff --git a/src/node_http2.cc b/src/node_http2.cc
index 633d2389c7..ce5523a9d2 100644
--- a/src/node_http2.cc
+++ b/src/node_http2.cc
@@ -1256,10 +1256,7 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   CHECK_LE(offset, session->stream_buf_.len);
   CHECK_LE(offset + buf.len, session->stream_buf_.len);
 
-  Local<Object> buffer =
-      Buffer::New(env, session->stream_buf_ab_, offset, nread).ToLocalChecked();
-
-  stream->CallJSOnreadMethod(nread, buffer);
+  stream->CallJSOnreadMethod(nread, session->stream_buf_ab_, offset);
 }
 
 
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);
 }
 
diff --git a/src/stream_base.h b/src/stream_base.h
index d8e6df960f..039009e072 100644
--- a/src/stream_base.h
+++ b/src/stream_base.h
@@ -264,7 +264,9 @@ class StreamBase : public StreamResource {
   virtual bool IsIPCPipe();
   virtual int GetFD();
 
-  void CallJSOnreadMethod(ssize_t nread, v8::Local<v8::Object> buf);
+  void CallJSOnreadMethod(ssize_t nread,
+                          v8::Local<v8::ArrayBuffer> ab,
+                          size_t offset = 0);
 
   // This is named `stream_env` to avoid name clashes, because a lot of
   // subclasses are also `BaseObject`s.
@@ -326,12 +328,20 @@ class StreamBase : public StreamResource {
       const v8::FunctionCallbackInfo<v8::Value>& args)>
   static void JSMethod(const v8::FunctionCallbackInfo<v8::Value>& args);
 
+  // Internal, used only in StreamBase methods + env.cc.
+  enum StreamBaseStateFields {
+    kReadBytesOrError,
+    kArrayBufferOffset,
+    kNumStreamBaseStateFields
+  };
+
  private:
   Environment* env_;
   EmitToJSStreamListener default_listener_;
 
   friend class WriteWrap;
   friend class ShutdownWrap;
+  friend class Environment;  // For kNumStreamBaseStateFields.
 };
 
 
diff --git a/src/stream_wrap.cc b/src/stream_wrap.cc
index 9ccace435c..a3c45b940a 100644
--- a/src/stream_wrap.cc
+++ b/src/stream_wrap.cc
@@ -80,6 +80,11 @@ void LibuvStreamWrap::Initialize(Local<Object> target,
   target->Set(writeWrapString,
               ww->GetFunction(env->context()).ToLocalChecked());
   env->set_write_wrap_template(ww->InstanceTemplate());
+
+  NODE_DEFINE_CONSTANT(target, kReadBytesOrError);
+  NODE_DEFINE_CONSTANT(target, kArrayBufferOffset);
+  target->Set(context, FIXED_ONE_BYTE_STRING(env->isolate(), "streamBaseState"),
+              env->stream_base_state().GetJSArray()).FromJust();
 }