src: remove usage of AllocatedBuffer from stream_*

Signed-off-by: Darshan Sen <darshan.sen@postman.com>

PR-URL: https://github.com/nodejs/node/pull/40293
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

src/stream_base-inl.h

@@ -3,7 +3,6 @@
 
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
-#include "allocated_buffer-inl.h"
 #include "async_wrap-inl.h"
 #include "base_object-inl.h"
 #include "node.h"
@@ -270,9 +269,9 @@ ShutdownWrap* ShutdownWrap::FromObject(
   return FromObject(base_obj->object());
 }
 
-void WriteWrap::SetAllocatedStorage(AllocatedBuffer&& storage) {
-  CHECK_NULL(storage_.data());
-  storage_ = std::move(storage);
+void WriteWrap::SetBackingStore(std::unique_ptr<v8::BackingStore> bs) {
+  CHECK(!backing_store_);
+  backing_store_ = std::move(bs);
 }
 
 void StreamReq::Done(int status, const char* error_str) {

src/stream_base.cc

@@ -19,6 +19,7 @@ namespace node {
 
 using v8::Array;
 using v8::ArrayBuffer;
+using v8::BackingStore;
 using v8::ConstructorBehavior;
 using v8::Context;
 using v8::DontDelete;
@@ -29,6 +30,7 @@ using v8::FunctionCallbackInfo;
 using v8::FunctionTemplate;
 using v8::HandleScope;
 using v8::Integer;
+using v8::Isolate;
 using v8::Local;
 using v8::MaybeLocal;
 using v8::Object;
@@ -80,6 +82,8 @@ void StreamBase::SetWriteResult(const StreamWriteResult& res) {
 
 int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
+  Isolate* isolate = env->isolate();
+  Local<Context> context = env->context();
 
   CHECK(args[0]->IsObject());
   CHECK(args[1]->IsArray());
@@ -102,21 +106,21 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
   if (!all_buffers) {
     // Determine storage size first
     for (size_t i = 0; i < count; i++) {
-      Local<Value> chunk = chunks->Get(env->context(), i * 2).ToLocalChecked();
+      Local<Value> chunk = chunks->Get(context, i * 2).ToLocalChecked();
 
       if (Buffer::HasInstance(chunk))
         continue;
         // Buffer chunk, no additional storage required
 
       // String chunk
-      Local<String> string = chunk->ToString(env->context()).ToLocalChecked();
-      enum encoding encoding = ParseEncoding(env->isolate(),
-          chunks->Get(env->context(), i * 2 + 1).ToLocalChecked());
+      Local<String> string = chunk->ToString(context).ToLocalChecked();
+      enum encoding encoding = ParseEncoding(isolate,
+          chunks->Get(context, i * 2 + 1).ToLocalChecked());
       size_t chunk_size;
       if (encoding == UTF8 && string->Length() > 65535 &&
-          !StringBytes::Size(env->isolate(), string, encoding).To(&chunk_size))
+          !StringBytes::Size(isolate, string, encoding).To(&chunk_size))
         return 0;
-      else if (!StringBytes::StorageSize(env->isolate(), string, encoding)
+      else if (!StringBytes::StorageSize(isolate, string, encoding)
                     .To(&chunk_size))
         return 0;
       storage_size += chunk_size;
@@ -126,20 +130,22 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
       return UV_ENOBUFS;
   } else {
     for (size_t i = 0; i < count; i++) {
-      Local<Value> chunk = chunks->Get(env->context(), i).ToLocalChecked();
+      Local<Value> chunk = chunks->Get(context, i).ToLocalChecked();
       bufs[i].base = Buffer::Data(chunk);
       bufs[i].len = Buffer::Length(chunk);
     }
   }
 
-  AllocatedBuffer storage;
-  if (storage_size > 0)
-    storage = AllocatedBuffer::AllocateManaged(env, storage_size);
+  std::unique_ptr<BackingStore> bs;
+  if (storage_size > 0) {
+    NoArrayBufferZeroFillScope no_zero_fill_scope(env->isolate_data());
+    bs = ArrayBuffer::NewBackingStore(isolate, storage_size);
+  }
 
   offset = 0;
   if (!all_buffers) {
     for (size_t i = 0; i < count; i++) {
-      Local<Value> chunk = chunks->Get(env->context(), i * 2).ToLocalChecked();
+      Local<Value> chunk = chunks->Get(context, i * 2).ToLocalChecked();
 
       // Write buffer
       if (Buffer::HasInstance(chunk)) {
@@ -150,13 +156,14 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
 
       // Write string
       CHECK_LE(offset, storage_size);
-      char* str_storage = storage.data() + offset;
-      size_t str_size = storage.size() - offset;
+      char* str_storage =
+          static_cast<char*>(bs ? bs->Data() : nullptr) + offset;
+      size_t str_size = (bs ? bs->ByteLength() : 0) - offset;
 
-      Local<String> string = chunk->ToString(env->context()).ToLocalChecked();
-      enum encoding encoding = ParseEncoding(env->isolate(),
-          chunks->Get(env->context(), i * 2 + 1).ToLocalChecked());
-      str_size = StringBytes::Write(env->isolate(),
+      Local<String> string = chunk->ToString(context).ToLocalChecked();
+      enum encoding encoding = ParseEncoding(isolate,
+          chunks->Get(context, i * 2 + 1).ToLocalChecked());
+      str_size = StringBytes::Write(isolate,
                                     str_storage,
                                     str_size,
                                     string,
@@ -169,9 +176,8 @@ int StreamBase::Writev(const FunctionCallbackInfo<Value>& args) {
 
   StreamWriteResult res = Write(*bufs, count, nullptr, req_wrap_obj);
   SetWriteResult(res);
-  if (res.wrap != nullptr && storage_size > 0) {
-    res.wrap->SetAllocatedStorage(std::move(storage));
-  }
+  if (res.wrap != nullptr && storage_size > 0)
+    res.wrap->SetBackingStore(std::move(bs));
   return res.err;
 }
 
@@ -216,6 +222,7 @@ int StreamBase::WriteBuffer(const FunctionCallbackInfo<Value>& args) {
 template <enum encoding enc>
 int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
+  Isolate* isolate = env->isolate();
   CHECK(args[0]->IsObject());
   CHECK(args[1]->IsString());
 
@@ -230,9 +237,9 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
   // computing their actual size, rather than tripling the storage.
   size_t storage_size;
   if (enc == UTF8 && string->Length() > 65535 &&
-      !StringBytes::Size(env->isolate(), string, enc).To(&storage_size))
+      !StringBytes::Size(isolate, string, enc).To(&storage_size))
     return 0;
-  else if (!StringBytes::StorageSize(env->isolate(), string, enc)
+  else if (!StringBytes::StorageSize(isolate, string, enc)
                 .To(&storage_size))
     return 0;
 
@@ -248,7 +255,7 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
   bool try_write = storage_size <= sizeof(stack_storage) &&
                    (!IsIPCPipe() || send_handle_obj.IsEmpty());
   if (try_write) {
-    data_size = StringBytes::Write(env->isolate(),
+    data_size = StringBytes::Write(isolate,
                                    stack_storage,
                                    storage_size,
                                    string,
@@ -274,18 +281,20 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
     CHECK_EQ(count, 1);
   }
 
-  AllocatedBuffer data;
+  std::unique_ptr<BackingStore> bs;
 
   if (try_write) {
     // Copy partial data
-    data = AllocatedBuffer::AllocateManaged(env, buf.len);
-    memcpy(data.data(), buf.base, buf.len);
+    NoArrayBufferZeroFillScope no_zero_fill_scope(env->isolate_data());
+    bs = ArrayBuffer::NewBackingStore(isolate, buf.len);
+    memcpy(static_cast<char*>(bs->Data()), buf.base, buf.len);
     data_size = buf.len;
   } else {
     // Write it
-    data = AllocatedBuffer::AllocateManaged(env, storage_size);
-    data_size = StringBytes::Write(env->isolate(),
-                                   data.data(),
+    NoArrayBufferZeroFillScope no_zero_fill_scope(env->isolate_data());
+    bs = ArrayBuffer::NewBackingStore(isolate, storage_size);
+    data_size = StringBytes::Write(isolate,
+                                   static_cast<char*>(bs->Data()),
                                    storage_size,
                                    string,
                                    enc);
@@ -293,7 +302,7 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
 
   CHECK_LE(data_size, storage_size);
 
-  buf = uv_buf_init(data.data(), data_size);
+  buf = uv_buf_init(static_cast<char*>(bs->Data()), data_size);
 
   uv_stream_t* send_handle = nullptr;
 
@@ -312,9 +321,8 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
   res.bytes += synchronously_written;
 
   SetWriteResult(res);
-  if (res.wrap != nullptr) {
-    res.wrap->SetAllocatedStorage(std::move(data));
-  }
+  if (res.wrap != nullptr)
+    res.wrap->SetBackingStore(std::move(bs));
 
   return res.err;
 }
@@ -511,16 +519,17 @@ 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 AllocatedBuffer::AllocateManaged(env, suggested_size).release();
+  return env->allocate_managed_buffer(suggested_size);
 }
 
 void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
   CHECK_NOT_NULL(stream_);
   StreamBase* stream = static_cast<StreamBase*>(stream_);
   Environment* env = stream->stream_env();
-  HandleScope handle_scope(env->isolate());
+  Isolate* isolate = env->isolate();
+  HandleScope handle_scope(isolate);
   Context::Scope context_scope(env->context());
-  AllocatedBuffer buf(env, buf_);
+  std::unique_ptr<BackingStore> bs = env->release_managed_buffer(buf_);
 
   if (nread <= 0)  {
     if (nread < 0)
@@ -528,10 +537,10 @@ void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
     return;
   }
 
-  CHECK_LE(static_cast<size_t>(nread), buf.size());
-  buf.Resize(nread);
+  CHECK_LE(static_cast<size_t>(nread), bs->ByteLength());
+  bs = BackingStore::Reallocate(isolate, std::move(bs), nread);
 
-  stream->CallJSOnreadMethod(nread, buf.ToArrayBuffer());
+  stream->CallJSOnreadMethod(nread, ArrayBuffer::New(isolate, std::move(bs)));
 }
 
 

src/stream_base.h

@@ -4,7 +4,6 @@
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
 #include "env.h"
-#include "allocated_buffer.h"
 #include "async_wrap.h"
 #include "node.h"
 #include "util.h"
@@ -90,7 +89,7 @@ class ShutdownWrap : public StreamReq {
 
 class WriteWrap : public StreamReq {
  public:
-  inline void SetAllocatedStorage(AllocatedBuffer&& storage);
+  inline void SetBackingStore(std::unique_ptr<v8::BackingStore> bs);
 
   inline WriteWrap(
       StreamBase* stream,
@@ -105,7 +104,7 @@ class WriteWrap : public StreamReq {
   void OnDone(int status) override;
 
  private:
-  AllocatedBuffer storage_;
+  std::unique_ptr<v8::BackingStore> backing_store_;
 };
 
 

src/stream_pipe.cc

@@ -1,11 +1,11 @@
 #include "stream_pipe.h"
-#include "allocated_buffer-inl.h"
 #include "stream_base-inl.h"
 #include "node_buffer.h"
 #include "util-inl.h"
 
 namespace node {
 
+using v8::BackingStore;
 using v8::Context;
 using v8::Function;
 using v8::FunctionCallbackInfo;
@@ -118,13 +118,13 @@ uv_buf_t StreamPipe::ReadableListener::OnStreamAlloc(size_t suggested_size) {
   StreamPipe* pipe = ContainerOf(&StreamPipe::readable_listener_, this);
   size_t size = std::min(suggested_size, pipe->wanted_data_);
   CHECK_GT(size, 0);
-  return AllocatedBuffer::AllocateManaged(pipe->env(), size).release();
+  return pipe->env()->allocate_managed_buffer(size);
 }
 
 void StreamPipe::ReadableListener::OnStreamRead(ssize_t nread,
                                                 const uv_buf_t& buf_) {
   StreamPipe* pipe = ContainerOf(&StreamPipe::readable_listener_, this);
-  AllocatedBuffer buf(pipe->env(), buf_);
+  std::unique_ptr<BackingStore> bs = pipe->env()->release_managed_buffer(buf_);
   if (nread < 0) {
     // EOF or error; stop reading and pass the error to the previous listener
     // (which might end up in JS).
@@ -144,19 +144,20 @@ void StreamPipe::ReadableListener::OnStreamRead(ssize_t nread,
     return;
   }
 
-  pipe->ProcessData(nread, std::move(buf));
+  pipe->ProcessData(nread, std::move(bs));
 }
 
-void StreamPipe::ProcessData(size_t nread, AllocatedBuffer&& buf) {
+void StreamPipe::ProcessData(size_t nread,
+                             std::unique_ptr<BackingStore> bs) {
   CHECK(uses_wants_write_ || pending_writes_ == 0);
-  uv_buf_t buffer = uv_buf_init(buf.data(), nread);
+  uv_buf_t buffer = uv_buf_init(static_cast<char*>(bs->Data()), nread);
   StreamWriteResult res = sink()->Write(&buffer, 1);
   pending_writes_++;
   if (!res.async) {
     writable_listener_.OnStreamAfterWrite(nullptr, res.err);
   } else {
     is_reading_ = false;
-    res.wrap->SetAllocatedStorage(std::move(buf));
+    res.wrap->SetBackingStore(std::move(bs));
     if (source() != nullptr)
       source()->ReadStop();
   }

src/stream_pipe.h

@@ -4,7 +4,6 @@
 #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
 
 #include "stream_base.h"
-#include "allocated_buffer.h"
 
 namespace node {
 
@@ -43,7 +42,7 @@ class StreamPipe : public AsyncWrap {
   // `OnStreamWantsWrite()` support.
   size_t wanted_data_ = 0;
 
-  void ProcessData(size_t nread, AllocatedBuffer&& buf);
+  void ProcessData(size_t nread, std::unique_ptr<v8::BackingStore> bs);
 
   class ReadableListener : public StreamListener {
    public: