diff --git a/lib/dgram.js b/lib/dgram.js index 26d1e1d11a0e76..dba1bc024272eb 100644 --- a/lib/dgram.js +++ b/lib/dgram.js @@ -232,7 +232,9 @@ Socket.prototype.bind = function(port_, address_ /* , callback */) { this.on('listening', onListening); } - if (port instanceof UDP) { + if (port !== null && + typeof port === 'object' && + typeof port.recvStart === 'function') { replaceHandle(this, port); startListening(this); return this; diff --git a/src/udp_wrap.cc b/src/udp_wrap.cc index 8b6a1cdea1bffd..d714cfa902d5f6 100644 --- a/src/udp_wrap.cc +++ b/src/udp_wrap.cc @@ -69,18 +69,57 @@ SendWrap::SendWrap(Environment* env, } -inline bool SendWrap::have_callback() const { +bool SendWrap::have_callback() const { return have_callback_; } +UDPListener::~UDPListener() { + if (wrap_ != nullptr) + wrap_->set_listener(nullptr); +} + +UDPWrapBase::~UDPWrapBase() { + set_listener(nullptr); +} + +UDPListener* UDPWrapBase::listener() const { + CHECK_NOT_NULL(listener_); + return listener_; +} + +void UDPWrapBase::set_listener(UDPListener* listener) { + if (listener_ != nullptr) + listener_->wrap_ = nullptr; + listener_ = listener; + if (listener_ != nullptr) { + CHECK_NULL(listener_->wrap_); + listener_->wrap_ = this; + } +} + +UDPWrapBase* UDPWrapBase::FromObject(Local obj) { + CHECK_GT(obj->InternalFieldCount(), UDPWrapBase::kUDPWrapBaseField); + return static_cast( + obj->GetAlignedPointerFromInternalField(UDPWrapBase::kUDPWrapBaseField)); +} + +void UDPWrapBase::AddMethods(Environment* env, Local t) { + env->SetProtoMethod(t, "recvStart", RecvStart); + env->SetProtoMethod(t, "recvStop", RecvStop); +} UDPWrap::UDPWrap(Environment* env, Local object) : HandleWrap(env, object, reinterpret_cast(&handle_), AsyncWrap::PROVIDER_UDPWRAP) { + object->SetAlignedPointerInInternalField( + UDPWrapBase::kUDPWrapBaseField, static_cast(this)); + int r = uv_udp_init(env->event_loop(), &handle_); CHECK_EQ(r, 0); // can't fail anyway + + set_listener(this); } @@ -91,7 +130,8 @@ void UDPWrap::Initialize(Local target, Environment* env = Environment::GetCurrent(context); Local t = env->NewFunctionTemplate(New); - t->InstanceTemplate()->SetInternalFieldCount(UDPWrap::kInternalFieldCount); + t->InstanceTemplate()->SetInternalFieldCount( + UDPWrapBase::kInternalFieldCount); Local udpString = FIXED_ONE_BYTE_STRING(env->isolate(), "UDP"); t->SetClassName(udpString); @@ -112,6 +152,7 @@ void UDPWrap::Initialize(Local target, Local(), attributes); + UDPWrapBase::AddMethods(env, t); env->SetProtoMethod(t, "open", Open); env->SetProtoMethod(t, "bind", Bind); env->SetProtoMethod(t, "connect", Connect); @@ -120,8 +161,6 @@ void UDPWrap::Initialize(Local target, env->SetProtoMethod(t, "connect6", Connect6); env->SetProtoMethod(t, "send6", Send6); env->SetProtoMethod(t, "disconnect", Disconnect); - env->SetProtoMethod(t, "recvStart", RecvStart); - env->SetProtoMethod(t, "recvStop", RecvStop); env->SetProtoMethod(t, "getpeername", GetSockOrPeerName); env->SetProtoMethod(t, "getsockname", @@ -220,6 +259,9 @@ void UDPWrap::DoBind(const FunctionCallbackInfo& args, int family) { flags); } + if (err == 0) + wrap->listener()->OnAfterBind(); + args.GetReturnValue().Set(err); } @@ -464,14 +506,10 @@ void UDPWrap::DoSend(const FunctionCallbackInfo& args, int family) { CHECK(args[3]->IsBoolean()); } - Local req_wrap_obj = args[0].As(); Local chunks = args[1].As(); // it is faster to fetch the length of the // array in js-land size_t count = args[2].As()->Value(); - const bool have_callback = sendto ? args[5]->IsTrue() : args[3]->IsTrue(); - - size_t msg_size = 0; MaybeStackBuffer bufs(count); @@ -483,7 +521,6 @@ void UDPWrap::DoSend(const FunctionCallbackInfo& args, int family) { size_t length = Buffer::Length(chunk); bufs[i] = uv_buf_init(Buffer::Data(chunk), length); - msg_size += length; } int err = 0; @@ -493,14 +530,36 @@ void UDPWrap::DoSend(const FunctionCallbackInfo& args, int family) { const unsigned short port = args[3].As()->Value(); node::Utf8Value address(env->isolate(), args[4]); err = sockaddr_for_family(family, address.out(), port, &addr_storage); - if (err == 0) { + if (err == 0) addr = reinterpret_cast(&addr_storage); - } } - uv_buf_t* bufs_ptr = *bufs; - if (err == 0 && !UNLIKELY(env->options()->test_udp_no_try_send)) { - err = uv_udp_try_send(&wrap->handle_, bufs_ptr, count, addr); + if (err == 0) { + wrap->current_send_req_wrap_ = args[0].As(); + wrap->current_send_has_callback_ = + sendto ? args[5]->IsTrue() : args[3]->IsTrue(); + + err = wrap->Send(*bufs, count, addr); + + wrap->current_send_req_wrap_.Clear(); + wrap->current_send_has_callback_ = false; + } + + args.GetReturnValue().Set(err); +} + +ssize_t UDPWrap::Send(uv_buf_t* bufs_ptr, + size_t count, + const sockaddr* addr) { + if (IsHandleClosing()) return UV_EBADF; + + size_t msg_size = 0; + for (size_t i = 0; i < count; i++) + msg_size += bufs_ptr[i].len; + + int err = 0; + if (!UNLIKELY(env()->options()->test_udp_no_try_send)) { + err = uv_udp_try_send(&handle_, bufs_ptr, count, addr); if (err == UV_ENOSYS || err == UV_EAGAIN) { err = 0; } else if (err >= 0) { @@ -518,28 +577,41 @@ void UDPWrap::DoSend(const FunctionCallbackInfo& args, int family) { CHECK_EQ(static_cast(err), msg_size); // + 1 so that the JS side can distinguish 0-length async sends from // 0-length sync sends. - args.GetReturnValue().Set(static_cast(msg_size) + 1); - return; + return msg_size + 1; } } } if (err == 0) { - AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope(wrap); - SendWrap* req_wrap = new SendWrap(env, req_wrap_obj, have_callback); - req_wrap->msg_size = msg_size; - - err = req_wrap->Dispatch(uv_udp_send, - &wrap->handle_, - bufs_ptr, - count, - addr, - OnSend); + AsyncHooks::DefaultTriggerAsyncIdScope trigger_scope(this); + ReqWrap* req_wrap = listener()->CreateSendWrap(msg_size); + if (req_wrap == nullptr) return UV_ENOSYS; + + err = req_wrap->Dispatch( + uv_udp_send, + &handle_, + bufs_ptr, + count, + addr, + uv_udp_send_cb{[](uv_udp_send_t* req, int status) { + UDPWrap* self = ContainerOf(&UDPWrap::handle_, req->handle); + self->listener()->OnSendDone( + ReqWrap::from_req(req), status); + }}); if (err) delete req_wrap; } - args.GetReturnValue().Set(err); + return err; +} + + +ReqWrap* UDPWrap::CreateSendWrap(size_t msg_size) { + SendWrap* req_wrap = new SendWrap(env(), + current_send_req_wrap_, + current_send_has_callback_); + req_wrap->msg_size = msg_size; + return req_wrap; } @@ -553,31 +625,46 @@ void UDPWrap::Send6(const FunctionCallbackInfo& args) { } -void UDPWrap::RecvStart(const FunctionCallbackInfo& args) { - UDPWrap* wrap; - ASSIGN_OR_RETURN_UNWRAP(&wrap, - args.Holder(), - args.GetReturnValue().Set(UV_EBADF)); - int err = uv_udp_recv_start(&wrap->handle_, OnAlloc, OnRecv); +AsyncWrap* UDPWrap::GetAsyncWrap() { + return this; +} + +int UDPWrap::GetPeerName(sockaddr* name, int* namelen) { + return uv_udp_getpeername(&handle_, name, namelen); +} + +int UDPWrap::GetSockName(sockaddr* name, int* namelen) { + return uv_udp_getsockname(&handle_, name, namelen); +} + +void UDPWrapBase::RecvStart(const FunctionCallbackInfo& args) { + UDPWrapBase* wrap = UDPWrapBase::FromObject(args.Holder()); + args.GetReturnValue().Set(wrap == nullptr ? UV_EBADF : wrap->RecvStart()); +} + +int UDPWrap::RecvStart() { + if (IsHandleClosing()) return UV_EBADF; + int err = uv_udp_recv_start(&handle_, OnAlloc, OnRecv); // UV_EALREADY means that the socket is already bound but that's okay if (err == UV_EALREADY) err = 0; - args.GetReturnValue().Set(err); + return err; } -void UDPWrap::RecvStop(const FunctionCallbackInfo& args) { - UDPWrap* wrap; - ASSIGN_OR_RETURN_UNWRAP(&wrap, - args.Holder(), - args.GetReturnValue().Set(UV_EBADF)); - int r = uv_udp_recv_stop(&wrap->handle_); - args.GetReturnValue().Set(r); +void UDPWrapBase::RecvStop(const FunctionCallbackInfo& args) { + UDPWrapBase* wrap = UDPWrapBase::FromObject(args.Holder()); + args.GetReturnValue().Set(wrap == nullptr ? UV_EBADF : wrap->RecvStop()); +} + +int UDPWrap::RecvStop() { + if (IsHandleClosing()) return UV_EBADF; + return uv_udp_recv_stop(&handle_); } -void UDPWrap::OnSend(uv_udp_send_t* req, int status) { - std::unique_ptr req_wrap{static_cast(req->data)}; +void UDPWrap::OnSendDone(ReqWrap* req, int status) { + std::unique_ptr req_wrap{static_cast(req)}; if (req_wrap->have_callback()) { Environment* env = req_wrap->env(); HandleScope handle_scope(env->isolate()); @@ -594,19 +681,30 @@ void UDPWrap::OnSend(uv_udp_send_t* req, int status) { void UDPWrap::OnAlloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { - UDPWrap* wrap = static_cast(handle->data); - *buf = wrap->env()->AllocateManaged(suggested_size).release(); + UDPWrap* wrap = ContainerOf(&UDPWrap::handle_, + reinterpret_cast(handle)); + *buf = wrap->listener()->OnAlloc(suggested_size); +} + +uv_buf_t UDPWrap::OnAlloc(size_t suggested_size) { + return env()->AllocateManaged(suggested_size).release(); } void UDPWrap::OnRecv(uv_udp_t* handle, ssize_t nread, - const uv_buf_t* buf_, - const struct sockaddr* addr, + const uv_buf_t* buf, + const sockaddr* addr, unsigned int flags) { - UDPWrap* wrap = static_cast(handle->data); - Environment* env = wrap->env(); + UDPWrap* wrap = ContainerOf(&UDPWrap::handle_, handle); + wrap->listener()->OnRecv(nread, *buf, addr, flags); +} - AllocatedBuffer buf(env, *buf_); +void UDPWrap::OnRecv(ssize_t nread, + const uv_buf_t& buf_, + const sockaddr* addr, + unsigned int flags) { + Environment* env = this->env(); + AllocatedBuffer buf(env, buf_); if (nread == 0 && addr == nullptr) { return; } @@ -614,23 +712,22 @@ void UDPWrap::OnRecv(uv_udp_t* handle, HandleScope handle_scope(env->isolate()); Context::Scope context_scope(env->context()); - Local wrap_obj = wrap->object(); Local argv[] = { Integer::New(env->isolate(), nread), - wrap_obj, + object(), Undefined(env->isolate()), Undefined(env->isolate()) }; if (nread < 0) { - wrap->MakeCallback(env->onmessage_string(), arraysize(argv), argv); + MakeCallback(env->onmessage_string(), arraysize(argv), argv); return; } buf.Resize(nread); argv[2] = buf.ToBuffer().ToLocalChecked(); argv[3] = AddressToJS(env, addr); - wrap->MakeCallback(env->onmessage_string(), arraysize(argv), argv); + MakeCallback(env->onmessage_string(), arraysize(argv), argv); } MaybeLocal UDPWrap::Instantiate(Environment* env, diff --git a/src/udp_wrap.h b/src/udp_wrap.h index 2026dd1dee1eb1..7afd9784b0ac9e 100644 --- a/src/udp_wrap.h +++ b/src/udp_wrap.h @@ -25,14 +25,101 @@ #if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS #include "handle_wrap.h" +#include "req_wrap.h" #include "uv.h" #include "v8.h" namespace node { -class Environment; +class UDPWrapBase; -class UDPWrap: public HandleWrap { +// A listener that can be attached to an `UDPWrapBase` object and generally +// manages its I/O activity. This is similar to `StreamListener`. +class UDPListener { + public: + virtual ~UDPListener(); + + // Called right before data is received from the socket. Must return a + // buffer suitable for reading data into, that is then passed to OnRecv. + virtual uv_buf_t OnAlloc(size_t suggested_size) = 0; + + // Called right after data is received from the socket, and includes + // information about the source address. If `nread` is negative, an error + // has occurred, and it represents a libuv error code. + virtual void OnRecv(ssize_t nread, + const uv_buf_t& buf, + const sockaddr* addr, + unsigned int flags) = 0; + + // Called when an asynchronous request for writing data is created. + // The `msg_size` value contains the total size of the data to be sent, + // but may be ignored by the implementation of this Method. + // The return value is later passed to OnSendDone. + virtual ReqWrap* CreateSendWrap(size_t msg_size) = 0; + + // Called when an asynchronous request for writing data has finished. + // If status is negative, an error has occurred, and it represents a libuv + // error code. + virtual void OnSendDone(ReqWrap* wrap, int status) = 0; + + // Optional callback that is called after the socket has been bound. + virtual void OnAfterBind() {} + + inline UDPWrapBase* udp() const { return wrap_; } + + protected: + UDPWrapBase* wrap_ = nullptr; + + friend class UDPWrapBase; +}; + +class UDPWrapBase { + public: + // While UDPWrapBase itself does not extend from HandleWrap, classes + // derived from it will (like UDPWrap) + enum InternalFields { + kUDPWrapBaseField = HandleWrap::kInternalFieldCount, + kInternalFieldCount + }; + virtual ~UDPWrapBase(); + + // Start emitting OnAlloc() + OnRecv() events on the listener. + virtual int RecvStart() = 0; + + // Stop emitting OnAlloc() + OnRecv() events on the listener. + virtual int RecvStop() = 0; + + // Send a chunk of data over this socket. This may call CreateSendWrap() + // on the listener if an async transmission is necessary. + virtual ssize_t Send(uv_buf_t* bufs, + size_t nbufs, + const sockaddr* addr) = 0; + + // Stores the sockaddr for the peer in `name`. + virtual int GetPeerName(sockaddr* name, int* namelen) = 0; + + // Stores the sockaddr for the local socket in `name`. + virtual int GetSockName(sockaddr* name, int* namelen) = 0; + + // Returns an AsyncWrap object with the same lifetime as this object. + virtual AsyncWrap* GetAsyncWrap() = 0; + + void set_listener(UDPListener* listener); + UDPListener* listener() const; + + static UDPWrapBase* FromObject(v8::Local obj); + + static void RecvStart(const v8::FunctionCallbackInfo& args); + static void RecvStop(const v8::FunctionCallbackInfo& args); + static void AddMethods(Environment* env, v8::Local t); + + private: + UDPListener* listener_ = nullptr; +}; + +class UDPWrap final : public HandleWrap, + public UDPWrapBase, + public UDPListener { public: enum SocketType { SOCKET @@ -51,8 +138,6 @@ class UDPWrap: public HandleWrap { static void Connect6(const v8::FunctionCallbackInfo& args); static void Send6(const v8::FunctionCallbackInfo& args); static void Disconnect(const v8::FunctionCallbackInfo& args); - static void RecvStart(const v8::FunctionCallbackInfo& args); - static void RecvStop(const v8::FunctionCallbackInfo& args); static void AddMembership(const v8::FunctionCallbackInfo& args); static void DropMembership(const v8::FunctionCallbackInfo& args); static void AddSourceSpecificMembership( @@ -68,6 +153,25 @@ class UDPWrap: public HandleWrap { static void SetTTL(const v8::FunctionCallbackInfo& args); static void BufferSize(const v8::FunctionCallbackInfo& args); + // UDPListener implementation + uv_buf_t OnAlloc(size_t suggested_size) override; + void OnRecv(ssize_t nread, + const uv_buf_t& buf, + const sockaddr* addr, + unsigned int flags) override; + ReqWrap* CreateSendWrap(size_t msg_size) override; + void OnSendDone(ReqWrap* wrap, int status) override; + + // UDPWrapBase implementation + int RecvStart() override; + int RecvStop() override; + ssize_t Send(uv_buf_t* bufs, + size_t nbufs, + const sockaddr* addr) override; + int GetPeerName(sockaddr* name, int* namelen) override; + int GetSockName(sockaddr* name, int* namelen) override; + AsyncWrap* GetAsyncWrap() override; + static v8::MaybeLocal Instantiate(Environment* env, AsyncWrap* parent, SocketType type); @@ -99,7 +203,6 @@ class UDPWrap: public HandleWrap { static void OnAlloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf); - static void OnSend(uv_udp_send_t* req, int status); static void OnRecv(uv_udp_t* handle, ssize_t nread, const uv_buf_t* buf, @@ -107,6 +210,9 @@ class UDPWrap: public HandleWrap { unsigned int flags); uv_udp_t handle_; + + bool current_send_has_callback_; + v8::Local current_send_req_wrap_; }; } // namespace node