From 14d5351a81f26e308695d486851015162dd69706 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Mon, 17 Aug 2026 08:23:41 +0200 Subject: [PATCH] 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 --- Docs/UDPStreamer.md | 44 +++++ .../UDPStreamerClient/UDPStreamerClient.cpp | 11 +- .../UDPStreamerClient/UDPStreamerClient.h | 7 +- .../Interfaces/UDPStream/UDPSClient.cpp | 15 +- .../Interfaces/UDPStream/UDPSClient.h | 6 +- .../UDPStreamerClientGTest.cpp | 5 + .../UDPStreamerClientTest.cpp | 183 ++++++++++++++++++ .../UDPStreamerClient/UDPStreamerClientTest.h | 8 + 8 files changed, 270 insertions(+), 9 deletions(-) diff --git a/Docs/UDPStreamer.md b/Docs/UDPStreamer.md index 28eb48f..661dd3b 100644 --- a/Docs/UDPStreamer.md +++ b/Docs/UDPStreamer.md @@ -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: ``` diff --git a/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.cpp b/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.cpp index f93feef..d420fea 100644 --- a/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.cpp +++ b/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.cpp @@ -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(port), static_cast(dataPort)); + static_cast(port), static_cast(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(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); } diff --git a/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h b/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h index 26be648..c5df749 100644 --- a/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h +++ b/Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.h @@ -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. */ diff --git a/Source/Components/Interfaces/UDPStream/UDPSClient.cpp b/Source/Components/Interfaces/UDPStream/UDPSClient.cpp index 8c9ed05..8a5598c 100644 --- a/Source/Components/Interfaces/UDPStream/UDPSClient.cpp +++ b/Source/Components/Interfaces/UDPStream/UDPSClient.cpp @@ -85,6 +85,9 @@ bool UDPSClient::Initialise(StructuredDataI &data) { uint32 dpU32 = static_cast(serverPort) + 1u; (void) data.Read("DataPort", dpU32); dataPort = static_cast(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(dataPort)); + "UDPSClient: Joined multicast group %s on port %u via interface %s.", + multicastGroup.Buffer(), static_cast(dataPort), + (multicastInterface.Size() > 0u) ? multicastInterface.Buffer() : "default"); // Now open the TCP control connection and announce ourselves if (!tcpSocket.Open()) { diff --git a/Source/Components/Interfaces/UDPStream/UDPSClient.h b/Source/Components/Interfaces/UDPStream/UDPSClient.h index 7eec743..6393336 100644 --- a/Source/Components/Interfaces/UDPStream/UDPSClient.h +++ b/Source/Components/Interfaces/UDPStream/UDPSClient.h @@ -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; diff --git a/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientGTest.cpp b/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientGTest.cpp index 6ae2b37..5e0b0c9 100644 --- a/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientGTest.cpp +++ b/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientGTest.cpp @@ -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()); +} diff --git a/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.cpp b/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.cpp index 2e5dfd9..78cb1c2 100644 --- a/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.cpp +++ b/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.cpp @@ -26,11 +26,14 @@ /* Standard header includes */ /*---------------------------------------------------------------------------*/ #include +#include +#include /*---------------------------------------------------------------------------*/ /* 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 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(dataSocket.GetWriteHandle()); + ok = (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_IF, + &localIf, static_cast(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(recvBuf), recvSize); + if (ok) { + const UDPSPacketHeader *hdr = reinterpret_cast(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(sizeof(buf)); + ok = clientConn->Write(reinterpret_cast(buf), sendSize); + } + + Sleep::MSec(100u); + + ReferenceT 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(sizeof(buf)); + ok = dataSocket.Write(reinterpret_cast(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; +} diff --git a/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.h b/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.h index f4c0f87..921c9da 100644 --- a/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.h +++ b/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.h @@ -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_ */