src: unify ReqWrap libuv calling

This allows easier tracking of whether there are active `ReqWrap`s.

Many thanks for Stephen Belanger for reviewing the original version of
this commit in the Ayo.js project.

Refs: ayojs/ayo#85
PR-URL: #19377
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>
Reviewed-By: James M Snell <jasnell@gmail.com>

src/cares_wrap.cc

@@ -515,7 +515,7 @@ ChannelWrap::~ChannelWrap() {
 void ChannelWrap::CleanupTimer() {
   if (timer_handle_ == nullptr) return;
 
-  env()->CloseHandle(timer_handle_, [](uv_timer_t* handle){ delete handle; });
+  env()->CloseHandle(timer_handle_, [](uv_timer_t* handle) { delete handle; });
   timer_handle_ = nullptr;
 }
 
@@ -1927,13 +1927,11 @@ void GetAddrInfo(const FunctionCallbackInfo<Value>& args) {
   hints.ai_socktype = SOCK_STREAM;
   hints.ai_flags = flags;
 
-  int err = uv_getaddrinfo(env->event_loop(),
-                           req_wrap->req(),
-                           AfterGetAddrInfo,
-                           *hostname,
-                           nullptr,
-                           &hints);
-  req_wrap->Dispatched();
+  int err = req_wrap->Dispatch(uv_getaddrinfo,
+                               AfterGetAddrInfo,
+                               *hostname,
+                               nullptr,
+                               &hints);
   if (err)
     delete req_wrap;
 
@@ -1957,12 +1955,10 @@ void GetNameInfo(const FunctionCallbackInfo<Value>& args) {
 
   GetNameInfoReqWrap* req_wrap = new GetNameInfoReqWrap(env, req_wrap_obj);
 
-  int err = uv_getnameinfo(env->event_loop(),
-                           req_wrap->req(),
-                           AfterGetNameInfo,
-                           (struct sockaddr*)&addr,
-                           NI_NAMEREQD);
-  req_wrap->Dispatched();
+  int err = req_wrap->Dispatch(uv_getnameinfo,
+                               AfterGetNameInfo,
+                               reinterpret_cast<struct sockaddr*>(&addr),
+                               NI_NAMEREQD);
   if (err)
     delete req_wrap;
 

src/node_file.cc

@@ -89,6 +89,11 @@ using v8::Value;
   TRACE_EVENT_END(TRACING_CATEGORY_NODE2(fs, sync), TRACE_NAME(syscall),   \
   ##__VA_ARGS__);
 
+// We sometimes need to convert a C++ lambda function to a raw C-style function.
+// This is helpful, because ReqWrap::Dispatch() does not recognize lambda
+// functions, and thus does not wrap them properly.
+typedef void(*uv_fs_callback_t)(uv_fs_t*);
+
 // The FileHandle object wraps a file descriptor and will close it on garbage
 // collection if necessary. If that happens, a process warning will be
 // emitted (or a fatal exception will occur if the fd cannot be closed.)
@@ -216,7 +221,7 @@ inline MaybeLocal<Promise> FileHandle::ClosePromise() {
   if (!closed_ && !closing_) {
     closing_ = true;
     CloseReq* req = new CloseReq(env(), promise, object());
-    auto AfterClose = [](uv_fs_t* req) {
+    auto AfterClose = uv_fs_callback_t{[](uv_fs_t* req) {
       CloseReq* close = static_cast<CloseReq*>(req->data);
       CHECK_NE(close, nullptr);
       close->file_handle()->AfterClose();
@@ -227,9 +232,8 @@ inline MaybeLocal<Promise> FileHandle::ClosePromise() {
         close->Resolve();
       }
       delete close;
-    };
-    req->Dispatched();
-    int ret = uv_fs_close(env()->event_loop(), req->req(), fd_, AfterClose);
+    }};
+    int ret = req->Dispatch(uv_fs_close, fd_, AfterClose);
     if (ret < 0) {
       req->Reject(UVException(isolate, ret, "close"));
       delete req;
@@ -309,17 +313,15 @@ int FileHandle::ReadStart() {
     recommended_read = read_length_;
 
   read_wrap->buffer_ = EmitAlloc(recommended_read);
-  read_wrap->Dispatched();
 
   current_read_ = std::move(read_wrap);
 
-  uv_fs_read(env()->event_loop(),
-             current_read_->req(),
-             fd_,
-             &current_read_->buffer_,
-             1,
-             read_offset_,
-             [](uv_fs_t* req) {
+  current_read_->Dispatch(uv_fs_read,
+                          fd_,
+                          &current_read_->buffer_,
+                          1,
+                          read_offset_,
+                          uv_fs_callback_t{[](uv_fs_t* req) {
     FileHandle* handle;
     {
       FileHandleReadWrap* req_wrap = FileHandleReadWrap::from_req(req);
@@ -342,8 +344,10 @@ int FileHandle::ReadStart() {
     // once we’re exiting the current scope.
     constexpr size_t wanted_freelist_fill = 100;
     auto& freelist = handle->env()->file_handle_read_wrap_freelist();
-    if (freelist.size() < wanted_freelist_fill)
+    if (freelist.size() < wanted_freelist_fill) {
+      read_wrap->Reset();
       freelist.emplace_back(std::move(read_wrap));
+    }
 
     if (result >= 0) {
       // Read at most as many bytes as we originally planned to.
@@ -370,7 +374,7 @@ int FileHandle::ReadStart() {
     // Start over, if EmitRead() didn’t tell us to stop.
     if (handle->reading_)
       handle->ReadStart();
-  });
+  }});
 
   return 0;
 }
@@ -389,8 +393,7 @@ ShutdownWrap* FileHandle::CreateShutdownWrap(Local<Object> object) {
 int FileHandle::DoShutdown(ShutdownWrap* req_wrap) {
   FileHandleCloseWrap* wrap = static_cast<FileHandleCloseWrap*>(req_wrap);
   closing_ = true;
-  wrap->Dispatched();
-  uv_fs_close(env()->event_loop(), wrap->req(), fd_, [](uv_fs_t* req) {
+  wrap->Dispatch(uv_fs_close, fd_, uv_fs_callback_t{[](uv_fs_t* req) {
     FileHandleCloseWrap* wrap = static_cast<FileHandleCloseWrap*>(
         FileHandleCloseWrap::from_req(req));
     FileHandle* handle = static_cast<FileHandle*>(wrap->stream());
@@ -399,7 +402,7 @@ int FileHandle::DoShutdown(ShutdownWrap* req_wrap) {
     int result = req->result;
     uv_fs_req_cleanup(req);
     wrap->Done(result);
-  });
+  }});
 
   return 0;
 }
@@ -616,8 +619,7 @@ inline FSReqBase* AsyncDestCall(Environment* env,
     enum encoding enc, uv_fs_cb after, Func fn, Args... fn_args) {
   CHECK_NE(req_wrap, nullptr);
   req_wrap->Init(syscall, dest, len, enc);
-  int err = fn(env->event_loop(), req_wrap->req(), fn_args..., after);
-  req_wrap->Dispatched();
+  int err = req_wrap->Dispatch(fn, fn_args..., after);
   if (err < 0) {
     uv_fs_t* uv_req = req_wrap->req();
     uv_req->result = err;

src/pipe_wrap.cc

@@ -224,11 +224,10 @@ void PipeWrap::Connect(const FunctionCallbackInfo<Value>& args) {
 
   ConnectWrap* req_wrap =
       new ConnectWrap(env, req_wrap_obj, AsyncWrap::PROVIDER_PIPECONNECTWRAP);
-  uv_pipe_connect(req_wrap->req(),
-                  &wrap->handle_,
-                  *name,
-                  AfterConnect);
-  req_wrap->Dispatched();
+  req_wrap->Dispatch(uv_pipe_connect,
+                     &wrap->handle_,
+                     *name,
+                     AfterConnect);
 
   args.GetReturnValue().Set(0);  // uv_pipe_connect() doesn't return errors.
 }

src/req_wrap-inl.h

@@ -43,6 +43,110 @@ void ReqWrap<T>::Cancel() {
   uv_cancel(reinterpret_cast<uv_req_t*>(&req_));
 }
 
+// Below is dark template magic designed to invoke libuv functions that
+// initialize uv_req_t instances in a unified fashion, to allow easier
+// tracking of active/inactive requests.
+
+// Invoke a generic libuv function that initializes uv_req_t instances.
+// This is, unfortunately, necessary since they come in three different
+// variants that can not all be invoked in the same way:
+// - int uv_foo(uv_loop_t* loop, uv_req_t* request, ...);
+// - int uv_foo(uv_req_t* request, ...);
+// - void uv_foo(uv_req_t* request, ...);
+template <typename ReqT, typename T>
+struct CallLibuvFunction;
+
+// Detect `int uv_foo(uv_loop_t* loop, uv_req_t* request, ...);`.
+template <typename ReqT, typename... Args>
+struct CallLibuvFunction<ReqT, int(*)(uv_loop_t*, ReqT*, Args...)> {
+  using T = int(*)(uv_loop_t*, ReqT*, Args...);
+  template <typename... PassedArgs>
+  static int Call(T fn, uv_loop_t* loop, ReqT* req, PassedArgs... args) {
+    return fn(loop, req, args...);
+  }
+};
+
+// Detect `int uv_foo(uv_req_t* request, ...);`.
+template <typename ReqT, typename... Args>
+struct CallLibuvFunction<ReqT, int(*)(ReqT*, Args...)> {
+  using T = int(*)(ReqT*, Args...);
+  template <typename... PassedArgs>
+  static int Call(T fn, uv_loop_t* loop, ReqT* req, PassedArgs... args) {
+    return fn(req, args...);
+  }
+};
+
+// Detect `void uv_foo(uv_req_t* request, ...);`.
+template <typename ReqT, typename... Args>
+struct CallLibuvFunction<ReqT, void(*)(ReqT*, Args...)> {
+  using T = void(*)(ReqT*, Args...);
+  template <typename... PassedArgs>
+  static int Call(T fn, uv_loop_t* loop, ReqT* req, PassedArgs... args) {
+    fn(req, args...);
+    return 0;
+  }
+};
+
+// This is slightly darker magic: This template is 'applied' to each parameter
+// passed to the libuv function. If the parameter type (aka `T`) is a
+// function type, it is assumed that this it is the request callback, and a
+// wrapper that calls the original callback is created.
+// If not, the parameter is passed through verbatim.
+template <typename ReqT, typename T>
+struct MakeLibuvRequestCallback {
+  static T For(ReqWrap<ReqT>* req_wrap, T v) {
+    static_assert(!std::is_function<T>::value,
+                  "MakeLibuvRequestCallback missed a callback");
+    return v;
+  }
+};
+
+// Match the `void callback(uv_req_t*, ...);` signature that all libuv
+// callbacks use.
+template <typename ReqT, typename... Args>
+struct MakeLibuvRequestCallback<ReqT, void(*)(ReqT*, Args...)> {
+  using F = void(*)(ReqT* req, Args... args);
+
+  static void Wrapper(ReqT* req, Args... args) {
+    ReqWrap<ReqT>* req_wrap = ContainerOf(&ReqWrap<ReqT>::req_, req);
+    F original_callback = reinterpret_cast<F>(req_wrap->original_callback_);
+    original_callback(req, args...);
+  }
+
+  static F For(ReqWrap<ReqT>* req_wrap, F v) {
+    CHECK_EQ(req_wrap->original_callback_, nullptr);
+    req_wrap->original_callback_ =
+        reinterpret_cast<typename ReqWrap<ReqT>::callback_t>(v);
+    return Wrapper;
+  }
+};
+
+template <typename T>
+template <typename LibuvFunction, typename... Args>
+int ReqWrap<T>::Dispatch(LibuvFunction fn, Args... args) {
+  Dispatched();
+
+  // This expands as:
+  //
+  // return fn(env()->event_loop(), req(), arg1, arg2, Wrapper, arg3, ...)
+  //           ^                                       ^        ^
+  //           |                                       |        |
+  //           \-- Omitted if `fn` has no              |        |
+  //               first `uv_loop_t*` argument         |        |
+  //                                                   |        |
+  //     A function callback whose first argument      |        |
+  //     matches the libuv request type is replaced ---/        |
+  //     by the `Wrapper` method defined above                  |
+  //                                                            |
+  //            Other (non-function) arguments are passed  -----/
+  //            through verbatim
+  return CallLibuvFunction<T, LibuvFunction>::Call(
+      fn,
+      env()->event_loop(),
+      req(),
+      MakeLibuvRequestCallback<T, Args>::For(this, args)...);
+}
+
 }  // namespace node
 
 #endif  // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS

src/req_wrap.h

@@ -17,17 +17,28 @@ class ReqWrap : public AsyncWrap {
                  v8::Local<v8::Object> object,
                  AsyncWrap::ProviderType provider);
   inline ~ReqWrap() override;
-  inline void Dispatched();  // Call this after the req has been dispatched.
+  // Call this after the req has been dispatched, if that did not already
+  // happen by using Dispatch().
+  inline void Dispatched();
   T* req() { return &req_; }
   inline void Cancel();
 
   static ReqWrap* from_req(T* req);
 
+  template <typename LibuvFunction, typename... Args>
+  inline int Dispatch(LibuvFunction fn, Args... args);
+
  private:
   friend class Environment;
   friend int GenDebugSymbols();
+  template <typename ReqT, typename U>
+  friend struct MakeLibuvRequestCallback;
+
   ListNode<ReqWrap> req_wrap_queue_;
 
+  typedef void (*callback_t)();
+  callback_t original_callback_ = nullptr;
+
  protected:
   // req_wrap_queue_ needs to be at a fixed offset from the start of the class
   // because it is used by ContainerOf to calculate the address of the embedding

src/tcp_wrap.cc

@@ -287,11 +287,10 @@ void TCPWrap::Connect(const FunctionCallbackInfo<Value>& args) {
     AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope(wrap);
     ConnectWrap* req_wrap =
         new ConnectWrap(env, req_wrap_obj, AsyncWrap::PROVIDER_TCPCONNECTWRAP);
-    err = uv_tcp_connect(req_wrap->req(),
-                         &wrap->handle_,
-                         reinterpret_cast<const sockaddr*>(&addr),
-                         AfterConnect);
-    req_wrap->Dispatched();
+    err = req_wrap->Dispatch(uv_tcp_connect,
+                             &wrap->handle_,
+                             reinterpret_cast<const sockaddr*>(&addr),
+                             AfterConnect);
     if (err)
       delete req_wrap;
   }
@@ -323,11 +322,10 @@ void TCPWrap::Connect6(const FunctionCallbackInfo<Value>& args) {
     AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope(wrap);
     ConnectWrap* req_wrap =
         new ConnectWrap(env, req_wrap_obj, AsyncWrap::PROVIDER_TCPCONNECTWRAP);
-    err = uv_tcp_connect(req_wrap->req(),
-                         &wrap->handle_,
-                         reinterpret_cast<const sockaddr*>(&addr),
-                         AfterConnect);
-    req_wrap->Dispatched();
+    err = req_wrap->Dispatch(uv_tcp_connect,
+                             &wrap->handle_,
+                             reinterpret_cast<const sockaddr*>(&addr),
+                             AfterConnect);
     if (err)
       delete req_wrap;
   }

src/udp_wrap.cc

@@ -380,15 +380,14 @@ void UDPWrap::DoSend(const FunctionCallbackInfo<Value>& args, int family) {
   }
 
   if (err == 0) {
-    err = uv_udp_send(req_wrap->req(),
-                      &wrap->handle_,
-                      *bufs,
-                      count,
-                      reinterpret_cast<const sockaddr*>(&addr),
-                      OnSend);
+    err = req_wrap->Dispatch(uv_udp_send,
+                             &wrap->handle_,
+                             *bufs,
+                             count,
+                             reinterpret_cast<const sockaddr*>(&addr),
+                             OnSend);
   }
 
-  req_wrap->Dispatched();
   if (err)
     delete req_wrap;