Martino Ferrari
231 lines
9.2 KiB
C++
231 lines
9.2 KiB
C++
#ifndef UDPS_CLIENT_H_
|
|
#define UDPS_CLIENT_H_
|
|
|
|
/**
|
|
* @file UDPSClient.h
|
|
* @brief Auto-reconnecting UDPS receiver client (C++ MARTe2 library class).
|
|
*
|
|
* UDPSClient runs its own background thread (via SingleThreadService).
|
|
* On start it connects to a UDPSServer, receives CONFIG + DATA packets, and
|
|
* reassembles fragmented updates. If the server goes silent for longer than
|
|
* SilenceTimeout, it automatically disconnects and retries.
|
|
*
|
|
* Unicast mode: sends CONNECT to server UDP port; receives on an ephemeral UDP port.
|
|
* Multicast mode: TCP to server port for CONFIG; joins UDP multicast group for DATA.
|
|
*
|
|
* Threading: UDPSClientListener callbacks are invoked from the internal receive
|
|
* thread — the implementation must be thread-safe with respect to the caller.
|
|
*/
|
|
|
|
#include "BasicTCPSocket.h"
|
|
#include "BasicUDPSocket.h"
|
|
#include "EmbeddedServiceMethodBinderI.h"
|
|
#include "ExecutionInfo.h"
|
|
#include "HighResolutionTimer.h"
|
|
#include "InternetHost.h"
|
|
#include "SingleThreadService.h"
|
|
#include "StreamString.h"
|
|
#include "StructuredDataI.h"
|
|
#include "UDPSProtocol.h"
|
|
|
|
namespace MARTe {
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Listener interface
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* @brief Callback interface for UDPSClient events.
|
|
*
|
|
* Implement this interface and pass an instance to UDPSClient::SetListener().
|
|
* All callbacks are invoked from the UDPSClient internal thread.
|
|
*/
|
|
class UDPSClientListener {
|
|
public:
|
|
virtual ~UDPSClientListener() {}
|
|
|
|
/** Called when a fully-reassembled CONFIG payload is available. */
|
|
virtual void OnUDPSConfig(const uint8 *payload, uint32 payloadSize) {}
|
|
|
|
/** Called when a fully-reassembled DATA payload is available. */
|
|
virtual void OnUDPSData(const uint8 *payload, uint32 payloadSize) {}
|
|
|
|
/**
|
|
* @brief Called for every received DATA datagram (fragment).
|
|
* @param counter DATA packet counter from the UDPS header.
|
|
* @param nBytes Raw datagram size (header + payload).
|
|
* @param complete True iff this fragment completed the DATA reassembly.
|
|
*/
|
|
virtual void OnUDPSFragment(uint32 counter, uint32 nBytes, bool complete) {}
|
|
|
|
/** Called when the connection to the server has been established. */
|
|
virtual void OnUDPSConnected() {}
|
|
|
|
/** Called when the connection has been lost or closed. */
|
|
virtual void OnUDPSDisconnected() {}
|
|
};
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// UDPSClient
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* @brief UDPS receiver client with auto-reconnect and fragment reassembly.
|
|
*/
|
|
class UDPSClient : public EmbeddedServiceMethodBinderI {
|
|
public:
|
|
|
|
/** Maximum number of concurrent in-flight reassembly slots. */
|
|
static const uint32 UDPS_CLIENT_MAX_REASSEMBLY_SLOTS = 4u;
|
|
|
|
/** Maximum size (bytes) of a single reassembled packet payload. A UDPS
|
|
* update for one source is fragmented into MaxPayloadSize chunks; this is
|
|
* the ceiling on the reassembled total, so it bounds the largest multi-
|
|
* fragment packet the client can deliver. Sized for large array bursts
|
|
* (e.g. 8x10000 float32 ~= 320 KiB) with headroom; stays well within the
|
|
* 256-fragment span the recvMask[32] tracks at typical chunk sizes. */
|
|
static const uint32 UDPS_CLIENT_MAX_PACKET_BYTES = 1048576u; // 1 MiB
|
|
|
|
/** Default silence timeout before reconnect (seconds). */
|
|
static const uint32 UDPS_CLIENT_DEFAULT_SILENCE_TIMEOUT_S = 5u;
|
|
|
|
/** Default delay between reconnect attempts (seconds). */
|
|
static const uint32 UDPS_CLIENT_DEFAULT_RECONNECT_DELAY_S = 2u;
|
|
|
|
/** Default maximum payload size (bytes, excluding 17-byte header). */
|
|
static const uint32 UDPS_CLIENT_DEFAULT_MAX_PAYLOAD = 1400u;
|
|
|
|
/** Default OS UDP receive socket buffer size (bytes). The Linux default
|
|
* (rmem_default, typically ~208 KiB) is easily overrun by high-throughput
|
|
* sources (e.g. multi-hundred-KiB bursts every few ms), causing silent
|
|
* kernel-level datagram drops. 4 MiB gives generous burst headroom. */
|
|
static const uint32 UDPS_CLIENT_DEFAULT_RECV_BUFFER = 4194304u; // 4 MiB
|
|
|
|
UDPSClient();
|
|
virtual ~UDPSClient();
|
|
|
|
/**
|
|
* @brief Read configuration from a StructuredDataI node.
|
|
*
|
|
* Expected keys:
|
|
* - ServerAddr (char*) Server IPv4 address. Required.
|
|
* - Port (uint16) Server UDP port (unicast) or TCP listen port (multicast). Required.
|
|
* - MulticastGroup (char*) IPv4 multicast address; presence enables multicast mode.
|
|
* - DataPort (uint16) UDP multicast data port (defaults to Port+1).
|
|
* - SilenceTimeout (uint32) Seconds of no data before reconnect. Default 5.
|
|
* - ReconnectDelay (uint32) Seconds to wait between reconnect attempts. Default 2.
|
|
* - MaxPayloadSize (uint32) Max payload bytes per datagram, excluding header. Default 1400.
|
|
* - CPUMask (uint32) CPU affinity mask for the receive thread. Default 0xFFFFFFFF.
|
|
* - StackSize (uint32) Stack size for the receive thread. Default 65536.
|
|
* - RecvBufferSize (uint32) OS UDP receive socket buffer size (bytes). Default 4 MiB.
|
|
*/
|
|
bool Initialise(StructuredDataI &data);
|
|
|
|
/**
|
|
* @brief Register the event listener. Must be called before Start().
|
|
*/
|
|
void SetListener(UDPSClientListener *listener);
|
|
|
|
/**
|
|
* @brief Start the receive thread.
|
|
* @return true on success.
|
|
*/
|
|
bool Start();
|
|
|
|
/**
|
|
* @brief Stop the receive thread and close all sockets.
|
|
* @return true on success.
|
|
*/
|
|
bool Stop();
|
|
|
|
/**
|
|
* @brief Internal thread entry point (EmbeddedServiceMethodBinderI).
|
|
* @details Do NOT call directly.
|
|
*/
|
|
virtual ErrorManagement::ErrorType Execute(ExecutionInfo &info);
|
|
|
|
private:
|
|
|
|
// -------------------------------------------------------------------------
|
|
// Fragment reassembly slot
|
|
// -------------------------------------------------------------------------
|
|
struct UDPSReassemblySlot {
|
|
uint32 counter; ///< Packet counter this slot belongs to
|
|
uint8 type; ///< UDPS_TYPE_DATA or UDPS_TYPE_CONFIG
|
|
uint16 totalFragments; ///< Expected fragment count
|
|
uint16 receivedFragments; ///< How many we have so far
|
|
uint8 recvMask[32]; ///< Bitmask: bit f set iff fragment f received
|
|
uint8 payload[UDPS_CLIENT_MAX_PACKET_BYTES]; ///< Assembled payload buffer
|
|
uint64 firstSeenTicks; ///< For GC (2 s stale detection)
|
|
bool active; ///< Slot in use
|
|
uint32 chunkSize; ///< Payload bytes per fragment (from first fragment)
|
|
uint32 assembledBytes; ///< Exact total payload bytes placed so far
|
|
};
|
|
|
|
// -------------------------------------------------------------------------
|
|
// Connection state machine helpers
|
|
// -------------------------------------------------------------------------
|
|
bool Connect();
|
|
void Disconnect();
|
|
bool ReceiveAndProcess();
|
|
|
|
bool ConnectUnicast();
|
|
bool ConnectMulticast();
|
|
|
|
/** Set the OS receive buffer size (SO_RCVBUF) on a UDP socket's raw handle. */
|
|
void SetRecvBufferSize(BasicUDPSocket &sock);
|
|
|
|
void ProcessDatagram(const uint8 *buf, uint32 size);
|
|
/** Read one full UDPS frame (header + payload) from the TCP control socket. */
|
|
bool ReceiveTCPFrame();
|
|
/** Read exactly n bytes from the TCP control socket. */
|
|
bool ReadExactTCP(uint8 *dst, uint32 n);
|
|
/** @return true iff this fragment completed the reassembly (payload delivered). */
|
|
bool PlaceFragment(const UDPSPacketHeader *hdr, const uint8 *payload, uint32 payloadBytes);
|
|
void GcReassemblySlots();
|
|
void DeliverAssembled(UDPSReassemblySlot &slot);
|
|
|
|
// -------------------------------------------------------------------------
|
|
// Configuration
|
|
// -------------------------------------------------------------------------
|
|
StreamString serverAddr;
|
|
uint16 serverPort;
|
|
StreamString multicastGroup;
|
|
uint16 dataPort;
|
|
bool useMulticast;
|
|
uint64 silenceTimeoutTicks;
|
|
uint64 reconnectDelayTicks;
|
|
uint32 maxPayloadSize;
|
|
uint32 cpuMask;
|
|
uint32 stackSize;
|
|
uint32 recvBufferSize;
|
|
|
|
// -------------------------------------------------------------------------
|
|
// Runtime state
|
|
// -------------------------------------------------------------------------
|
|
UDPSClientListener *listener;
|
|
SingleThreadService threadService;
|
|
bool connected;
|
|
uint64 lastDataTicks; ///< Ticks at last received DATA/CONFIG
|
|
uint64 disconnectTick; ///< Ticks when we disconnected (for delay)
|
|
|
|
// Unicast
|
|
BasicUDPSocket recvSocket; ///< Bound to ephemeral port; receives DATA
|
|
uint16 localPort; ///< Ephemeral port we're listening on
|
|
|
|
// Multicast
|
|
BasicTCPSocket tcpSocket; ///< TCP connection to server (for CONNECT + CONFIG)
|
|
BasicUDPSocket mcastSocket; ///< Joined multicast group
|
|
|
|
// Reassembly
|
|
UDPSReassemblySlot reassemblySlots[UDPS_CLIENT_MAX_REASSEMBLY_SLOTS];
|
|
uint64 lastGcTicks; ///< Ticks at last GC run
|
|
|
|
// Receive scratch buffer
|
|
uint8 recvBuf[65535u + UDPS_HEADER_SIZE];
|
|
};
|
|
|
|
} // namespace MARTe
|
|
|
|
#endif // UDPS_CLIENT_H_
|