fix: UDPSClient uses two-arg Join so multicast receiver lands on the right interface
UDPSClient::ConnectMulticast was calling the single-arg BasicUDPSocket::Join, which forwards NULL as the local interface and lets the kernel bind to INADDR_ANY. On a multi-homed host (or when the server sends on loopback via Interface = "127.0.0.1") the client joins the wrong interface and silently receives nothing. Fix: read the optional Interface key inside the useMulticast block in UDPSClient::Initialise; in ConnectMulticast call the two-arg Join(group, interface) when Interface is set, and fall back to the one-arg call otherwise to preserve the existing INADDR_ANY behaviour for configs that omit it. Forward the new optional Interface key through UDPStreamerClient (read from DataSource config, written into the UDPSClient ConfigurationDatabase only when non-empty). Extend the "Joined multicast group" log to report the interface name or "default". Regression test TestExecute_MulticastReceivesDataOnInterface: mock TCP control listener + multicast UDP DATA socket with IP_MULTICAST_IF set to 127.0.0.1, verifying a uint32 value of 424242 reaches DataSource signal memory. Confirmed FAILED without the Join fix and PASSED with it. Suite: 133/133 (was 132/132 before this commit). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
61a2aa3988
commit
14d5351a81
@@ -298,6 +298,50 @@ PrepareNextState() ← opens UDP server socket, starts background threa
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## UDPStreamerClient DataSource
|
||||
|
||||
`UDPStreamerClient` is a MARTe2 **input** DataSource that receives signals from a `UDPStreamer`
|
||||
server. Transport, fragment reassembly, and auto-reconnect are delegated to `UDPSClient`; the
|
||||
DataSource only decodes CONFIG/DATA payloads into real-time signal memory.
|
||||
|
||||
### Configuration
|
||||
|
||||
```
|
||||
+ClientDS = {
|
||||
Class = UDPStreamerClient
|
||||
ServerAddress = "192.168.1.10" // UDPStreamer server IP
|
||||
Port = 44500 // Server port
|
||||
|
||||
// Multicast (optional — omit for unicast)
|
||||
MulticastGroup = "239.0.0.1"
|
||||
DataPort = 44501 // UDP data port (default: Port+1)
|
||||
Interface = "192.168.1.10" // See table below
|
||||
|
||||
MaxPayloadSize = 1400
|
||||
|
||||
Signals = {
|
||||
Counter = { Type = uint32 }
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Parameters
|
||||
|
||||
| Parameter | Type | Default | Description |
|
||||
| --------------- | ------- | ---------- | ----------- |
|
||||
| `ServerAddress` | string | 127.0.0.1 | IPv4 address of the `UDPStreamer` server. |
|
||||
| `Port` | uint16 | 44500 | Server UDP port (unicast) or TCP control port (multicast). |
|
||||
| `MulticastGroup`| string | *(absent)* | IPv4 multicast address. Presence enables multicast mode. |
|
||||
| `DataPort` | uint16 | Port+1 | UDP port for multicast DATA datagrams. |
|
||||
| `Interface` | string | *(absent)* | Local IPv4 dotted-quad address (e.g. `"127.0.0.1"`) of the interface to join the multicast group on. **Optional**: omitting it uses the default-route interface (INADDR_ANY), which silently receives nothing if the server sends on a different interface. Not an interface name — `"eth0"` is invalid. |
|
||||
| `MaxPayloadSize`| uint32 | 1400 | Max payload bytes per datagram (must match the server). |
|
||||
| `SilenceTimeout`| float32 | 1.0 | Seconds of no data before auto-reconnect. 0 disables. |
|
||||
| `KeepAliveInterval` | uint32 | 15 | Seconds between unicast keepalive ACKs. 0 disables. |
|
||||
| `CPUMask` | uint32 | 0xFFFFFFFF | CPU affinity for the background receiver thread. |
|
||||
| `StackSize` | uint32 | default | Stack size in bytes for the receiver thread. |
|
||||
|
||||
With `MaxPayloadSize = 1400`, a single 1000-element float32 signal produces:
|
||||
|
||||
```
|
||||
|
||||
@@ -244,10 +244,14 @@ bool UDPStreamerClient::Initialise(StructuredDataI &data) {
|
||||
dp = port + UDPS_CLIENT_DEFAULT_DP_OFFSET;
|
||||
}
|
||||
dataPort = dp;
|
||||
StreamString ifaceStr = "";
|
||||
(void) data.Read("Interface", ifaceStr);
|
||||
multicastInterface = ifaceStr;
|
||||
REPORT_ERROR(ErrorManagement::Information,
|
||||
"Multicast mode: group=%s, server=%s, controlPort=%u, dataPort=%u.",
|
||||
"Multicast mode: group=%s, server=%s, controlPort=%u, dataPort=%u, interface=%s.",
|
||||
multicastGroup.Buffer(), serverAddress.Buffer(),
|
||||
static_cast<uint32>(port), static_cast<uint32>(dataPort));
|
||||
static_cast<uint32>(port), static_cast<uint32>(dataPort),
|
||||
(multicastInterface.Size() > 0u) ? multicastInterface.Buffer() : "default");
|
||||
}
|
||||
else {
|
||||
useMulticast = false;
|
||||
@@ -263,6 +267,9 @@ bool UDPStreamerClient::Initialise(StructuredDataI &data) {
|
||||
if (ok && useMulticast) {
|
||||
ok = cdb.Write("MulticastGroup", multicastGroup);
|
||||
if (ok) { ok = cdb.Write("DataPort", static_cast<uint32>(dataPort)); }
|
||||
if (ok && (multicastInterface.Size() > 0u)) {
|
||||
ok = cdb.Write("Interface", multicastInterface);
|
||||
}
|
||||
}
|
||||
if (ok) { ok = cdb.Write("MaxPayloadSize", maxPayloadSize); }
|
||||
if (ok) { ok = cdb.Write("KeepAliveInterval", keepAliveInterval); }
|
||||
|
||||
@@ -178,9 +178,10 @@ private:
|
||||
float32 silenceTimeout; /**< Seconds of no data before reconnect (sub-second allowed, 0 disables). */
|
||||
uint32 cpuMask; /**< Background thread CPU affinity. */
|
||||
uint32 stackSize; /**< Background thread stack size. */
|
||||
StreamString multicastGroup; /**< Multicast group IP; empty = unicast. */
|
||||
uint16 dataPort; /**< UDP port for DATA datagrams (multicast). */
|
||||
bool useMulticast; /**< True when MulticastGroup is set. */
|
||||
StreamString multicastGroup; /**< Multicast group IP; empty = unicast. */
|
||||
StreamString multicastInterface; /**< Local IPv4 address for multicast join; empty = INADDR_ANY. */
|
||||
uint16 dataPort; /**< UDP port for DATA datagrams (multicast). */
|
||||
bool useMulticast; /**< True when MulticastGroup is set. */
|
||||
|
||||
/* Signal metadata */
|
||||
uint32 numSigs; /**< Number of signals. */
|
||||
|
||||
@@ -85,6 +85,9 @@ bool UDPSClient::Initialise(StructuredDataI &data) {
|
||||
uint32 dpU32 = static_cast<uint32>(serverPort) + 1u;
|
||||
(void) data.Read("DataPort", dpU32);
|
||||
dataPort = static_cast<uint16>(dpU32);
|
||||
StreamString iface;
|
||||
(void) data.Read("Interface", iface);
|
||||
multicastInterface = iface;
|
||||
}
|
||||
|
||||
float32 silenceS = UDPS_CLIENT_DEFAULT_SILENCE_TIMEOUT_S;
|
||||
@@ -311,7 +314,12 @@ bool UDPSClient::ConnectMulticast() {
|
||||
return false;
|
||||
}
|
||||
|
||||
ok = mcastSocket.Join(multicastGroup.Buffer());
|
||||
if (multicastInterface.Size() > 0u) {
|
||||
ok = mcastSocket.Join(multicastGroup.Buffer(), multicastInterface.Buffer());
|
||||
}
|
||||
else {
|
||||
ok = mcastSocket.Join(multicastGroup.Buffer());
|
||||
}
|
||||
if (!ok) {
|
||||
REPORT_ERROR_STATIC(ErrorManagement::Warning,
|
||||
"UDPSClient: Could not join multicast group %s.",
|
||||
@@ -320,8 +328,9 @@ bool UDPSClient::ConnectMulticast() {
|
||||
return false;
|
||||
}
|
||||
REPORT_ERROR_STATIC(ErrorManagement::Information,
|
||||
"UDPSClient: Joined multicast group %s on port %u.",
|
||||
multicastGroup.Buffer(), static_cast<uint32>(dataPort));
|
||||
"UDPSClient: Joined multicast group %s on port %u via interface %s.",
|
||||
multicastGroup.Buffer(), static_cast<uint32>(dataPort),
|
||||
(multicastInterface.Size() > 0u) ? multicastInterface.Buffer() : "default");
|
||||
|
||||
// Now open the TCP control connection and announce ourselves
|
||||
if (!tcpSocket.Open()) {
|
||||
|
||||
@@ -116,7 +116,10 @@ public:
|
||||
* - ServerAddr (char*) Server IPv4 address. Required.
|
||||
* - Port (uint16) Server UDP port (unicast) or TCP listen port (multicast). Required.
|
||||
* - MulticastGroup (char*) IPv4 multicast address; presence enables multicast mode.
|
||||
* - Interface (char*) Network interface for multicast join (e.g. "lo"). Required when MulticastGroup is set.
|
||||
* - Interface (char*) Local IPv4 dotted-quad address (e.g. "127.0.0.1") of the interface on
|
||||
* which to join the multicast group. Optional; omitting it uses the
|
||||
* default-route interface (INADDR_ANY), which silently receives nothing
|
||||
* if the server sends on a different interface.
|
||||
* - DataPort (uint16) UDP multicast data port (defaults to Port+1).
|
||||
* - SilenceTimeout (float32) Seconds of no data before reconnect. Default 1.0.
|
||||
* Sub-second values allowed; 0 disables the check.
|
||||
@@ -201,6 +204,7 @@ private:
|
||||
StreamString serverAddr;
|
||||
uint16 serverPort;
|
||||
StreamString multicastGroup;
|
||||
StreamString multicastInterface;
|
||||
uint16 dataPort;
|
||||
bool useMulticast;
|
||||
uint64 silenceTimeoutTicks;
|
||||
|
||||
@@ -145,3 +145,8 @@ TEST(UDPStreamerClientGTest, TestExecute_ConnectConfigDataEndToEnd) {
|
||||
UDPStreamerClientTest test;
|
||||
ASSERT_TRUE(test.TestExecute_ConnectConfigDataEndToEnd());
|
||||
}
|
||||
|
||||
TEST(UDPStreamerClientGTest, TestExecute_MulticastReceivesDataOnInterface) {
|
||||
UDPStreamerClientTest test;
|
||||
ASSERT_TRUE(test.TestExecute_MulticastReceivesDataOnInterface());
|
||||
}
|
||||
|
||||
@@ -26,11 +26,14 @@
|
||||
/* Standard header includes */
|
||||
/*---------------------------------------------------------------------------*/
|
||||
#include <string.h>
|
||||
#include <netinet/in.h>
|
||||
#include <sys/socket.h>
|
||||
|
||||
/*---------------------------------------------------------------------------*/
|
||||
/* Project header includes */
|
||||
/*---------------------------------------------------------------------------*/
|
||||
#include "AdvancedErrorManagement.h"
|
||||
#include "BasicTCPSocket.h"
|
||||
#include "BasicUDPSocket.h"
|
||||
#include "ConfigurationDatabase.h"
|
||||
#include "GAM.h"
|
||||
@@ -1491,3 +1494,183 @@ bool UDPStreamerClientTest::TestExecute_ConnectConfigDataEndToEnd() {
|
||||
ObjectRegistryDatabase::Instance()->Purge();
|
||||
return ok;
|
||||
}
|
||||
|
||||
bool UDPStreamerClientTest::TestExecute_MulticastReceivesDataOnInterface() {
|
||||
using namespace MARTe;
|
||||
|
||||
static const uint16 controlPort = 44730u;
|
||||
static const uint16 dataPort = 44731u;
|
||||
static const char8 *const mcGroup = "239.0.0.7";
|
||||
static const char8 *const mcIface = "127.0.0.1";
|
||||
|
||||
static const char8 *const cfg =
|
||||
"+Test = {\n"
|
||||
" Class = RealTimeApplication\n"
|
||||
" +Functions = {\n"
|
||||
" Class = ReferenceContainer\n"
|
||||
" +Reader = {\n"
|
||||
" Class = UDPStreamerClientTestGAM\n"
|
||||
" InputSignals = {\n"
|
||||
" Counter = { DataSource = ClientDS Type = uint32 }\n"
|
||||
" }\n"
|
||||
" OutputSignals = {\n"
|
||||
" Counter = { DataSource = DDB Type = uint32 }\n"
|
||||
" }\n"
|
||||
" }\n"
|
||||
" }\n"
|
||||
" +Data = {\n"
|
||||
" Class = ReferenceContainer\n"
|
||||
" DefaultDataSource = DDB\n"
|
||||
" +DDB = { Class = GAMDataSource }\n"
|
||||
" +ClientDS = {\n"
|
||||
" Class = UDPStreamerClient\n"
|
||||
" ServerAddress = \"127.0.0.1\"\n"
|
||||
" Port = 44730\n"
|
||||
" MulticastGroup = \"239.0.0.7\"\n"
|
||||
" DataPort = 44731\n"
|
||||
" Interface = \"127.0.0.1\"\n"
|
||||
" MaxPayloadSize = 1400\n"
|
||||
" Signals = {\n"
|
||||
" Counter = { Type = uint32 }\n"
|
||||
" }\n"
|
||||
" }\n"
|
||||
" +Timings = { Class = TimingDataSource }\n"
|
||||
" }\n"
|
||||
" +States = {\n"
|
||||
" Class = ReferenceContainer\n"
|
||||
" +State1 = {\n"
|
||||
" Class = RealTimeState\n"
|
||||
" +Threads = {\n"
|
||||
" Class = ReferenceContainer\n"
|
||||
" +Thread1 = {\n"
|
||||
" Class = RealTimeThread\n"
|
||||
" Functions = { Reader }\n"
|
||||
" }\n"
|
||||
" }\n"
|
||||
" }\n"
|
||||
" }\n"
|
||||
" +Scheduler = {\n"
|
||||
" Class = GAMScheduler\n"
|
||||
" TimingDataSource = Timings\n"
|
||||
" }\n"
|
||||
"}\n";
|
||||
|
||||
ReferenceT<RealTimeApplication> app = LoadApplication(cfg);
|
||||
bool ok = app.IsValid();
|
||||
|
||||
/* Open a TCP listener for the control port BEFORE PrepareNextState so we
|
||||
* never miss the client's CONNECT. */
|
||||
BasicTCPSocket tcpListener;
|
||||
if (ok) {
|
||||
ok = tcpListener.Open() && tcpListener.Listen(controlPort, 5);
|
||||
}
|
||||
|
||||
if (ok) {
|
||||
ok = (app->PrepareNextState("State1") == ErrorManagement::NoError);
|
||||
}
|
||||
|
||||
/* Open the multicast data socket aimed at the group, with IP_MULTICAST_IF
|
||||
* set to 127.0.0.1 so the datagram leaves on loopback — exactly what
|
||||
* UDPSServer does. */
|
||||
BasicUDPSocket dataSocket;
|
||||
if (ok) {
|
||||
ok = dataSocket.Open();
|
||||
}
|
||||
if (ok) {
|
||||
struct in_addr localIf;
|
||||
localIf.s_addr = inet_addr(mcIface);
|
||||
int fd = static_cast<int>(dataSocket.GetWriteHandle());
|
||||
ok = (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_IF,
|
||||
&localIf, static_cast<socklen_t>(sizeof(localIf))) == 0);
|
||||
}
|
||||
if (ok) {
|
||||
ok = dataSocket.Connect(mcGroup, dataPort);
|
||||
}
|
||||
|
||||
/* Accept the client's TCP CONNECT. */
|
||||
BasicTCPSocket *clientConn = NULL_PTR(BasicTCPSocket *);
|
||||
if (ok) {
|
||||
clientConn = tcpListener.WaitConnection(TimeoutType(2000u));
|
||||
ok = (clientConn != NULL_PTR(BasicTCPSocket *));
|
||||
}
|
||||
|
||||
/* Read the CONNECT packet from the accepted TCP connection. */
|
||||
if (ok) {
|
||||
uint8 recvBuf[64u];
|
||||
uint32 recvSize = UDPS_HEADER_SIZE;
|
||||
ok = clientConn->Read(reinterpret_cast<char8 *>(recvBuf), recvSize);
|
||||
if (ok) {
|
||||
const UDPSPacketHeader *hdr = reinterpret_cast<const UDPSPacketHeader *>(recvBuf);
|
||||
ok = (hdr->magic == UDPS_MAGIC) && (hdr->type == UDPS_TYPE_CONNECT);
|
||||
}
|
||||
}
|
||||
|
||||
/* Send CONFIG: one scalar "Counter" uint32 signal, unquantised. */
|
||||
if (ok) {
|
||||
UDPSSignalDescriptor desc;
|
||||
(void) memset(&desc, 0, sizeof(desc));
|
||||
(void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u);
|
||||
desc.typeCode = UDPS_TYPECODE_UINT32;
|
||||
desc.numRows = 1u;
|
||||
desc.numCols = 1u;
|
||||
|
||||
uint8 buf[UDPS_HEADER_SIZE + 4u + UDPS_SIGNAL_DESC_SIZE + 1u];
|
||||
const uint32 configPayloadBytes = 4u + UDPS_SIGNAL_DESC_SIZE + 1u;
|
||||
UDPSBuildHeader(buf, UDPS_TYPE_CONFIG, 1u, 0u, 1u, configPayloadBytes);
|
||||
uint32 numSigs = 1u;
|
||||
(void) memcpy(buf + UDPS_HEADER_SIZE, &numSigs, 4u);
|
||||
(void) memcpy(buf + UDPS_HEADER_SIZE + 4u, &desc, UDPS_SIGNAL_DESC_SIZE);
|
||||
buf[UDPS_HEADER_SIZE + 4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT;
|
||||
|
||||
uint32 sendSize = static_cast<uint32>(sizeof(buf));
|
||||
ok = clientConn->Write(reinterpret_cast<const char8 *>(buf), sendSize);
|
||||
}
|
||||
|
||||
Sleep::MSec(100u);
|
||||
|
||||
ReferenceT<UDPStreamerClient> ds;
|
||||
if (ok) {
|
||||
ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS");
|
||||
ok = ds.IsValid();
|
||||
}
|
||||
|
||||
/* Send DATA over UDP multicast. Each attempt uses a fresh packet counter
|
||||
* so UDPSClient's reassembly layer does not drop retransmissions. */
|
||||
bool gotValue = false;
|
||||
for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) {
|
||||
SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds);
|
||||
|
||||
uint8 buf[UDPS_HEADER_SIZE + 8u + 4u];
|
||||
const uint32 dataPayloadBytes = 8u + 4u;
|
||||
UDPSBuildHeader(buf, UDPS_TYPE_DATA, 2u + attempt, 0u, 1u, dataPayloadBytes);
|
||||
(void) memset(buf + UDPS_HEADER_SIZE, 0, 8u);
|
||||
uint32 value = 424242u;
|
||||
(void) memcpy(buf + UDPS_HEADER_SIZE + 8u, &value, 4u);
|
||||
|
||||
uint32 sendSize = static_cast<uint32>(sizeof(buf));
|
||||
ok = dataSocket.Write(reinterpret_cast<const char8 *>(buf), sendSize);
|
||||
|
||||
if (ok && JoinSynchroniseThread(syncArgs)) {
|
||||
void *sigMem = NULL_PTR(void *);
|
||||
if (ds->GetSignalMemoryBuffer(0u, 0u, sigMem)) {
|
||||
uint32 decoded = 0u;
|
||||
(void) memcpy(&decoded, sigMem, sizeof(uint32));
|
||||
gotValue = (decoded == 424242u);
|
||||
}
|
||||
}
|
||||
else if (!ok) {
|
||||
(void) JoinSynchroniseThread(syncArgs);
|
||||
}
|
||||
}
|
||||
ok = ok && gotValue;
|
||||
|
||||
if (clientConn != NULL_PTR(BasicTCPSocket *)) {
|
||||
(void) clientConn->Close();
|
||||
delete clientConn;
|
||||
}
|
||||
(void) tcpListener.Close();
|
||||
(void) dataSocket.Close();
|
||||
Sleep::MSec(50u);
|
||||
ObjectRegistryDatabase::Instance()->Purge();
|
||||
return ok;
|
||||
}
|
||||
|
||||
@@ -158,6 +158,14 @@ public:
|
||||
* UDP sockets, mirroring the server side of the wire protocol.
|
||||
*/
|
||||
bool TestExecute_ConnectConfigDataEndToEnd();
|
||||
|
||||
/**
|
||||
* @brief Regression test: verifies that UDPStreamerClient receives multicast
|
||||
* DATA when Interface is set to "127.0.0.1", ensuring the two-argument
|
||||
* Join(group, interface) path in UDPSClient::ConnectMulticast is taken.
|
||||
* This test fails without the UDPSClient multicastInterface fix.
|
||||
*/
|
||||
bool TestExecute_MulticastReceivesDataOnInterface();
|
||||
};
|
||||
|
||||
#endif /* UDPSTREAMERCLIENTTEST_H_ */
|
||||
|
||||
Reference in New Issue
Block a user