stream: create write request objects lazily

Every stream write created a WriteWrap JS object up front, even though
most writes complete synchronously via uv_try_write() and never use it.

Let stream_base_commons pass null instead of a request object.
StreamBase::Write() already creates the wrap object only when the write
does not complete synchronously; return that object to JS (which
attaches oncomplete/callback to it) and a plain error code otherwise.

Writes that complete synchronously now cross the JS/C++ boundary once
and allocate nothing. Callers that pass in a request object
(child_process IPC, webstreams adapters) behave as before.

Since Http2Stream::DoWrite() can invoke the completion callback
synchronously - before JS has attached oncomplete - such completions
are now recorded on the request object's writeStatus field and replayed
by stream_base_commons after dispatch. This also replaces a Has() plus
name-based MakeCallback() pair with a single Get().

Also pre-create the JS fields of WriteWrap instances in the object
template, as was already done for ShutdownWrap, so that they are
in-object properties.

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

Author: mcollina

lib/internal/stream_base_commons.js

@@ -8,11 +8,10 @@ const {
 const { Buffer } = require('buffer');
 const { FastBuffer } = require('internal/buffer');
 const {
-  WriteWrap,
   kReadBytesOrError,
   kArrayBufferOffset,
   kBytesWritten,
-  kLastWriteWasAsync,
+  kLastWriteErr,
   streamBaseState,
 } = internalBinding('stream_wrap');
 const { UV_EOF } = internalBinding('uv');
@@ -43,41 +42,6 @@ const kBuffer = Symbol('kBuffer');
 const kBufferGen = Symbol('kBufferGen');
 const kBufferCb = Symbol('kBufferCb');
 
-function handleWriteReq(req, data, encoding) {
-  const { handle } = req;
-
-  switch (encoding) {
-    case 'buffer':
-    {
-      const ret = handle.writeBuffer(req, data);
-      if (streamBaseState[kLastWriteWasAsync])
-        req.buffer = data;
-      return ret;
-    }
-    case 'latin1':
-    case 'binary':
-      return handle.writeLatin1String(req, data);
-    case 'utf8':
-    case 'utf-8':
-      return handle.writeUtf8String(req, data);
-    case 'ascii':
-      return handle.writeAsciiString(req, data);
-    case 'ucs2':
-    case 'ucs-2':
-    case 'utf16le':
-    case 'utf-16le':
-      return handle.writeUcs2String(req, data);
-    default:
-    {
-      const buffer = Buffer.from(data, encoding);
-      const ret = handle.writeBuffer(req, buffer);
-      if (streamBaseState[kLastWriteWasAsync])
-        req.buffer = buffer;
-      return ret;
-    }
-  }
-}
-
 function onWriteComplete(status) {
   debug('onWriteComplete', status, this.error);
 
@@ -105,21 +69,8 @@ function onWriteComplete(status) {
     this.callback(null);
 }
 
-function createWriteWrap(handle, callback) {
-  const req = new WriteWrap();
-
-  req.handle = handle;
-  req.oncomplete = onWriteComplete;
-  req.async = false;
-  req.bytes = 0;
-  req.buffer = null;
-  req.callback = callback;
-
-  return req;
-}
-
 function writevGeneric(self, data, cb) {
-  const req = createWriteWrap(self[kHandle], cb);
+  const handle = self[kHandle];
   const allBuffers = data.allBuffers;
   let chunks;
   if (allBuffers) {
@@ -134,33 +85,87 @@ function writevGeneric(self, data, cb) {
       chunks[i * 2 + 1] = entry.encoding;
     }
   }
-  const err = req.handle.writev(req, chunks, allBuffers);
-
-  // Retain chunks
-  if (err === 0) req._chunks = chunks;
+  const ret = handle.writev(null, chunks, allBuffers);
 
-  afterWriteDispatched(req, err, cb);
-  return req;
+  return afterWriteDispatched(handle, ret, chunks, cb);
 }
 
 function writeGeneric(self, data, encoding, cb) {
-  const req = createWriteWrap(self[kHandle], cb);
-  const err = handleWriteReq(req, data, encoding);
+  const handle = self[kHandle];
+  let ret;
+  let buffer = null;
 
-  afterWriteDispatched(req, err, cb);
-  return req;
+  switch (encoding) {
+    case 'buffer':
+      buffer = data;
+      ret = handle.writeBuffer(null, data);
+      break;
+    case 'latin1':
+    case 'binary':
+      ret = handle.writeLatin1String(null, data);
+      break;
+    case 'utf8':
+    case 'utf-8':
+      ret = handle.writeUtf8String(null, data);
+      break;
+    case 'ascii':
+      ret = handle.writeAsciiString(null, data);
+      break;
+    case 'ucs2':
+    case 'ucs-2':
+    case 'utf16le':
+    case 'utf-16le':
+      ret = handle.writeUcs2String(null, data);
+      break;
+    default:
+      buffer = Buffer.from(data, encoding);
+      ret = handle.writeBuffer(null, buffer);
+      break;
+  }
+
+  return afterWriteDispatched(handle, ret, buffer, cb);
 }
 
-function afterWriteDispatched(req, err, cb) {
-  req.bytes = streamBaseState[kBytesWritten];
-  req.async = !!streamBaseState[kLastWriteWasAsync];
+// `ret` is either a numeric error code (when the write - or its failure -
+// completed synchronously and no write request was created) or the WriteWrap
+// object of a dispatched write.
+function afterWriteDispatched(handle, ret, buffer, cb) {
+  const bytes = streamBaseState[kBytesWritten];
+
+  if (typeof ret === 'number') {
+    // The write (or its failure) completed synchronously.
+    if (ret !== 0)
+      cb(new ErrnoException(ret, 'write'));
+    else if (typeof cb === 'function')
+      cb();
+    return { async: false, bytes };
+  }
 
-  if (err !== 0)
-    return cb(new ErrnoException(err, 'write', req.error));
+  const req = ret;
+  const err = streamBaseState[kLastWriteErr];
+  if (err !== 0) {
+    cb(new ErrnoException(err, 'write', req.error));
+    return { async: false, bytes };
+  }
 
-  if (!req.async && typeof req.callback === 'function') {
-    req.callback();
+  req.handle = handle;
+  req.oncomplete = onWriteComplete;
+  req.callback = cb;
+  req.async = true;
+  req.bytes = bytes;
+  // Retain the data (or chunks) being written until the write completes.
+  req.buffer = buffer;
+
+  // If the write completed synchronously inside the write call, before
+  // `oncomplete` could be attached, the completion status has been recorded;
+  // deliver it now.
+  const status = req.writeStatus;
+  if (status !== null) {
+    req.writeStatus = null;
+    req.oncomplete(status);
   }
+
+  return req;
 }
 
 function onStreamRead(arrayBuffer) {

src/env_properties.h

@@ -404,6 +404,7 @@
   V(writable_string, "writable")                                               \
   V(write_host_object_string, "_writeHostObject")                              \
   V(write_queue_size_string, "writeQueueSize")                                 \
+  V(write_status_string, "writeStatus")                                        \
   V(zlib_string, "zlib")                                                       \
   V(zstd_string, "zstd")
 

src/stream_base.cc

@@ -175,17 +175,37 @@ int StreamBase::Shutdown(const FunctionCallbackInfo<Value>& args) {
 void StreamBase::SetWriteResult(const StreamWriteResult& res) {
   env_->stream_base_state()[kBytesWritten] = res.bytes;
   env_->stream_base_state()[kLastWriteWasAsync] = res.async;
+  env_->stream_base_state()[kLastWriteErr] = res.err;
+}
+
+// Finish a JS-initiated write. When the caller did not pass a request
+// object (`lazy_req`), the write wrap object exists only if the write did
+// not complete synchronously; hand it to JS so that it can attach its
+// completion callback. Otherwise the numeric error code (via JSMethod) is
+// all JS needs.
+int StreamBase::FinishWrite(const v8::FunctionCallbackInfo<v8::Value>& args,
+                            const StreamWriteResult& res,
+                            bool lazy_req) {
+  SetWriteResult(res);
+  if (lazy_req && res.wrap_obj) {
+    args.GetReturnValue().Set(res.wrap_obj->object());
+    return kReturnValueSet;
+  }
+  return res.err;
 }
 
 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());
 
-  Local<Object> req_wrap_obj = args[0].As<Object>();
+  // When no request object is passed in, one is created by Write() only if
+  // the write does not complete synchronously; see FinishWrite().
+  const bool lazy_req = !args[0]->IsObject();
+  Local<Object> req_wrap_obj;
+  if (!lazy_req) req_wrap_obj = args[0].As<Object>();
   Local<Array> chunks = args[1].As<Array>();
   bool all_buffers = args[2]->IsTrue();
 
@@ -287,20 +307,19 @@ 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->SetBackingStore(std::move(bs));
-  return res.err;
+  return FinishWrite(args, res, lazy_req);
 }
 
-
 int StreamBase::WriteBuffer(const FunctionCallbackInfo<Value>& args) {
-  CHECK(args[0]->IsObject());
   CHECK(args[1]->IsUint8Array());
 
   Environment* env = Environment::GetCurrent(args);
 
-  Local<Object> req_wrap_obj = args[0].As<Object>();
+  bool lazy_req = !args[0]->IsObject();
+  Local<Object> req_wrap_obj;
+  if (!lazy_req) req_wrap_obj = args[0].As<Object>();
   uv_buf_t buf;
   buf.base = Buffer::Data(args[1]);
   buf.len = Buffer::Length(args[1]);
@@ -310,6 +329,16 @@ int StreamBase::WriteBuffer(const FunctionCallbackInfo<Value>& args) {
   if (args[2]->IsObject() && IsIPCPipe()) {
     Local<Object> send_handle_obj = args[2].As<Object>();
 
+    if (lazy_req) {
+      // Sending a handle requires a request object up front to reference it.
+      if (!env->write_wrap_template()
+               ->NewInstance(env->context())
+               .ToLocal(&req_wrap_obj)) {
+        return UV_EBUSY;
+      }
+      StreamReq::ResetObject(req_wrap_obj);
+    }
+
     HandleWrap* wrap;
     ASSIGN_OR_RETURN_UNWRAP(&wrap, send_handle_obj, UV_EINVAL);
     send_handle = reinterpret_cast<uv_stream_t*>(wrap->GetHandle());
@@ -323,20 +352,18 @@ int StreamBase::WriteBuffer(const FunctionCallbackInfo<Value>& args) {
   }
 
   StreamWriteResult res = Write(&buf, 1, send_handle, req_wrap_obj);
-  SetWriteResult(res);
-
-  return res.err;
+  return FinishWrite(args, res, lazy_req);
 }
 
-
 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());
 
-  Local<Object> req_wrap_obj = args[0].As<Object>();
+  const bool lazy_req = !args[0]->IsObject();
+  Local<Object> req_wrap_obj;
+  if (!lazy_req) req_wrap_obj = args[0].As<Object>();
   Local<String> string = args[1].As<String>();
   Local<Object> send_handle_obj;
   if (args[2]->IsObject())
@@ -417,6 +444,16 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
   uv_stream_t* send_handle = nullptr;
 
   if (IsIPCPipe() && !send_handle_obj.IsEmpty()) {
+    if (lazy_req && req_wrap_obj.IsEmpty()) {
+      // Sending a handle requires a request object up front to reference it.
+      if (!env->write_wrap_template()
+               ->NewInstance(env->context())
+               .ToLocal(&req_wrap_obj)) {
+        return UV_EBUSY;
+      }
+      StreamReq::ResetObject(req_wrap_obj);
+    }
+
     HandleWrap* wrap;
     ASSIGN_OR_RETURN_UNWRAP(&wrap, send_handle_obj, UV_EINVAL);
     send_handle = reinterpret_cast<uv_stream_t*>(wrap->GetHandle());
@@ -432,11 +469,10 @@ int StreamBase::WriteString(const FunctionCallbackInfo<Value>& args) {
   StreamWriteResult res = Write(&buf, 1, send_handle, req_wrap_obj, try_write);
   res.bytes += synchronously_written;
 
-  SetWriteResult(res);
   if (res.wrap != nullptr)
     res.wrap->SetBackingStore(std::move(bs));
 
-  return res.err;
+  return FinishWrite(args, res, lazy_req);
 }
 
 
@@ -662,7 +698,8 @@ void StreamBase::JSMethod(const FunctionCallbackInfo<Value>& args) {
   if (!wrap->IsAlive()) return args.GetReturnValue().Set(UV_EINVAL);
 
   AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope(wrap->GetAsyncWrap());
-  args.GetReturnValue().Set((wrap->*Method)(args));
+  int ret = (wrap->*Method)(args);
+  if (ret != kReturnValueSet) args.GetReturnValue().Set(ret);
 }
 
 int StreamResource::DoTryWrite(uv_buf_t** bufs, size_t* count) {
@@ -773,20 +810,50 @@ void ReportWritesToJSStreamListener::OnStreamAfterReqFinished(
   CHECK(!async_wrap->persistent().IsEmpty());
   Local<Object> req_wrap_obj = async_wrap->object();
 
+  Local<Value> oncomplete;
+  if (!req_wrap_obj->Get(env->context(), env->oncomplete_string())
+           .ToLocal(&oncomplete)) {
+    return;
+  }
+
+  const char* msg = stream->Error();
+
+  if (!oncomplete->IsFunction()) {
+    // The completion callback has not been attached yet: the write finished
+    // synchronously inside the JS write call, before the request object was
+    // returned to JS. Record the status so that JS can replay the callback.
+    if (req_wrap_obj
+            ->Set(env->context(),
+                  env->write_status_string(),
+                  Integer::New(env->isolate(), status))
+            .IsNothing()) {
+      return;
+    }
+    if (msg != nullptr) {
+      if (req_wrap_obj
+              ->Set(env->context(),
+                    env->error_string(),
+                    OneByteString(env->isolate(), msg))
+              .IsNothing()) {
+        return;
+      }
+      stream->ClearError();
+    }
+    return;
+  }
+
   Local<Value> argv[] = {
     Integer::New(env->isolate(), status),
     stream->GetObject(),
     Undefined(env->isolate())
   };
 
-  const char* msg = stream->Error();
   if (msg != nullptr) {
     argv[2] = OneByteString(env->isolate(), msg);
     stream->ClearError();
   }
 
-  if (req_wrap_obj->Has(env->context(), env->oncomplete_string()).FromJust())
-    async_wrap->MakeCallback(env->oncomplete_string(), arraysize(argv), argv);
+  async_wrap->MakeCallback(oncomplete.As<Function>(), arraysize(argv), argv);
 }
 
 void ReportWritesToJSStreamListener::OnStreamAfterWrite(