From c52f1266c2e870f0fba089ed6b71215efa8f15bd Mon Sep 17 00:00:00 2001 From: Anna Henningsen Date: Mon, 12 Mar 2018 15:02:28 +0100 Subject: [PATCH] src: enable `StreamPipe` for generic `StreamBase`s Originally landed in the nodejs/quic repo and used there for file sending. Original review: ``` PR-URL: https://github.com/nodejs/quic/pull/150 Reviewed-By: James M Snell ``` --- src/stream_pipe.cc | 56 +++++++++++++++++++++++++++++++++++----------- src/stream_pipe.h | 7 +++--- 2 files changed, 47 insertions(+), 16 deletions(-) diff --git a/src/stream_pipe.cc b/src/stream_pipe.cc index d405c4d5cbeaed..5f7514b1b84790 100644 --- a/src/stream_pipe.cc +++ b/src/stream_pipe.cc @@ -25,7 +25,7 @@ StreamPipe::StreamPipe(StreamBase* source, source->PushStreamListener(&readable_listener_); sink->PushStreamListener(&writable_listener_); - CHECK(sink->HasWantsWrite()); + uses_wants_write_ = sink->HasWantsWrite(); // Set up links between this object and the source/sink objects. // In particular, this makes sure that they are garbage collected as a group, @@ -66,7 +66,8 @@ void StreamPipe::Unpipe(bool is_in_deletion) { is_closed_ = true; is_reading_ = false; source()->RemoveStreamListener(&readable_listener_); - sink()->RemoveStreamListener(&writable_listener_); + if (pending_writes_ == 0) + sink()->RemoveStreamListener(&writable_listener_); if (is_in_deletion) return; @@ -126,13 +127,16 @@ void StreamPipe::ReadableListener::OnStreamRead(ssize_t nread, // EOF or error; stop reading and pass the error to the previous listener // (which might end up in JS). pipe->is_eof_ = true; + // Cache `sink()` here because the previous listener might do things + // that eventually lead to an `Unpipe()` call. + StreamBase* sink = pipe->sink(); stream()->ReadStop(); CHECK_NOT_NULL(previous_listener_); previous_listener_->OnStreamRead(nread, uv_buf_init(nullptr, 0)); // If we’re not writing, close now. Otherwise, we’ll do that in // `OnStreamAfterWrite()`. - if (!pipe->is_writing_) { - pipe->ShutdownWritable(); + if (pipe->pending_writes_ == 0) { + sink->Shutdown(); pipe->Unpipe(); } return; @@ -142,12 +146,13 @@ void StreamPipe::ReadableListener::OnStreamRead(ssize_t nread, } void StreamPipe::ProcessData(size_t nread, AllocatedBuffer&& buf) { + CHECK(uses_wants_write_ || pending_writes_ == 0); uv_buf_t buffer = uv_buf_init(buf.data(), nread); StreamWriteResult res = sink()->Write(&buffer, 1); + pending_writes_++; if (!res.async) { writable_listener_.OnStreamAfterWrite(nullptr, res.err); } else { - is_writing_ = true; is_reading_ = false; res.wrap->SetAllocatedStorage(std::move(buf)); if (source() != nullptr) @@ -155,19 +160,26 @@ void StreamPipe::ProcessData(size_t nread, AllocatedBuffer&& buf) { } } -void StreamPipe::ShutdownWritable() { - sink()->Shutdown(); -} - void StreamPipe::WritableListener::OnStreamAfterWrite(WriteWrap* w, int status) { StreamPipe* pipe = ContainerOf(&StreamPipe::writable_listener_, this); - pipe->is_writing_ = false; + pipe->pending_writes_--; + if (pipe->is_closed_) { + if (pipe->pending_writes_ == 0) { + Environment* env = pipe->env(); + HandleScope handle_scope(env->isolate()); + Context::Scope context_scope(env->context()); + pipe->MakeCallback(env->oncomplete_string(), 0, nullptr).ToLocalChecked(); + stream()->RemoveStreamListener(this); + } + return; + } + if (pipe->is_eof_) { HandleScope handle_scope(pipe->env()->isolate()); InternalCallbackScope callback_scope(pipe, InternalCallbackScope::kSkipTaskQueues); - pipe->ShutdownWritable(); + pipe->sink()->Shutdown(); pipe->Unpipe(); return; } @@ -179,6 +191,10 @@ void StreamPipe::WritableListener::OnStreamAfterWrite(WriteWrap* w, prev->OnStreamAfterWrite(w, status); return; } + + if (!pipe->uses_wants_write_) { + OnStreamWantsWrite(65536); + } } void StreamPipe::WritableListener::OnStreamAfterShutdown(ShutdownWrap* w, @@ -202,6 +218,7 @@ void StreamPipe::WritableListener::OnStreamDestroy() { StreamPipe* pipe = ContainerOf(&StreamPipe::writable_listener_, this); pipe->sink_destroyed_ = true; pipe->is_eof_ = true; + pipe->pending_writes_ = 0; pipe->Unpipe(); } @@ -242,8 +259,7 @@ void StreamPipe::Start(const FunctionCallbackInfo& args) { StreamPipe* pipe; ASSIGN_OR_RETURN_UNWRAP(&pipe, args.Holder()); pipe->is_closed_ = false; - if (pipe->wanted_data_ > 0) - pipe->writable_listener_.OnStreamWantsWrite(pipe->wanted_data_); + pipe->writable_listener_.OnStreamWantsWrite(65536); } void StreamPipe::Unpipe(const FunctionCallbackInfo& args) { @@ -252,6 +268,18 @@ void StreamPipe::Unpipe(const FunctionCallbackInfo& args) { pipe->Unpipe(); } +void StreamPipe::IsClosed(const FunctionCallbackInfo& args) { + StreamPipe* pipe; + ASSIGN_OR_RETURN_UNWRAP(&pipe, args.Holder()); + args.GetReturnValue().Set(pipe->is_closed_); +} + +void StreamPipe::PendingWrites(const FunctionCallbackInfo& args) { + StreamPipe* pipe; + ASSIGN_OR_RETURN_UNWRAP(&pipe, args.Holder()); + args.GetReturnValue().Set(pipe->pending_writes_); +} + namespace { void InitializeStreamPipe(Local target, @@ -266,6 +294,8 @@ void InitializeStreamPipe(Local target, FIXED_ONE_BYTE_STRING(env->isolate(), "StreamPipe"); env->SetProtoMethod(pipe, "unpipe", StreamPipe::Unpipe); env->SetProtoMethod(pipe, "start", StreamPipe::Start); + env->SetProtoMethod(pipe, "isClosed", StreamPipe::IsClosed); + env->SetProtoMethod(pipe, "pendingWrites", StreamPipe::PendingWrites); pipe->Inherit(AsyncWrap::GetConstructorTemplate(env)); pipe->SetClassName(stream_pipe_string); pipe->InstanceTemplate()->SetInternalFieldCount(1); diff --git a/src/stream_pipe.h b/src/stream_pipe.h index 0e155006102eaa..e22abab0115c8a 100644 --- a/src/stream_pipe.h +++ b/src/stream_pipe.h @@ -17,6 +17,8 @@ class StreamPipe : public AsyncWrap { static void New(const v8::FunctionCallbackInfo& args); static void Start(const v8::FunctionCallbackInfo& args); static void Unpipe(const v8::FunctionCallbackInfo& args); + static void IsClosed(const v8::FunctionCallbackInfo& args); + static void PendingWrites(const v8::FunctionCallbackInfo& args); SET_NO_MEMORY_INFO() SET_MEMORY_INFO_NAME(StreamPipe) @@ -26,14 +28,13 @@ class StreamPipe : public AsyncWrap { inline StreamBase* source(); inline StreamBase* sink(); - inline void ShutdownWritable(); - + int pending_writes_ = 0; bool is_reading_ = false; - bool is_writing_ = false; bool is_eof_ = false; bool is_closed_ = true; bool sink_destroyed_ = false; bool source_destroyed_ = false; + bool uses_wants_write_ = false; // Set a default value so that when we’re coming from Start(), we know // that we don’t want to read just yet.