diff options
| -rw-r--r-- | doc/api/dgram.md | 33 | ||||
| -rw-r--r-- | lib/dgram.js | 49 | ||||
| -rw-r--r-- | src/udp_wrap.cc | 42 | ||||
| -rw-r--r-- | src/udp_wrap.h | 7 | ||||
| -rw-r--r-- | test/internet/test-dgram-multicast-ssm-multi-process.js | 229 | ||||
| -rw-r--r-- | test/internet/test-dgram-multicast-ssmv6-multi-process.js | 229 |
6 files changed, 589 insertions, 0 deletions
diff --git a/doc/api/dgram.md b/doc/api/dgram.md index 9efd14c89e..894f64f36e 100644 --- a/doc/api/dgram.md +++ b/doc/api/dgram.md @@ -123,6 +123,21 @@ if (cluster.isMaster) { } ``` +### socket.addSourceSpecificMembership(sourceAddress, groupAddress\[, multicastInterface\]) +<!-- YAML +added: REPLACEME +--> +* `sourceAddress` {string} +* `groupAddress` {string} +* `multicastInterface` {string} + +Tells the kernel to join a source-specific multicast channel at the given +`sourceAddress` and `groupAddress`, using the `multicastInterface` with the +`IP_ADD_SOURCE_MEMBERSHIP` socket option. If the `multicastInterface` argument +is not specified, the operating system will choose one interface and will add +membership to it. To add membership to every available interface, call +`socket.addSourceSpecificMembership()` multiple times, once per interface. + ### socket.address() <!-- YAML added: v0.1.99 @@ -297,6 +312,24 @@ never have reason to call this. If `multicastInterface` is not specified, the operating system will attempt to drop membership on all valid interfaces. +### socket.dropSourceSpecificMembership(sourceAddress, groupAddress\[, multicastInterface\]) +<!-- YAML +added: REPLACEME +--> + +* `sourceAddress` {string} +* `groupAddress` {string} +* `multicastInterface` {string} + +Instructs the kernel to leave a source-specific multicast channel at the given +`sourceAddress` and `groupAddress` using the `IP_DROP_SOURCE_MEMBERSHIP` +socket option. This method is automatically called by the kernel when the +socket is closed or the process terminates, so most apps will never have +reason to call this. + +If `multicastInterface` is not specified, the operating system will attempt to +drop membership on all valid interfaces. + ### socket.getRecvBufferSize() <!-- YAML added: v8.7.0 diff --git a/lib/dgram.js b/lib/dgram.js index cc61d4d274..d29d1e7b19 100644 --- a/lib/dgram.js +++ b/lib/dgram.js @@ -832,6 +832,55 @@ Socket.prototype.dropMembership = function(multicastAddress, } }; +Socket.prototype.addSourceSpecificMembership = function(sourceAddress, + groupAddress, + interfaceAddress) { + healthCheck(this); + + if (typeof sourceAddress !== 'string') { + throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'sourceAddress', + 'string'); + } + + if (typeof groupAddress !== 'string') { + throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'groupAddress', + 'string'); + } + + const err = + this[kStateSymbol].handle.addSourceSpecificMembership(sourceAddress, + groupAddress, + interfaceAddress); + if (err) { + throw errnoException(err, 'addSourceSpecificMembership'); + } +}; + + +Socket.prototype.dropSourceSpecificMembership = function(sourceAddress, + groupAddress, + interfaceAddress) { + healthCheck(this); + + if (typeof sourceAddress !== 'string') { + throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'sourceAddress', + 'string'); + } + + if (typeof groupAddress !== 'string') { + throw new errors.TypeError('ERR_INVALID_ARG_TYPE', 'groupAddress', + 'string'); + } + + const err = + this[kStateSymbol].handle.dropSourceSpecificMembership(sourceAddress, + groupAddress, + interfaceAddress); + if (err) { + throw errnoException(err, 'dropSourceSpecificMembership'); + } +}; + function healthCheck(socket) { if (!socket[kStateSymbol].handle) { diff --git a/src/udp_wrap.cc b/src/udp_wrap.cc index 64c4c8b304..4a66ce0a1f 100644 --- a/src/udp_wrap.cc +++ b/src/udp_wrap.cc @@ -128,6 +128,10 @@ void UDPWrap::Initialize(Local<Object> target, GetSockOrPeerName<UDPWrap, uv_udp_getsockname>); env->SetProtoMethod(t, "addMembership", AddMembership); env->SetProtoMethod(t, "dropMembership", DropMembership); + env->SetProtoMethod(t, "addSourceSpecificMembership", + AddSourceSpecificMembership); + env->SetProtoMethod(t, "dropSourceSpecificMembership", + DropSourceSpecificMembership); env->SetProtoMethod(t, "setMulticastInterface", SetMulticastInterface); env->SetProtoMethod(t, "setMulticastTTL", SetMulticastTTL); env->SetProtoMethod(t, "setMulticastLoopback", SetMulticastLoopback); @@ -397,6 +401,44 @@ void UDPWrap::DropMembership(const FunctionCallbackInfo<Value>& args) { SetMembership(args, UV_LEAVE_GROUP); } +void UDPWrap::SetSourceMembership(const FunctionCallbackInfo<Value>& args, + uv_membership membership) { + UDPWrap* wrap; + ASSIGN_OR_RETURN_UNWRAP(&wrap, + args.Holder(), + args.GetReturnValue().Set(UV_EBADF)); + + CHECK_EQ(args.Length(), 3); + + node::Utf8Value source_address(args.GetIsolate(), args[0]); + node::Utf8Value group_address(args.GetIsolate(), args[1]); + node::Utf8Value iface(args.GetIsolate(), args[2]); + + if (*iface == nullptr) return; + const char* iface_cstr = *iface; + if (args[2]->IsUndefined() || args[2]->IsNull()) { + iface_cstr = nullptr; + } + + int err = uv_udp_set_source_membership(&wrap->handle_, + *group_address, + iface_cstr, + *source_address, + membership); + args.GetReturnValue().Set(err); +} + +void UDPWrap::AddSourceSpecificMembership( + const FunctionCallbackInfo<Value>& args) { + SetSourceMembership(args, UV_JOIN_GROUP); +} + + +void UDPWrap::DropSourceSpecificMembership( + const FunctionCallbackInfo<Value>& args) { + SetSourceMembership(args, UV_LEAVE_GROUP); +} + void UDPWrap::DoSend(const FunctionCallbackInfo<Value>& args, int family) { Environment* env = Environment::GetCurrent(args); diff --git a/src/udp_wrap.h b/src/udp_wrap.h index fb2a9362cd..f79fdd9109 100644 --- a/src/udp_wrap.h +++ b/src/udp_wrap.h @@ -55,6 +55,10 @@ class UDPWrap: public HandleWrap { static void RecvStop(const v8::FunctionCallbackInfo<v8::Value>& args); static void AddMembership(const v8::FunctionCallbackInfo<v8::Value>& args); static void DropMembership(const v8::FunctionCallbackInfo<v8::Value>& args); + static void AddSourceSpecificMembership( + const v8::FunctionCallbackInfo<v8::Value>& args); + static void DropSourceSpecificMembership( + const v8::FunctionCallbackInfo<v8::Value>& args); static void SetMulticastInterface( const v8::FunctionCallbackInfo<v8::Value>& args); static void SetMulticastTTL(const v8::FunctionCallbackInfo<v8::Value>& args); @@ -88,6 +92,9 @@ class UDPWrap: public HandleWrap { int family); static void SetMembership(const v8::FunctionCallbackInfo<v8::Value>& args, uv_membership membership); + static void SetSourceMembership( + const v8::FunctionCallbackInfo<v8::Value>& args, + uv_membership membership); static void OnAlloc(uv_handle_t* handle, size_t suggested_size, diff --git a/test/internet/test-dgram-multicast-ssm-multi-process.js b/test/internet/test-dgram-multicast-ssm-multi-process.js new file mode 100644 index 0000000000..1771a4204c --- /dev/null +++ b/test/internet/test-dgram-multicast-ssm-multi-process.js @@ -0,0 +1,229 @@ +'use strict'; +const common = require('../common'); +// Skip test in FreeBSD jails. +if (common.inFreeBSDJail) + common.skip('In a FreeBSD jail'); + +const assert = require('assert'); +const dgram = require('dgram'); +const fork = require('child_process').fork; +const networkInterfaces = require('os').networkInterfaces(); +const GROUP_ADDRESS = '232.1.1.1'; +const TIMEOUT = common.platformTimeout(5000); +const messages = [ + Buffer.from('First message to send'), + Buffer.from('Second message to send'), + Buffer.from('Third message to send'), + Buffer.from('Fourth message to send') +]; +const workers = {}; +const listeners = 3; +let listening, sendSocket, done, timer, dead; + +let sourceAddress = null; + +// Take the first non-internal interface as the IPv4 address for binding. +// Ideally, this should favor internal/private interfaces. +get_sourceAddress: for (const name in networkInterfaces) { + const interfaces = networkInterfaces[name]; + for (let i = 0; i < interfaces.length; i++) { + const localInterface = interfaces[i]; + if (!localInterface.internal && localInterface.family === 'IPv4') { + sourceAddress = localInterface.address; + break get_sourceAddress; + } + } +} +assert.ok(sourceAddress); + +function launchChildProcess() { + const worker = fork(__filename, ['child']); + workers[worker.pid] = worker; + + worker.messagesReceived = []; + + // Handle the death of workers. + worker.on('exit', function(code) { + // Don't consider this the true death if the worker has finished + // successfully or if the exit code is 0. + if (worker.isDone || code === 0) { + return; + } + + dead += 1; + console.error('[PARENT] Worker %d died. %d dead of %d', + worker.pid, + dead, + listeners); + + if (dead === listeners) { + console.error('[PARENT] All workers have died.'); + console.error('[PARENT] Fail'); + assert.fail(); + } + }); + + worker.on('message', function(msg) { + if (msg.listening) { + listening += 1; + + if (listening === listeners) { + // All child process are listening, so start sending. + sendSocket.sendNext(); + } + return; + } + if (msg.message) { + worker.messagesReceived.push(msg.message); + + if (worker.messagesReceived.length === messages.length) { + done += 1; + worker.isDone = true; + console.error('[PARENT] %d received %d messages total.', + worker.pid, + worker.messagesReceived.length); + } + + if (done === listeners) { + console.error('[PARENT] All workers have received the ' + + 'required number of messages. Will now compare.'); + + Object.keys(workers).forEach(function(pid) { + const worker = workers[pid]; + + let count = 0; + + worker.messagesReceived.forEach(function(buf) { + for (let i = 0; i < messages.length; ++i) { + if (buf.toString() === messages[i].toString()) { + count++; + break; + } + } + }); + + console.error('[PARENT] %d received %d matching messages.', + worker.pid, count); + + assert.strictEqual(count, messages.length); + }); + + clearTimeout(timer); + console.error('[PARENT] Success'); + killChildren(workers); + } + } + }); +} + +function killChildren(children) { + Object.keys(children).forEach(function(key) { + const child = children[key]; + child.kill(); + }); +} + +if (process.argv[2] !== 'child') { + listening = 0; + dead = 0; + let i = 0; + done = 0; + + // Exit the test if it doesn't succeed within TIMEOUT. + timer = setTimeout(function() { + console.error('[PARENT] Responses were not received within %d ms.', + TIMEOUT); + console.error('[PARENT] Fail'); + + killChildren(workers); + + assert.fail(); + }, TIMEOUT); + + // Launch child processes. + for (let x = 0; x < listeners; x++) { + launchChildProcess(x); + } + + sendSocket = dgram.createSocket('udp4'); + + // The socket is actually created async now. + sendSocket.on('listening', function() { + sendSocket.setTTL(1); + sendSocket.setBroadcast(true); + sendSocket.setMulticastTTL(1); + sendSocket.setMulticastLoopback(true); + sendSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS); + }); + + sendSocket.on('close', function() { + console.error('[PARENT] sendSocket closed'); + }); + + sendSocket.sendNext = function() { + const buf = messages[i++]; + + if (!buf) { + try { sendSocket.close(); } catch {} + return; + } + + sendSocket.send( + buf, + 0, + buf.length, + common.PORT, + GROUP_ADDRESS, + function(err) { + assert.ifError(err); + console.error('[PARENT] sent "%s" to %s:%s', + buf.toString(), + GROUP_ADDRESS, common.PORT); + process.nextTick(sendSocket.sendNext); + } + ); + }; +} + +if (process.argv[2] === 'child') { + const receivedMessages = []; + const listenSocket = dgram.createSocket({ + type: 'udp4', + reuseAddr: true + }); + + listenSocket.on('listening', function() { + listenSocket.setMulticastLoopback(true); + listenSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS); + + listenSocket.on('message', function(buf, rinfo) { + console.error('[CHILD] %s received "%s" from %j', process.pid, + buf.toString(), rinfo); + + receivedMessages.push(buf); + + process.send({ message: buf.toString() }); + + if (receivedMessages.length === messages.length) { + // .dropSourceSpecificMembership() not strictly needed, + // it is here as a sanity check. + listenSocket.dropSourceSpecificMembership(sourceAddress, GROUP_ADDRESS); + process.nextTick(function() { + listenSocket.close(); + }); + } + }); + + listenSocket.on('close', function() { + // HACK: Wait to exit the process to ensure that the parent + // process has had time to receive all messages via process.send() + // This may be indicative of some other issue. + setTimeout(function() { + process.exit(); + }, common.platformTimeout(1000)); + }); + process.send({ listening: true }); + }); + + listenSocket.bind(common.PORT); +} diff --git a/test/internet/test-dgram-multicast-ssmv6-multi-process.js b/test/internet/test-dgram-multicast-ssmv6-multi-process.js new file mode 100644 index 0000000000..88bb563364 --- /dev/null +++ b/test/internet/test-dgram-multicast-ssmv6-multi-process.js @@ -0,0 +1,229 @@ +'use strict'; +const common = require('../common'); +// Skip test in FreeBSD jails. +if (common.inFreeBSDJail) + common.skip('In a FreeBSD jail'); + +const assert = require('assert'); +const dgram = require('dgram'); +const fork = require('child_process').fork; +const networkInterfaces = require('os').networkInterfaces(); +const GROUP_ADDRESS = 'ff3e::1234'; +const TIMEOUT = common.platformTimeout(5000); +const messages = [ + Buffer.from('First message to send'), + Buffer.from('Second message to send'), + Buffer.from('Third message to send'), + Buffer.from('Fourth message to send') +]; +const workers = {}; +const listeners = 3; +let listening, sendSocket, done, timer, dead; + +let sourceAddress = null; + +// Take the first non-internal interface as the IPv6 address for binding. +// Ideally, this should check favor internal/private interfaces. +get_sourceAddress: for (const name in networkInterfaces) { + const interfaces = networkInterfaces[name]; + for (let i = 0; i < interfaces.length; i++) { + const localInterface = interfaces[i]; + if (!localInterface.internal && localInterface.family === 'IPv6') { + sourceAddress = localInterface.address; + break get_sourceAddress; + } + } +} +assert.ok(sourceAddress); + +function launchChildProcess() { + const worker = fork(__filename, ['child']); + workers[worker.pid] = worker; + + worker.messagesReceived = []; + + // Handle the death of workers. + worker.on('exit', function(code) { + // Don't consider this the true death if the worker has finished + // successfully or if the exit code is 0. + if (worker.isDone || code === 0) { + return; + } + + dead += 1; + console.error('[PARENT] Worker %d died. %d dead of %d', + worker.pid, + dead, + listeners); + + if (dead === listeners) { + console.error('[PARENT] All workers have died.'); + console.error('[PARENT] Fail'); + assert.fail(); + } + }); + + worker.on('message', function(msg) { + if (msg.listening) { + listening += 1; + + if (listening === listeners) { + // All child process are listening, so start sending. + sendSocket.sendNext(); + } + return; + } + if (msg.message) { + worker.messagesReceived.push(msg.message); + + if (worker.messagesReceived.length === messages.length) { + done += 1; + worker.isDone = true; + console.error('[PARENT] %d received %d messages total.', + worker.pid, + worker.messagesReceived.length); + } + + if (done === listeners) { + console.error('[PARENT] All workers have received the ' + + 'required number of messages. Will now compare.'); + + Object.keys(workers).forEach(function(pid) { + const worker = workers[pid]; + + let count = 0; + + worker.messagesReceived.forEach(function(buf) { + for (let i = 0; i < messages.length; ++i) { + if (buf.toString() === messages[i].toString()) { + count++; + break; + } + } + }); + + console.error('[PARENT] %d received %d matching messages.', + worker.pid, count); + + assert.strictEqual(count, messages.length); + }); + + clearTimeout(timer); + console.error('[PARENT] Success'); + killChildren(workers); + } + } + }); +} + +function killChildren(children) { + Object.keys(children).forEach(function(key) { + const child = children[key]; + child.kill(); + }); +} + +if (process.argv[2] !== 'child') { + listening = 0; + dead = 0; + let i = 0; + done = 0; + + // Exit the test if it doesn't succeed within TIMEOUT. + timer = setTimeout(function() { + console.error('[PARENT] Responses were not received within %d ms.', + TIMEOUT); + console.error('[PARENT] Fail'); + + killChildren(workers); + + assert.fail(); + }, TIMEOUT); + + // Launch child processes. + for (let x = 0; x < listeners; x++) { + launchChildProcess(x); + } + + sendSocket = dgram.createSocket('udp6'); + + // The socket is actually created async now. + sendSocket.on('listening', function() { + sendSocket.setTTL(1); + sendSocket.setBroadcast(true); + sendSocket.setMulticastTTL(1); + sendSocket.setMulticastLoopback(true); + sendSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS); + }); + + sendSocket.on('close', function() { + console.error('[PARENT] sendSocket closed'); + }); + + sendSocket.sendNext = function() { + const buf = messages[i++]; + + if (!buf) { + try { sendSocket.close(); } catch {} + return; + } + + sendSocket.send( + buf, + 0, + buf.length, + common.PORT, + GROUP_ADDRESS, + function(err) { + assert.ifError(err); + console.error('[PARENT] sent "%s" to %s:%s', + buf.toString(), + GROUP_ADDRESS, common.PORT); + process.nextTick(sendSocket.sendNext); + } + ); + }; +} + +if (process.argv[2] === 'child') { + const receivedMessages = []; + const listenSocket = dgram.createSocket({ + type: 'udp6', + reuseAddr: true + }); + + listenSocket.on('listening', function() { + listenSocket.setMulticastLoopback(true); + listenSocket.addSourceSpecificMembership(sourceAddress, GROUP_ADDRESS); + + listenSocket.on('message', function(buf, rinfo) { + console.error('[CHILD] %s received "%s" from %j', process.pid, + buf.toString(), rinfo); + + receivedMessages.push(buf); + + process.send({ message: buf.toString() }); + + if (receivedMessages.length === messages.length) { + // .dropSourceSpecificMembership() not strictly needed, + // it is here as a sanity check. + listenSocket.dropSourceSpecificMembership(sourceAddress, GROUP_ADDRESS); + process.nextTick(function() { + listenSocket.close(); + }); + } + }); + + listenSocket.on('close', function() { + // HACK: Wait to exit the process to ensure that the parent + // process has had time to receive all messages via process.send() + // This may be indicative of some other issue. + setTimeout(function() { + process.exit(); + }, common.platformTimeout(1000)); + }); + process.send({ listening: true }); + }); + + listenSocket.bind(common.PORT); +} |