MARTe-Integrated-Components/Docs/UDPStreamer.md

UDPStreamer DataSource

UDPStreamer is a MARTe2 output DataSource that streams signals from a real-time application to a single connected UDP client. It is fully asynchronous from the RT thread: the RT cycle only performs a fast spinlock + memcpy, while all network I/O runs on a dedicated background thread.

Key Features

• Zero-copy RT path — Synchronise() only locks, copies signal memory, and posts a semaphore.
• Unicast and multicast — unicast (default): single client at a time, new CONNECT replaces the previous session. Multicast: multiple clients receive data simultaneously by joining a multicast group; control traffic uses a TCP listener.
• Packet fragmentation — large payloads are split into ≤ MaxPayloadSize-byte datagrams, each with a header carrying fragment index and total count so the client can reassemble them.
• Signal quantization — float32/float64 signals can be linearly quantized to uint8, int8, uint16, or int16 on the wire, reducing bandwidth significantly.
• Temporal arrays — signals with NumberOfElements > 1 can carry per-sample time metadata via TimeMode and TimeSignal, enabling high-frequency burst transmission (e.g. 1 000 samples per RT cycle at 1 MSps).
• Publishing modes — Strict (one packet per RT cycle), Accumulate (batch N snapshots then flush on size or time limit), Decimate (send every Nth cycle).

---

Configuration

+Streamer = {
    Class           = UDPStreamer

    // Network
    Port            = 44500        // UDP port (unicast) or TCP control port (multicast)
    MaxPayloadSize  = 1400         // Maximum bytes per UDP datagram (default: 1400)
                                   //   Must be > 17 (header size). Tune for MTU.

    // Multicast (optional — omit for unicast mode)
    MulticastGroup  = "239.0.0.1"  // IPv4 multicast address (224.0.0.0/4)
    Interface       = "eth0"       // Multicast-bound interface (mandatory when MulticastGroup is set)
    DataPort        = 44501        // UDP port for multicast DATA (default: Port+1)

    // Publishing mode (optional)
    PublishingMode  = "Strict"     // Strict | Accumulate | Decimate
    // For Accumulate mode:
    MinRefreshRate  = 120.0        // Flush frequency in Hz (required for Accumulate)
    // For Decimate mode:
    Ratio           = 10           // Send 1 packet every N RT cycles (required for Decimate)

    // Background thread (optional)
    CPUMask         = 0x2          // CPU affinity mask for the network thread
    StackSize       = 1048576      // Stack size in bytes (default: 1 MiB)

    Signals = {
        // ── Scalar signal ────────────────────────────────────────────────────
        Time = {
            Type = uint32
            Unit = "us"            // Optional: physical unit string (informational)
        }

        // ── Float signal with quantization ───────────────────────────────────
        Pressure = {
            Type         = float32
            Unit         = "Pa"
            RangeMin     = 0.0         // Required when QuantizedType is set
            RangeMax     = 1000000.0   // Required when QuantizedType is set
            QuantizedType = uint16     // none | uint8 | int8 | uint16 | int16
        }

        // ── Temporal array (packed burst) ────────────────────────────────────
        Channel1 = {
            Type               = float32
            NumberOfDimensions = 1
            NumberOfElements   = 1000         // N samples per RT cycle
            Unit               = "V"
            TimeMode           = FirstSample  // see Time Modes below
            TimeSignal         = Time         // name of a scalar signal in this DataSource
            SamplingRate       = 1000000.0    // Hz — used by client to reconstruct timestamps
        }
    }
}

Top-level Parameters

Parameter	Type	Default	Description
Port	uint16	44500	UDP server port (unicast) or TCP control port (multicast). Values ≤ 1024 produce a warning.
MaxPayloadSize	uint32	1400	Max payload bytes per UDP datagram (min 18)
MulticastGroup	string	(absent)	IPv4 multicast address (e.g. "239.0.0.1"). Must be in 224.0.0.0/4. Absent or empty = unicast mode.
Interface	string	(absent)	Network interface for multicast binding (e.g. "eth0"). Mandatory when MulticastGroup is set.
DataPort	uint16	Port+1	UDP port for multicast DATA datagrams. Ignored in unicast mode.
PublishingMode	string	Strict	Strict: send every RT cycle. Accumulate: batch until size/time limit. Decimate: send every Nth cycle.
MinRefreshRate	float64	—	Flush frequency in Hz. Required when PublishingMode = Accumulate.
Ratio	uint32	—	Send 1 packet every Ratio RT cycles. Required when PublishingMode = Decimate.
CPUMask	uint32	0xFFFFFFFF (any)	Background thread CPU affinity bitmask
StackSize	uint32	MARTe2 default	Background thread stack size in bytes

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
QuantizedType	string	none	float32/float64 only
TimeMode	string	PacketTime	Signals with NumberOfElements > 1
TimeSignal	string	—	Required when TimeMode ≠ PacketTime
SamplingRate	float64	0.0	Required when TimeMode = FirstSample or LastSample

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]
uint16	uint16	16-bit	Maps [RangeMin, RangeMax] → [0, 65 535]
int16	int16	16-bit	Maps [RangeMin, RangeMax] → [-32 767, 32 767]

Quantization formula (unsigned, e.g. uint16):

normalized = clamp((value - RangeMin) / (RangeMax - RangeMin), 0.0, 1.0)
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.
LastSample	Same as FirstSample but TimeSignal is the timestamp of element [N-1].	Scalar TimeSignal; SamplingRate > 0.

---

Network Modes

Unicast (default)

The server opens a single UDP socket on Port. The client initiates the session by sending a CONNECT packet to that port. The server replies with a CONFIG packet on the same socket and subsequently sends DATA packets directly to the client's address. One client at a time; a new CONNECT evicts the previous client.

Multicast

Enabled by setting MulticastGroup to a valid IPv4 multicast address (224.0.0.0/4). The Interface parameter is mandatory and specifies the network interface to bind.

The server opens a TCP listener on Port for control traffic and a UDP socket aimed at MulticastGroup:DataPort for data traffic. The client:

1. Connects to Port via TCP and sends a CONNECT packet.
2. Receives the CONFIG packet over TCP.
3. Joins the multicast group (MulticastGroup:DataPort) to receive DATA packets.

Multiple clients may receive data simultaneously by joining the same group.

---

Publishing Modes

Strict (default)

Sends one DATA packet for every Synchronise() call (every RT cycle). Simplest and lowest latency.

Accumulate

Batches multiple RT-cycle snapshots into a single DATA packet. All signals (scalars and arrays) are accumulated: one full snapshot per RT cycle. The batch is flushed when either:

• Size condition: adding one more sample would exceed MaxPayloadSize.
• Time condition: 1/MinRefreshRate seconds have elapsed since the last flush.

The maximum batch count is computed automatically from MaxPayloadSize and the total wire size of all signals. Scalar signals with Unit="us" or "ns" are auto-promoted as the per-sample FullArray time reference for all other scalars.

Requires MinRefreshRate (Hz) to be set.

Decimate

Sends one DATA packet every Ratio RT cycles, dropping intermediate cycles. Only the most recent snapshot at the Nth cycle is sent.

Requires Ratio (≥ 1) to be set. Ratio = 1 is equivalent to Strict mode (a warning is logged).

---

Broker

UDPStreamer uses MemoryMapSynchronisedOutputBroker for output signals. This broker is called automatically by the MARTe2 scheduler after all GAMs in the thread have executed.

Note: Input signals are not supported — GetBrokerName() returns "" for InputSignals.

---

Lifecycle

PrepareNextState()          ← opens UDP server socket, starts background thread
    ↓
[RT thread, each cycle]
    GAMs execute            ← write into Streamer signal memory via broker
    Synchronise()           ← spinlock + memcpy to readyBuffer + post dataSem
    ↓
[Background thread]
    Poll serverSocket       ← receive CONNECT / DISCONNECT / ACK
    Wait dataSem            ← woken by Synchronise()
    QuantizeAndSerialize()  ← build wire payload
    SendFragmented()        ← send DATA fragments to client

---

Performance Notes

• The RT path (Synchronise()) performs only: FastLock() + memcpy + FastUnLock() + EventSem.Post(). No socket calls, no heap allocation.
• readyBuffer and wireBuffer are allocated once in AllocateMemory().
• If no client is connected the background thread skips serialisation entirely.
• Packet loss is tolerated silently. ACK tracking is reserved for future use.

---

Example: minimal scalar streaming (unicast)

+Data = {
    Class = ReferenceContainer

    +Streamer = {
        Class = UDPStreamer
        Port = 44500
        Signals = {
            Counter = { Type = uint32 }
            Voltage = { Type = float32; Unit = "V"; RangeMin = -10.0; RangeMax = 10.0; QuantizedType = uint16 }
        }
    }
}

Example: multicast with accumulation

+Streamer = {
    Class            = UDPStreamer
    Port             = 44500           // TCP control port
    MulticastGroup   = "239.0.0.1"     // Enables multicast mode
    Interface        = "eth0"          // Mandatory for multicast
    DataPort         = 44501           // UDP data port (default: Port+1)
    MaxPayloadSize   = 1400
    PublishingMode   = "Accumulate"
    MinRefreshRate   = 60.0            // Flush at least 60 times/s

    Signals = {
        Time    = { Type = uint32; Unit = "us" }
        Voltage = { Type = float32; Unit = "V"; RangeMin = -10.0; RangeMax = 10.0; QuantizedType = uint16 }
    }
}

Example: high-frequency burst

+Streamer = {
    Class          = UDPStreamer
    Port           = 44500
    MaxPayloadSize = 1400

    Signals = {
        T0 = { Type = uint32; Unit = "us" }

        Ch1 = {
            Type               = float32
            NumberOfDimensions = 1
            NumberOfElements   = 1000        // 1000 samples per RT cycle
            Unit               = "V"
            TimeMode           = FirstSample
            TimeSignal         = T0
            SamplingRate       = 1000000.0   // 1 MSps
        }
    }
}

With MaxPayloadSize = 1400, a single 1000-element float32 signal produces:

payload = 8 B (HRT timestamp) + 4 B (T0/uint32) + 4000 B (float32×1000) = 4012 B
fragments = ceil(4012 / 1383) = 3

Example: decimated output

+Streamer = {
    Class          = UDPStreamer
    Port           = 44500
    PublishingMode = "Decimate"
    Ratio          = 10                // Send 1 packet every 10 RT cycles

    Signals = {
        Time     = { Type = uint32; Unit = "us" }
        Position = { Type = float64; Unit = "mm" }
    }
}