diff --git a/benchmark/net/tcp-raw-c2s.js b/benchmark/net/tcp-raw-c2s.js index 5b174f50e81c2f..24e35906c925f6 100644 --- a/benchmark/net/tcp-raw-c2s.js +++ b/benchmark/net/tcp-raw-c2s.js @@ -23,7 +23,11 @@ function main({ dur, len, type }) { TCPConnectWrap, constants: TCPConstants, } = common.binding('tcp_wrap'); - const { WriteWrap } = common.binding('stream_wrap'); + const { + WriteWrap, + kReadBytesOrError, + streamBaseState, + } = common.binding('stream_wrap'); const PORT = common.PORT; const serverHandle = new TCP(TCPConstants.SERVER); @@ -55,9 +59,7 @@ function main({ dur, len, type }) { if (!buffer) fail('read'); - // Don't slice the buffer. The point of this is to isolate, not - // simulate real traffic. - bytes += buffer.byteLength; + bytes += streamBaseState[kReadBytesOrError]; }; clientHandle.readStart(); diff --git a/benchmark/net/tcp-raw-pipe.js b/benchmark/net/tcp-raw-pipe.js index fbaa21b5a3097e..36ddb822e04e97 100644 --- a/benchmark/net/tcp-raw-pipe.js +++ b/benchmark/net/tcp-raw-pipe.js @@ -23,7 +23,12 @@ function main({ dur, len, type }) { TCPConnectWrap, constants: TCPConstants, } = common.binding('tcp_wrap'); - const { WriteWrap } = common.binding('stream_wrap'); + const { + WriteWrap, + kReadBytesOrError, + kArrayBufferOffset, + streamBaseState, + } = common.binding('stream_wrap'); const PORT = common.PORT; function fail(err, syscall) { @@ -50,9 +55,11 @@ function main({ dur, len, type }) { if (!buffer) fail('read'); + const nread = streamBaseState[kReadBytesOrError]; + const offset = streamBaseState[kArrayBufferOffset]; const writeReq = new WriteWrap(); writeReq.async = false; - err = clientHandle.writeBuffer(writeReq, Buffer.from(buffer)); + err = clientHandle.writeBuffer(writeReq, Buffer.from(buffer, offset, nread)); if (err) fail(err, 'write'); @@ -94,7 +101,7 @@ function main({ dur, len, type }) { if (!buffer) fail('read'); - bytes += buffer.byteLength; + bytes += streamBaseState[kReadBytesOrError]; }; connectReq.oncomplete = function(err) { diff --git a/benchmark/net/tcp-raw-s2c.js b/benchmark/net/tcp-raw-s2c.js index 3ca03529fcea9f..a847b8c28dcd59 100644 --- a/benchmark/net/tcp-raw-s2c.js +++ b/benchmark/net/tcp-raw-s2c.js @@ -23,7 +23,11 @@ function main({ dur, len, type }) { TCPConnectWrap, constants: TCPConstants, } = common.binding('tcp_wrap'); - const { WriteWrap } = common.binding('stream_wrap'); + const { + WriteWrap, + kReadBytesOrError, + streamBaseState, + } = common.binding('stream_wrap'); const PORT = common.PORT; const serverHandle = new TCP(TCPConstants.SERVER); @@ -116,9 +120,7 @@ function main({ dur, len, type }) { if (!buffer) fail('read'); - // Don't slice the buffer. The point of this is to isolate, not - // simulate real traffic. - bytes += buffer.byteLength; + bytes += streamBaseState[kReadBytesOrError]; }; clientHandle.readStart(); diff --git a/lib/internal/stream_base_commons.js b/lib/internal/stream_base_commons.js index 6d144f8a0fa696..7bf3a2bab6383d 100644 --- a/lib/internal/stream_base_commons.js +++ b/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) { diff --git a/src/env.cc b/src/env.cc index d6a859e7a35452..98f79e0bdcf41d 100644 --- a/src/env.cc +++ b/src/env.cc @@ -767,6 +767,105 @@ std::unique_ptr Environment::release_managed_buffer( return bs; } +uv_buf_t StreamReadSlab::Allocate(Isolate* isolate, size_t suggested) { + DCHECK_GT(suggested, 0); + // Reads always get the full `suggested` size: handing out a smaller + // remainder would shrink the read() buffer and fragment large reads into + // more system calls, which costs more than the slab tail it saves. + size_t remaining = current_.bs ? current_.size() - current_.offset : 0; + if (remaining < suggested) { + // Retire the current slab; it stays alive through `retired_` if reads + // are still pending on it, or through JS views over it otherwise. + if (current_.bs && current_.pending > 0) + retired_.push_back(std::move(current_)); + current_ = Slab(); + std::unique_ptr bs = ArrayBuffer::NewBackingStore( + isolate, + std::max(kSlabSize, suggested), + BackingStoreInitializationMode::kUninitialized); + current_.bs = std::move(bs); + } + char* base = current_.data() + current_.offset; + current_.offset += suggested; + current_.last_base = base; + current_.last_end = current_.offset; + current_.pending++; + return uv_buf_init(base, suggested); +} + +StreamReadSlab::Slab* StreamReadSlab::FindSlab(const char* base) { + if (current_.Contains(base)) return ¤t_; + for (Slab& slab : retired_) + if (slab.Contains(base)) return &slab; + return nullptr; +} + +void StreamReadSlab::CompleteReservation(Slab* slab, + const uv_buf_t& buf, + size_t used) { + DCHECK_GT(slab->pending, 0); + slab->pending--; + if (buf.base == slab->last_base && slab->offset == slab->last_end) { + // This was the most recent reservation and nothing was reserved after + // it: rewind the unused remainder so it can be reserved again. + slab->offset = (buf.base - slab->data()) + used; + slab->last_end = slab->offset; + } + if (slab != ¤t_ && slab->pending == 0) { + for (auto it = retired_.begin(); it != retired_.end(); ++it) { + if (&*it == slab) { + retired_.erase(it); + break; + } + } + } +} + +bool StreamReadSlab::Commit(Isolate* isolate, + const uv_buf_t& buf, + size_t nread, + Local* ab, + size_t* offset) { + Slab* slab = FindSlab(buf.base); + if (slab == nullptr) return false; + DCHECK_LE(nread, buf.len); + + // A partial read would leave the rest of its reservation as waste once + // the slab retires - memory that counts towards V8's external memory and + // drives up GC frequency. If most of the reservation would be wasted, + // give the read a right-sized copy instead (as if it had never been read + // into the slab) and return its reservation in full. Reads that (mostly) + // fill their reservation get a zero-copy view into the slab. + if (nread < buf.len - buf.len / 4 && buf.base == slab->last_base && + slab->offset == slab->last_end) { + std::unique_ptr bs = ArrayBuffer::NewBackingStore( + isolate, nread, BackingStoreInitializationMode::kUninitialized); + memcpy(bs->Data(), buf.base, nread); + *ab = ArrayBuffer::New(isolate, std::move(bs)); + *offset = 0; + CompleteReservation(slab, buf, 0); + return true; + } + + *offset = buf.base - slab->data(); + if (slab->ab.IsEmpty()) { + *ab = ArrayBuffer::New(isolate, slab->bs); + slab->ab.Reset(isolate, *ab); + } else { + *ab = slab->ab.Get(isolate); + } + CompleteReservation(slab, buf, nread); + return true; +} + +bool StreamReadSlab::Release(const uv_buf_t& buf) { + if (buf.base == nullptr) return true; + Slab* slab = FindSlab(buf.base); + if (slab == nullptr) return false; + CompleteReservation(slab, buf, 0); + return true; +} + std::string Environment::GetExecPath(const std::vector& argv) { char exec_path_buf[2 * PATH_MAX]; size_t exec_path_len = sizeof(exec_path_buf); diff --git a/src/env.h b/src/env.h index c2caf979023811..980c5b9b276f58 100644 --- a/src/env.h +++ b/src/env.h @@ -266,6 +266,63 @@ struct ContextInfo { class EnabledDebugList; +// A bump allocator for stream read buffers. Reads reserve a chunk of the +// current slab and, once completed, are handed to JS as a view (ArrayBuffer + +// offset) over the slab, so that no per-read allocation or copy is needed. +// Unused reservation space is rewound when a read returns fewer bytes than +// were reserved. Slabs with reads still pending when a new slab is started +// (possible when multiple reads are in flight, e.g. on Windows) are kept +// alive in `retired_` until those reads complete. +class StreamReadSlab { + public: + // Sized to match the read buffer size that libuv suggests for stream + // reads: a slab typically serves a single large read (still avoiding the + // copy that right-sizing the buffer would need), or many small ones. + // Larger slabs amortize allocations further, but stay alive (pinned by + // chunk views) long enough to be promoted to V8's old generation, where + // their external memory is only reclaimed by major GCs. + static constexpr size_t kSlabSize = 64 * 1024; + + // Reserve `suggested` bytes. Starts a new slab if the current one does + // not have enough space left. + uv_buf_t Allocate(v8::Isolate* isolate, size_t suggested); + // Commit a completed read of `nread` bytes into the buffer previously + // returned by Allocate(), rewinding the unused remainder of the + // reservation if possible. Returns the slab's ArrayBuffer and the offset + // of `buf.base` within it. Returns false if the buffer was not allocated + // from this slab (e.g. reads rerouted from another stream listener). + bool Commit(v8::Isolate* isolate, + const uv_buf_t& buf, + size_t nread, + v8::Local* ab, + size_t* offset); + // Return an unused reservation (failed or empty read). Returns false if + // the buffer was not allocated from this slab. + bool Release(const uv_buf_t& buf); + + private: + struct Slab { + std::shared_ptr bs; + v8::Global ab; + size_t offset = 0; // Bump pointer. + size_t pending = 0; // Reservations not yet committed/released. + char* last_base = nullptr; // Most recent reservation... + size_t last_end = 0; // ...and the bump pointer after it. + + char* data() const { return static_cast(bs->Data()); } + size_t size() const { return bs->ByteLength(); } + bool Contains(const char* p) const { + return bs && p >= data() && p < data() + size(); + } + }; + + Slab* FindSlab(const char* base); + void CompleteReservation(Slab* slab, const uv_buf_t& buf, size_t used); + + Slab current_; + std::vector retired_; +}; + namespace per_process { extern std::shared_ptr system_environment; } @@ -1042,6 +1099,10 @@ class Environment final : public MemoryRetainer { uv_buf_t allocate_managed_buffer(const size_t suggested_size); std::unique_ptr release_managed_buffer(const uv_buf_t& buf); + StreamReadSlab& stream_read_slab() { + return stream_read_slab_; + } + void AddUnmanagedFd(int fd); void RemoveUnmanagedFd(int fd); @@ -1257,6 +1318,9 @@ class Environment final : public MemoryRetainer { std::unordered_map> released_allocated_buffers_; + // Used by EmitToJSStreamListener to allocate stream read buffers. + StreamReadSlab stream_read_slab_; + v8::CpuProfiler* cpu_profiler_ = nullptr; std::vector pending_profiles_; }; diff --git a/src/env_properties.h b/src/env_properties.h index e3179287dce443..91e47cd36a6b99 100644 --- a/src/env_properties.h +++ b/src/env_properties.h @@ -388,7 +388,8 @@ V(wrap_string, "wrap") \ V(writable_string, "writable") \ V(write_host_object_string, "_writeHostObject") \ - V(write_queue_size_string, "writeQueueSize") + V(write_queue_size_string, "writeQueueSize") \ + V(write_status_string, "writeStatus") #define PER_ISOLATE_TEMPLATE_PROPERTIES(V) \ V(a_record_template, v8::DictionaryTemplate) \ diff --git a/src/stream_base.cc b/src/stream_base.cc index 370b8f682eadae..1bd955427e404c 100644 --- a/src/stream_base.cc +++ b/src/stream_base.cc @@ -175,6 +175,23 @@ int StreamBase::Shutdown(const FunctionCallbackInfo& 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& 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& args) { @@ -182,10 +199,13 @@ int StreamBase::Writev(const FunctionCallbackInfo& args) { Isolate* isolate = env->isolate(); Local context = env->context(); - CHECK(args[0]->IsObject()); CHECK(args[1]->IsArray()); - Local req_wrap_obj = args[0].As(); + // 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 req_wrap_obj; + if (!lazy_req) req_wrap_obj = args[0].As(); Local chunks = args[1].As(); bool all_buffers = args[2]->IsTrue(); @@ -287,20 +307,19 @@ int StreamBase::Writev(const FunctionCallbackInfo& 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& args) { - CHECK(args[0]->IsObject()); CHECK(args[1]->IsUint8Array()); Environment* env = Environment::GetCurrent(args); - Local req_wrap_obj = args[0].As(); + bool lazy_req = !args[0]->IsObject(); + Local req_wrap_obj; + if (!lazy_req) req_wrap_obj = args[0].As(); 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& args) { if (args[2]->IsObject() && IsIPCPipe()) { Local send_handle_obj = args[2].As(); + 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(wrap->GetHandle()); @@ -323,20 +352,18 @@ int StreamBase::WriteBuffer(const FunctionCallbackInfo& args) { } StreamWriteResult res = Write(&buf, 1, send_handle, req_wrap_obj); - SetWriteResult(res); - - return res.err; + return FinishWrite(args, res, lazy_req); } - template int StreamBase::WriteString(const FunctionCallbackInfo& args) { Environment* env = Environment::GetCurrent(args); Isolate* isolate = env->isolate(); - CHECK(args[0]->IsObject()); CHECK(args[1]->IsString()); - Local req_wrap_obj = args[0].As(); + const bool lazy_req = !args[0]->IsObject(); + Local req_wrap_obj; + if (!lazy_req) req_wrap_obj = args[0].As(); Local string = args[1].As(); Local send_handle_obj; if (args[2]->IsObject()) @@ -417,6 +444,16 @@ int StreamBase::WriteString(const FunctionCallbackInfo& 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(wrap->GetHandle()); @@ -432,11 +469,10 @@ int StreamBase::WriteString(const FunctionCallbackInfo& 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& 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) { @@ -684,7 +721,7 @@ void StreamResource::ClearError() { uv_buf_t EmitToJSStreamListener::OnStreamAlloc(size_t suggested_size) { CHECK_NOT_NULL(stream_); Environment* env = static_cast(stream_)->stream_env(); - return env->allocate_managed_buffer(suggested_size); + return env->stream_read_slab().Allocate(env->isolate(), suggested_size); } void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { @@ -694,23 +731,33 @@ void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) { Isolate* isolate = env->isolate(); HandleScope handle_scope(isolate); Context::Scope context_scope(env->context()); - std::unique_ptr bs = env->release_managed_buffer(buf_); if (nread <= 0) { + if (!env->stream_read_slab().Release(buf_)) + env->release_managed_buffer(buf_); if (nread < 0) stream->CallJSOnreadMethod(nread, Local()); return; } - CHECK_LE(static_cast(nread), bs->ByteLength()); - if (static_cast(nread) != bs->ByteLength()) { - std::unique_ptr old_bs = std::move(bs); - bs = ArrayBuffer::NewBackingStore( - isolate, nread, BackingStoreInitializationMode::kUninitialized); - memcpy(bs->Data(), old_bs->Data(), nread); + CHECK_LE(static_cast(nread), buf_.len); + size_t offset = 0; + Local ab; + if (!env->stream_read_slab().Commit(isolate, buf_, nread, &ab, &offset)) { + // The buffer was allocated through allocate_managed_buffer() by another + // stream listener (e.g. StreamPipe's) whose read was rerouted here after + // that listener was removed. + std::unique_ptr bs = env->release_managed_buffer(buf_); + CHECK_LE(static_cast(nread), bs->ByteLength()); + if (static_cast(nread) != bs->ByteLength()) { + std::unique_ptr old_bs = std::move(bs); + bs = ArrayBuffer::NewBackingStore( + isolate, nread, BackingStoreInitializationMode::kUninitialized); + memcpy(bs->Data(), old_bs->Data(), nread); + } + ab = ArrayBuffer::New(isolate, std::move(bs)); } - - stream->CallJSOnreadMethod(nread, ArrayBuffer::New(isolate, std::move(bs))); + stream->CallJSOnreadMethod(nread, ab, offset); } @@ -760,20 +807,50 @@ void ReportWritesToJSStreamListener::OnStreamAfterReqFinished( CHECK(!async_wrap->persistent().IsEmpty()); Local req_wrap_obj = async_wrap->object(); + Local 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 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(), arraysize(argv), argv); } void ReportWritesToJSStreamListener::OnStreamAfterWrite( diff --git a/src/stream_base.h b/src/stream_base.h index be00134eb1fc77..b9c36a5bf6d8f3 100644 --- a/src/stream_base.h +++ b/src/stream_base.h @@ -10,6 +10,8 @@ #include "v8.h" +#include // INT_MIN + namespace node { // Forward declarations @@ -405,14 +407,22 @@ class StreamBase : public StreamResource { kArrayBufferOffset, kBytesWritten, kLastWriteWasAsync, + kLastWriteErr, kNumStreamBaseStateFields }; private: + // Sentinel return value for JS methods that have set their own (object) + // return value and must not have it overwritten by JSMethod(). + static constexpr int kReturnValueSet = INT_MIN; + Environment* env_; EmitToJSStreamListener default_listener_; void SetWriteResult(const StreamWriteResult& res); + int FinishWrite(const v8::FunctionCallbackInfo& args, + const StreamWriteResult& res, + bool lazy_req); static void AddAccessor(v8::Isolate* isolate, v8::Local sig, enum v8::PropertyAttribute attributes, diff --git a/src/stream_wrap.cc b/src/stream_wrap.cc index b41f6ac7494730..8c6cc0c0aae8e9 100644 --- a/src/stream_wrap.cc +++ b/src/stream_wrap.cc @@ -95,6 +95,20 @@ void LibuvStreamWrap::Initialize(Local target, Local ww = FunctionTemplate::New(isolate, IsConstructCallCallback); ww->InstanceTemplate()->SetInternalFieldCount(WriteWrap::kInternalFieldCount); + // Pre-create the fields that JS attaches to write requests, so that they + // are in-object properties and the object shape stays monomorphic. + ww->InstanceTemplate()->Set(env->oncomplete_string(), v8::Null(isolate)); + ww->InstanceTemplate()->Set(FIXED_ONE_BYTE_STRING(isolate, "callback"), + v8::Null(isolate)); + ww->InstanceTemplate()->Set(env->handle_string(), v8::Null(isolate)); + ww->InstanceTemplate()->Set(FIXED_ONE_BYTE_STRING(isolate, "async"), + v8::False(isolate)); + ww->InstanceTemplate()->Set(FIXED_ONE_BYTE_STRING(isolate, "bytes"), + v8::Integer::New(isolate, 0)); + ww->InstanceTemplate()->Set(env->buffer_string(), v8::Null(isolate)); + // Slot for a completion that fires before JS attaches `oncomplete`; see + // ReportWritesToJSStreamListener::OnStreamAfterReqFinished(). + ww->InstanceTemplate()->Set(env->write_status_string(), v8::Null(isolate)); ww->Inherit(AsyncWrap::GetConstructorTemplate(env)); SetConstructorFunction(context, target, "WriteWrap", ww); env->set_write_wrap_template(ww->InstanceTemplate()); @@ -103,6 +117,7 @@ void LibuvStreamWrap::Initialize(Local target, NODE_DEFINE_CONSTANT(target, kArrayBufferOffset); NODE_DEFINE_CONSTANT(target, kBytesWritten); NODE_DEFINE_CONSTANT(target, kLastWriteWasAsync); + NODE_DEFINE_CONSTANT(target, kLastWriteErr); target ->Set(context, FIXED_ONE_BYTE_STRING(isolate, "streamBaseState"),