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: https://github.com/ayojs/ayo/pull/85
PR-URL: https://github.com/nodejs/node/pull/19377
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>
Reviewed-By: James M Snell <jasnell@gmail.com>
This commit is contained in:
Anna Henningsen 2017-09-25 21:16:02 +02:00
parent 17e289eca8
commit cac8496c2f
No known key found for this signature in database
GPG Key ID: 9C63F3A6CD2AD8F9
7 changed files with 165 additions and 56 deletions

View File

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

View File

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

View File

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

View File

@ -43,6 +43,110 @@ void ReqWrap<T>::Cancel() {
uv_cancel(reinterpret_cast<uv_req_t*>(&req_)); 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 } // namespace node
#endif // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS #endif // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS

View File

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

View File

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

View File

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