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
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| ---------------- | ------ | --------- | ------------------------------------------- |
| `Port` | uint16 | 44500 | UDP server port |
| `MaxPayloadSize` | uint32 | 1400 | Max payload bytes per UDP datagram (min 18) |
| `CPUMask` | uint32 | 0 (any) | Background thread CPU affinity |
@@ -76,7 +76,7 @@ thread.
### Per-signal Parameters
| Parameter | Type | Default | Applies to |
|-----------|------|---------|------------|
| --------------- | ------- | ------------ | -------------------------------------------------------- |
| `Unit` | string | `""` | Any type — informational, forwarded to client in CONFIG |
| `RangeMin` | float64 | 0.0 | float32/float64 with `QuantizedType` |
| `RangeMax` | float64 | 1.0 | float32/float64 with `QuantizedType` |
@@ -88,7 +88,7 @@ thread.
### Quantization Types
| Value | Wire type | Bit depth | Notes |
|-------|-----------|-----------|-------|
| -------- | -------------- | --------- | ------------------------------------------------- |
| `none` | same as source | — | Raw copy, no quantization |
| `uint8` | uint8 | 8-bit | Maps `[RangeMin, RangeMax]` → `[0, 255]` |
| `int8` | int8 | 8-bit | Maps `[RangeMin, RangeMax]` → `[-127, 127]` |
@@ -105,7 +105,7 @@ wire_value = (uint16)(normalized × 65535)
### Time Modes
| Value | Meaning | Requirements |
|-------|---------|--------------|
| ------------- | ------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------- |
| `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`. |
| `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:
| Capability | Component | Purpose |
|---|---|---|
| --------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------- |
| **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 |
| **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.
Capabilities accessible over TCP (port 8080 by default):
- `DISCOVER` — enumerate all signals with type and alias metadata
- `TRACE` — enable/disable high-speed UDP telemetry per signal (with decimation)
- `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
| Document | Contents |
|---|---|
| ----------------------------- | -------------------------------------------------------------- |
| `Docs/Protocol.md` | UDPS binary wire protocol specification |
| `Docs/UDPStreamer.md` | UDPStreamer DataSource configuration reference |
| `Docs/SineArrayGAM.md` | SineArrayGAM configuration reference |
Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp
+162 -142
View File
@@ -21,6 +21,8 @@
* methods, such as those inline could be defined on the header file, instead.
*/
#include "ErrorType.h"
#include "StreamString.h"
#define DLL_API
/*---------------------------------------------------------------------------*/
@@ -37,10 +39,7 @@
#include "EmbeddedThreadI.h"
#include "GlobalObjectsDatabase.h"
#include "HighResolutionTimer.h"
#include "MemoryMapSynchronisedOutputBroker.h"
#include "MemoryOperationsHelper.h"
#include "Sleep.h"
#include "Threads.h"
#include "UDPStreamer.h"
/*---------------------------------------------------------------------------*/
@@ -52,7 +51,8 @@ namespace MARTe {
/** Default port used when none is specified. */
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;
/** Maximum pending TCP connections on the listener backlog. */
@@ -80,10 +80,8 @@ static const uint32 UDPS_TIMESTAMP_BYTES = 8u;
/* Method definitions */
/*---------------------------------------------------------------------------*/
UDPStreamer::UDPStreamer() :
MemoryDataSourceI(),
EmbeddedServiceMethodBinderI(),
executor(*this) {
UDPStreamer::UDPStreamer()
: MemoryDataSourceI(), EmbeddedServiceMethodBinderI(), executor(*this) {
port = UDPS_DEFAULT_PORT;
maxPayloadSize = UDPS_DEFAULT_MAX_PAYLOAD;
cpuMask = 0xFFFFFFFFu;
@@ -231,15 +229,13 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
(void)data.Read("PublishingMode", publishStr);
if ((publishStr.Size() == 0u) || (publishStr == "Strict")) {
publishMode = UDPStreamerPublishStrict;
}
else if (publishStr == "Accumulate") {
} else if (publishStr == "Accumulate") {
publishMode = UDPStreamerPublishAccumulate;
}
else if (publishStr == "Decimate") {
} else if (publishStr == "Decimate") {
publishMode = UDPStreamerPublishDecimate;
}
else {
REPORT_ERROR(ErrorManagement::ParametersError,
} else {
REPORT_ERROR(
ErrorManagement::ParametersError,
"Unknown PublishingMode '%s'. Allowed: Strict|Accumulate|Decimate.",
publishStr.Buffer());
ok = false;
@@ -250,18 +246,19 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
/* MinRefreshRate controls the time-based flush: flush when
* (now - lastPublishTs) >= flushPeriodTicks, or when adding one more
* sample would overflow MaxPayloadSize. Whichever fires first. */
if (!data.Read("MinRefreshRate", minRefreshRate) || (minRefreshRate <= 0.0)) {
REPORT_ERROR(ErrorManagement::ParametersError,
if (!data.Read("MinRefreshRate", minRefreshRate) ||
(minRefreshRate <= 0.0)) {
REPORT_ERROR(
ErrorManagement::ParametersError,
"MinRefreshRate > 0 is required when PublishingMode = Accumulate.");
ok = false;
}
else {
} else {
float64 hrtFreq = static_cast<float64>(HighResolutionTimer::Frequency());
flushPeriodTicks = static_cast<uint64>(hrtFreq / minRefreshRate);
REPORT_ERROR(ErrorManagement::Information,
REPORT_ERROR(
ErrorManagement::Information,
"Accumulate mode: MinRefreshRate=%.1f Hz, flushPeriodTicks=%llu.",
minRefreshRate,
static_cast<unsigned long long>(flushPeriodTicks));
minRefreshRate, static_cast<unsigned long long>(flushPeriodTicks));
}
}
@@ -272,11 +269,11 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
REPORT_ERROR(ErrorManagement::ParametersError,
"Ratio >= 1 is required when PublishingMode = Decimate.");
ok = false;
}
else {
} else {
decimateRatio = ratio;
if (decimateRatio == 1u) {
REPORT_ERROR(ErrorManagement::Warning,
REPORT_ERROR(
ErrorManagement::Warning,
"Decimate mode with Ratio=1 is equivalent to Strict mode.");
}
REPORT_ERROR(ErrorManagement::Information,
@@ -299,6 +296,14 @@ bool UDPStreamer::Initialise(StructuredDataI &data) {
if (data.Read("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;
if (data.Read("ClientTimeout", clientTimeout)) {
@@ -407,20 +412,15 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
if (signalsDatabase.Read("QuantizedType", quantStr)) {
if (quantStr == "uint8") {
signalInfos[i].quantType = UDPStreamerQuantUint8;
}
else if (quantStr == "int8") {
} else if (quantStr == "int8") {
signalInfos[i].quantType = UDPStreamerQuantInt8;
}
else if (quantStr == "uint16") {
} else if (quantStr == "uint16") {
signalInfos[i].quantType = UDPStreamerQuantUint16;
}
else if (quantStr == "int16") {
} else if (quantStr == "int16") {
signalInfos[i].quantType = UDPStreamerQuantInt16;
}
else if (quantStr == "none") {
} else if (quantStr == "none") {
signalInfos[i].quantType = UDPStreamerQuantNone;
}
else {
} else {
REPORT_ERROR(ErrorManagement::ParametersError,
"Signal %s: unknown QuantizedType '%s'. "
"Allowed: none|uint8|int8|uint16|int16.",
@@ -449,17 +449,13 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
}
if (timeModeStr == "PacketTime") {
signalInfos[i].timeMode = UDPStreamerTimePacket;
}
else if (timeModeStr == "FullArray") {
} else if (timeModeStr == "FullArray") {
signalInfos[i].timeMode = UDPStreamerTimeFullArray;
}
else if (timeModeStr == "FirstSample") {
} else if (timeModeStr == "FirstSample") {
signalInfos[i].timeMode = UDPStreamerTimeFirstSample;
}
else if (timeModeStr == "LastSample") {
} else if (timeModeStr == "LastSample") {
signalInfos[i].timeMode = UDPStreamerTimeLastSample;
}
else {
} else {
REPORT_ERROR(ErrorManagement::ParametersError,
"Signal %s: unknown TimeMode '%s'. "
"Allowed: PacketTime|FullArray|FirstSample|LastSample.",
@@ -477,8 +473,7 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
"TimeMode != PacketTime.",
signalInfos[i].name.Buffer());
ok = false;
}
else {
} else {
timeSignalNames[i] = tsName;
/* Index resolved in pass 3 */
signalInfos[i].timeSignalIdx = UDPS_NO_TIME_SIGNAL;
@@ -520,10 +515,10 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
}
}
if (!found) {
REPORT_ERROR(ErrorManagement::ParametersError,
REPORT_ERROR(
ErrorManagement::ParametersError,
"Signal %s: TimeSignal '%s' not found among declared signals.",
signalInfos[i].name.Buffer(),
timeSignalNames[i].Buffer());
signalInfos[i].name.Buffer(), timeSignalNames[i].Buffer());
ok = false;
}
}
@@ -567,12 +562,11 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
"Signal %s: FullArray TimeMode requires TimeSignal "
"%s to have the same NumberOfElements (%u vs %u).",
signalInfos[i].name.Buffer(),
signalInfos[tsIdx].name.Buffer(),
tsElems, signalInfos[i].numElements);
signalInfos[tsIdx].name.Buffer(), tsElems,
signalInfos[i].numElements);
ok = false;
}
}
else if ((signalInfos[i].timeMode == UDPStreamerTimeFirstSample) ||
} else if ((signalInfos[i].timeMode == UDPStreamerTimeFirstSample) ||
(signalInfos[i].timeMode == UDPStreamerTimeLastSample)) {
if (tsElems != 1u) {
REPORT_ERROR(ErrorManagement::ParametersError,
@@ -605,9 +599,11 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
*/
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;
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].unit == "us") || (signalInfos[i].unit == "ns")) {
primaryTsIdx = i;
@@ -615,7 +611,8 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
}
}
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) {
TypeDescriptor td = signalInfos[i].type;
if ((td == UnsignedInteger32Bit) || (td == UnsignedInteger64Bit) ||
@@ -639,8 +636,8 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
signalInfos[i].accumulated = true;
singleCycleWireBytes += signalInfos[i].wireByteSize;
/* Auto-assign time reference for non-primary, non-time scalars */
if ((signalInfos[i].numElements == 1u) &&
(i != primaryTsIdx) && (primaryTsIdx != UDPS_NO_TIME_SIGNAL) &&
if ((signalInfos[i].numElements == 1u) && (i != primaryTsIdx) &&
(primaryTsIdx != UDPS_NO_TIME_SIGNAL) &&
(signalInfos[i].timeMode == UDPStreamerTimePacket)) {
signalInfos[i].timeMode = UDPStreamerTimeFullArray;
signalInfos[i].timeSignalIdx = primaryTsIdx;
@@ -655,13 +652,13 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
if (ok) {
/* 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) {
REPORT_ERROR(ErrorManagement::ParametersError,
"Accumulate mode: even a single sample (%u B) exceeds "
"MaxPayloadSize (%u B).",
ACCUM_HEADER + singleCycleWireBytes,
maxPayloadSize);
ACCUM_HEADER + singleCycleWireBytes, maxPayloadSize);
ok = false;
}
}
@@ -674,8 +671,8 @@ bool UDPStreamer::SetConfiguredDatabase(StructuredDataI &data) {
REPORT_ERROR(ErrorManagement::Information,
"Accumulate mode: singleCycleWireBytes=%u, "
"maxBatchCount=%u, maxPayloadSize=%u, totalWireBytes=%u.",
singleCycleWireBytes,
maxBatchCount, maxPayloadSize, totalWireBytes);
singleCycleWireBytes, maxBatchCount, maxPayloadSize,
totalWireBytes);
}
}
@@ -695,15 +692,18 @@ bool UDPStreamer::AllocateMemory() {
HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap();
/* In Accumulate mode, readyBuffer / scratchBuffer hold maxBatchCount consecutive
* snapshots instead of a single one. */
/* HI-3: use 64-bit arithmetic to prevent overflow in maxBatchCount * totalSrcBytes */
/* In Accumulate mode, readyBuffer / scratchBuffer hold maxBatchCount
* consecutive snapshots instead of a single one. */
/* HI-3: use 64-bit arithmetic to prevent overflow in maxBatchCount *
* totalSrcBytes */
uint64 readyBufSize64 = (maxBatchCount > 0u)
? (static_cast<uint64>(maxBatchCount) * static_cast<uint64>(totalSrcBytes))
? (static_cast<uint64>(maxBatchCount) *
static_cast<uint64>(totalSrcBytes))
: static_cast<uint64>(totalSrcBytes);
if (readyBufSize64 > 0xFFFFFFFFu) {
REPORT_ERROR(ErrorManagement::FatalError,
"Accumulate buffer size overflow (maxBatchCount=%u * totalSrcBytes=%u).",
"Accumulate buffer size overflow (maxBatchCount=%u * "
"totalSrcBytes=%u).",
maxBatchCount, totalSrcBytes);
return false;
}
@@ -712,7 +712,8 @@ bool UDPStreamer::AllocateMemory() {
/* readyBuffer: copy of signal memory shared with background thread */
readyBuffer = reinterpret_cast<uint8 *>(heap->Malloc(readyBufSize));
if (readyBuffer == NULL_PTR(uint8 *)) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate readyBuffer.");
REPORT_ERROR(ErrorManagement::FatalError,
"Could not allocate readyBuffer.");
return false;
}
(void)MemoryOperationsHelper::Set(readyBuffer, 0, readyBufSize);
@@ -720,7 +721,8 @@ bool UDPStreamer::AllocateMemory() {
/* scratchBuffer: background-thread-private copy for serialization */
scratchBuffer = reinterpret_cast<uint8 *>(heap->Malloc(readyBufSize));
if (scratchBuffer == NULL_PTR(uint8 *)) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate scratchBuffer.");
REPORT_ERROR(ErrorManagement::FatalError,
"Could not allocate scratchBuffer.");
return false;
}
(void)MemoryOperationsHelper::Set(scratchBuffer, 0, readyBufSize);
@@ -744,11 +746,13 @@ bool UDPStreamer::AllocateMemory() {
/* --- Accumulate-mode extra buffers --- */
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;
accumBuffer = reinterpret_cast<uint8 *>(heap->Malloc(accumBufSize));
if (accumBuffer == NULL_PTR(uint8 *)) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not allocate accumBuffer.");
REPORT_ERROR(ErrorManagement::FatalError,
"Could not allocate accumBuffer.");
return false;
}
(void)MemoryOperationsHelper::Set(accumBuffer, 0, accumBufSize);
@@ -776,7 +780,8 @@ bool UDPStreamer::AllocateMemory() {
readyFill = 0u;
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);
}
@@ -799,7 +804,8 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName,
ok = server.Start();
/* 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) {
uint32 configBufSize = 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 32u + 1u;
HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap();
@@ -808,14 +814,12 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName,
uint32 cfgPayloadSize = 0u;
if (BuildConfigPayload(cfgBuf, configBufSize, cfgPayloadSize)) {
(void)server.SendConfig(cfgBuf, cfgPayloadSize);
}
else {
} else {
REPORT_ERROR(ErrorManagement::Warning,
"Could not build initial CONFIG payload.");
}
heap->Free(reinterpret_cast<void *&>(cfgBuf));
}
else {
} else {
REPORT_ERROR(ErrorManagement::Warning,
"Could not allocate CONFIG buffer.");
}
@@ -867,8 +871,8 @@ bool UDPStreamer::Synchronise() {
/* HI-3: if accumFill reached maxBatchCount, force-flush before writing */
if (accumFill >= maxBatchCount) {
uint32 filled = accumFill;
(void) MemoryOperationsHelper::Copy(
readyBuffer, accumBuffer, filled * totalSrcBytes);
(void)MemoryOperationsHelper::Copy(readyBuffer, accumBuffer,
filled * totalSrcBytes);
(void)MemoryOperationsHelper::Copy(
reinterpret_cast<uint8 *>(readyTimestamps),
reinterpret_cast<const uint8 *>(accumTimestamps),
@@ -887,7 +891,8 @@ bool UDPStreamer::Synchronise() {
uint32 filled = accumFill;
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 */
uint32 curPayload = ACCUM_HEADER + filled * singleCycleWireBytes;
uint32 nextPayload = curPayload + singleCycleWireBytes;
@@ -896,8 +901,8 @@ bool UDPStreamer::Synchronise() {
if (sizeCondition || timeCondition) {
bufMutex.FastLock(TTInfiniteWait);
(void) MemoryOperationsHelper::Copy(
readyBuffer, accumBuffer, filled * totalSrcBytes);
(void)MemoryOperationsHelper::Copy(readyBuffer, accumBuffer,
filled * totalSrcBytes);
(void)MemoryOperationsHelper::Copy(
reinterpret_cast<uint8 *>(readyTimestamps),
reinterpret_cast<const uint8 *>(accumTimestamps),
@@ -910,8 +915,7 @@ bool UDPStreamer::Synchronise() {
lastPublishTs = ts;
(void)dataSem.Post();
}
}
else if (publishMode == UDPStreamerPublishDecimate) {
} else if (publishMode == UDPStreamerPublishDecimate) {
/* --- Decimate path ---
* Post dataSem only every decimateRatio calls. */
decimateCounter++;
@@ -923,8 +927,7 @@ bool UDPStreamer::Synchronise() {
bufMutex.FastUnLock();
(void)dataSem.Post();
}
}
else {
} else {
/* --- Strict path: post every call --- */
bufMutex.FastLock(TTInfiniteWait);
(void)MemoryOperationsHelper::Copy(readyBuffer, memory, totalSrcBytes);
@@ -941,8 +944,11 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
if (info.GetStage() == ExecutionInfo::StartupStage) {
const char8 *modeStr = "Strict";
if (publishMode == UDPStreamerPublishAccumulate) { modeStr = "Accumulate"; }
else if (publishMode == UDPStreamerPublishDecimate) { modeStr = "Decimate"; }
if (publishMode == UDPStreamerPublishAccumulate) {
modeStr = "Accumulate";
} else if (publishMode == UDPStreamerPublishDecimate) {
modeStr = "Decimate";
}
REPORT_ERROR(ErrorManagement::Information,
"UDPStreamer background thread started (port %u, mode %s).",
static_cast<uint32>(port), modeStr);
@@ -960,7 +966,8 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
dataSem.ResetWait(TimeoutType(UDPS_DATA_WAIT_MS));
bool dataReady = (waitErr == ErrorManagement::NoError);
/* --- Poll for incoming control commands (CONNECT / DISCONNECT / ACK) --- */
/* --- Poll for incoming control commands (CONNECT / DISCONNECT / ACK) ---
*/
server.ServiceClients();
if (dataReady && server.HasClients()) {
@@ -973,8 +980,8 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
bufMutex.FastLock(TTInfiniteWait);
fill = readyFill;
if (fill > 0u) {
(void) MemoryOperationsHelper::Copy(
scratchBuffer, readyBuffer, fill * totalSrcBytes);
(void)MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer,
fill * totalSrcBytes);
(void)MemoryOperationsHelper::Copy(
reinterpret_cast<uint8 *>(scratchTimestamps),
reinterpret_cast<const uint8 *>(readyTimestamps),
@@ -984,8 +991,8 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
if (fill > 0u) {
SerializeAccumulated(scratchBuffer, scratchTimestamps, fill);
uint32 sendBytes = UDPS_TIMESTAMP_BYTES + 4u +
fill * singleCycleWireBytes;
uint32 sendBytes =
UDPS_TIMESTAMP_BYTES + 4u + fill * singleCycleWireBytes;
packetCounter++;
if (!server.SendData(packetCounter, wireBuffer, sendBytes)) {
REPORT_ERROR(ErrorManagement::Warning,
@@ -993,13 +1000,12 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
packetCounter);
}
}
}
else {
} else {
/* --- Single-snapshot send (Strict or Decimate) --- */
uint64 ts = 0u;
bufMutex.FastLock(TTInfiniteWait);
(void) MemoryOperationsHelper::Copy(
scratchBuffer, readyBuffer, totalSrcBytes);
(void)MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer,
totalSrcBytes);
ts = syncTimestamp;
bufMutex.FastUnLock();
@@ -1036,8 +1042,10 @@ void UDPStreamer::SerializeAccumulated(const uint8 *src,
*/
uint8 *dst = wireBuffer;
/* 8-byte packet-level HRT timestamp = timestamp of the first (oldest) sample */
(void) MemoryOperationsHelper::Copy(dst, &timestamps[0u], UDPS_TIMESTAMP_BYTES);
/* 8-byte packet-level HRT timestamp = timestamp of the first (oldest) sample
*/
(void)MemoryOperationsHelper::Copy(dst, &timestamps[0u],
UDPS_TIMESTAMP_BYTES);
dst += UDPS_TIMESTAMP_BYTES;
/* 4-byte sample count */
@@ -1049,17 +1057,20 @@ void UDPStreamer::SerializeAccumulated(const uint8 *src,
const bool isSrcFloat32 = (signalInfos[i].type == Float32Bit);
const float64 rMin = signalInfos[i].rangeMin;
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 */
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) {
(void) MemoryOperationsHelper::Copy(dst, slotSrc, signalInfos[i].srcByteSize);
(void)MemoryOperationsHelper::Copy(dst, slotSrc,
signalInfos[i].srcByteSize);
dst += signalInfos[i].srcByteSize;
}
else {
} else {
const uint8 *s = slotSrc;
for (uint32 e = 0u; e < nelems; e++) {
float64 rawVal = 0.0;
@@ -1068,18 +1079,22 @@ void UDPStreamer::SerializeAccumulated(const uint8 *src,
(void)MemoryOperationsHelper::Copy(&f32, s, 4u);
rawVal = static_cast<float64>(f32);
s += 4u;
}
else {
} else {
(void)MemoryOperationsHelper::Copy(&rawVal, s, 8u);
s += 8u;
}
float64 norm = (rawVal - rMin) / rRange;
if (norm < 0.0) { norm = 0.0; }
if (norm > 1.0) { norm = 1.0; }
if (norm < 0.0) {
norm = 0.0;
}
if (norm > 1.0) {
norm = 1.0;
}
switch (signalInfos[i].quantType) {
case UDPStreamerQuantUint8: {
uint8 q = static_cast<uint8>(norm * 255.0);
*dst = q; dst += 1u;
*dst = q;
dst += 1u;
break;
}
case UDPStreamerQuantInt8: {
@@ -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) {
payloadSize = 0u;
@@ -1135,7 +1148,8 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf,
if (nameLen >= UDPS_MAX_SIGNAL_NAME) {
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;
/* Type code: 1 byte */
@@ -1184,7 +1198,8 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf,
if (unitLen >= UDPS_MAX_UNIT_LEN) {
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;
payloadSize += UDPS_SIGNAL_DESC_SIZE;
@@ -1214,8 +1229,7 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
/* Raw copy */
(void)MemoryOperationsHelper::Copy(dst, src, signalInfos[i].srcByteSize);
dst += signalInfos[i].srcByteSize;
}
else {
} else {
float64 rMin = signalInfos[i].rangeMin;
float64 rRange = signalInfos[i].rangeMax - rMin;
if (rRange == 0.0) {
@@ -1232,16 +1246,19 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
(void)MemoryOperationsHelper::Copy(&f32, s, 4u);
rawVal = static_cast<float64>(f32);
s += 4u;
}
else {
} else {
(void)MemoryOperationsHelper::Copy(&rawVal, s, 8u);
s += 8u;
}
/* Normalize and clamp to [0.0, 1.0] */
float64 norm = (rawVal - rMin) / rRange;
if (norm < 0.0) { norm = 0.0; }
if (norm > 1.0) { norm = 1.0; }
if (norm < 0.0) {
norm = 0.0;
}
if (norm > 1.0) {
norm = 1.0;
}
switch (signalInfos[i].quantType) {
case UDPStreamerQuantUint8: {
@@ -1278,34 +1295,37 @@ void UDPStreamer::QuantizeAndSerialize(const uint8 *srcBuf, uint64 timestamp) {
uint8 UDPStreamer::TypeDescriptorToCode(TypeDescriptor td) {
uint8 code = UDPS_TYPECODE_UNKNOWN;
if (td == UnsignedInteger8Bit) { code = UDPS_TYPECODE_UINT8; }
else if (td == SignedInteger8Bit) { code = UDPS_TYPECODE_INT8; }
else if (td == UnsignedInteger16Bit) { code = UDPS_TYPECODE_UINT16; }
else if (td == SignedInteger16Bit) { code = UDPS_TYPECODE_INT16; }
else if (td == UnsignedInteger32Bit) { 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; }
if (td == UnsignedInteger8Bit) {
code = UDPS_TYPECODE_UINT8;
} else if (td == SignedInteger8Bit) {
code = UDPS_TYPECODE_INT8;
} else if (td == UnsignedInteger16Bit) {
code = UDPS_TYPECODE_UINT16;
} else if (td == SignedInteger16Bit) {
code = UDPS_TYPECODE_INT16;
} else if (td == UnsignedInteger32Bit) {
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;
}
uint16 UDPStreamer::GetPort() const {
return port;
}
uint16 UDPStreamer::GetPort() const { return port; }
uint32 UDPStreamer::GetMaxPayloadSize() const {
return maxPayloadSize;
}
uint32 UDPStreamer::GetMaxPayloadSize() const { return maxPayloadSize; }
bool UDPStreamer::IsClientConnected() const {
return server.HasClients();
}
bool UDPStreamer::IsClientConnected() const { return server.HasClients(); }
bool UDPStreamer::IsMulticast() const {
return server.IsMulticast();
}
bool UDPStreamer::IsMulticast() const { return server.IsMulticast(); }
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
* | Parameter | Type | Default | Description |
* |-----------------|---------|---------|-------------|
* |-----------------|---------|------------------|-------------|
* | 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. |
* | 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. |
* | 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. |
Source/Components/Interfaces/UDPStream/UDPSServer.cpp
+102 -115
View File
@@ -6,12 +6,13 @@
#include "UDPSServer.h"
#include "AdvancedErrorManagement.h"
#include "ErrorType.h"
#include "HighResolutionTimer.h"
#include "MemoryOperationsHelper.h"
#include "StreamString.h"
#include <sys/select.h>
#include <poll.h>
#include <sys/select.h>
namespace MARTe {
@@ -20,18 +21,10 @@ namespace MARTe {
// ---------------------------------------------------------------------------
UDPSServer::UDPSServer()
: port(0u),
maxPayloadSize(UDPS_SERVER_DEFAULT_MAX_PAYLOAD),
dataPort(0u),
useMulticast(false),
clientTimeoutTicks(0u),
numUnicastClients(0u),
numTCPClients(0u),
cachedConfig(NULL_PTR(uint8 *)),
cachedConfigSize(0u),
sendBuf(NULL_PTR(uint8 *)),
sendBufCapacity(0u),
configCounter(0u),
: port(0u), maxPayloadSize(UDPS_SERVER_DEFAULT_MAX_PAYLOAD), dataPort(0u),
useMulticast(false), clientTimeoutTicks(0u), numUnicastClients(0u),
numTCPClients(0u), cachedConfig(NULL_PTR(uint8 *)), cachedConfigSize(0u),
sendBuf(NULL_PTR(uint8 *)), sendBufCapacity(0u), configCounter(0u),
started(false) {
for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) {
@@ -47,9 +40,7 @@ UDPSServer::UDPSServer()
}
}
UDPSServer::~UDPSServer() {
(void) Stop();
}
UDPSServer::~UDPSServer() { (void)Stop(); }
// ---------------------------------------------------------------------------
// Initialise
@@ -59,14 +50,14 @@ bool UDPSServer::Initialise(StructuredDataI &data) {
uint32 portU32 = 0u;
if (data.Read("Port", portU32)) {
port = static_cast<uint16>(portU32);
}
else {
} else {
REPORT_ERROR_STATIC(ErrorManagement::ParametersError,
"UDPSServer: Port not specified.");
return false;
}
/* 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;
if (data.Read("MulticastGroup", mcGroup) && (mcGroup.Size() > 0u)) {
@@ -79,11 +70,18 @@ bool UDPSServer::Initialise(StructuredDataI &data) {
(void)data.Read("DataPort", dpU32);
dataPort = static_cast<uint16>(dpU32);
if (dataPort == port) {
REPORT_ERROR_STATIC(ErrorManagement::ParametersError,
REPORT_ERROR_STATIC(
ErrorManagement::ParametersError,
"UDPSServer: DataPort (%u) must differ from Port (%u).",
static_cast<uint32>(dataPort), static_cast<uint32>(port));
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;
@@ -92,8 +90,8 @@ bool UDPSServer::Initialise(StructuredDataI &data) {
uint32 timeoutSecs = UDPS_SERVER_DEFAULT_CLIENT_TIMEOUT_S;
(void)data.Read("ClientTimeout", timeoutSecs);
clientTimeoutTicks = (timeoutSecs > 0u)
? (static_cast<uint64>(timeoutSecs) * HighResolutionTimer::Frequency())
clientTimeoutTicks = (timeoutSecs > 0u) ? (static_cast<uint64>(timeoutSecs) *
HighResolutionTimer::Frequency())
: 0u;
return true;
@@ -128,19 +126,16 @@ bool UDPSServer::Start() {
tcpListener.SetBlocking(false);
}
// UDP data socket connected to multicast group
if (ok) {
ok = dataSocket.Open();
}
if (ok) {
ok = dataSocket.Connect(multicastGroup.Buffer(), dataPort);
}
ok &= dataSocket.Open();
ok &= dataSocket.Join(multicastGroup.Buffer(), interface.Buffer());
ok &= dataSocket.Connect(multicastGroup.Buffer(), dataPort);
if (!ok) {
REPORT_ERROR_STATIC(ErrorManagement::FatalError,
REPORT_ERROR_STATIC(
ErrorManagement::FatalError,
"UDPSServer: Failed to open multicast sockets on port %u.",
static_cast<uint32>(port));
}
}
else {
} else {
// Unicast send socket (unconnected; SetDestination per Write)
ok = uniSendSocket.Open();
if (!ok) {
@@ -155,7 +150,8 @@ bool UDPSServer::Start() {
ok = serverSocket.Listen(port); // UDP "listen" = bind
}
if (!ok) {
REPORT_ERROR_STATIC(ErrorManagement::FatalError,
REPORT_ERROR_STATIC(
ErrorManagement::FatalError,
"UDPSServer: Failed to bind receive socket on port %u.",
static_cast<uint32>(port));
}
@@ -235,11 +231,9 @@ void UDPSServer::ServiceClients() {
pfd.fd = static_cast<int>(tcpListener.GetReadHandle());
pfd.events = POLLIN;
pfd.revents = 0;
bool pending = (::poll(&pfd, 1u, 0) > 0) &&
((pfd.revents & POLLIN) != 0);
BasicTCPSocket *newConn = pending
? tcpListener.WaitConnection(0u)
: NULL_PTR(BasicTCPSocket *);
bool pending = (::poll(&pfd, 1u, 0) > 0) && ((pfd.revents & POLLIN) != 0);
BasicTCPSocket *newConn =
pending ? tcpListener.WaitConnection(0u) : NULL_PTR(BasicTCPSocket *);
if (newConn != NULL_PTR(BasicTCPSocket *)) {
// Find free TCP client slot
uint32 freeSlot = UDPS_SERVER_MAX_TCP_CLIENTS;
@@ -251,9 +245,9 @@ void UDPSServer::ServiceClients() {
}
if (freeSlot < UDPS_SERVER_MAX_TCP_CLIENTS) {
HandleMulticastTCPConnect(newConn, freeSlot);
}
else {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
} else {
REPORT_ERROR_STATIC(
ErrorManagement::Warning,
"UDPSServer: TCP client table full, rejecting new connection.");
(void)newConn->Close();
delete newConn;
@@ -268,20 +262,25 @@ void UDPSServer::ServiceClients() {
}
// Non-blocking peek
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_ZERO(&rset);
FD_SET(fd, &rset);
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);
if (nready > 0) {
uint8 pktBuf[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)) {
REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: TCP client disconnected (slot %u).", i);
"UDPSServer: TCP client disconnected (slot %u).",
i);
EvictTCPClient(i);
continue;
}
@@ -290,21 +289,23 @@ void UDPSServer::ServiceClients() {
reinterpret_cast<const UDPSPacketHeader *>(pktBuf);
if ((hdr->magic == UDPS_MAGIC) &&
(hdr->type == UDPS_TYPE_DISCONNECT)) {
REPORT_ERROR_STATIC(ErrorManagement::Information,
REPORT_ERROR_STATIC(
ErrorManagement::Information,
"UDPSServer: TCP client sent DISCONNECT (slot %u).", i);
EvictTCPClient(i);
}
}
}
}
}
else {
} else {
// Unicast: poll serverSocket for CONNECT / DISCONNECT / ACK
if (!serverSocket.IsValid()) {
return;
}
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_ZERO(&rset);
FD_SET(fd, &rset);
@@ -313,7 +314,8 @@ void UDPSServer::ServiceClients() {
while (nready > 0) {
uint8 pktBuf[UDPS_HEADER_SIZE + 4u];
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)) {
break;
}
@@ -325,11 +327,9 @@ void UDPSServer::ServiceClients() {
InternetHost src = serverSocket.GetSource();
if (hdr->type == UDPS_TYPE_CONNECT) {
HandleUnicastConnect(src);
}
else if (hdr->type == UDPS_TYPE_DISCONNECT) {
} else if (hdr->type == UDPS_TYPE_DISCONNECT) {
HandleUnicastDisconnect(src);
}
else if (hdr->type == UDPS_TYPE_ACK) {
} else if (hdr->type == UDPS_TYPE_ACK) {
HandleUnicastAck(src);
}
@@ -373,14 +373,14 @@ bool UDPSServer::SendConfig(const uint8 *payload, uint32 payloadSize) {
bool sent = SendFragmentedTCP(*tcpClients[i], UDPS_TYPE_CONFIG,
configCounter, payload, payloadSize);
if (!sent) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
REPORT_ERROR_STATIC(
ErrorManagement::Warning,
"UDPSServer: CONFIG send failed to TCP client %u, evicting.", i);
EvictTCPClient(i);
ok = false;
}
}
}
else {
} else {
// Send CONFIG to each unicast client
for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) {
if (!unicastClients[i].active) {
@@ -406,7 +406,8 @@ bool UDPSServer::SendConfig(const uint8 *payload, uint32 payloadSize) {
// SendData
// ---------------------------------------------------------------------------
bool UDPSServer::SendData(uint32 counter, const uint8 *payload, uint32 payloadSize) {
bool UDPSServer::SendData(uint32 counter, const uint8 *payload,
uint32 payloadSize) {
if (!started) {
return false;
}
@@ -415,15 +416,15 @@ bool UDPSServer::SendData(uint32 counter, const uint8 *payload, uint32 payloadSi
if (useMulticast) {
// Single multicast write (no dest needed — socket already connected)
bool sent = SendFragmentedUDP(dataSocket, NULL_PTR(InternetHost *),
UDPS_TYPE_DATA, counter, payload, payloadSize);
bool sent =
SendFragmentedUDP(dataSocket, NULL_PTR(InternetHost *), UDPS_TYPE_DATA,
counter, payload, payloadSize);
if (!sent) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"UDPSServer: DATA send to multicast group failed.");
ok = false;
}
}
else {
} else {
for (uint32 i = 0u; i < UDPS_SERVER_MAX_UNICAST_CLIENTS; i++) {
if (!unicastClients[i].active) {
continue;
@@ -470,7 +471,8 @@ bool UDPSServer::AddStaticClient(const char8 *ip, uint16 port_) {
}
if (freeSlot >= UDPS_SERVER_MAX_UNICAST_CLIENTS) {
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_));
return false;
}
@@ -489,8 +491,8 @@ bool UDPSServer::AddStaticClient(const char8 *ip, uint16 port_) {
numUnicastClients++;
REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: Static client added: %s:%u.",
ip, static_cast<uint32>(port_));
"UDPSServer: Static client added: %s:%u.", ip,
static_cast<uint32>(port_));
return true;
}
@@ -513,35 +515,24 @@ uint32 UDPSServer::GetClientCount() const {
return count;
}
bool UDPSServer::HasClients() const {
return (GetClientCount() > 0u);
}
bool UDPSServer::HasClients() const { return (GetClientCount() > 0u); }
bool UDPSServer::IsMulticast() const {
return useMulticast;
}
bool UDPSServer::IsMulticast() const { return useMulticast; }
uint16 UDPSServer::GetPort() const {
return port;
}
uint16 UDPSServer::GetPort() const { return port; }
uint32 UDPSServer::GetMaxPayloadSize() const {
return maxPayloadSize;
}
uint32 UDPSServer::GetMaxPayloadSize() const { return maxPayloadSize; }
// ---------------------------------------------------------------------------
// Private: SendFragmentedUDP
// ---------------------------------------------------------------------------
bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock,
InternetHost *dest,
uint8 type,
uint32 counter,
const uint8 *payload,
uint32 payloadSize) {
bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock, InternetHost *dest,
uint8 type, uint32 counter,
const uint8 *payload, uint32 payloadSize) {
uint32 maxChunk = maxPayloadSize; // payload bytes per fragment (excl. header)
uint32 totalFrags = (payloadSize == 0u) ? 1u :
((payloadSize + maxChunk - 1u) / maxChunk);
uint32 totalFrags =
(payloadSize == 0u) ? 1u : ((payloadSize + maxChunk - 1u) / maxChunk);
bool ok = true;
uint32 offs = 0u;
@@ -552,15 +543,12 @@ bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock,
chunkSize = maxChunk;
}
UDPSBuildHeader(sendBuf, type, counter,
static_cast<uint16>(f),
static_cast<uint16>(totalFrags),
chunkSize);
UDPSBuildHeader(sendBuf, type, counter, static_cast<uint16>(f),
static_cast<uint16>(totalFrags), chunkSize);
if (chunkSize > 0u) {
(void)MemoryOperationsHelper::Copy(sendBuf + UDPS_HEADER_SIZE,
payload + offs,
chunkSize);
payload + offs, chunkSize);
}
uint32 sendSize = UDPS_HEADER_SIZE + chunkSize;
@@ -586,14 +574,12 @@ bool UDPSServer::SendFragmentedUDP(BasicUDPSocket &sock,
// Private: SendFragmentedTCP
// ---------------------------------------------------------------------------
bool UDPSServer::SendFragmentedTCP(BasicTCPSocket &sock,
uint8 type,
uint32 counter,
const uint8 *payload,
bool UDPSServer::SendFragmentedTCP(BasicTCPSocket &sock, uint8 type,
uint32 counter, const uint8 *payload,
uint32 payloadSize) {
uint32 maxChunk = maxPayloadSize;
uint32 totalFrags = (payloadSize == 0u) ? 1u :
((payloadSize + maxChunk - 1u) / maxChunk);
uint32 totalFrags =
(payloadSize == 0u) ? 1u : ((payloadSize + maxChunk - 1u) / maxChunk);
bool ok = true;
uint32 offs = 0u;
@@ -604,15 +590,12 @@ bool UDPSServer::SendFragmentedTCP(BasicTCPSocket &sock,
chunkSize = maxChunk;
}
UDPSBuildHeader(sendBuf, type, counter,
static_cast<uint16>(f),
static_cast<uint16>(totalFrags),
chunkSize);
UDPSBuildHeader(sendBuf, type, counter, static_cast<uint16>(f),
static_cast<uint16>(totalFrags), chunkSize);
if (chunkSize > 0u) {
(void)MemoryOperationsHelper::Copy(sendBuf + UDPS_HEADER_SIZE,
payload + offs,
chunkSize);
payload + offs, chunkSize);
}
uint32 sendSize = UDPS_HEADER_SIZE + chunkSize;
@@ -660,8 +643,8 @@ void UDPSServer::HandleUnicastConnect(const InternetHost &src) {
uint16 srcPort = src.GetPort();
REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: CONNECT from %s:%u.",
srcAddr, static_cast<uint32>(srcPort));
"UDPSServer: CONNECT from %s:%u.", srcAddr,
static_cast<uint32>(srcPort));
// Check if this client is already known
uint32 existing = FindUnicastClient(srcAddr, srcPort);
@@ -693,9 +676,9 @@ void UDPSServer::HandleUnicastConnect(const InternetHost &src) {
unicastClients[slot].ipAddr,
static_cast<uint32>(unicastClients[slot].clientPort));
EvictUnicastClient(slot);
}
else {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
} else {
REPORT_ERROR_STATIC(
ErrorManagement::Warning,
"UDPSServer: All slots occupied by static clients; rejecting %s:%u.",
srcAddr, static_cast<uint32>(srcPort));
return;
@@ -736,8 +719,8 @@ void UDPSServer::HandleUnicastDisconnect(const InternetHost &src) {
uint32 slot = FindUnicastClient(srcAddr, srcPort);
if (slot < UDPS_SERVER_MAX_UNICAST_CLIENTS) {
REPORT_ERROR_STATIC(ErrorManagement::Information,
"UDPSServer: DISCONNECT from %s:%u.",
srcAddr, static_cast<uint32>(srcPort));
"UDPSServer: DISCONNECT from %s:%u.", srcAddr,
static_cast<uint32>(srcPort));
EvictUnicastClient(slot);
}
}
@@ -826,7 +809,8 @@ void UDPSServer::EvictUnicastClient(uint32 idx) {
// Private: HandleMulticastTCPConnect
// ---------------------------------------------------------------------------
void UDPSServer::HandleMulticastTCPConnect(BasicTCPSocket *newClient, uint32 idx) {
void UDPSServer::HandleMulticastTCPConnect(BasicTCPSocket *newClient,
uint32 idx) {
if (idx >= UDPS_SERVER_MAX_TCP_CLIENTS) {
return;
}
@@ -835,16 +819,19 @@ void UDPSServer::HandleMulticastTCPConnect(BasicTCPSocket *newClient, uint32 idx
numTCPClients++;
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
if (cachedConfig != NULL_PTR(uint8 *)) {
configCounter++;
bool sent = SendFragmentedTCP(*newClient, UDPS_TYPE_CONFIG,
configCounter, cachedConfig, cachedConfigSize);
bool sent = SendFragmentedTCP(*newClient, UDPS_TYPE_CONFIG, configCounter,
cachedConfig, cachedConfigSize);
if (!sent) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"UDPSServer: Failed to send CONFIG to new TCP client (slot %u).", idx);
REPORT_ERROR_STATIC(
ErrorManagement::Warning,
"UDPSServer: Failed to send CONFIG to new TCP client (slot %u).",
idx);
EvictTCPClient(idx);
}
}
Source/Components/Interfaces/UDPStream/UDPSServer.h
+1
View File
@@ -235,6 +235,7 @@ private:
uint16 port;
uint32 maxPayloadSize;
StreamString multicastGroup;
StreamString interface;
uint16 dataPort;
bool useMulticast;
uint64 clientTimeoutTicks; ///< 0 = disabled
vgcore.26968
BIN
View File
Binary file not shown.