martino/MARTe-Integrated-Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added interface and added join to multicast

Browse Source
This commit is contained in:
Martino Ferrari
2026-07-25 12:26:51 +02:00
parent 2d5ca20ae4
commit 3e0a481c13
7 changed files with 1854 additions and 1844 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Docs/UDPStreamer.md
+4 -4
View File
@@ -67,7 +67,7 @@ thread.
### Top-level Parameters ### Top-level Parameters
| Parameter | Type | Default | Description | | Parameter | Type | Default | Description |
|-----------|------|---------|-------------| | ---------------- | ------ | --------- | ------------------------------------------- |
| `Port` | uint16 | 44500 | UDP server port | | `Port` | uint16 | 44500 | UDP server port |
| `MaxPayloadSize` | uint32 | 1400 | Max payload bytes per UDP datagram (min 18) | | `MaxPayloadSize` | uint32 | 1400 | Max payload bytes per UDP datagram (min 18) |
| `CPUMask` | uint32 | 0 (any) | Background thread CPU affinity | | `CPUMask` | uint32 | 0 (any) | Background thread CPU affinity |
@@ -76,7 +76,7 @@ thread.
### Per-signal Parameters ### Per-signal Parameters
| Parameter | Type | Default | Applies to | | Parameter | Type | Default | Applies to |
|-----------|------|---------|------------| | --------------- | ------- | ------------ | -------------------------------------------------------- |
| `Unit` | string | `""` | Any type — informational, forwarded to client in CONFIG | | `Unit` | string | `""` | Any type — informational, forwarded to client in CONFIG |
| `RangeMin` | float64 | 0.0 | float32/float64 with `QuantizedType` | | `RangeMin` | float64 | 0.0 | float32/float64 with `QuantizedType` |
| `RangeMax` | float64 | 1.0 | float32/float64 with `QuantizedType` | | `RangeMax` | float64 | 1.0 | float32/float64 with `QuantizedType` |
@@ -88,7 +88,7 @@ thread.
### Quantization Types ### Quantization Types
| Value | Wire type | Bit depth | Notes | | Value | Wire type | Bit depth | Notes |
|-------|-----------|-----------|-------| | -------- | -------------- | --------- | ------------------------------------------------- |
| `none` | same as source | — | Raw copy, no quantization | | `none` | same as source | — | Raw copy, no quantization |
| `uint8` | uint8 | 8-bit | Maps `[RangeMin, RangeMax]` → `[0, 255]` | | `uint8` | uint8 | 8-bit | Maps `[RangeMin, RangeMax]` → `[0, 255]` |
| `int8` | int8 | 8-bit | Maps `[RangeMin, RangeMax]` → `[-127, 127]` | | `int8` | int8 | 8-bit | Maps `[RangeMin, RangeMax]` → `[-127, 127]` |
@@ -105,7 +105,7 @@ wire_value = (uint16)(normalized × 65535)
### Time Modes ### Time Modes
| Value | Meaning | Requirements | | Value | Meaning | Requirements |
|-------|---------|--------------| | ------------- | ------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------- |
| `PacketTime` | The HRT counter captured at `Synchronise()` time is used as the packet timestamp. No per-signal time metadata. | — | | `PacketTime` | The HRT counter captured at `Synchronise()` time is used as the packet timestamp. No per-signal time metadata. | — |
| `FullArray` | `TimeSignal` carries one timestamp per element (same `NumberOfElements`). | `TimeSignal` must have the same `NumberOfElements`. | | `FullArray` | `TimeSignal` carries one timestamp per element (same `NumberOfElements`). | `TimeSignal` must have the same `NumberOfElements`. |
| `FirstSample` | `TimeSignal` is a scalar giving the timestamp of element `[0]`. Elements `[1..N-1]` are inferred at `1/SamplingRate` intervals. | Scalar `TimeSignal`; `SamplingRate > 0`. | | `FirstSample` | `TimeSignal` is a scalar giving the timestamp of element `[0]`. Elements `[1..N-1]` are inferred at `1/SamplingRate` intervals. | Scalar `TimeSignal`; `SamplingRate > 0`. |
README.md
+3 -2
View File
@@ -10,7 +10,7 @@ for control applications built with [MARTe2](https://vcis.f4e.europa.eu/marte2-d
This repository integrates two complementary capabilities: This repository integrates two complementary capabilities:
| Capability | Component | Purpose | | Capability | Component | Purpose |
|---|---|---| | --------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------- |
| **Signal streaming** | `UDPStreamer` DataSource | Continuously stream selected signals to a browser-based oscilloscope over UDP | | **Signal streaming** | `UDPStreamer` DataSource | Continuously stream selected signals to a browser-based oscilloscope over UDP |
| **Signal debugging** | `DebugService` Interface | On-demand signal tracing, value forcing, and conditional breakpoints — zero application code changes required | | **Signal debugging** | `DebugService` Interface | On-demand signal tracing, value forcing, and conditional breakpoints — zero application code changes required |
| **Sine generation** | `SineArrayGAM` | Generate continuous sine-wave arrays for testing and simulation | | **Sine generation** | `SineArrayGAM` | Generate continuous sine-wave arrays for testing and simulation |
@@ -81,6 +81,7 @@ Instruments a running MARTe2 application **without modifying its source code**.
afterward the application transparently uses the wrapped brokers. afterward the application transparently uses the wrapped brokers.
Capabilities accessible over TCP (port 8080 by default): Capabilities accessible over TCP (port 8080 by default):
- `DISCOVER` — enumerate all signals with type and alias metadata - `DISCOVER` — enumerate all signals with type and alias metadata
- `TRACE` — enable/disable high-speed UDP telemetry per signal (with decimation) - `TRACE` — enable/disable high-speed UDP telemetry per signal (with decimation)
- `FORCE` / `UNFORCE` — inject persistent values into signals on the RT path - `FORCE` / `UNFORCE` — inject persistent values into signals on the RT path
@@ -225,7 +226,7 @@ Open `http://localhost:9090`, explore the object tree, trace signals, force valu
## Documentation ## Documentation
| Document | Contents | | Document | Contents |
|---|---| | ----------------------------- | -------------------------------------------------------------- |
| `Docs/Protocol.md` | UDPS binary wire protocol specification | | `Docs/Protocol.md` | UDPS binary wire protocol specification |
| `Docs/UDPStreamer.md` | UDPStreamer DataSource configuration reference | | `Docs/UDPStreamer.md` | UDPStreamer DataSource configuration reference |
| `Docs/SineArrayGAM.md` | SineArrayGAM configuration reference | | `Docs/SineArrayGAM.md` | SineArrayGAM configuration reference |
Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp
+223 -203
View File
@@ -21,6 +21,8 @@
* methods, such as those inline could be defined on the header file, instead. * methods, such as those inline could be defined on the header file, instead.
*/ */
#include "ErrorType.h"
#include "StreamString.h"
#define DLL_API #define DLL_API
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
@@ -37,10 +39,7 @@
#include "EmbeddedThreadI.h" #include "EmbeddedThreadI.h"
#include "GlobalObjectsDatabase.h" #include "GlobalObjectsDatabase.h"
#include "HighResolutionTimer.h" #include "HighResolutionTimer.h"
#include "MemoryMapSynchronisedOutputBroker.h"
#include "MemoryOperationsHelper.h" #include "MemoryOperationsHelper.h"
#include "Sleep.h"
#include "Threads.h"
#include "UDPStreamer.h" #include "UDPStreamer.h"
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
@@ -52,7 +51,8 @@ namespace MARTe {
/** Default port used when none is specified. */ /** Default port used when none is specified. */
static const uint16 UDPS_DEFAULT_PORT = 44500u; static const uint16 UDPS_DEFAULT_PORT = 44500u;
/** Default data port offset: dataPort = port + this value when DataPort is not specified. */ /** Default data port offset: dataPort = port + this value when DataPort is not
* specified. */
static const uint16 UDPS_DEFAULT_DATA_PORT_OFFSET = 1u; static const uint16 UDPS_DEFAULT_DATA_PORT_OFFSET = 1u;
/** Maximum pending TCP connections on the listener backlog. */ /** Maximum pending TCP connections on the listener backlog. */
@@ -80,10 +80,8 @@ static const uint32 UDPS_TIMESTAMP_BYTES = 8u;
/* Method definitions */ /* Method definitions */
/*---------------------------------------------------------------------------*/ /*---------------------------------------------------------------------------*/
UDPStreamer::UDPStreamer() : UDPStreamer::UDPStreamer()
MemoryDataSourceI(), : MemoryDataSourceI(), EmbeddedServiceMethodBinderI(), executor(*this) {
EmbeddedServiceMethodBinderI(),
executor(*this) {
port = UDPS_DEFAULT_PORT; port = UDPS_DEFAULT_PORT;
maxPayloadSize = UDPS_DEFAULT_MAX_PAYLOAD; maxPayloadSize = UDPS_DEFAULT_MAX_PAYLOAD;
cpuMask = 0xFFFFFFFFu; cpuMask = 0xFFFFFFFFu;
@@ -121,7 +119,7 @@ UDPStreamer::UDPStreamer() :
/*lint -e{1551} Destructor must guarantee thread and socket cleanup. */ /*lint -e{1551} Destructor must guarantee thread and socket cleanup. */
UDPStreamer::~UDPStreamer() { UDPStreamer::~UDPStreamer() {
/* Unblock the background thread's dataSem wait so it can exit */ /* Unblock the background thread's dataSem wait so it can exit */
(void) dataSem.Post(); (void)dataSem.Post();
if (executor.GetStatus() != EmbeddedThreadI::OffState) { if (executor.GetStatus() != EmbeddedThreadI::OffState) {
if (!executor.Stop()) { if (!executor.Stop()) {
@@ -134,7 +132,7 @@ UDPStreamer::~UDPStreamer() {
} }
} }
(void) server.Stop(); (void)server.Stop();
HeapI *heapAccum = GlobalObjectsDatabase::Instance()->GetStandardHeap(); HeapI *heapAccum = GlobalObjectsDatabase::Instance()->GetStandardHeap();
if (accumBuffer != NULL_PTR(uint8 *)) { if (accumBuffer != NULL_PTR(uint8 *)) {
@@ -169,7 +167,7 @@ UDPStreamer::~UDPStreamer() {
heap->Free(reinterpret_cast<void *&>(wireBuffer)); heap->Free(reinterpret_cast<void *&>(wireBuffer));
} }
(void) dataSem.Close(); (void)dataSem.Close();
} }
bool UDPStreamer::Initialise(StructuredDataI &data) { bool UDPStreamer::Initialise(StructuredDataI &data) {
@@ -228,18 +226,16 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
if (ok) { if (ok) {
StreamString publishStr = ""; StreamString publishStr = "";
(void) data.Read("PublishingMode", publishStr); (void)data.Read("PublishingMode", publishStr);
if ((publishStr.Size() == 0u) || (publishStr == "Strict")) { if ((publishStr.Size() == 0u) || (publishStr == "Strict")) {
publishMode = UDPStreamerPublishStrict; publishMode = UDPStreamerPublishStrict;
} } else if (publishStr == "Accumulate") {
else if (publishStr == "Accumulate") {
publishMode = UDPStreamerPublishAccumulate; publishMode = UDPStreamerPublishAccumulate;
} } else if (publishStr == "Decimate") {
else if (publishStr == "Decimate") {
publishMode = UDPStreamerPublishDecimate; publishMode = UDPStreamerPublishDecimate;
} } else {
else { REPORT_ERROR(
REPORT_ERROR(ErrorManagement::ParametersError, ErrorManagement::ParametersError,
"Unknown PublishingMode '%s'. Allowed: Strict|Accumulate|Decimate.", "Unknown PublishingMode '%s'. Allowed: Strict|Accumulate|Decimate.",
publishStr.Buffer()); publishStr.Buffer());
ok = false; ok = false;
@@ -250,18 +246,19 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
/* MinRefreshRate controls the time-based flush: flush when /* MinRefreshRate controls the time-based flush: flush when
* (now - lastPublishTs) >= flushPeriodTicks, or when adding one more * (now - lastPublishTs) >= flushPeriodTicks, or when adding one more
* sample would overflow MaxPayloadSize. Whichever fires first. */ * sample would overflow MaxPayloadSize. Whichever fires first. */
if (!data.Read("MinRefreshRate", minRefreshRate) || (minRefreshRate <= 0.0)) { if (!data.Read("MinRefreshRate", minRefreshRate) ||
REPORT_ERROR(ErrorManagement::ParametersError, (minRefreshRate <= 0.0)) {
REPORT_ERROR(
ErrorManagement::ParametersError,
"MinRefreshRate > 0 is required when PublishingMode = Accumulate."); "MinRefreshRate > 0 is required when PublishingMode = Accumulate.");
ok = false; ok = false;
} } else {
else {
float64 hrtFreq = static_cast<float64>(HighResolutionTimer::Frequency()); float64 hrtFreq = static_cast<float64>(HighResolutionTimer::Frequency());
flushPeriodTicks = static_cast<uint64>(hrtFreq / minRefreshRate); flushPeriodTicks = static_cast<uint64>(hrtFreq / minRefreshRate);
REPORT_ERROR(ErrorManagement::Information, REPORT_ERROR(
ErrorManagement::Information,
"Accumulate mode: MinRefreshRate=%.1f Hz, flushPeriodTicks=%llu.", "Accumulate mode: MinRefreshRate=%.1f Hz, flushPeriodTicks=%llu.",
minRefreshRate, minRefreshRate, static_cast<unsigned long long>(flushPeriodTicks));
static_cast<unsigned long long>(flushPeriodTicks));
} }
} }
@@ -272,11 +269,11 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
REPORT_ERROR(ErrorManagement::ParametersError, REPORT_ERROR(ErrorManagement::ParametersError,
"Ratio >= 1 is required when PublishingMode = Decimate."); "Ratio >= 1 is required when PublishingMode = Decimate.");
ok = false; ok = false;
} } else {
else {
decimateRatio = ratio; decimateRatio = ratio;
if (decimateRatio == 1u) { if (decimateRatio == 1u) {
REPORT_ERROR(ErrorManagement::Warning, REPORT_ERROR(
ErrorManagement::Warning,
"Decimate mode with Ratio=1 is equivalent to Strict mode."); "Decimate mode with Ratio=1 is equivalent to Strict mode.");
} }
REPORT_ERROR(ErrorManagement::Information, REPORT_ERROR(ErrorManagement::Information,
@@ -289,20 +286,28 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
// Build server config with already-resolved Port and MaxPayloadSize so // Build server config with already-resolved Port and MaxPayloadSize so
// UDPSServer::Initialise() always sees them, even when defaults were used. // UDPSServer::Initialise() always sees them, even when defaults were used.
ConfigurationDatabase serverCfg; ConfigurationDatabase serverCfg;
(void) serverCfg.Write("Port", static_cast<uint32>(port)); (void)serverCfg.Write("Port", static_cast<uint32>(port));
(void) serverCfg.Write("MaxPayloadSize", maxPayloadSize); (void)serverCfg.Write("MaxPayloadSize", maxPayloadSize);
// Forward optional multicast / timeout params if present in caller's data. // Forward optional multicast / timeout params if present in caller's data.
StreamString mcGroup; StreamString mcGroup;
if (data.Read("MulticastGroup", mcGroup) && (mcGroup.Size() > 0u)) { if (data.Read("MulticastGroup", mcGroup) && (mcGroup.Size() > 0u)) {
(void) serverCfg.Write("MulticastGroup", mcGroup.Buffer()); (void)serverCfg.Write("MulticastGroup", mcGroup.Buffer());
uint32 dp = 0u; uint32 dp = 0u;
if (data.Read("DataPort", dp)) { if (data.Read("DataPort", dp)) {
(void) serverCfg.Write("DataPort", dp); (void)serverCfg.Write("DataPort", dp);
}
StreamString iface;
if (data.Read("Interface", iface)) {
(void)serverCfg.Write("Interface", iface);
} else {
ok = false;
REPORT_ERROR(ErrorManagement::InitialisationError,
"Missing mandatory interface for multicasting");
} }
} }
uint32 clientTimeout = 0u; uint32 clientTimeout = 0u;
if (data.Read("ClientTimeout", clientTimeout)) { if (data.Read("ClientTimeout", clientTimeout)) {
(void) serverCfg.Write("ClientTimeout", clientTimeout); (void)serverCfg.Write("ClientTimeout", clientTimeout);
} }
ok = server.Initialise(serverCfg); ok = server.Initialise(serverCfg);
} }
@@ -358,16 +363,16 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
timeSignalNames[i] = ""; timeSignalNames[i] = "";
uint8 ndims = 0u; uint8 ndims = 0u;
(void) GetSignalNumberOfDimensions(i, ndims); (void)GetSignalNumberOfDimensions(i, ndims);
signalInfos[i].numDimensions = ndims; signalInfos[i].numDimensions = ndims;
uint32 nelems = 1u; uint32 nelems = 1u;
(void) GetSignalNumberOfElements(i, nelems); (void)GetSignalNumberOfElements(i, nelems);
signalInfos[i].numElements = nelems; signalInfos[i].numElements = nelems;
signalInfos[i].numCols = nelems; signalInfos[i].numCols = nelems;
uint32 bsz = 0u; uint32 bsz = 0u;
(void) GetSignalByteSize(i, bsz); (void)GetSignalByteSize(i, bsz);
signalInfos[i].srcByteSize = bsz; signalInfos[i].srcByteSize = bsz;
signalInfos[i].bufferOffset = totalSrcBytes; signalInfos[i].bufferOffset = totalSrcBytes;
totalSrcBytes += bsz; totalSrcBytes += bsz;
@@ -377,7 +382,7 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
* Note: DataSourceI::AddSignals() leaves signalsDatabase positioned at the * Note: DataSourceI::AddSignals() leaves signalsDatabase positioned at the
* "Signals" node. We must reset to root before navigating. */ * "Signals" node. We must reset to root before navigating. */
if (ok) { if (ok) {
(void) signalsDatabase.MoveToRoot(); (void)signalsDatabase.MoveToRoot();
bool moved = signalsDatabase.MoveRelative("Signals"); bool moved = signalsDatabase.MoveRelative("Signals");
if (!moved) { if (!moved) {
REPORT_ERROR(ErrorManagement::FatalError, REPORT_ERROR(ErrorManagement::FatalError,
@@ -395,32 +400,27 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
/* Unit */ /* Unit */
StreamString unit = ""; StreamString unit = "";
(void) signalsDatabase.Read("Unit", unit); (void)signalsDatabase.Read("Unit", unit);
signalInfos[i].unit = unit; signalInfos[i].unit = unit;
/* Range */ /* Range */
(void) signalsDatabase.Read("RangeMin", signalInfos[i].rangeMin); (void)signalsDatabase.Read("RangeMin", signalInfos[i].rangeMin);
(void) signalsDatabase.Read("RangeMax", signalInfos[i].rangeMax); (void)signalsDatabase.Read("RangeMax", signalInfos[i].rangeMax);
/* QuantizedType */ /* QuantizedType */
StreamString quantStr = ""; StreamString quantStr = "";
if (signalsDatabase.Read("QuantizedType", quantStr)) { if (signalsDatabase.Read("QuantizedType", quantStr)) {
if (quantStr == "uint8") { if (quantStr == "uint8") {
signalInfos[i].quantType = UDPStreamerQuantUint8; signalInfos[i].quantType = UDPStreamerQuantUint8;
} } else if (quantStr == "int8") {
else if (quantStr == "int8") {
signalInfos[i].quantType = UDPStreamerQuantInt8; signalInfos[i].quantType = UDPStreamerQuantInt8;
} } else if (quantStr == "uint16") {
else if (quantStr == "uint16") {
signalInfos[i].quantType = UDPStreamerQuantUint16; signalInfos[i].quantType = UDPStreamerQuantUint16;
} } else if (quantStr == "int16") {
else if (quantStr == "int16") {
signalInfos[i].quantType = UDPStreamerQuantInt16; signalInfos[i].quantType = UDPStreamerQuantInt16;
} } else if (quantStr == "none") {
else if (quantStr == "none") {
signalInfos[i].quantType = UDPStreamerQuantNone; signalInfos[i].quantType = UDPStreamerQuantNone;
} } else {
else {
REPORT_ERROR(ErrorManagement::ParametersError, REPORT_ERROR(ErrorManagement::ParametersError,
"Signal %s: unknown QuantizedType '%s'. " "Signal %s: unknown QuantizedType '%s'. "
"Allowed: none|uint8|int8|uint16|int16.", "Allowed: none|uint8|int8|uint16|int16.",
@@ -443,23 +443,19 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
/* TimeMode */ /* TimeMode */
if (ok) { if (ok) {
StreamString timeModeStr; StreamString timeModeStr;
(void) signalsDatabase.Read("TimeMode", timeModeStr); (void)signalsDatabase.Read("TimeMode", timeModeStr);
if (timeModeStr.Size() == 0u) { if (timeModeStr.Size() == 0u) {
timeModeStr = "PacketTime"; timeModeStr = "PacketTime";
} }
if (timeModeStr == "PacketTime") { if (timeModeStr == "PacketTime") {
signalInfos[i].timeMode = UDPStreamerTimePacket; signalInfos[i].timeMode = UDPStreamerTimePacket;
} } else if (timeModeStr == "FullArray") {
else if (timeModeStr == "FullArray") {
signalInfos[i].timeMode = UDPStreamerTimeFullArray; signalInfos[i].timeMode = UDPStreamerTimeFullArray;
} } else if (timeModeStr == "FirstSample") {
else if (timeModeStr == "FirstSample") {
signalInfos[i].timeMode = UDPStreamerTimeFirstSample; signalInfos[i].timeMode = UDPStreamerTimeFirstSample;
} } else if (timeModeStr == "LastSample") {
else if (timeModeStr == "LastSample") {
signalInfos[i].timeMode = UDPStreamerTimeLastSample; signalInfos[i].timeMode = UDPStreamerTimeLastSample;
} } else {
else {
REPORT_ERROR(ErrorManagement::ParametersError, REPORT_ERROR(ErrorManagement::ParametersError,
"Signal %s: unknown TimeMode '%s'. " "Signal %s: unknown TimeMode '%s'. "
"Allowed: PacketTime|FullArray|FirstSample|LastSample.", "Allowed: PacketTime|FullArray|FirstSample|LastSample.",
@@ -477,8 +473,7 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
"TimeMode != PacketTime.", "TimeMode != PacketTime.",
signalInfos[i].name.Buffer()); signalInfos[i].name.Buffer());
ok = false; ok = false;
} } else {
else {
timeSignalNames[i] = tsName; timeSignalNames[i] = tsName;
/* Index resolved in pass 3 */ /* Index resolved in pass 3 */
signalInfos[i].timeSignalIdx = UDPS_NO_TIME_SIGNAL; signalInfos[i].timeSignalIdx = UDPS_NO_TIME_SIGNAL;
@@ -487,7 +482,7 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
/* SamplingRate */ /* SamplingRate */
if (ok) { if (ok) {
(void) signalsDatabase.Read("SamplingRate", signalInfos[i].samplingRate); (void)signalsDatabase.Read("SamplingRate", signalInfos[i].samplingRate);
bool needsRate = (signalInfos[i].timeMode == UDPStreamerTimeFirstSample || bool needsRate = (signalInfos[i].timeMode == UDPStreamerTimeFirstSample ||
signalInfos[i].timeMode == UDPStreamerTimeLastSample); signalInfos[i].timeMode == UDPStreamerTimeLastSample);
if (needsRate && (signalInfos[i].samplingRate <= 0.0)) { if (needsRate && (signalInfos[i].samplingRate <= 0.0)) {
@@ -499,11 +494,11 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
} }
} }
(void) signalsDatabase.MoveToAncestor(1u); (void)signalsDatabase.MoveToAncestor(1u);
} }
if (ok || true) { /* always attempt to restore navigation */ if (ok || true) { /* always attempt to restore navigation */
(void) signalsDatabase.MoveToAncestor(1u); (void)signalsDatabase.MoveToAncestor(1u);
} }
/* --- Pass 3: resolve TimeSignal names to signal indices --- */ /* --- Pass 3: resolve TimeSignal names to signal indices --- */
@@ -520,10 +515,10 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
} }
} }
if (!found) { if (!found) {
REPORT_ERROR(ErrorManagement::ParametersError, REPORT_ERROR(
ErrorManagement::ParametersError,
"Signal %s: TimeSignal '%s' not found among declared signals.", "Signal %s: TimeSignal '%s' not found among declared signals.",
signalInfos[i].name.Buffer(), signalInfos[i].name.Buffer(), timeSignalNames[i].Buffer());
timeSignalNames[i].Buffer());
ok = false; ok = false;
} }
} }
@@ -567,12 +562,11 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
"Signal %s: FullArray TimeMode requires TimeSignal " "Signal %s: FullArray TimeMode requires TimeSignal "
"%s to have the same NumberOfElements (%u vs %u).", "%s to have the same NumberOfElements (%u vs %u).",
signalInfos[i].name.Buffer(), signalInfos[i].name.Buffer(),
signalInfos[tsIdx].name.Buffer(), signalInfos[tsIdx].name.Buffer(), tsElems,
tsElems, signalInfos[i].numElements); signalInfos[i].numElements);
ok = false; ok = false;
} }
} } else if ((signalInfos[i].timeMode == UDPStreamerTimeFirstSample) ||
else if ((signalInfos[i].timeMode == UDPStreamerTimeFirstSample) ||
(signalInfos[i].timeMode == UDPStreamerTimeLastSample)) { (signalInfos[i].timeMode == UDPStreamerTimeLastSample)) {
if (tsElems != 1u) { if (tsElems != 1u) {
REPORT_ERROR(ErrorManagement::ParametersError, REPORT_ERROR(ErrorManagement::ParametersError,
@@ -605,9 +599,11 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
*/ */
if (ok && (publishMode == UDPStreamerPublishAccumulate)) { if (ok && (publishMode == UDPStreamerPublishAccumulate)) {
/* Find primary time signal: prefer Unit="us"/"ns", fall back to first integer scalar */ /* Find primary time signal: prefer Unit="us"/"ns", fall back to first
* integer scalar */
uint32 primaryTsIdx = UDPS_NO_TIME_SIGNAL; uint32 primaryTsIdx = UDPS_NO_TIME_SIGNAL;
for (uint32 i = 0u; i < numSigs && (primaryTsIdx == UDPS_NO_TIME_SIGNAL); i++) { for (uint32 i = 0u; i < numSigs && (primaryTsIdx == UDPS_NO_TIME_SIGNAL);
i++) {
if (signalInfos[i].numElements == 1u) { if (signalInfos[i].numElements == 1u) {
if ((signalInfos[i].unit == "us") || (signalInfos[i].unit == "ns")) { if ((signalInfos[i].unit == "us") || (signalInfos[i].unit == "ns")) {
primaryTsIdx = i; primaryTsIdx = i;
@@ -615,7 +611,8 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
} }
} }
if (primaryTsIdx == UDPS_NO_TIME_SIGNAL) { if (primaryTsIdx == UDPS_NO_TIME_SIGNAL) {
for (uint32 i = 0u; i < numSigs && (primaryTsIdx == UDPS_NO_TIME_SIGNAL); i++) { for (uint32 i = 0u; i < numSigs && (primaryTsIdx == UDPS_NO_TIME_SIGNAL);
i++) {
if (signalInfos[i].numElements == 1u) { if (signalInfos[i].numElements == 1u) {
TypeDescriptor td = signalInfos[i].type; TypeDescriptor td = signalInfos[i].type;
if ((td == UnsignedInteger32Bit) || (td == UnsignedInteger64Bit) || if ((td == UnsignedInteger32Bit) || (td == UnsignedInteger64Bit) ||
@@ -639,8 +636,8 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
signalInfos[i].accumulated = true; signalInfos[i].accumulated = true;
singleCycleWireBytes += signalInfos[i].wireByteSize; singleCycleWireBytes += signalInfos[i].wireByteSize;
/* Auto-assign time reference for non-primary, non-time scalars */ /* Auto-assign time reference for non-primary, non-time scalars */
if ((signalInfos[i].numElements == 1u) && if ((signalInfos[i].numElements == 1u) && (i != primaryTsIdx) &&
(i != primaryTsIdx) && (primaryTsIdx != UDPS_NO_TIME_SIGNAL) && (primaryTsIdx != UDPS_NO_TIME_SIGNAL) &&
(signalInfos[i].timeMode == UDPStreamerTimePacket)) { (signalInfos[i].timeMode == UDPStreamerTimePacket)) {
signalInfos[i].timeMode = UDPStreamerTimeFullArray; signalInfos[i].timeMode = UDPStreamerTimeFullArray;
signalInfos[i].timeSignalIdx = primaryTsIdx; signalInfos[i].timeSignalIdx = primaryTsIdx;
@@ -655,13 +652,13 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
if (ok) { if (ok) {
/* DATA payload: [8 HRT][4 numSamples][numSamples × singleCycle] */ /* DATA payload: [8 HRT][4 numSamples][numSamples × singleCycle] */
static const uint32 ACCUM_HEADER = UDPS_TIMESTAMP_BYTES + 4u; /* 12 bytes */ static const uint32 ACCUM_HEADER =
UDPS_TIMESTAMP_BYTES + 4u; /* 12 bytes */
if ((ACCUM_HEADER + singleCycleWireBytes) > maxPayloadSize) { if ((ACCUM_HEADER + singleCycleWireBytes) > maxPayloadSize) {
REPORT_ERROR(ErrorManagement::ParametersError, REPORT_ERROR(ErrorManagement::ParametersError,
"Accumulate mode: even a single sample (%u B) exceeds " "Accumulate mode: even a single sample (%u B) exceeds "
"MaxPayloadSize (%u B).", "MaxPayloadSize (%u B).",
ACCUM_HEADER + singleCycleWireBytes, ACCUM_HEADER + singleCycleWireBytes, maxPayloadSize);
maxPayloadSize);
ok = false; ok = false;
} }
} }
@@ -674,8 +671,8 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
REPORT_ERROR(ErrorManagement::Information, REPORT_ERROR(ErrorManagement::Information,
"Accumulate mode: singleCycleWireBytes=%u, " "Accumulate mode: singleCycleWireBytes=%u, "
"maxBatchCount=%u, maxPayloadSize=%u, totalWireBytes=%u.", "maxBatchCount=%u, maxPayloadSize=%u, totalWireBytes=%u.",
singleCycleWireBytes, singleCycleWireBytes, maxBatchCount, maxPayloadSize,
maxBatchCount, maxPayloadSize, totalWireBytes); totalWireBytes);
} }
} }
@@ -695,15 +692,18 @@ bool UDPStreamer::AllocateMemory() {
HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap(); HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap();
/* In Accumulate mode, readyBuffer / scratchBuffer hold maxBatchCount consecutive /* In Accumulate mode, readyBuffer / scratchBuffer hold maxBatchCount
* snapshots instead of a single one. */ * consecutive snapshots instead of a single one. */
/* HI-3: use 64-bit arithmetic to prevent overflow in maxBatchCount * totalSrcBytes */ /* HI-3: use 64-bit arithmetic to prevent overflow in maxBatchCount *
* totalSrcBytes */
uint64 readyBufSize64 = (maxBatchCount > 0u) uint64 readyBufSize64 = (maxBatchCount > 0u)
? (static_cast<uint64>(maxBatchCount) * static_cast<uint64>(totalSrcBytes)) ? (static_cast<uint64>(maxBatchCount) *
static_cast<uint64>(totalSrcBytes))
: static_cast<uint64>(totalSrcBytes); : static_cast<uint64>(totalSrcBytes);
if (readyBufSize64 > 0xFFFFFFFFu) { if (readyBufSize64 > 0xFFFFFFFFu) {
REPORT_ERROR(ErrorManagement::FatalError, REPORT_ERROR(ErrorManagement::FatalError,
"Accumulate buffer size overflow (maxBatchCount=%u * totalSrcBytes=%u).", "Accumulate buffer size overflow (maxBatchCount=%u * "
"totalSrcBytes=%u).",
maxBatchCount, totalSrcBytes); maxBatchCount, totalSrcBytes);
return false; return false;
} }
@@ -712,18 +712,20 @@ bool UDPStreamer::AllocateMemory() {
/* readyBuffer: copy of signal memory shared with background thread */ /* readyBuffer: copy of signal memory shared with background thread */
readyBuffer = reinterpret_cast<uint8 *>(heap->Malloc(readyBufSize)); readyBuffer = reinterpret_cast<uint8 *>(heap->Malloc(readyBufSize));
if (readyBuffer == NULL_PTR(uint8 *)) { if (readyBuffer == NULL_PTR(uint8 *)) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate readyBuffer."); REPORT_ERROR(ErrorManagement::FatalError,
"Could not allocate readyBuffer.");
return false; return false;
} }
(void) MemoryOperationsHelper::Set(readyBuffer, 0, readyBufSize); (void)MemoryOperationsHelper::Set(readyBuffer, 0, readyBufSize);
/* scratchBuffer: background-thread-private copy for serialization */ /* scratchBuffer: background-thread-private copy for serialization */
scratchBuffer = reinterpret_cast<uint8 *>(heap->Malloc(readyBufSize)); scratchBuffer = reinterpret_cast<uint8 *>(heap->Malloc(readyBufSize));
if (scratchBuffer == NULL_PTR(uint8 *)) { if (scratchBuffer == NULL_PTR(uint8 *)) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate scratchBuffer."); REPORT_ERROR(ErrorManagement::FatalError,
"Could not allocate scratchBuffer.");
return false; return false;
} }
(void) MemoryOperationsHelper::Set(scratchBuffer, 0, readyBufSize); (void)MemoryOperationsHelper::Set(scratchBuffer, 0, readyBufSize);
/* wireBuffer: serialized/quantized payload for transmission */ /* wireBuffer: serialized/quantized payload for transmission */
wireBuffer = reinterpret_cast<uint8 *>(heap->Malloc(totalWireBytes)); wireBuffer = reinterpret_cast<uint8 *>(heap->Malloc(totalWireBytes));
@@ -731,7 +733,7 @@ bool UDPStreamer::AllocateMemory() {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate wireBuffer."); REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate wireBuffer.");
return false; return false;
} }
(void) MemoryOperationsHelper::Set(wireBuffer, 0, totalWireBytes); (void)MemoryOperationsHelper::Set(wireBuffer, 0, totalWireBytes);
/* Update buffer offsets to match actual MemoryDataSourceI layout */ /* Update buffer offsets to match actual MemoryDataSourceI layout */
for (uint32 i = 0u; i < numSigs; i++) { for (uint32 i = 0u; i < numSigs; i++) {
@@ -744,14 +746,16 @@ bool UDPStreamer::AllocateMemory() {
/* --- Accumulate-mode extra buffers --- */ /* --- Accumulate-mode extra buffers --- */
if (maxBatchCount > 0u) { if (maxBatchCount > 0u) {
/* Linear fill buffer: RT thread writes one snapshot per slot (0..maxBatchCount-1) */ /* Linear fill buffer: RT thread writes one snapshot per slot
* (0..maxBatchCount-1) */
uint32 accumBufSize = maxBatchCount * totalSrcBytes; uint32 accumBufSize = maxBatchCount * totalSrcBytes;
accumBuffer = reinterpret_cast<uint8 *>(heap->Malloc(accumBufSize)); accumBuffer = reinterpret_cast<uint8 *>(heap->Malloc(accumBufSize));
if (accumBuffer == NULL_PTR(uint8 *)) { if (accumBuffer == NULL_PTR(uint8 *)) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate accumBuffer."); REPORT_ERROR(ErrorManagement::FatalError,
"Could not allocate accumBuffer.");
return false; return false;
} }
(void) MemoryOperationsHelper::Set(accumBuffer, 0, accumBufSize); (void)MemoryOperationsHelper::Set(accumBuffer, 0, accumBufSize);
/* Per-slot HRT timestamp arrays */ /* Per-slot HRT timestamp arrays */
accumTimestamps = new uint64[maxBatchCount]; accumTimestamps = new uint64[maxBatchCount];
@@ -765,18 +769,19 @@ bool UDPStreamer::AllocateMemory() {
return false; return false;
} }
uint32 tsBytes = maxBatchCount * static_cast<uint32>(sizeof(uint64)); uint32 tsBytes = maxBatchCount * static_cast<uint32>(sizeof(uint64));
(void) MemoryOperationsHelper::Set( (void)MemoryOperationsHelper::Set(
reinterpret_cast<uint8 *>(accumTimestamps), 0, tsBytes); reinterpret_cast<uint8 *>(accumTimestamps), 0, tsBytes);
(void) MemoryOperationsHelper::Set( (void)MemoryOperationsHelper::Set(
reinterpret_cast<uint8 *>(readyTimestamps), 0, tsBytes); reinterpret_cast<uint8 *>(readyTimestamps), 0, tsBytes);
(void) MemoryOperationsHelper::Set( (void)MemoryOperationsHelper::Set(
reinterpret_cast<uint8 *>(scratchTimestamps), 0, tsBytes); reinterpret_cast<uint8 *>(scratchTimestamps), 0, tsBytes);
accumFill = 0u; accumFill = 0u;
readyFill = 0u; readyFill = 0u;
REPORT_ERROR(ErrorManagement::Information, REPORT_ERROR(ErrorManagement::Information,
"Accumulate buffers: maxBatchCount=%u, accumBufSize=%u B, readyBufSize=%u B.", "Accumulate buffers: maxBatchCount=%u, accumBufSize=%u B, "
"readyBufSize=%u B.",
maxBatchCount, accumBufSize, readyBufSize); maxBatchCount, accumBufSize, readyBufSize);
} }
@@ -799,7 +804,8 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName,
ok = server.Start(); ok = server.Start();
/* Build the CONFIG payload and cache it in the server so any CONNECT client /* Build the CONFIG payload and cache it in the server so any CONNECT client
* receives it immediately. The config is static for the lifetime of this state. */ * receives it immediately. The config is static for the lifetime of this
* state. */
if (ok) { if (ok) {
uint32 configBufSize = 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 32u + 1u; uint32 configBufSize = 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 32u + 1u;
HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap(); HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap();
@@ -807,15 +813,13 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName,
if (cfgBuf != NULL_PTR(uint8 *)) { if (cfgBuf != NULL_PTR(uint8 *)) {
uint32 cfgPayloadSize = 0u; uint32 cfgPayloadSize = 0u;
if (BuildConfigPayload(cfgBuf, configBufSize, cfgPayloadSize)) { if (BuildConfigPayload(cfgBuf, configBufSize, cfgPayloadSize)) {
(void) server.SendConfig(cfgBuf, cfgPayloadSize); (void)server.SendConfig(cfgBuf, cfgPayloadSize);
} } else {
else {
REPORT_ERROR(ErrorManagement::Warning, REPORT_ERROR(ErrorManagement::Warning,
"Could not build initial CONFIG payload."); "Could not build initial CONFIG payload.");
} }
heap->Free(reinterpret_cast<void *&>(cfgBuf)); heap->Free(reinterpret_cast<void *&>(cfgBuf));
} } else {
else {
REPORT_ERROR(ErrorManagement::Warning, REPORT_ERROR(ErrorManagement::Warning,
"Could not allocate CONFIG buffer."); "Could not allocate CONFIG buffer.");
} }
@@ -867,9 +871,9 @@ bool UDPStreamer::Synchronise() {
/* HI-3: if accumFill reached maxBatchCount, force-flush before writing */ /* HI-3: if accumFill reached maxBatchCount, force-flush before writing */
if (accumFill >= maxBatchCount) { if (accumFill >= maxBatchCount) {
uint32 filled = accumFill; uint32 filled = accumFill;
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(readyBuffer, accumBuffer,
readyBuffer, accumBuffer, filled * totalSrcBytes); filled * totalSrcBytes);
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(
reinterpret_cast<uint8 *>(readyTimestamps), reinterpret_cast<uint8 *>(readyTimestamps),
reinterpret_cast<const uint8 *>(accumTimestamps), reinterpret_cast<const uint8 *>(accumTimestamps),
filled * static_cast<uint32>(sizeof(uint64))); filled * static_cast<uint32>(sizeof(uint64)));
@@ -877,17 +881,18 @@ bool UDPStreamer::Synchronise() {
accumFill = 0u; accumFill = 0u;
lastPublishTs = ts; lastPublishTs = ts;
bufMutex.FastUnLock(); bufMutex.FastUnLock();
(void) dataSem.Post(); (void)dataSem.Post();
bufMutex.FastLock(TTInfiniteWait); bufMutex.FastLock(TTInfiniteWait);
} }
uint8 *slot = accumBuffer + (accumFill * totalSrcBytes); uint8 *slot = accumBuffer + (accumFill * totalSrcBytes);
(void) MemoryOperationsHelper::Copy(slot, memory, totalSrcBytes); (void)MemoryOperationsHelper::Copy(slot, memory, totalSrcBytes);
accumTimestamps[accumFill] = ts; accumTimestamps[accumFill] = ts;
accumFill++; accumFill++;
uint32 filled = accumFill; uint32 filled = accumFill;
bufMutex.FastUnLock(); bufMutex.FastUnLock();
/* Check flush conditions (volatile read of lastPublishTs is safe on x86). */ /* Check flush conditions (volatile read of lastPublishTs is safe on x86).
*/
static const uint32 ACCUM_HEADER = UDPS_TIMESTAMP_BYTES + 4u; /* 12 bytes */ static const uint32 ACCUM_HEADER = UDPS_TIMESTAMP_BYTES + 4u; /* 12 bytes */
uint32 curPayload = ACCUM_HEADER + filled * singleCycleWireBytes; uint32 curPayload = ACCUM_HEADER + filled * singleCycleWireBytes;
uint32 nextPayload = curPayload + singleCycleWireBytes; uint32 nextPayload = curPayload + singleCycleWireBytes;
@@ -896,9 +901,9 @@ bool UDPStreamer::Synchronise() {
if (sizeCondition || timeCondition) { if (sizeCondition || timeCondition) {
bufMutex.FastLock(TTInfiniteWait); bufMutex.FastLock(TTInfiniteWait);
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(readyBuffer, accumBuffer,
readyBuffer, accumBuffer, filled * totalSrcBytes); filled * totalSrcBytes);
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(
reinterpret_cast<uint8 *>(readyTimestamps), reinterpret_cast<uint8 *>(readyTimestamps),
reinterpret_cast<const uint8 *>(accumTimestamps), reinterpret_cast<const uint8 *>(accumTimestamps),
filled * static_cast<uint32>(sizeof(uint64))); filled * static_cast<uint32>(sizeof(uint64)));
@@ -908,29 +913,27 @@ bool UDPStreamer::Synchronise() {
/* Reset the time-based deadline (volatile write). */ /* Reset the time-based deadline (volatile write). */
lastPublishTs = ts; lastPublishTs = ts;
(void) dataSem.Post(); (void)dataSem.Post();
} }
} } else if (publishMode == UDPStreamerPublishDecimate) {
else if (publishMode == UDPStreamerPublishDecimate) {
/* --- Decimate path --- /* --- Decimate path ---
* Post dataSem only every decimateRatio calls. */ * Post dataSem only every decimateRatio calls. */
decimateCounter++; decimateCounter++;
if (decimateCounter >= decimateRatio) { if (decimateCounter >= decimateRatio) {
decimateCounter = 0u; decimateCounter = 0u;
bufMutex.FastLock(TTInfiniteWait); bufMutex.FastLock(TTInfiniteWait);
(void) MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes); (void)MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes);
syncTimestamp = ts; syncTimestamp = ts;
bufMutex.FastUnLock(); bufMutex.FastUnLock();
(void) dataSem.Post(); (void)dataSem.Post();
} }
} } else {
else {
/* --- Strict path: post every call --- */ /* --- Strict path: post every call --- */
bufMutex.FastLock(TTInfiniteWait); bufMutex.FastLock(TTInfiniteWait);
(void) MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes); (void)MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes);
syncTimestamp = ts; syncTimestamp = ts;
bufMutex.FastUnLock(); bufMutex.FastUnLock();
(void) dataSem.Post(); (void)dataSem.Post();
} }
return true; return true;
@@ -941,8 +944,11 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
if (info.GetStage() == ExecutionInfo::StartupStage) { if (info.GetStage() == ExecutionInfo::StartupStage) {
const char8 *modeStr = "Strict"; const char8 *modeStr = "Strict";
if (publishMode == UDPStreamerPublishAccumulate) { modeStr = "Accumulate"; } if (publishMode == UDPStreamerPublishAccumulate) {
else if (publishMode == UDPStreamerPublishDecimate) { modeStr = "Decimate"; } modeStr = "Accumulate";
} else if (publishMode == UDPStreamerPublishDecimate) {
modeStr = "Decimate";
}
REPORT_ERROR(ErrorManagement::Information, REPORT_ERROR(ErrorManagement::Information,
"UDPStreamer background thread started (port %u, mode %s).", "UDPStreamer background thread started (port %u, mode %s).",
static_cast<uint32>(port), modeStr); static_cast<uint32>(port), modeStr);
@@ -960,7 +966,8 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
dataSem.ResetWait(TimeoutType(UDPS_DATA_WAIT_MS)); dataSem.ResetWait(TimeoutType(UDPS_DATA_WAIT_MS));
bool dataReady = (waitErr == ErrorManagement::NoError); bool dataReady = (waitErr == ErrorManagement::NoError);
/* --- Poll for incoming control commands (CONNECT / DISCONNECT / ACK) --- */ /* --- Poll for incoming control commands (CONNECT / DISCONNECT / ACK) ---
*/
server.ServiceClients(); server.ServiceClients();
if (dataReady && server.HasClients()) { if (dataReady && server.HasClients()) {
@@ -973,9 +980,9 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
bufMutex.FastLock(TTInfiniteWait); bufMutex.FastLock(TTInfiniteWait);
fill = readyFill; fill = readyFill;
if (fill > 0u) { if (fill > 0u) {
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer,
scratchBuffer, readyBuffer, fill * totalSrcBytes); fill * totalSrcBytes);
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(
reinterpret_cast<uint8 *>(scratchTimestamps), reinterpret_cast<uint8 *>(scratchTimestamps),
reinterpret_cast<const uint8 *>(readyTimestamps), reinterpret_cast<const uint8 *>(readyTimestamps),
fill * static_cast<uint32>(sizeof(uint64))); fill * static_cast<uint32>(sizeof(uint64)));
@@ -984,8 +991,8 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
if (fill > 0u) { if (fill > 0u) {
SerializeAccumulated(scratchBuffer, scratchTimestamps, fill); SerializeAccumulated(scratchBuffer, scratchTimestamps, fill);
uint32 sendBytes = UDPS_TIMESTAMP_BYTES + 4u + uint32 sendBytes =
fill * singleCycleWireBytes; UDPS_TIMESTAMP_BYTES + 4u + fill * singleCycleWireBytes;
packetCounter++; packetCounter++;
if (!server.SendData(packetCounter, wireBuffer, sendBytes)) { if (!server.SendData(packetCounter, wireBuffer, sendBytes)) {
REPORT_ERROR(ErrorManagement::Warning, REPORT_ERROR(ErrorManagement::Warning,
@@ -993,13 +1000,12 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
packetCounter); packetCounter);
} }
} }
} } else {
else {
/* --- Single-snapshot send (Strict or Decimate) --- */ /* --- Single-snapshot send (Strict or Decimate) --- */
uint64 ts = 0u; uint64 ts = 0u;
bufMutex.FastLock(TTInfiniteWait); bufMutex.FastLock(TTInfiniteWait);
(void) MemoryOperationsHelper::Copy( (void)MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer,
scratchBuffer, readyBuffer, totalSrcBytes); totalSrcBytes);
ts = syncTimestamp; ts = syncTimestamp;
bufMutex.FastUnLock(); bufMutex.FastUnLock();
@@ -1016,7 +1022,7 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
} }
if (info.GetStage() == ExecutionInfo::TerminationStage) { if (info.GetStage() == ExecutionInfo::TerminationStage) {
(void) server.Stop(); (void)server.Stop();
REPORT_ERROR(ErrorManagement::Information, REPORT_ERROR(ErrorManagement::Information,
"UDPStreamer background thread terminated."); "UDPStreamer background thread terminated.");
} }
@@ -1036,12 +1042,14 @@ void UDPStreamer::SerializeAccumulated(const uint8 *src,
*/ */
uint8 *dst = wireBuffer; uint8 *dst = wireBuffer;
/* 8-byte packet-level HRT timestamp = timestamp of the first (oldest) sample */ /* 8-byte packet-level HRT timestamp = timestamp of the first (oldest) sample
(void) MemoryOperationsHelper::Copy(dst, &timestamps[0u], UDPS_TIMESTAMP_BYTES); */
(void)MemoryOperationsHelper::Copy(dst, &timestamps[0u],
UDPS_TIMESTAMP_BYTES);
dst += UDPS_TIMESTAMP_BYTES; dst += UDPS_TIMESTAMP_BYTES;
/* 4-byte sample count */ /* 4-byte sample count */
(void) MemoryOperationsHelper::Copy(dst, &numSamples, 4u); (void)MemoryOperationsHelper::Copy(dst, &numSamples, 4u);
dst += 4u; dst += 4u;
for (uint32 i = 0u; i < numSigs; i++) { for (uint32 i = 0u; i < numSigs; i++) {
@@ -1049,54 +1057,61 @@ void UDPStreamer::SerializeAccumulated(const uint8 *src,
const bool isSrcFloat32 = (signalInfos[i].type == Float32Bit); const bool isSrcFloat32 = (signalInfos[i].type == Float32Bit);
const float64 rMin = signalInfos[i].rangeMin; const float64 rMin = signalInfos[i].rangeMin;
float64 rRange = signalInfos[i].rangeMax - rMin; float64 rRange = signalInfos[i].rangeMax - rMin;
if (rRange == 0.0) { rRange = 1.0; } if (rRange == 0.0) {
rRange = 1.0;
}
/* Pack one snapshot (all elements) from each slot, in order */ /* Pack one snapshot (all elements) from each slot, in order */
for (uint32 k = 0u; k < numSamples; k++) { for (uint32 k = 0u; k < numSamples; k++) {
const uint8 *slotSrc = src + (k * totalSrcBytes) + signalInfos[i].bufferOffset; const uint8 *slotSrc =
src + (k * totalSrcBytes) + signalInfos[i].bufferOffset;
if (signalInfos[i].quantType == UDPStreamerQuantNone) { if (signalInfos[i].quantType == UDPStreamerQuantNone) {
(void) MemoryOperationsHelper::Copy(dst, slotSrc, signalInfos[i].srcByteSize); (void)MemoryOperationsHelper::Copy(dst, slotSrc,
signalInfos[i].srcByteSize);
dst += signalInfos[i].srcByteSize; dst += signalInfos[i].srcByteSize;
} } else {
else {
const uint8 *s = slotSrc; const uint8 *s = slotSrc;
for (uint32 e = 0u; e < nelems; e++) { for (uint32 e = 0u; e < nelems; e++) {
float64 rawVal = 0.0; float64 rawVal = 0.0;
if (isSrcFloat32) { if (isSrcFloat32) {
float32 f32 = 0.0f; float32 f32 = 0.0f;
(void) MemoryOperationsHelper::Copy(&f32, s, 4u); (void)MemoryOperationsHelper::Copy(&f32, s, 4u);
rawVal = static_cast<float64>(f32); rawVal = static_cast<float64>(f32);
s += 4u; s += 4u;
} } else {
else { (void)MemoryOperationsHelper::Copy(&rawVal, s, 8u);
(void) MemoryOperationsHelper::Copy(&rawVal, s, 8u);
s += 8u; s += 8u;
} }
float64 norm = (rawVal - rMin) / rRange; float64 norm = (rawVal - rMin) / rRange;
if (norm < 0.0) { norm = 0.0; } if (norm < 0.0) {
if (norm > 1.0) { norm = 1.0; } norm = 0.0;
}
if (norm > 1.0) {
norm = 1.0;
}
switch (signalInfos[i].quantType) { switch (signalInfos[i].quantType) {
case UDPStreamerQuantUint8: { case UDPStreamerQuantUint8: {
uint8 q = static_cast<uint8>(norm * 255.0); uint8 q = static_cast<uint8>(norm * 255.0);
*dst = q; dst += 1u; *dst = q;
dst += 1u;
break; break;
} }
case UDPStreamerQuantInt8: { case UDPStreamerQuantInt8: {
int8 q = static_cast<int8>((norm * 254.0) - 127.0); int8 q = static_cast<int8>((norm * 254.0) - 127.0);
(void) MemoryOperationsHelper::Copy(dst, &q, 1u); (void)MemoryOperationsHelper::Copy(dst, &q, 1u);
dst += 1u; dst += 1u;
break; break;
} }
case UDPStreamerQuantUint16: { case UDPStreamerQuantUint16: {
uint16 q = static_cast<uint16>(norm * 65535.0); uint16 q = static_cast<uint16>(norm * 65535.0);
(void) MemoryOperationsHelper::Copy(dst, &q, 2u); (void)MemoryOperationsHelper::Copy(dst, &q, 2u);
dst += 2u; dst += 2u;
break; break;
} }
case UDPStreamerQuantInt16: { case UDPStreamerQuantInt16: {
int16 q = static_cast<int16>((norm * 65534.0) - 32767.0); int16 q = static_cast<int16>((norm * 65534.0) - 32767.0);
(void) MemoryOperationsHelper::Copy(dst, &q, 2u); (void)MemoryOperationsHelper::Copy(dst, &q, 2u);
dst += 2u; dst += 2u;
break; break;
} }
@@ -1109,9 +1124,7 @@ void UDPStreamer::SerializeAccumulated(const uint8 *src,
} }
} }
bool UDPStreamer::BuildConfigPayload(uint8 *buf, uint32 bufSize,
bool UDPStreamer::BuildConfigPayload(uint8 *buf,
uint32 bufSize,
uint32 &payloadSize) { uint32 &payloadSize) {
payloadSize = 0u; payloadSize = 0u;
@@ -1119,7 +1132,7 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf,
if ((payloadSize + 4u) > bufSize) { if ((payloadSize + 4u) > bufSize) {
return false; return false;
} }
(void) MemoryOperationsHelper::Copy(buf + payloadSize, &numSigs, 4u); (void)MemoryOperationsHelper::Copy(buf + payloadSize, &numSigs, 4u);
payloadSize += 4u; payloadSize += 4u;
for (uint32 i = 0u; i < numSigs; i++) { for (uint32 i = 0u; i < numSigs; i++) {
@@ -1130,12 +1143,13 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf,
uint8 *p = buf + payloadSize; uint8 *p = buf + payloadSize;
/* Name: 64 bytes, zero-padded */ /* Name: 64 bytes, zero-padded */
(void) MemoryOperationsHelper::Set(p, 0, UDPS_MAX_SIGNAL_NAME); (void)MemoryOperationsHelper::Set(p, 0, UDPS_MAX_SIGNAL_NAME);
uint32 nameLen = static_cast<uint32>(signalInfos[i].name.Size()); uint32 nameLen = static_cast<uint32>(signalInfos[i].name.Size());
if (nameLen >= UDPS_MAX_SIGNAL_NAME) { if (nameLen >= UDPS_MAX_SIGNAL_NAME) {
nameLen = UDPS_MAX_SIGNAL_NAME - 1u; nameLen = UDPS_MAX_SIGNAL_NAME - 1u;
} }
(void) MemoryOperationsHelper::Copy(p, signalInfos[i].name.Buffer(), nameLen); (void)MemoryOperationsHelper::Copy(p, signalInfos[i].name.Buffer(),
nameLen);
p += UDPS_MAX_SIGNAL_NAME; p += UDPS_MAX_SIGNAL_NAME;
/* Type code: 1 byte */ /* Type code: 1 byte */
@@ -1151,19 +1165,19 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf,
p += 1u; p += 1u;
/* numRows: 4 bytes */ /* numRows: 4 bytes */
(void) MemoryOperationsHelper::Copy(p, &signalInfos[i].numRows, 4u); (void)MemoryOperationsHelper::Copy(p, &signalInfos[i].numRows, 4u);
p += 4u; p += 4u;
/* numCols: 4 bytes */ /* numCols: 4 bytes */
(void) MemoryOperationsHelper::Copy(p, &signalInfos[i].numCols, 4u); (void)MemoryOperationsHelper::Copy(p, &signalInfos[i].numCols, 4u);
p += 4u; p += 4u;
/* rangeMin: 8 bytes (float64) */ /* rangeMin: 8 bytes (float64) */
(void) MemoryOperationsHelper::Copy(p, &signalInfos[i].rangeMin, 8u); (void)MemoryOperationsHelper::Copy(p, &signalInfos[i].rangeMin, 8u);
p += 8u; p += 8u;
/* rangeMax: 8 bytes (float64) */ /* rangeMax: 8 bytes (float64) */
(void) MemoryOperationsHelper::Copy(p, &signalInfos[i].rangeMax, 8u); (void)MemoryOperationsHelper::Copy(p, &signalInfos[i].rangeMax, 8u);
p += 8u; p += 8u;
/* timeMode: 1 byte */ /* timeMode: 1 byte */
@@ -1171,20 +1185,21 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf,
p += 1u; p += 1u;
/* samplingRate: 8 bytes (float64) */ /* samplingRate: 8 bytes (float64) */
(void) MemoryOperationsHelper::Copy(p, &signalInfos[i].samplingRate, 8u); (void)MemoryOperationsHelper::Copy(p, &signalInfos[i].samplingRate, 8u);
p += 8u; p += 8u;
/* timeSignalIdx: 4 bytes */ /* timeSignalIdx: 4 bytes */
(void) MemoryOperationsHelper::Copy(p, &signalInfos[i].timeSignalIdx, 4u); (void)MemoryOperationsHelper::Copy(p, &signalInfos[i].timeSignalIdx, 4u);
p += 4u; p += 4u;
/* Unit: 32 bytes, zero-padded */ /* Unit: 32 bytes, zero-padded */
(void) MemoryOperationsHelper::Set(p, 0, UDPS_MAX_UNIT_LEN); (void)MemoryOperationsHelper::Set(p, 0, UDPS_MAX_UNIT_LEN);
uint32 unitLen = static_cast<uint32>(signalInfos[i].unit.Size()); uint32 unitLen = static_cast<uint32>(signalInfos[i].unit.Size());
if (unitLen >= UDPS_MAX_UNIT_LEN) { if (unitLen >= UDPS_MAX_UNIT_LEN) {
unitLen = UDPS_MAX_UNIT_LEN - 1u; unitLen = UDPS_MAX_UNIT_LEN - 1u;
} }
(void) MemoryOperationsHelper::Copy(p, signalInfos[i].unit.Buffer(), unitLen); (void)MemoryOperationsHelper::Copy(p, signalInfos[i].unit.Buffer(),
unitLen);
p += UDPS_MAX_UNIT_LEN; p += UDPS_MAX_UNIT_LEN;
payloadSize += UDPS_SIGNAL_DESC_SIZE; payloadSize += UDPS_SIGNAL_DESC_SIZE;
@@ -1204,7 +1219,7 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
uint8 *dst = wireBuffer; uint8 *dst = wireBuffer;
/* 8-byte packet timestamp */ /* 8-byte packet timestamp */
(void) MemoryOperationsHelper::Copy(dst, &timestamp, UDPS_TIMESTAMP_BYTES); (void)MemoryOperationsHelper::Copy(dst, &timestamp, UDPS_TIMESTAMP_BYTES);
dst += UDPS_TIMESTAMP_BYTES; dst += UDPS_TIMESTAMP_BYTES;
for (uint32 i = 0u; i < numSigs; i++) { for (uint32 i = 0u; i < numSigs; i++) {
@@ -1212,10 +1227,9 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
if (signalInfos[i].quantType == UDPStreamerQuantNone) { if (signalInfos[i].quantType == UDPStreamerQuantNone) {
/* Raw copy */ /* Raw copy */
(void) MemoryOperationsHelper::Copy(dst, src, signalInfos[i].srcByteSize); (void)MemoryOperationsHelper::Copy(dst, src, signalInfos[i].srcByteSize);
dst += signalInfos[i].srcByteSize; dst += signalInfos[i].srcByteSize;
} } else {
else {
float64 rMin = signalInfos[i].rangeMin; float64 rMin = signalInfos[i].rangeMin;
float64 rRange = signalInfos[i].rangeMax - rMin; float64 rRange = signalInfos[i].rangeMax - rMin;
if (rRange == 0.0) { if (rRange == 0.0) {
@@ -1229,19 +1243,22 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
float64 rawVal = 0.0; float64 rawVal = 0.0;
if (isSrcFloat32) { if (isSrcFloat32) {
float32 f32 = 0.0f; float32 f32 = 0.0f;
(void) MemoryOperationsHelper::Copy(&f32, s, 4u); (void)MemoryOperationsHelper::Copy(&f32, s, 4u);
rawVal = static_cast<float64>(f32); rawVal = static_cast<float64>(f32);
s += 4u; s += 4u;
} } else {
else { (void)MemoryOperationsHelper::Copy(&rawVal, s, 8u);
(void) MemoryOperationsHelper::Copy(&rawVal, s, 8u);
s += 8u; s += 8u;
} }
/* Normalize and clamp to [0.0, 1.0] */ /* Normalize and clamp to [0.0, 1.0] */
float64 norm = (rawVal - rMin) / rRange; float64 norm = (rawVal - rMin) / rRange;
if (norm < 0.0) { norm = 0.0; } if (norm < 0.0) {
if (norm > 1.0) { norm = 1.0; } norm = 0.0;
}
if (norm > 1.0) {
norm = 1.0;
}
switch (signalInfos[i].quantType) { switch (signalInfos[i].quantType) {
case UDPStreamerQuantUint8: { case UDPStreamerQuantUint8: {
@@ -1252,19 +1269,19 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
} }
case UDPStreamerQuantInt8: { case UDPStreamerQuantInt8: {
int8 q = static_cast<int8>((norm * 254.0) - 127.0); int8 q = static_cast<int8>((norm * 254.0) - 127.0);
(void) MemoryOperationsHelper::Copy(dst, &q, 1u); (void)MemoryOperationsHelper::Copy(dst, &q, 1u);
dst += 1u; dst += 1u;
break; break;
} }
case UDPStreamerQuantUint16: { case UDPStreamerQuantUint16: {
uint16 q = static_cast<uint16>(norm * 65535.0); uint16 q = static_cast<uint16>(norm * 65535.0);
(void) MemoryOperationsHelper::Copy(dst, &q, 2u); (void)MemoryOperationsHelper::Copy(dst, &q, 2u);
dst += 2u; dst += 2u;
break; break;
} }
case UDPStreamerQuantInt16: { case UDPStreamerQuantInt16: {
int16 q = static_cast<int16>((norm * 65534.0) - 32767.0); int16 q = static_cast<int16>((norm * 65534.0) - 32767.0);
(void) MemoryOperationsHelper::Copy(dst, &q, 2u); (void)MemoryOperationsHelper::Copy(dst, &q, 2u);
dst += 2u; dst += 2u;
break; break;
} }
@@ -1278,34 +1295,37 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
uint8 UDPStreamer::TypeDescriptorToCode(TypeDescriptor td) { uint8 UDPStreamer::TypeDescriptorToCode(TypeDescriptor td) {
uint8 code = UDPS_TYPECODE_UNKNOWN; uint8 code = UDPS_TYPECODE_UNKNOWN;
if (td == UnsignedInteger8Bit) { code = UDPS_TYPECODE_UINT8; } if (td == UnsignedInteger8Bit) {
else if (td == SignedInteger8Bit) { code = UDPS_TYPECODE_INT8; } code = UDPS_TYPECODE_UINT8;
else if (td == UnsignedInteger16Bit) { code = UDPS_TYPECODE_UINT16; } } else if (td == SignedInteger8Bit) {
else if (td == SignedInteger16Bit) { code = UDPS_TYPECODE_INT16; } code = UDPS_TYPECODE_INT8;
else if (td == UnsignedInteger32Bit) { code = UDPS_TYPECODE_UINT32; } } else if (td == UnsignedInteger16Bit) {
else if (td == SignedInteger32Bit) { code = UDPS_TYPECODE_INT32; } code = UDPS_TYPECODE_UINT16;
else if (td == UnsignedInteger64Bit) { code = UDPS_TYPECODE_UINT64; } } else if (td == SignedInteger16Bit) {
else if (td == SignedInteger64Bit) { code = UDPS_TYPECODE_INT64; } code = UDPS_TYPECODE_INT16;
else if (td == Float32Bit) { code = UDPS_TYPECODE_FLOAT32; } } else if (td == UnsignedInteger32Bit) {
else if (td == Float64Bit) { code = UDPS_TYPECODE_FLOAT64; } code = UDPS_TYPECODE_UINT32;
} else if (td == SignedInteger32Bit) {
code = UDPS_TYPECODE_INT32;
} else if (td == UnsignedInteger64Bit) {
code = UDPS_TYPECODE_UINT64;
} else if (td == SignedInteger64Bit) {
code = UDPS_TYPECODE_INT64;
} else if (td == Float32Bit) {
code = UDPS_TYPECODE_FLOAT32;
} else if (td == Float64Bit) {
code = UDPS_TYPECODE_FLOAT64;
}
return code; return code;
} }
uint16 UDPStreamer::GetPort() const { uint16 UDPStreamer::GetPort() const { return port; }
return port;
}
uint32 UDPStreamer::GetMaxPayloadSize() const { uint32 UDPStreamer::GetMaxPayloadSize() const { return maxPayloadSize; }
return maxPayloadSize;
}
bool UDPStreamer::IsClientConnected() const { bool UDPStreamer::IsClientConnected() const { return server.HasClients(); }
return server.HasClients();
}
bool UDPStreamer::IsMulticast() const { bool UDPStreamer::IsMulticast() const { return server.IsMulticast(); }
return server.IsMulticast();
}
CLASS_REGISTER(UDPStreamer, "1.0") CLASS_REGISTER(UDPStreamer, "1.0")
Source/Components/DataSources/UDPStreamer/UDPStreamer.h
+2 -1
View File
@@ -141,9 +141,10 @@ struct UDPStreamerSignalInfo {
* *
* @par Top-level configuration parameters * @par Top-level configuration parameters
* | Parameter | Type | Default | Description | * | Parameter | Type | Default | Description |
* |-----------------|---------|---------|-------------| * |-----------------|---------|------------------|-------------|
* | Port | uint16 | 44500 | TCP control port (multicast) or UDP server port (unicast). Values ≤ 1024 produce a warning. | * | Port | uint16 | 44500 | TCP control port (multicast) or UDP server port (unicast). Values ≤ 1024 produce a warning. |
* | MulticastGroup | string | *(absent)* | **Enables multicast mode.** IPv4 multicast address, e.g. `"239.0.0.1"`. Must be in 224.0.0.0/4. Absent or empty = unicast. | * | MulticastGroup | string | *(absent)* | **Enables multicast mode.** IPv4 multicast address, e.g. `"239.0.0.1"`. Must be in 224.0.0.0/4. Absent or empty = unicast. |
* | Interface | string | *(absent)* | Multicast binded interface **ONLY FOR MULTICAST** |
* | DataPort | uint16 | Port+1 | UDP port for multicast DATA datagrams. Ignored in unicast mode. Must be non-zero and differ from Port. | * | DataPort | uint16 | Port+1 | UDP port for multicast DATA datagrams. Ignored in unicast mode. Must be non-zero and differ from Port. |
* | MaxPayloadSize | uint32 | 1400 | Maximum bytes of signal payload per UDP datagram (excluding the 17-byte header). Larger signals are fragmented. | * | MaxPayloadSize | uint32 | 1400 | Maximum bytes of signal payload per UDP datagram (excluding the 17-byte header). Larger signals are fragmented. |
* | PublishingMode | string | Strict | `Strict`: send one packet every Synchronise() call. `Auto`: rate-limited; flush only when MinRefreshRate interval has elapsed. | * | PublishingMode | string | Strict | `Strict`: send one packet every Synchronise() call. `Auto`: rate-limited; flush only when MinRefreshRate interval has elapsed. |
Source/Components/Interfaces/UDPStream/UDPSServer.cpp
+118 -131
View File
@@ -6,12 +6,13 @@
#include "UDPSServer.h" #include "UDPSServer.h"
#include "AdvancedErrorManagement.h" #include "AdvancedErrorManagement.h"
#include "ErrorType.h"
#include "HighResolutionTimer.h" #include "HighResolutionTimer.h"
#include "MemoryOperationsHelper.h" #include "MemoryOperationsHelper.h"
#include "StreamString.h" #include "StreamString.h"
#include <sys/select.h>
#include <poll.h> #include <poll.h>
#include <sys/select.h>
namespace MARTe { namespace MARTe {
@@ -20,18 +21,10 @@ namespace MARTe {
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
UDPSServer::UDPSServer() UDPSServer::UDPSServer()
: port(0u), : port(0u), maxPayloadSize(UDPS_SERVER_DEFAULT_MAX_PAYLOAD), dataPort(0u),
maxPayloadSize(UDPS_SERVER_DEFAULT_MAX_PAYLOAD), useMulticast(false), clientTimeoutTicks(0u), numUnicastClients(0u),
dataPort(0u), numTCPClients(0u), cachedConfig(NULL_PTR(uint8 *)), cachedConfigSize(0u),
useMulticast(false), sendBuf(NULL_PTR(uint8 *)), sendBufCapacity(0u), configCounter(0u),
clientTimeoutTicks(0u),
numUnicastClients(0u),
numTCPClients(0u),
cachedConfig(NULL_PTR(uint8 *)),
cachedConfigSize(0u),
sendBuf(NULL_PTR(uint8 *)),
sendBufCapacity(0u),
configCounter(0u),
started(false) { started(false) {
for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) { for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) {
@@ -47,9 +40,7 @@ UDPSServer::UDPSServer()
} }
} }
UDPSServer::~UDPSServer() { UDPSServer::~UDPSServer() { (void)Stop(); }
(void) Stop();
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Initialise // Initialise
@@ -59,14 +50,14 @@ bool UDPSServer::Initialise(StructuredDataI &data) {
uint32 portU32 = 0u; uint32 portU32 = 0u;
if (data.Read("Port", portU32)) { if (data.Read("Port", portU32)) {
port = static_cast<uint16>(portU32); port = static_cast<uint16>(portU32);
} } else {
else {
REPORT_ERROR_STATIC(ErrorManagement::ParametersError, REPORT_ERROR_STATIC(ErrorManagement::ParametersError,
"UDPSServer: Port not specified."); "UDPSServer: Port not specified.");
return false; return false;
} }
/* port == 0 is valid: unicast push-only mode (no serverSocket bind, /* port == 0 is valid: unicast push-only mode (no serverSocket bind,
* no CONNECT/DISCONNECT/ACK reception). Clients added via AddStaticClient(). */ * no CONNECT/DISCONNECT/ACK reception). Clients added via AddStaticClient().
*/
StreamString mcGroup; StreamString mcGroup;
if (data.Read("MulticastGroup", mcGroup) && (mcGroup.Size() > 0u)) { if (data.Read("MulticastGroup", mcGroup) && (mcGroup.Size() > 0u)) {
@@ -76,24 +67,31 @@ bool UDPSServer::Initialise(StructuredDataI &data) {
if (useMulticast) { if (useMulticast) {
uint32 dpU32 = static_cast<uint32>(port) + 1u; uint32 dpU32 = static_cast<uint32>(port) + 1u;
(void) data.Read("DataPort", dpU32); (void)data.Read("DataPort", dpU32);
dataPort = static_cast<uint16>(dpU32); dataPort = static_cast<uint16>(dpU32);
if (dataPort == port) { if (dataPort == port) {
REPORT_ERROR_STATIC(ErrorManagement::ParametersError, REPORT_ERROR_STATIC(
ErrorManagement::ParametersError,
"UDPSServer: DataPort (%u) must differ from Port (%u).", "UDPSServer: DataPort (%u) must differ from Port (%u).",
static_cast<uint32>(dataPort), static_cast<uint32>(port)); static_cast<uint32>(dataPort), static_cast<uint32>(port));
return false; return false;
} }
if (!data.Read("Interface", interface)) {
REPORT_ERROR_STATIC(
ErrorManagement::ParametersError,
"Missing mandatory `Interface` field for multicast operations");
return false;
}
} }
uint32 mps = UDPS_SERVER_DEFAULT_MAX_PAYLOAD; uint32 mps = UDPS_SERVER_DEFAULT_MAX_PAYLOAD;
(void) data.Read("MaxPayloadSize", mps); (void)data.Read("MaxPayloadSize", mps);
maxPayloadSize = mps; maxPayloadSize = mps;
uint32 timeoutSecs = UDPS_SERVER_DEFAULT_CLIENT_TIMEOUT_S; uint32 timeoutSecs = UDPS_SERVER_DEFAULT_CLIENT_TIMEOUT_S;
(void) data.Read("ClientTimeout", timeoutSecs); (void)data.Read("ClientTimeout", timeoutSecs);
clientTimeoutTicks = (timeoutSecs > 0u) clientTimeoutTicks = (timeoutSecs > 0u) ? (static_cast<uint64>(timeoutSecs) *
? (static_cast<uint64>(timeoutSecs) * HighResolutionTimer::Frequency()) HighResolutionTimer::Frequency())
: 0u; : 0u;
return true; return true;
@@ -128,19 +126,16 @@ bool UDPSServer::Start() {
tcpListener.SetBlocking(false); tcpListener.SetBlocking(false);
} }
// UDP data socket connected to multicast group // UDP data socket connected to multicast group
if (ok) { ok &= dataSocket.Open();
ok = dataSocket.Open(); ok &= dataSocket.Join(multicastGroup.Buffer(), interface.Buffer());
} ok &= dataSocket.Connect(multicastGroup.Buffer(), dataPort);
if (ok) {
ok = dataSocket.Connect(multicastGroup.Buffer(), dataPort);
}
if (!ok) { if (!ok) {
REPORT_ERROR_STATIC(ErrorManagement::FatalError, REPORT_ERROR_STATIC(
ErrorManagement::FatalError,
"UDPSServer: Failed to open multicast sockets on port %u.", "UDPSServer: Failed to open multicast sockets on port %u.",
static_cast<uint32>(port)); static_cast<uint32>(port));
} }
} } else {
else {
// Unicast send socket (unconnected; SetDestination per Write) // Unicast send socket (unconnected; SetDestination per Write)
ok = uniSendSocket.Open(); ok = uniSendSocket.Open();
if (!ok) { if (!ok) {
@@ -155,7 +150,8 @@ bool UDPSServer::Start() {
ok = serverSocket.Listen(port); // UDP "listen" = bind ok = serverSocket.Listen(port); // UDP "listen" = bind
} }
if (!ok) { if (!ok) {
REPORT_ERROR_STATIC(ErrorManagement::FatalError, REPORT_ERROR_STATIC(
ErrorManagement::FatalError,
"UDPSServer: Failed to bind receive socket on port %u.", "UDPSServer: Failed to bind receive socket on port %u.",
static_cast<uint32>(port)); static_cast<uint32>(port));
} }
@@ -189,16 +185,16 @@ bool UDPSServer::Stop() {
// Close sockets // Close sockets
if (serverSocket.IsValid()) { if (serverSocket.IsValid()) {
(void) serverSocket.Close(); (void)serverSocket.Close();
} }
if (uniSendSocket.IsValid()) { if (uniSendSocket.IsValid()) {
(void) uniSendSocket.Close(); (void)uniSendSocket.Close();
} }
if (tcpListener.IsValid()) { if (tcpListener.IsValid()) {
(void) tcpListener.Close(); (void)tcpListener.Close();
} }
if (dataSocket.IsValid()) { if (dataSocket.IsValid()) {
(void) dataSocket.Close(); (void)dataSocket.Close();
} }
// Free heap buffers // Free heap buffers
@@ -235,11 +231,9 @@ void UDPSServer::ServiceClients() {
pfd.fd = static_cast<int>(tcpListener.GetReadHandle()); pfd.fd = static_cast<int>(tcpListener.GetReadHandle());
pfd.events = POLLIN; pfd.events = POLLIN;
pfd.revents = 0; pfd.revents = 0;
bool pending = (::poll(&pfd, 1u, 0) > 0) && bool pending = (::poll(&pfd, 1u, 0) > 0) && ((pfd.revents & POLLIN) != 0);
((pfd.revents & POLLIN) != 0); BasicTCPSocket *newConn =
BasicTCPSocket *newConn = pending pending ? tcpListener.WaitConnection(0u) : NULL_PTR(BasicTCPSocket *);
? tcpListener.WaitConnection(0u)
: NULL_PTR(BasicTCPSocket *);
if (newConn != NULL_PTR(BasicTCPSocket *)) { if (newConn != NULL_PTR(BasicTCPSocket *)) {
// Find free TCP client slot // Find free TCP client slot
uint32 freeSlot = UDPS_SERVER_MAX_TCP_CLIENTS; uint32 freeSlot = UDPS_SERVER_MAX_TCP_CLIENTS;
@@ -251,11 +245,11 @@ void UDPSServer::ServiceClients() {
} }
if (freeSlot < UDPS_SERVER_MAX_TCP_CLIENTS) { if (freeSlot < UDPS_SERVER_MAX_TCP_CLIENTS) {
HandleMulticastTCPConnect(newConn, freeSlot); HandleMulticastTCPConnect(newConn, freeSlot);
} } else {
else { REPORT_ERROR_STATIC(
REPORT_ERROR_STATIC(ErrorManagement::Warning, ErrorManagement::Warning,
"UDPSServer: TCP client table full, rejecting new connection."); "UDPSServer: TCP client table full, rejecting new connection.");
(void) newConn->Close(); (void)newConn->Close();
delete newConn; delete newConn;
} }
} }
@@ -268,20 +262,25 @@ void UDPSServer::ServiceClients() {
} }
// Non-blocking peek // Non-blocking peek
int fd = tcpClients[i]->GetReadHandle(); int fd = tcpClients[i]->GetReadHandle();
if (fd < 0 || fd >= FD_SETSIZE) { continue; /* HI-6: skip FDs outside select range */ } if (fd < 0 || fd >= FD_SETSIZE) {
continue; /* HI-6: skip FDs outside select range */
}
fd_set rset; fd_set rset;
FD_ZERO(&rset); FD_ZERO(&rset);
FD_SET(fd, &rset); FD_SET(fd, &rset);
struct timeval tv; struct timeval tv;
tv.tv_sec = 0; tv.tv_usec = 0; tv.tv_sec = 0;
tv.tv_usec = 0;
int nready = select(fd + 1, &rset, NULL, NULL, &tv); int nready = select(fd + 1, &rset, NULL, NULL, &tv);
if (nready > 0) { if (nready > 0) {
uint8 pktBuf[UDPS_HEADER_SIZE]; uint8 pktBuf[UDPS_HEADER_SIZE];
uint32 recvSize = UDPS_HEADER_SIZE; uint32 recvSize = UDPS_HEADER_SIZE;
bool recvOk = tcpClients[i]->Read(reinterpret_cast<char8 *>(pktBuf), recvSize); bool recvOk =
tcpClients[i]->Read(reinterpret_cast<char8 *>(pktBuf), recvSize);
if (!recvOk || (recvSize == 0u)) { if (!recvOk || (recvSize == 0u)) {
REPORT_ERROR_STATIC(ErrorManagement::Information, REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: TCP client disconnected (slot %u).", i); "UDPSServer: TCP client disconnected (slot %u).",
i);
EvictTCPClient(i); EvictTCPClient(i);
continue; continue;
} }
@@ -290,21 +289,23 @@ void UDPSServer::ServiceClients() {
reinterpret_cast<const UDPSPacketHeader *>(pktBuf); reinterpret_cast<const UDPSPacketHeader *>(pktBuf);
if ((hdr->magic == UDPS_MAGIC) && if ((hdr->magic == UDPS_MAGIC) &&
(hdr->type == UDPS_TYPE_DISCONNECT)) { (hdr->type == UDPS_TYPE_DISCONNECT)) {
REPORT_ERROR_STATIC(ErrorManagement::Information, REPORT_ERROR_STATIC(
ErrorManagement::Information,
"UDPSServer: TCP client sent DISCONNECT (slot %u).", i); "UDPSServer: TCP client sent DISCONNECT (slot %u).", i);
EvictTCPClient(i); EvictTCPClient(i);
} }
} }
} }
} }
} } else {
else {
// Unicast: poll serverSocket for CONNECT / DISCONNECT / ACK // Unicast: poll serverSocket for CONNECT / DISCONNECT / ACK
if (!serverSocket.IsValid()) { if (!serverSocket.IsValid()) {
return; return;
} }
int fd = serverSocket.GetReadHandle(); int fd = serverSocket.GetReadHandle();
if (fd < 0 || fd >= FD_SETSIZE) { return; /* HI-6 */ } if (fd < 0 || fd >= FD_SETSIZE) {
return; /* HI-6 */
}
fd_set rset; fd_set rset;
FD_ZERO(&rset); FD_ZERO(&rset);
FD_SET(fd, &rset); FD_SET(fd, &rset);
@@ -313,7 +314,8 @@ void UDPSServer::ServiceClients() {
while (nready > 0) { while (nready > 0) {
uint8 pktBuf[UDPS_HEADER_SIZE + 4u]; uint8 pktBuf[UDPS_HEADER_SIZE + 4u];
uint32 recvSize = static_cast<uint32>(sizeof(pktBuf)); uint32 recvSize = static_cast<uint32>(sizeof(pktBuf));
bool recvOk = serverSocket.Read(reinterpret_cast<char8 *>(pktBuf), recvSize); bool recvOk =
serverSocket.Read(reinterpret_cast<char8 *>(pktBuf), recvSize);
if (!recvOk || (recvSize < UDPS_HEADER_SIZE)) { if (!recvOk || (recvSize < UDPS_HEADER_SIZE)) {
break; break;
} }
@@ -325,11 +327,9 @@ void UDPSServer::ServiceClients() {
InternetHost src = serverSocket.GetSource(); InternetHost src = serverSocket.GetSource();
if (hdr->type == UDPS_TYPE_CONNECT) { if (hdr->type == UDPS_TYPE_CONNECT) {
HandleUnicastConnect(src); HandleUnicastConnect(src);
} } else if (hdr->type == UDPS_TYPE_DISCONNECT) {
else if (hdr->type == UDPS_TYPE_DISCONNECT) {
HandleUnicastDisconnect(src); HandleUnicastDisconnect(src);
} } else if (hdr->type == UDPS_TYPE_ACK) {
else if (hdr->type == UDPS_TYPE_ACK) {
HandleUnicastAck(src); HandleUnicastAck(src);
} }
@@ -358,7 +358,7 @@ bool UDPSServer::SendConfig(const uint8 *payload, uint32 payloadSize) {
} }
// Cache for future CONNECT clients // Cache for future CONNECT clients
(void) CacheConfig(payload, payloadSize); (void)CacheConfig(payload, payloadSize);
configCounter++; configCounter++;
@@ -373,14 +373,14 @@ bool UDPSServer::SendConfig(const uint8 *payload, uint32 payloadSize) {
bool sent = SendFragmentedTCP(*tcpClients[i], UDPS_TYPE_CONFIG, bool sent = SendFragmentedTCP(*tcpClients[i], UDPS_TYPE_CONFIG,
configCounter, payload, payloadSize); configCounter, payload, payloadSize);
if (!sent) { if (!sent) {
REPORT_ERROR_STATIC(ErrorManagement::Warning, REPORT_ERROR_STATIC(
ErrorManagement::Warning,
"UDPSServer: CONFIG send failed to TCP client %u, evicting.", i); "UDPSServer: CONFIG send failed to TCP client %u, evicting.", i);
EvictTCPClient(i); EvictTCPClient(i);
ok = false; ok = false;
} }
} }
} } else {
else {
// Send CONFIG to each unicast client // Send CONFIG to each unicast client
for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) { for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) {
if (!unicastClients[i].active) { if (!unicastClients[i].active) {
@@ -406,7 +406,8 @@ bool UDPSServer::SendConfig(const uint8 *payload, uint32 payloadSize) {
// SendData // SendData
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
bool UDPSServer::SendData(uint32 counter, const uint8 *payload, uint32 payloadSize) { bool UDPSServer::SendData(uint32 counter, const uint8 *payload,
uint32 payloadSize) {
if (!started) { if (!started) {
return false; return false;
} }
@@ -415,15 +416,15 @@ bool UDPSServer::SendData(uint32 counter, const uint8 *payload, uint32 payloadSi
if (useMulticast) { if (useMulticast) {
// Single multicast write (no dest needed — socket already connected) // Single multicast write (no dest needed — socket already connected)
bool sent = SendFragmentedUDP(dataSocket, NULL_PTR(InternetHost *), bool sent =
UDPS_TYPE_DATA, counter, payload, payloadSize); SendFragmentedUDP(dataSocket, NULL_PTR(InternetHost *), UDPS_TYPE_DATA,
counter, payload, payloadSize);
if (!sent) { if (!sent) {
REPORT_ERROR_STATIC(ErrorManagement::Warning, REPORT_ERROR_STATIC(ErrorManagement::Warning,
"UDPSServer: DATA send to multicast group failed."); "UDPSServer: DATA send to multicast group failed.");
ok = false; ok = false;
} }
} } else {
else {
for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) { for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) {
if (!unicastClients[i].active) { if (!unicastClients[i].active) {
continue; continue;
@@ -470,7 +471,8 @@ bool UDPSServer::AddStaticClient(const char8 *ip, uint16 port_) {
} }
if (freeSlot >= UDPS_SERVER_MAX_UNICAST_CLIENTS) { if (freeSlot >= UDPS_SERVER_MAX_UNICAST_CLIENTS) {
REPORT_ERROR_STATIC(ErrorManagement::Warning, REPORT_ERROR_STATIC(ErrorManagement::Warning,
"UDPSServer: Unicast client table full, cannot add static client %s:%u.", "UDPSServer: Unicast client table full, cannot add "
"static client %s:%u.",
ip, static_cast<uint32>(port_)); ip, static_cast<uint32>(port_));
return false; return false;
} }
@@ -489,8 +491,8 @@ bool UDPSServer::AddStaticClient(const char8 *ip, uint16 port_) {
numUnicastClients++; numUnicastClients++;
REPORT_ERROR_STATIC(ErrorManagement::Information, REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: Static client added: %s:%u.", "UDPSServer: Static client added: %s:%u.", ip,
ip, static_cast<uint32>(port_)); static_cast<uint32>(port_));
return true; return true;
} }
@@ -513,35 +515,24 @@ uint32 UDPSServer::GetClientCount() const {
return count; return count;
} }
bool UDPSServer::HasClients() const { bool UDPSServer::HasClients() const { return (GetClientCount() > 0u); }
return (GetClientCount() > 0u);
}
bool UDPSServer::IsMulticast() const { bool UDPSServer::IsMulticast() const { return useMulticast; }
return useMulticast;
}
uint16 UDPSServer::GetPort() const { uint16 UDPSServer::GetPort() const { return port; }
return port;
}
uint32 UDPSServer::GetMaxPayloadSize() const { uint32 UDPSServer::GetMaxPayloadSize() const { return maxPayloadSize; }
return maxPayloadSize;
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Private: SendFragmentedUDP // Private: SendFragmentedUDP
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock, bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock, InternetHost *dest,
InternetHost *dest, uint8 type, uint32 counter,
uint8 type, const uint8 *payload, uint32 payloadSize) {
uint32 counter,
const uint8 *payload,
uint32 payloadSize) {
uint32 maxChunk = maxPayloadSize; // payload bytes per fragment (excl. header) uint32 maxChunk = maxPayloadSize; // payload bytes per fragment (excl. header)
uint32 totalFrags = (payloadSize == 0u) ? 1u : uint32 totalFrags =
((payloadSize + maxChunk - 1u) / maxChunk); (payloadSize == 0u) ? 1u : ((payloadSize + maxChunk - 1u) / maxChunk);
bool ok = true; bool ok = true;
uint32 offs = 0u; uint32 offs = 0u;
@@ -552,21 +543,18 @@ bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock,
chunkSize = maxChunk; chunkSize = maxChunk;
} }
UDPSBuildHeader(sendBuf, type, counter, UDPSBuildHeader(sendBuf, type, counter, static_cast<uint16>(f),
static_cast<uint16>(f), static_cast<uint16>(totalFrags), chunkSize);
static_cast<uint16>(totalFrags),
chunkSize);
if (chunkSize > 0u) { if (chunkSize > 0u) {
(void) MemoryOperationsHelper::Copy(sendBuf + UDPS_HEADER_SIZE, (void)MemoryOperationsHelper::Copy(sendBuf + UDPS_HEADER_SIZE,
payload + offs, payload + offs, chunkSize);
chunkSize);
} }
uint32 sendSize = UDPS_HEADER_SIZE + chunkSize; uint32 sendSize = UDPS_HEADER_SIZE + chunkSize;
if (dest != NULL_PTR(InternetHost *)) { if (dest != NULL_PTR(InternetHost *)) {
(void) sock.SetDestination(*dest); (void)sock.SetDestination(*dest);
} }
ok = sock.Write(reinterpret_cast<const char8 *>(sendBuf), sendSize); ok = sock.Write(reinterpret_cast<const char8 *>(sendBuf), sendSize);
@@ -586,14 +574,12 @@ bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock,
// Private: SendFragmentedTCP // Private: SendFragmentedTCP
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
bool UDPSServer::SendFragmentedTCP(BasicTCPSocket &sock, bool UDPSServer::SendFragmentedTCP(BasicTCPSocket &sock, uint8 type,
uint8 type, uint32 counter, const uint8 *payload,
uint32 counter,
const uint8 *payload,
uint32 payloadSize) { uint32 payloadSize) {
uint32 maxChunk = maxPayloadSize; uint32 maxChunk = maxPayloadSize;
uint32 totalFrags = (payloadSize == 0u) ? 1u : uint32 totalFrags =
((payloadSize + maxChunk - 1u) / maxChunk); (payloadSize == 0u) ? 1u : ((payloadSize + maxChunk - 1u) / maxChunk);
bool ok = true; bool ok = true;
uint32 offs = 0u; uint32 offs = 0u;
@@ -604,15 +590,12 @@ bool UDPSServer::SendFragmentedTCP(BasicTCPSocket &sock,
chunkSize = maxChunk; chunkSize = maxChunk;
} }
UDPSBuildHeader(sendBuf, type, counter, UDPSBuildHeader(sendBuf, type, counter, static_cast<uint16>(f),
static_cast<uint16>(f), static_cast<uint16>(totalFrags), chunkSize);
static_cast<uint16>(totalFrags),
chunkSize);
if (chunkSize > 0u) { if (chunkSize > 0u) {
(void) MemoryOperationsHelper::Copy(sendBuf + UDPS_HEADER_SIZE, (void)MemoryOperationsHelper::Copy(sendBuf + UDPS_HEADER_SIZE,
payload + offs, payload + offs, chunkSize);
chunkSize);
} }
uint32 sendSize = UDPS_HEADER_SIZE + chunkSize; uint32 sendSize = UDPS_HEADER_SIZE + chunkSize;
@@ -660,8 +643,8 @@ void UDPSServer::HandleUnicastConnect(const InternetHost &src) {
uint16 srcPort = src.GetPort(); uint16 srcPort = src.GetPort();
REPORT_ERROR_STATIC(ErrorManagement::Information, REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: CONNECT from %s:%u.", "UDPSServer: CONNECT from %s:%u.", srcAddr,
srcAddr, static_cast<uint32>(srcPort)); static_cast<uint32>(srcPort));
// Check if this client is already known // Check if this client is already known
uint32 existing = FindUnicastClient(srcAddr, srcPort); uint32 existing = FindUnicastClient(srcAddr, srcPort);
@@ -671,7 +654,7 @@ void UDPSServer::HandleUnicastConnect(const InternetHost &src) {
if (cachedConfig != NULL_PTR(uint8 *)) { if (cachedConfig != NULL_PTR(uint8 *)) {
InternetHost dest(srcPort, srcAddr); InternetHost dest(srcPort, srcAddr);
configCounter++; configCounter++;
(void) SendFragmentedUDP(uniSendSocket, &dest, UDPS_TYPE_CONFIG, (void)SendFragmentedUDP(uniSendSocket, &dest, UDPS_TYPE_CONFIG,
configCounter, cachedConfig, cachedConfigSize); configCounter, cachedConfig, cachedConfigSize);
} }
return; return;
@@ -693,9 +676,9 @@ void UDPSServer::HandleUnicastConnect(const InternetHost &src) {
unicastClients[slot].ipAddr, unicastClients[slot].ipAddr,
static_cast<uint32>(unicastClients[slot].clientPort)); static_cast<uint32>(unicastClients[slot].clientPort));
EvictUnicastClient(slot); EvictUnicastClient(slot);
} } else {
else { REPORT_ERROR_STATIC(
REPORT_ERROR_STATIC(ErrorManagement::Warning, ErrorManagement::Warning,
"UDPSServer: All slots occupied by static clients; rejecting %s:%u.", "UDPSServer: All slots occupied by static clients; rejecting %s:%u.",
srcAddr, static_cast<uint32>(srcPort)); srcAddr, static_cast<uint32>(srcPort));
return; return;
@@ -719,7 +702,7 @@ void UDPSServer::HandleUnicastConnect(const InternetHost &src) {
if (cachedConfig != NULL_PTR(uint8 *)) { if (cachedConfig != NULL_PTR(uint8 *)) {
InternetHost dest(srcPort, srcAddr); InternetHost dest(srcPort, srcAddr);
configCounter++; configCounter++;
(void) SendFragmentedUDP(uniSendSocket, &dest, UDPS_TYPE_CONFIG, (void)SendFragmentedUDP(uniSendSocket, &dest, UDPS_TYPE_CONFIG,
configCounter, cachedConfig, cachedConfigSize); configCounter, cachedConfig, cachedConfigSize);
} }
} }
@@ -736,8 +719,8 @@ void UDPSServer::HandleUnicastDisconnect(const InternetHost &src) {
uint32 slot = FindUnicastClient(srcAddr, srcPort); uint32 slot = FindUnicastClient(srcAddr, srcPort);
if (slot < UDPS_SERVER_MAX_UNICAST_CLIENTS) { if (slot < UDPS_SERVER_MAX_UNICAST_CLIENTS) {
REPORT_ERROR_STATIC(ErrorManagement::Information, REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: DISCONNECT from %s:%u.", "UDPSServer: DISCONNECT from %s:%u.", srcAddr,
srcAddr, static_cast<uint32>(srcPort)); static_cast<uint32>(srcPort));
EvictUnicastClient(slot); EvictUnicastClient(slot);
} }
} }
@@ -826,7 +809,8 @@ void UDPSServer::EvictUnicastClient(uint32 idx) {
// Private: HandleMulticastTCPConnect // Private: HandleMulticastTCPConnect
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
void UDPSServer::HandleMulticastTCPConnect(BasicTCPSocket *newClient, uint32 idx) { void UDPSServer::HandleMulticastTCPConnect(BasicTCPSocket *newClient,
uint32 idx) {
if (idx >= UDPS_SERVER_MAX_TCP_CLIENTS) { if (idx >= UDPS_SERVER_MAX_TCP_CLIENTS) {
return; return;
} }
@@ -835,16 +819,19 @@ void UDPSServer::HandleMulticastTCPConnect(BasicTCPSocket *newClient, uint32 idx
numTCPClients++; numTCPClients++;
REPORT_ERROR_STATIC(ErrorManagement::Information, REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: Multicast TCP client connected (slot %u).", idx); "UDPSServer: Multicast TCP client connected (slot %u).",
idx);
// Send cached CONFIG over TCP // Send cached CONFIG over TCP
if (cachedConfig != NULL_PTR(uint8 *)) { if (cachedConfig != NULL_PTR(uint8 *)) {
configCounter++; configCounter++;
bool sent = SendFragmentedTCP(*newClient, UDPS_TYPE_CONFIG, bool sent = SendFragmentedTCP(*newClient, UDPS_TYPE_CONFIG, configCounter,
configCounter, cachedConfig, cachedConfigSize); cachedConfig, cachedConfigSize);
if (!sent) { if (!sent) {
REPORT_ERROR_STATIC(ErrorManagement::Warning, REPORT_ERROR_STATIC(
"UDPSServer: Failed to send CONFIG to new TCP client (slot %u).", idx); ErrorManagement::Warning,
"UDPSServer: Failed to send CONFIG to new TCP client (slot %u).",
idx);
EvictTCPClient(idx); EvictTCPClient(idx);
} }
} }
@@ -859,7 +846,7 @@ void UDPSServer::EvictTCPClient(uint32 idx) {
return; return;
} }
if (tcpClients[idx] != NULL_PTR(BasicTCPSocket *)) { if (tcpClients[idx] != NULL_PTR(BasicTCPSocket *)) {
(void) tcpClients[idx]->Close(); (void)tcpClients[idx]->Close();
delete tcpClients[idx]; delete tcpClients[idx];
tcpClients[idx] = NULL_PTR(BasicTCPSocket *); tcpClients[idx] = NULL_PTR(BasicTCPSocket *);
if (numTCPClients > 0u) { if (numTCPClients > 0u) {
@@ -887,7 +874,7 @@ bool UDPSServer::CacheConfig(const uint8 *payload, uint32 payloadSize) {
"UDPSServer: Failed to allocate cached CONFIG buffer."); "UDPSServer: Failed to allocate cached CONFIG buffer.");
return false; return false;
} }
(void) MemoryOperationsHelper::Copy(cachedConfig, payload, payloadSize); (void)MemoryOperationsHelper::Copy(cachedConfig, payload, payloadSize);
cachedConfigSize = payloadSize; cachedConfigSize = payloadSize;
return true; return true;
} }
Source/Components/Interfaces/UDPStream/UDPSServer.h
+1
View File
@@ -235,6 +235,7 @@ private:
uint16 port; uint16 port;
uint32 maxPayloadSize; uint32 maxPayloadSize;
StreamString multicastGroup; StreamString multicastGroup;
StreamString interface;
uint16 dataPort; uint16 dataPort;
bool useMulticast; bool useMulticast;
uint64 clientTimeoutTicks; ///< 0 = disabled uint64 clientTimeoutTicks; ///< 0 = disabled
vgcore.26968
BIN
View File
Binary file not shown.