src: simplify handles for libuv streams

Instead of passing along the handle object, just set it as a property on the stream handle object and let the read handler grab it from there. PR-URL: #18334 Reviewed-By: James M Snell <[email protected]> Reviewed-By: Anatoli Papirovski <[email protected]> Reviewed-By: Matteo Collina <[email protected]>
nodejs · Feb 26, 2018 · 684684e · 684684e
1 parent cb5ed45
commit 684684e
Show file tree

Hide file tree

Showing 7 changed files with 53 additions and 128 deletions.
diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js
@@ -455,7 +455,10 @@ function setupChannel(target, channel) {
   var jsonBuffer = '';
   var pendingHandle = null;
   channel.buffering = false;
-  channel.onread = function(nread, pool, recvHandle) {
+  channel.pendingHandle = null;
+  channel.onread = function(nread, pool) {
+    const recvHandle = channel.pendingHandle;
+    channel.pendingHandle = null;
     // TODO(bnoordhuis) Check that nread > 0.
     if (pool) {
       if (recvHandle)

diff --git a/src/env.h b/src/env.h
@@ -215,6 +215,7 @@ class ModuleWrap;
   V(owner_string, "owner")                                                    \
   V(parse_error_string, "Parse Error")                                        \
   V(path_string, "path")                                                      \
+  V(pending_handle_string, "pendingHandle")                                   \
   V(pbkdf2_error_string, "PBKDF2 Error")                                      \
   V(pid_string, "pid")                                                        \
   V(pipe_string, "pipe")                                                      \

diff --git a/src/stream_base-inl.h b/src/stream_base-inl.h
@@ -33,9 +33,7 @@ inline StreamListener::~StreamListener() {
 
 inline void StreamListener::PassReadErrorToPreviousListener(ssize_t nread) {
   CHECK_NE(previous_listener_, nullptr);
-  previous_listener_->OnStreamRead(nread,
-                                   uv_buf_init(nullptr, 0),
-                                   UV_UNKNOWN_HANDLE);
+  previous_listener_->OnStreamRead(nread, uv_buf_init(nullptr, 0));
 }
 
 
@@ -85,12 +83,10 @@ inline uv_buf_t StreamResource::EmitAlloc(size_t suggested_size) {
   return listener_->OnStreamAlloc(suggested_size);
 }
 
-inline void StreamResource::EmitRead(ssize_t nread,
-                                     const uv_buf_t& buf,
-                                     uv_handle_type pending) {
+inline void StreamResource::EmitRead(ssize_t nread, const uv_buf_t& buf) {
   if (nread > 0)
     bytes_read_ += static_cast<uint64_t>(nread);
-  listener_->OnStreamRead(nread, buf, pending);
+  listener_->OnStreamRead(nread, buf);
 }
 
 inline void StreamResource::EmitAfterWrite(WriteWrap* w, int status) {

diff --git a/src/stream_base.cc b/src/stream_base.cc
@@ -437,23 +437,17 @@ void StreamBase::AfterWrite(WriteWrap* req_wrap, int status) {
 }
 
 
-void StreamBase::CallJSOnreadMethod(ssize_t nread,
-                                    Local<Object> buf,
-                                    Local<Object> handle) {
+void StreamBase::CallJSOnreadMethod(ssize_t nread, Local<Object> buf) {
   Environment* env = env_;
 
   Local<Value> argv[] = {
     Integer::New(env->isolate(), nread),
-    buf,
-    handle
+    buf
   };
 
   if (argv[1].IsEmpty())
     argv[1] = Undefined(env->isolate());
 
-  if (argv[2].IsEmpty())
-    argv[2] = Undefined(env->isolate());
-
   AsyncWrap* wrap = GetAsyncWrap();
   CHECK_NE(wrap, nullptr);
   wrap->MakeCallback(env->onread_string(), arraysize(argv), argv);
@@ -495,19 +489,6 @@ uv_buf_t StreamListener::OnStreamAlloc(size_t suggested_size) {
   return uv_buf_init(Malloc(suggested_size), suggested_size);
 }
 
-void StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
-  // This cannot be virtual because it is just as valid to override the other
-  // OnStreamRead() callback.
-  CHECK(0 && "OnStreamRead() needs to be implemented");
-}
-
-void StreamListener::OnStreamRead(ssize_t nread,
-                                  const uv_buf_t& buf,
-                                  uv_handle_type pending) {
-  CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
-  OnStreamRead(nread, buf);
-}
-
 
 void EmitToJSStreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   CHECK_NE(stream_, nullptr);

diff --git a/src/stream_base.h b/src/stream_base.h
@@ -150,17 +150,8 @@ class StreamListener {
   // with base nullpptr in case of an error.
   // `nread` is the number of read bytes (which is at most the buffer length),
   // or, if negative, a libuv error code.
-  // The variant with a `uv_handle_type` argument is used by libuv-backed
-  // streams for handle transfers (e.g. passing net.Socket instances between
-  // cluster workers). For all other streams, overriding the simple variant
-  // should be sufficient.
-  // By default, the second variant crashes if `pending` is set and otherwise
-  // calls the simple variant.
   virtual void OnStreamRead(ssize_t nread,
                             const uv_buf_t& buf) = 0;
-  virtual void OnStreamRead(ssize_t nread,
-                            const uv_buf_t& buf,
-                            uv_handle_type pending);
 
   // This is called once a Write has finished. `status` may be 0 or,
   // if negative, a libuv error code.
@@ -229,9 +220,7 @@ class StreamResource {
   uv_buf_t EmitAlloc(size_t suggested_size);
   // Call the current listener's OnStreamRead() method and update the
   // stream's read byte counter.
-  void EmitRead(ssize_t nread,
-                const uv_buf_t& buf = uv_buf_init(nullptr, 0),
-                uv_handle_type pending = UV_UNKNOWN_HANDLE);
+  void EmitRead(ssize_t nread, const uv_buf_t& buf = uv_buf_init(nullptr, 0));
   // Call the current listener's OnStreamAfterWrite() method.
   void EmitAfterWrite(WriteWrap* w, int status);
 
@@ -260,10 +249,7 @@ class StreamBase : public StreamResource {
   virtual bool IsIPCPipe();
   virtual int GetFD();
 
-  void CallJSOnreadMethod(
-      ssize_t nread,
-      v8::Local<v8::Object> buf,
-      v8::Local<v8::Object> handle = v8::Local<v8::Object>());
+  void CallJSOnreadMethod(ssize_t nread, v8::Local<v8::Object> buf);
 
   // These are called by the respective {Write,Shutdown}Wrap class.
   virtual void AfterShutdown(ShutdownWrap* req, int status);

diff --git a/src/stream_wrap.cc b/src/stream_wrap.cc
@@ -93,7 +93,6 @@ LibuvStreamWrap::LibuvStreamWrap(Environment* env,
                  provider),
       StreamBase(env),
       stream_(stream) {
-  PushStreamListener(this);
 }
 
 
@@ -146,7 +145,13 @@ bool LibuvStreamWrap::IsIPCPipe() {
 
 
 int LibuvStreamWrap::ReadStart() {
-  return uv_read_start(stream(), OnAlloc, OnRead);
+  return uv_read_start(stream(), [](uv_handle_t* handle,
+                                    size_t suggested_size,
+                                    uv_buf_t* buf) {
+    static_cast<LibuvStreamWrap*>(handle->data)->OnUvAlloc(suggested_size, buf);
+  }, [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
+    static_cast<LibuvStreamWrap*>(stream->data)->OnUvRead(nread, buf);
+  });
 }
 
 
@@ -155,16 +160,11 @@ int LibuvStreamWrap::ReadStop() {
 }
 
 
-void LibuvStreamWrap::OnAlloc(uv_handle_t* handle,
-                              size_t suggested_size,
-                              uv_buf_t* buf) {
-  LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(handle->data);
-  HandleScope scope(wrap->env()->isolate());
-  Context::Scope context_scope(wrap->env()->context());
-
-  CHECK_EQ(wrap->stream(), reinterpret_cast<uv_stream_t*>(handle));
+void LibuvStreamWrap::OnUvAlloc(size_t suggested_size, uv_buf_t* buf) {
+  HandleScope scope(env()->isolate());
+  Context::Scope context_scope(env()->context());
 
-  *buf = wrap->EmitAlloc(suggested_size);
+  *buf = EmitAlloc(suggested_size);
 }
 
 
@@ -190,64 +190,47 @@ static Local<Object> AcceptHandle(Environment* env, LibuvStreamWrap* parent) {
 }
 
 
-void LibuvStreamWrap::OnStreamRead(ssize_t nread,
-                                   const uv_buf_t& buf,
-                                   uv_handle_type pending) {
-  HandleScope handle_scope(env()->isolate());
+void LibuvStreamWrap::OnUvRead(ssize_t nread, const uv_buf_t* buf) {
+  HandleScope scope(env()->isolate());
   Context::Scope context_scope(env()->context());
-
-  if (nread <= 0) {
-    free(buf.base);
-    if (nread < 0)
-      CallJSOnreadMethod(nread, Local<Object>());
-    return;
-  }
-
-  CHECK_LE(static_cast<size_t>(nread), buf.len);
-
-  Local<Object> pending_obj;
-
-  if (pending == UV_TCP) {
-    pending_obj = AcceptHandle<TCPWrap, uv_tcp_t>(env(), this);
-  } else if (pending == UV_NAMED_PIPE) {
-    pending_obj = AcceptHandle<PipeWrap, uv_pipe_t>(env(), this);
-  } else if (pending == UV_UDP) {
-    pending_obj = AcceptHandle<UDPWrap, uv_udp_t>(env(), this);
-  } else {
-    CHECK_EQ(pending, UV_UNKNOWN_HANDLE);
-  }
-
-  Local<Object> obj = Buffer::New(env(), buf.base, nread).ToLocalChecked();
-  CallJSOnreadMethod(nread, obj, pending_obj);
-}
-
-
-void LibuvStreamWrap::OnRead(uv_stream_t* handle,
-                             ssize_t nread,
-                             const uv_buf_t* buf) {
-  LibuvStreamWrap* wrap = static_cast<LibuvStreamWrap*>(handle->data);
-  HandleScope scope(wrap->env()->isolate());
-  Context::Scope context_scope(wrap->env()->context());
   uv_handle_type type = UV_UNKNOWN_HANDLE;
 
-  if (wrap->is_named_pipe_ipc() &&
-      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(handle)) > 0) {
-    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(handle));
+  if (is_named_pipe_ipc() &&
+      uv_pipe_pending_count(reinterpret_cast<uv_pipe_t*>(stream())) > 0) {
+    type = uv_pipe_pending_type(reinterpret_cast<uv_pipe_t*>(stream()));
   }
 
   // We should not be getting this callback if someone as already called
   // uv_close() on the handle.
-  CHECK_EQ(wrap->persistent().IsEmpty(), false);
+  CHECK_EQ(persistent().IsEmpty(), false);
 
   if (nread > 0) {
-    if (wrap->is_tcp()) {
+    if (is_tcp()) {
       NODE_COUNT_NET_BYTES_RECV(nread);
-    } else if (wrap->is_named_pipe()) {
+    } else if (is_named_pipe()) {
       NODE_COUNT_PIPE_BYTES_RECV(nread);
     }
+
+    Local<Object> pending_obj;
+
+    if (type == UV_TCP) {
+      pending_obj = AcceptHandle<TCPWrap, uv_tcp_t>(env(), this);
+    } else if (type == UV_NAMED_PIPE) {
+      pending_obj = AcceptHandle<PipeWrap, uv_pipe_t>(env(), this);
+    } else if (type == UV_UDP) {
+      pending_obj = AcceptHandle<UDPWrap, uv_udp_t>(env(), this);
+    } else {
+      CHECK_EQ(type, UV_UNKNOWN_HANDLE);
+    }
+
+    if (!pending_obj.IsEmpty()) {
+      object()->Set(env()->context(),
+                    env()->pending_handle_string(),
+                    pending_obj).FromJust();
+    }
   }
 
-  wrap->EmitRead(nread, *buf, type);
+  EmitRead(nread, *buf);
 }
 
 
@@ -373,11 +356,6 @@ void LibuvStreamWrap::AfterUvWrite(uv_write_t* req, int status) {
   req_wrap->Done(status);
 }
 
-
-void LibuvStreamWrap::AfterWrite(WriteWrap* w, int status) {
-  StreamBase::AfterWrite(w, status);
-}
-
 }  // namespace node
 
 NODE_BUILTIN_MODULE_CONTEXT_AWARE(stream_wrap,

diff --git a/src/stream_wrap.h b/src/stream_wrap.h
@@ -33,9 +33,7 @@
 
 namespace node {
 
-class LibuvStreamWrap : public HandleWrap,
-                        public StreamListener,
-                        public StreamBase {
+class LibuvStreamWrap : public HandleWrap, public StreamBase {
  public:
   static void Initialize(v8::Local<v8::Object> target,
                          v8::Local<v8::Value> unused,
@@ -93,30 +91,12 @@ class LibuvStreamWrap : public HandleWrap,
   static void SetBlocking(const v8::FunctionCallbackInfo<v8::Value>& args);
 
   // Callbacks for libuv
-  static void OnAlloc(uv_handle_t* handle,
-                      size_t suggested_size,
-                      uv_buf_t* buf);
+  void OnUvAlloc(size_t suggested_size, uv_buf_t* buf);
+  void OnUvRead(ssize_t nread, const uv_buf_t* buf);
 
-  static void OnRead(uv_stream_t* handle,
-                     ssize_t nread,
-                     const uv_buf_t* buf);
   static void AfterUvWrite(uv_write_t* req, int status);
   static void AfterUvShutdown(uv_shutdown_t* req, int status);
 
-  // Resource interface implementation
-  void OnStreamRead(ssize_t nread,
-                    const uv_buf_t& buf) override {
-    CHECK(0 && "must not be called");
-  }
-  void OnStreamRead(ssize_t nread,
-                    const uv_buf_t& buf,
-                    uv_handle_type pending) override;
-  void OnStreamAfterWrite(WriteWrap* w, int status) override {
-    previous_listener_->OnStreamAfterWrite(w, status);
-  }
-
-  void AfterWrite(WriteWrap* req_wrap, int status) override;
-
   uv_stream_t* const stream_;
 };