test: add common/udppair utility

Extracted from the QUIC PR. This adds a utility used to deterministically test UDP traffic. It is currently only used by the experimental QUIC implementation. Separated out on request to make review easier. PR-URL: #33380 Reviewed-By: Sam Roberts <[email protected]>

Author: jasnell

node.gyp

@@ -563,6 +563,7 @@
 'src/js_native_api_v8_internals.h',
 'src/js_stream.cc',
 'src/json_utils.cc',
+'src/js_udp_wrap.cc',
 'src/module_wrap.cc',
 'src/node.cc',
 'src/node_api.cc',

src/async_wrap.h

@@ -51,6 +51,7 @@ namespace node{
 V(HTTPINCOMINGMESSAGE) \
 V(HTTPCLIENTREQUEST) \
 V(JSSTREAM) \
+V(JSUDPWRAP) \
 V(MESSAGEPORT) \
 V(PIPECONNECTWRAP) \
 V(PIPESERVERWRAP) \

src/js_udp_wrap.cc

@@ -0,0 +1,218 @@
+#include"udp_wrap.h"
+#include"async_wrap-inl.h"
+#include"node_errors.h"
+#include"node_sockaddr-inl.h"
+
+#include<algorithm>
+
+namespacenode{
+
+using errors::TryCatchScope;
+using v8::Array;
+using v8::Context;
+using v8::FunctionCallbackInfo;
+using v8::FunctionTemplate;
+using v8::HandleScope;
+using v8::Int32;
+using v8::Local;
+using v8::Object;
+using v8::String;
+using v8::Value;
+
+// JSUDPWrap is a testing utility used by test/common/udppair.js
+// to simulate UDP traffic deterministically in Node.js tests.
+classJSUDPWrapfinal : public UDPWrapBase, public AsyncWrap{
+public:
+JSUDPWrap(Environment* env, Local<Object> obj);
+
+intRecvStart() override;
+intRecvStop() override;
+ssize_tSend(uv_buf_t* bufs,
+size_t nbufs,
+const sockaddr* addr) override;
+ SocketAddress GetPeerName() override;
+ SocketAddress GetSockName() override;
+ AsyncWrap* GetAsyncWrap() override{returnthis}
+
+staticvoidNew(const FunctionCallbackInfo<Value>& args);
+staticvoidEmitReceived(const FunctionCallbackInfo<Value>& args);
+staticvoidOnSendDone(const FunctionCallbackInfo<Value>& args);
+staticvoidOnAfterBind(const FunctionCallbackInfo<Value>& args);
+
+staticvoidInitialize(Local<Object> target,
+ Local<Value> unused,
+ Local<Context> context,
+void* priv);
+SET_NO_MEMORY_INFO()
+SET_MEMORY_INFO_NAME(JSUDPWrap)
+SET_SELF_SIZE(JSUDPWrap)
+};
+
+JSUDPWrap::JSUDPWrap(Environment* env, Local<Object> obj)
+ : AsyncWrap(env, obj, PROVIDER_JSUDPWRAP){
+MakeWeak();
+
+ obj->SetAlignedPointerInInternalField(
+kUDPWrapBaseField, static_cast<UDPWrapBase*>(this));
+}
+
+intJSUDPWrap::RecvStart(){
+ HandleScope scope(env()->isolate());
+ Context::Scope context_scope(env()->context());
+ TryCatchScope try_catch(env());
+ Local<Value> value;
+int32_t value_int = UV_EPROTO;
+if (!MakeCallback(env()->onreadstart_string(), 0, nullptr).ToLocal(&value) ||
+ !value->Int32Value(env()->context()).To(&value_int)){
+if (try_catch.HasCaught() && !try_catch.HasTerminated())
+errors::TriggerUncaughtException(env()->isolate(), try_catch);
+ }
+return value_int;
+}
+
+intJSUDPWrap::RecvStop(){
+ HandleScope scope(env()->isolate());
+ Context::Scope context_scope(env()->context());
+ TryCatchScope try_catch(env());
+ Local<Value> value;
+int32_t value_int = UV_EPROTO;
+if (!MakeCallback(env()->onreadstop_string(), 0, nullptr).ToLocal(&value) ||
+ !value->Int32Value(env()->context()).To(&value_int)){
+if (try_catch.HasCaught() && !try_catch.HasTerminated())
+errors::TriggerUncaughtException(env()->isolate(), try_catch);
+ }
+return value_int;
+}
+
+ssize_tJSUDPWrap::Send(uv_buf_t* bufs,
+size_t nbufs,
+const sockaddr* addr){
+ HandleScope scope(env()->isolate());
+ Context::Scope context_scope(env()->context());
+ TryCatchScope try_catch(env());
+ Local<Value> value;
+int64_t value_int = UV_EPROTO;
+size_t total_len = 0;
+
+ MaybeStackBuffer<Local<Value>, 16> buffers(nbufs);
+for (size_t i = 0; i < nbufs; i++){
+ buffers[i] = Buffer::Copy(env(), bufs[i].base, bufs[i].len)
+ .ToLocalChecked();
+ total_len += bufs[i].len;
+ }
+
+ Local<Value> args[] ={
+listener()->CreateSendWrap(total_len)->object(),
+Array::New(env()->isolate(), buffers.out(), nbufs),
+AddressToJS(env(), addr)
+ };
+
+if (!MakeCallback(env()->onwrite_string(), arraysize(args), args)
+ .ToLocal(&value) ||
+ !value->IntegerValue(env()->context()).To(&value_int)){
+if (try_catch.HasCaught() && !try_catch.HasTerminated())
+errors::TriggerUncaughtException(env()->isolate(), try_catch);
+ }
+return value_int;
+}
+
+SocketAddress JSUDPWrap::GetPeerName(){
+ SocketAddress ret;
+CHECK(SocketAddress::New(AF_INET, "127.0.0.1", 1337, &ret));
+return ret;
+}
+
+SocketAddress JSUDPWrap::GetSockName(){
+ SocketAddress ret;
+CHECK(SocketAddress::New(AF_INET, "127.0.0.1", 1337, &ret));
+return ret;
+}
+
+voidJSUDPWrap::New(const FunctionCallbackInfo<Value>& args){
+ Environment* env = Environment::GetCurrent(args);
+CHECK(args.IsConstructCall());
+newJSUDPWrap(env, args.Holder());
+}
+
+voidJSUDPWrap::EmitReceived(const FunctionCallbackInfo<Value>& args){
+ JSUDPWrap* wrap;
+ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
+ Environment* env = wrap->env();
+
+ ArrayBufferViewContents<char> buffer(args[0]);
+constchar* data = buffer.data();
+int len = buffer.length();
+
+CHECK(args[1]->IsInt32()); // family
+CHECK(args[2]->IsString()); // address
+CHECK(args[3]->IsInt32()); // port
+CHECK(args[4]->IsInt32()); // flags
+int family = args[1].As<Int32>()->Value() == 4 ? AF_INET : AF_INET6;
+ Utf8Value address(env->isolate(), args[2]);
+int port = args[3].As<Int32>()->Value();
+int flags = args[3].As<Int32>()->Value();
+
+ sockaddr_storage addr;
+CHECK_EQ(sockaddr_for_family(family, *address, port, &addr), 0);
+
+// Repeatedly ask the stream's owner for memory, copy the data that we
+// just read from JS into those buffers and emit them as reads.
+while (len != 0){
+uv_buf_t buf = wrap->listener()->OnAlloc(len);
+ssize_t avail = std::min<size_t>(buf.len, len);
+memcpy(buf.base, data, avail);
+ data += avail;
+ len -= avail;
+ wrap->listener()->OnRecv(
+ avail, buf, reinterpret_cast<sockaddr*>(&addr), flags);
+ }
+}
+
+voidJSUDPWrap::OnSendDone(const FunctionCallbackInfo<Value>& args){
+ JSUDPWrap* wrap;
+ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
+
+CHECK(args[0]->IsObject());
+CHECK(args[1]->IsInt32());
+ ReqWrap<uv_udp_send_t>* req_wrap;
+ASSIGN_OR_RETURN_UNWRAP(&req_wrap, args[0].As<Object>());
+int status = args[1].As<Int32>()->Value();
+
+ wrap->listener()->OnSendDone(req_wrap, status);
+}
+
+voidJSUDPWrap::OnAfterBind(const FunctionCallbackInfo<Value>& args){
+ JSUDPWrap* wrap;
+ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
+
+ wrap->listener()->OnAfterBind();
+}
+
+voidJSUDPWrap::Initialize(Local<Object> target,
+ Local<Value> unused,
+ Local<Context> context,
+void* priv){
+ Environment* env = Environment::GetCurrent(context);
+
+ Local<FunctionTemplate> t = env->NewFunctionTemplate(New);
+ Local<String> js_udp_wrap_string =
+FIXED_ONE_BYTE_STRING(env->isolate(), "JSUDPWrap");
+ t->SetClassName(js_udp_wrap_string);
+ t->InstanceTemplate()
+ ->SetInternalFieldCount(UDPWrapBase::kUDPWrapBaseField + 1);
+ t->Inherit(AsyncWrap::GetConstructorTemplate(env));
+
+UDPWrapBase::AddMethods(env, t);
+ env->SetProtoMethod(t, "emitReceived", EmitReceived);
+ env->SetProtoMethod(t, "onSendDone", OnSendDone);
+ env->SetProtoMethod(t, "onAfterBind", OnAfterBind);
+
+ target->Set(env->context(),
+ js_udp_wrap_string,
+ t->GetFunction(context).ToLocalChecked()).Check();
+}
+
+
+} // namespace node
+
+NODE_MODULE_CONTEXT_AWARE_INTERNAL(js_udp_wrap, node::JSUDPWrap::Initialize)

src/node_binding.cc

@@ -52,6 +52,7 @@
  V(http_parser_llhttp) \
  V(inspector) \
  V(js_stream) \
+ V(js_udp_wrap) \
  V(messaging) \
  V(module_wrap) \
  V(native_module) \

src/udp_wrap.h

@@ -215,6 +215,11 @@ class UDPWrap final : public HandleWrap,
  v8::Local<v8::Object> current_send_req_wrap_;
 };
 
+intsockaddr_for_family(int address_family,
+constchar* address,
+constunsignedshort port,
+ sockaddr_storage* addr);
+
 } // namespace node
 
 #endif// defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS

test/common/README.md

@@ -929,6 +929,19 @@ The realpath of the testing temporary directory.
 
 Deletes and recreates the testing temporary directory.
 
+## UDP pair helper
+
+The `common/udppair` module exports a function `makeUDPPair` and a class
+`FakeUDPWrap`.
+
+`FakeUDPWrap` emits `'send'` events when data is to be sent on it, and provides
+an `emitReceived()` API for actin as if data has been received on it.
+
+`makeUDPPair` returns an object `{clientSide, serverSide }` where each side
+is an `FakeUDPWrap` connected to the other side.
+
+There is no difference between cient or server side beyond their names.
+
 ## WPT Module
 
 ### `harness`

test/common/udppair.js

@@ -0,0 +1,102 @@
+/* eslint-disable node-core/require-common-first, node-core/required-modules */
+'use strict';
+const{ internalBinding }=require('internal/test/binding');
+const{ JSUDPWrap }=internalBinding('js_udp_wrap');
+constEventEmitter=require('events');
+
+// FakeUDPWrap is a testing utility that emulates a UDP connection
+// for the sake of making UDP tests more deterministic.
+classFakeUDPWrapextendsEventEmitter{
+constructor(){
+super();
+
+this._handle=newJSUDPWrap();
+
+this._handle.onreadstart=()=>this._startReading();
+this._handle.onreadstop=()=>this._stopReading();
+this._handle.onwrite=
+(wrap,buffers,addr)=>this._write(wrap,buffers,addr);
+this._handle.getsockname=(obj)=>{
+Object.assign(obj,{address: '127.0.0.1',family: 'IPv4',port: 1337});
+return0;
+};
+
+this.reading=false;
+this.bufferedReceived=[];
+this.emitBufferedImmediate=null;
+}
+
+_emitBuffered=()=>{
+if(!this.reading)return;
+if(this.bufferedReceived.length>0){
+this.emitReceived(this.bufferedReceived.shift());
+this.emitBufferedImmediate=setImmediate(this._emitBuffered);
+}else{
+this.emit('wantRead');
+}
+};
+
+_startReading(){
+this.reading=true;
+this.emitBufferedImmediate=setImmediate(this._emitBuffered);
+}
+
+_stopReading(){
+this.reading=false;
+clearImmediate(this.emitBufferedImmediate);
+}
+
+_write(wrap,buffers,addr){
+this.emit('send',{ buffers, addr });
+setImmediate(()=>this._handle.onSendDone(wrap,0));
+}
+
+afterBind(){
+this._handle.onAfterBind();
+}
+
+emitReceived(info){
+if(!this.reading){
+this.bufferedReceived.push(info);
+return;
+}
+
+const{
+ buffers,
+addr: {
+ family =4,
+ address ='127.0.0.1',
+ port =1337,
+},
+ flags =0
+}=info;
+
+letfamilyInt;
+switch(family){
+case'IPv4': familyInt=4;break;
+case'IPv6': familyInt=6;break;
+default: thrownewError('bad family');
+}
+
+for(constbufferofbuffers){
+this._handle.emitReceived(buffer,familyInt,address,port,flags);
+}
+}
+}
+
+functionmakeUDPPair(){
+constserverSide=newFakeUDPWrap();
+constclientSide=newFakeUDPWrap();
+
+serverSide.on('send',
+(chk)=>setImmediate(()=>clientSide.emitReceived(chk)));
+clientSide.on('send',
+(chk)=>setImmediate(()=>serverSide.emitReceived(chk)));
+
+return{ serverSide, clientSide };
+}
+
+module.exports={
+ FakeUDPWrap,
+ makeUDPPair
+};

test/sequential/test-async-wrap-getasyncid.js

@@ -45,6 +45,7 @@ const{getSystemErrorName } = require('util');
 deleteproviders.STREAMPIPE;
 deleteproviders.MESSAGEPORT;
 deleteproviders.WORKER;
+deleteproviders.JSUDPWRAP;
 if(!common.isMainThread)
 deleteproviders.INSPECTORJSBINDING;
 deleteproviders.KEYPAIRGENREQUEST;