diff --git a/CLAUDE.md b/CLAUDE.md index 7dce449..4278292 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -30,6 +30,9 @@ make -C Source/Components/DataSources/UDPStreamer -f Makefile.gcc cd Common/Client/go && go build ./... cd Client/debugger && go build ./... +# Standalone C UDPS client library (no MARTe2, libc + BSD sockets only) +cd Common/Client/c && make && make cxxcheck + # ImGui desktop client (not a MARTe2 component; needs SDL2) cd Client/streamhub && cmake -B build -DCMAKE_BUILD_TYPE=Release && cmake --build build @@ -81,7 +84,7 @@ Two independent data paths: 1. **Streaming path**: `UDPStreamer` DataSource serialises signals each RT cycle to UDPS binary packets (UDP 44500, unicast/multicast) → `StreamHub` (`Source/Applications/StreamHub/`, headless C++ app: ring buffers, LTTB decimation, trigger FSM) → WebSocket 8090 → browser (`Client/udpstreamer`, Go), native ImGui client (`Client/streamhub`), or native Qt client (`Client/streamhub-qt`). 2. **Debug path**: `DebugService` patches the `ClassRegistryDatabase` at `Initialise()` so subsequent `ConfigureApplication()` instantiates `DebugBrokerWrapper` around all `MemoryMap*Broker` types — no application changes. RT hot path goes through `DebugServiceI` (abstract singleton in `DebugServiceI.h`) for forcing/tracing/breakpoints. Exposes TCP 8080 (text commands), UDP 8081 (trace telemetry), works with `TcpLogger` on 8082. Web UI: `Client/debugger` (Go). -**Shared wire format**: `Common/UDP/UDPSProtocol.h` defines the UDPS binary protocol (17-byte packed header, 136-byte signal descriptors, little-endian). It is deliberately MARTe2-free so it's shared by C++ producers (`UDPStreamer`, `DebugService`), the C++ consumer (`Source/Components/Interfaces/UDPStream/UDPSClient`), and the Go decoder (`Common/Client/go/udpsprotocol`). Changes to the protocol must be mirrored across all of these, plus the JS client parsers. +**Shared wire format**: `Common/UDP/UDPSProtocol.h` defines the UDPS binary protocol (17-byte packed header, 136-byte signal descriptors, little-endian). It is deliberately MARTe2-free so it's shared by C++ producers (`UDPStreamer`, `DebugService`), the C++ consumer (`Source/Components/Interfaces/UDPStream/UDPSClient`), the Go decoder (`Common/Client/go/udpsprotocol`), and the standalone C client (`Common/Client/c`, which redeclares the constants rather than including this header, so it stays MARTe-free). Changes to the protocol must be mirrored across all of these, plus the JS client parsers. **StreamHub WebSocket protocol**: JSON text frames for commands/events, binary frames for data pushes — spec in `ARCHITECTURE.md` §6. The Go hub (`Client/udpstreamer`) and C++ StreamHub implement the identical protocol; both clients (browser JS and ImGui) must stay compatible with both. diff --git a/Common/Client/c/.gitignore b/Common/Client/c/.gitignore new file mode 100644 index 0000000..9ac152a --- /dev/null +++ b/Common/Client/c/.gitignore @@ -0,0 +1,5 @@ +*.o +*.a +udps_dump +.cxxcheck +.cxxcheck.cpp diff --git a/Common/Client/c/Makefile b/Common/Client/c/Makefile new file mode 100644 index 0000000..f7ad9c2 --- /dev/null +++ b/Common/Client/c/Makefile @@ -0,0 +1,46 @@ +# UDPS C client library — standalone, no MARTe2, no external dependencies. +# +# make build libudpsclient.a and the example +# make example build only the example +# make cxxcheck verify the header is usable from C++ +# make clean + +CC ?= cc +CXX ?= c++ +AR ?= ar +CFLAGS ?= -O2 -g +WARN = -Wall -Wextra -Wpedantic +STD = -std=c99 +CPPFLAGS += -I. + +# Old glibc (< 2.17) keeps clock_gettime in librt; harmless to add there. +LDLIBS ?= + +LIB = libudpsclient.a +OBJ = udps_client.o +EXAMPLE = udps_dump + +.PHONY: all example cxxcheck clean + +all: $(LIB) $(EXAMPLE) + +$(LIB): $(OBJ) + $(AR) rcs $@ $^ + +udps_client.o: udps_client.c udps_client.h + $(CC) $(STD) $(WARN) $(CFLAGS) $(CPPFLAGS) -c -o $@ $< + +example: $(EXAMPLE) + +$(EXAMPLE): example/udps_dump.c $(LIB) + $(CC) $(STD) $(WARN) $(CFLAGS) $(CPPFLAGS) -o $@ $< $(LIB) $(LDLIBS) + +# The header is C++-safe; this target keeps it that way. +cxxcheck: udps_client.h + echo '#include "udps_client.h"' > .cxxcheck.cpp + echo 'int main() { udps_client_config_t c; udps_client_config_init(&c); return 0; }' >> .cxxcheck.cpp + $(CXX) -std=c++11 -Wall -Wextra $(CPPFLAGS) -o .cxxcheck .cxxcheck.cpp $(LIB) $(LDLIBS) + ./.cxxcheck && rm -f .cxxcheck .cxxcheck.cpp + +clean: + rm -f $(LIB) $(OBJ) $(EXAMPLE) .cxxcheck .cxxcheck.cpp diff --git a/Common/Client/c/example/udps_dump.c b/Common/Client/c/example/udps_dump.c new file mode 100644 index 0000000..1fe4016 --- /dev/null +++ b/Common/Client/c/example/udps_dump.c @@ -0,0 +1,260 @@ +/** + * @file udps_dump.c + * @brief Example UDPS client: connects to a UDPStreamer and prints what arrives. + * + * Build with the Makefile in the parent directory, then for a unicast stream: + * + * ./udps_dump --host 127.0.0.1 --port 44500 + * + * or, for a multicast one: + * + * ./udps_dump --host 127.0.0.1 --port 44500 \ + * --multicast 239.0.0.1 --iface 127.0.0.1 + * + * Ctrl-C prints a summary of what was received. + */ + +#define _POSIX_C_SOURCE 200809L + +#include "udps_client.h" + +#include +#include +#include +#include +#include + +static volatile sig_atomic_t g_stop = 0; + +static void on_sigint(int sig) { + (void)sig; + g_stop = 1; +} + +typedef struct { + double print_interval; /**< Seconds between frame printouts. */ + double last_print; + uint64_t frames; + uint64_t max_frames; +} dump_state_t; + +static double now_wall(void) { + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9; +} + +static const char *time_mode_name(uint8_t m) { + switch (m) { + case UDPS_TIME_PACKET: return "packet"; + case UDPS_TIME_FULL_ARRAY: return "full-array"; + case UDPS_TIME_FIRST_SAMPLE: return "first-sample"; + case UDPS_TIME_LAST_SAMPLE: return "last-sample"; + default: return "?"; + } +} + +static const char *publish_mode_name(uint8_t m) { + switch (m) { + case UDPS_PUBLISH_STRICT: return "strict"; + case UDPS_PUBLISH_ACCUMULATE: return "accumulate"; + case UDPS_PUBLISH_DECIMATE: return "decimate"; + default: return "?"; + } +} + +static void on_config(const udps_signal_t *sigs, uint32_t n, uint8_t mode, + void *user) { + uint32_t i; + (void)user; + printf("\nCONFIG: %u signal(s), publish mode %s\n", (unsigned)n, + publish_mode_name(mode)); + printf(" %-3s %-24s %-8s %-10s %-8s %-10s %s\n", "#", "name", "type", + "shape", "unit", "rate[Hz]", "time-mode"); + for (i = 0u; i < n; i++) { + char shape[32]; + const udps_signal_t *s = &sigs[i]; + if (s->num_cols > 1u) { + snprintf(shape, sizeof shape, "%ux%u", (unsigned)s->num_rows, + (unsigned)s->num_cols); + } else { + snprintf(shape, sizeof shape, "%u", + (unsigned)udps_signal_num_elements(s)); + } + printf(" %-3u %-24s %-8s %-10s %-8s %-10.6g %s%s\n", (unsigned)i, + s->name, udps_type_name(s->type_code), shape, + (s->unit[0] != '\0') ? s->unit : "-", s->sampling_rate, + time_mode_name(s->time_mode), + (s->quant_type != UDPS_QUANT_NONE) ? " (quantised)" : ""); + } + fflush(stdout); +} + +static void on_data(const udps_frame_t *f, void *user) { + dump_state_t *st = (dump_state_t *)user; + uint32_t i; + double now; + + st->frames++; + now = now_wall(); + if ((now - st->last_print) < st->print_interval) { + return; /* Streams run far faster than a terminal can be read. */ + } + st->last_print = now; + + printf("\nframe #%lu t=%.6f samples=%u (%lu frames so far)\n", + (unsigned long)f->counter, f->recv_time, (unsigned)f->num_samples, + (unsigned long)st->frames); + for (i = 0u; i < f->num_signals; i++) { + const double *v = f->values[i].values; + uint32_t cnt = f->values[i].count; + double lo, hi; + uint32_t k; + if (cnt == 0u) { + continue; + } + lo = hi = v[0]; + for (k = 1u; k < cnt; k++) { + if (v[k] < lo) { + lo = v[k]; + } + if (v[k] > hi) { + hi = v[k]; + } + } + printf(" %-24s n=%-6u first=%-12.6g last=%-12.6g min=%-12.6g max=%-12.6g %s\n", + f->signals[i].name, (unsigned)cnt, v[0], v[cnt - 1u], lo, hi, + f->signals[i].unit); + } + fflush(stdout); +} + +static void on_event(udps_event_t ev, const char *detail, void *user) { + (void)user; + switch (ev) { + case UDPS_EVENT_CONNECTED: + printf("[connected to %s]\n", detail ? detail : ""); + break; + case UDPS_EVENT_DISCONNECTED: + printf("[disconnected: %s]\n", detail ? detail : ""); + break; + case UDPS_EVENT_ERROR: + fprintf(stderr, "[error] %s\n", detail ? detail : ""); + break; + default: + break; + } + fflush(stdout); +} + +static void usage(const char *argv0) { + printf("Usage: %s --host ADDR --port N [options]\n" + "\n" + " --host ADDR server address (default 127.0.0.1)\n" + " --port N server UDP port, or TCP control port in multicast\n" + " mode (default 44500)\n" + " --multicast GROUP join GROUP for data instead of unicast\n" + " --iface ADDR local interface address for the multicast join\n" + " --data-port N multicast data port (default: --port + 1)\n" + " --silence SEC reconnect after SEC without data (default 1, 0 off)\n" + " --interval SEC seconds between printouts (default 1)\n" + " --frames N exit after N frames (default: run until Ctrl-C)\n" + " --help this text\n", + argv0); +} + +int main(int argc, char **argv) { + udps_client_config_t cfg; + udps_client_t *cli; + dump_state_t st; + udps_stats_t stats; + struct sigaction sa; + const char *host = "127.0.0.1"; + int i; + + udps_client_config_init(&cfg); + cfg.server_port = 44500u; + + memset(&st, 0, sizeof st); + st.print_interval = 1.0; + + for (i = 1; i < argc; i++) { + const char *a = argv[i]; + const char *next = (i + 1 < argc) ? argv[i + 1] : NULL; + if (strcmp(a, "--help") == 0) { + usage(argv[0]); + return 0; + } + if (next == NULL) { + fprintf(stderr, "missing value for %s\n", a); + return 2; + } + if (strcmp(a, "--host") == 0) { + host = next; + } else if (strcmp(a, "--port") == 0) { + cfg.server_port = (uint16_t)atoi(next); + } else if (strcmp(a, "--multicast") == 0) { + cfg.multicast_group = next; + } else if (strcmp(a, "--iface") == 0) { + cfg.interface_addr = next; + } else if (strcmp(a, "--data-port") == 0) { + cfg.data_port = (uint16_t)atoi(next); + } else if (strcmp(a, "--silence") == 0) { + cfg.silence_timeout_s = atof(next); + } else if (strcmp(a, "--interval") == 0) { + st.print_interval = atof(next); + } else if (strcmp(a, "--frames") == 0) { + st.max_frames = (uint64_t)strtoull(next, NULL, 10); + } else { + fprintf(stderr, "unknown option %s\n", a); + usage(argv[0]); + return 2; + } + i++; + } + cfg.server_addr = host; + + cli = udps_client_create(&cfg); + if (cli == NULL) { + fprintf(stderr, "could not create client for %s:%u\n", host, + (unsigned)cfg.server_port); + return 1; + } + udps_client_set_callbacks(cli, on_config, on_data, on_event, &st); + + memset(&sa, 0, sizeof sa); + sa.sa_handler = on_sigint; + (void)sigaction(SIGINT, &sa, NULL); + (void)sigaction(SIGTERM, &sa, NULL); + + printf("listening to %s:%u%s%s ... (Ctrl-C to stop)\n", host, + (unsigned)cfg.server_port, + cfg.multicast_group ? " via multicast " : "", + cfg.multicast_group ? cfg.multicast_group : ""); + + while (!g_stop && (st.max_frames == 0u || st.frames < st.max_frames)) { + /* All the work — connecting, receiving, decoding, reconnecting — and + * every callback happens inside this call. */ + (void)udps_client_poll(cli, 200); + } + + udps_client_stats(cli, &stats); + printf("\n--- summary ---\n" + "packets %lu\n" + "bytes %.1f MiB\n" + "frames %lu\n" + "configs %lu\n" + "gaps %lu (datagrams lost)\n" + "dropped %lu (fragments)\n" + "reconnects %lu\n", + (unsigned long)stats.packets_received, + (double)stats.bytes_received / (1024.0 * 1024.0), + (unsigned long)stats.frames_delivered, + (unsigned long)stats.config_updates, + (unsigned long)stats.counter_gaps, + (unsigned long)stats.fragments_dropped, + (unsigned long)stats.reconnects); + + udps_client_destroy(cli); + return 0; +} diff --git a/Common/Client/c/udps_client.c b/Common/Client/c/udps_client.c new file mode 100644 index 0000000..f56e1e1 --- /dev/null +++ b/Common/Client/c/udps_client.c @@ -0,0 +1,1209 @@ +/** + * @file udps_client.c + * @brief Implementation of the standalone UDPS receiver library. + * + * Layering, bottom up: little-endian readers, fragment reassembly, CONFIG/DATA + * decoding, socket transport, and finally udps_client_poll() which stitches + * them together into the connect/receive/reconnect state machine. + */ + +#define _POSIX_C_SOURCE 200809L +/* struct ip_mreq is not in strict POSIX; _BSD_SOURCE is the pre-2.19 glibc name. */ +#define _DEFAULT_SOURCE 1 +#define _BSD_SOURCE 1 + +#include "udps_client.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/*---------------------------------------------------------------------------*/ +/* Tunable limits */ +/*---------------------------------------------------------------------------*/ + +/** Concurrent in-flight reassemblies. The server interleaves at most a couple. */ +#define UDPS_MAX_SLOTS 4 + +/** Fragments per update; also bounds the receive bitmask below. */ +#define UDPS_MAX_FRAGMENTS 512 +#define UDPS_FRAG_MASK_BYTES (UDPS_MAX_FRAGMENTS / 8) + +/** Largest datagram we can receive, header included. */ +#define UDPS_RX_BUF_BYTES (65535 + 17) + +/** An incomplete reassembly older than this is abandoned. */ +#define UDPS_SLOT_STALE_S 2.0 + +/** Datagrams drained per poll before housekeeping runs again. */ +#define UDPS_DRAIN_LIMIT 256 + +/** Refuse CONFIGs claiming more signals than this. */ +#define UDPS_MAX_SIGNALS 4096 + +/*---------------------------------------------------------------------------*/ +/* Little-endian primitives */ +/*---------------------------------------------------------------------------*/ + +static uint16_t rd_u16(const uint8_t *p) { + return (uint16_t)((uint16_t)p[0] | ((uint16_t)p[1] << 8)); +} + +static uint32_t rd_u32(const uint8_t *p) { + return (uint32_t)p[0] | ((uint32_t)p[1] << 8) | ((uint32_t)p[2] << 16) | + ((uint32_t)p[3] << 24); +} + +static uint64_t rd_u64(const uint8_t *p) { + return (uint64_t)rd_u32(p) | ((uint64_t)rd_u32(p + 4) << 32); +} + +static float rd_f32(const uint8_t *p) { + uint32_t bits = rd_u32(p); + float f; + memcpy(&f, &bits, sizeof f); + return f; +} + +static double rd_f64(const uint8_t *p) { + uint64_t bits = rd_u64(p); + double d; + memcpy(&d, &bits, sizeof d); + return d; +} + +static void wr_u16(uint8_t *p, uint16_t v) { + p[0] = (uint8_t)(v & 0xFFu); + p[1] = (uint8_t)(v >> 8); +} + +static void wr_u32(uint8_t *p, uint32_t v) { + p[0] = (uint8_t)(v & 0xFFu); + p[1] = (uint8_t)((v >> 8) & 0xFFu); + p[2] = (uint8_t)((v >> 16) & 0xFFu); + p[3] = (uint8_t)((v >> 24) & 0xFFu); +} + +/** Serialises a header into @p buf, which must hold UDPS_HEADER_SIZE bytes. */ +static void build_header(uint8_t *buf, uint8_t type, uint32_t counter, + uint16_t frag_idx, uint16_t total_frags, + uint32_t payload_bytes) { + wr_u32(buf, UDPS_MAGIC); + buf[4] = type; + wr_u32(buf + 5, counter); + wr_u16(buf + 9, frag_idx); + wr_u16(buf + 11, total_frags); + wr_u32(buf + 13, payload_bytes); +} + +/*---------------------------------------------------------------------------*/ +/* Clocks */ +/*---------------------------------------------------------------------------*/ + +static double now_mono(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9; +} + +static double now_wall(void) { + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + return (double)ts.tv_sec + (double)ts.tv_nsec * 1e-9; +} + +static void sleep_s(double s) { + struct timespec ts; + if (s <= 0.0) { + return; + } + ts.tv_sec = (time_t)s; + ts.tv_nsec = (long)((s - (double)ts.tv_sec) * 1e9); + (void)nanosleep(&ts, NULL); +} + +/*---------------------------------------------------------------------------*/ +/* Internal state */ +/*---------------------------------------------------------------------------*/ + +typedef struct { + int active; + uint32_t counter; + uint8_t type; + uint16_t total_fragments; + uint16_t received_fragments; + uint8_t mask[UDPS_FRAG_MASK_BYTES]; + uint8_t *payload; + size_t payload_cap; + uint32_t chunk_size; /**< Learned from fragment 0; fixes placement. */ + uint32_t assembled_bytes;/**< Highest byte written; the true payload size. */ + double first_seen; +} udps_slot_t; + +struct udps_client { + udps_client_config_t cfg; + /* Owned copies: the caller's strings need not outlive the create call. */ + char server_addr[128]; + char mcast_group[128]; + char iface_addr[128]; + + struct sockaddr_in server_sa; + int udp_fd; /**< Unicast receive socket, or joined multicast socket. */ + int tcp_fd; /**< Multicast control channel; -1 in unicast mode. */ + int connected; + int ever_connected; + double last_data; /**< Monotonic; drives the silence timeout. */ + double last_keepalive; + double disconnect_t; /**< Monotonic; 0 before the first attempt. */ + + udps_slot_t slots[UDPS_MAX_SLOTS]; + + udps_signal_t *sigs; + uint32_t num_sigs; + size_t sigs_cap; + uint8_t publish_mode; + + double *valbuf; /**< Arena holding one frame's doubles. */ + size_t valbuf_cap; + udps_signal_values_t *vals; + size_t vals_cap; + + uint8_t *rxbuf; + + uint32_t last_counter; + int have_counter; + + udps_stats_t stats; + char err[256]; + + udps_config_cb on_config; + udps_data_cb on_data; + udps_event_cb on_event; + void *user; +}; + +static void emit_event(udps_client_t *c, udps_event_t ev, const char *detail) { + if (c->on_event != NULL) { + c->on_event(ev, detail, c->user); + } +} + +/** Records an error, reports it, and always returns -1 for tail calls. */ +static int fail(udps_client_t *c, const char *fmt, ...) { + va_list ap; + va_start(ap, fmt); + vsnprintf(c->err, sizeof c->err, fmt, ap); + va_end(ap); + emit_event(c, UDPS_EVENT_ERROR, c->err); + return -1; +} + +/** Grows @p *buf to at least @p need bytes, preserving nothing. */ +static int ensure_cap(void **buf, size_t *cap, size_t need) { + void *p; + if (*cap >= need) { + return 0; + } + p = realloc(*buf, need); + if (p == NULL) { + return -1; + } + *buf = p; + *cap = need; + return 0; +} + +/*---------------------------------------------------------------------------*/ +/* Stateless helpers (public) */ +/*---------------------------------------------------------------------------*/ + +uint32_t udps_signal_num_elements(const udps_signal_t *s) { + uint32_t r, c, n; + if (s == NULL) { + return 0u; + } + r = (s->num_rows == 0u) ? 1u : s->num_rows; + c = (s->num_cols == 0u) ? 1u : s->num_cols; + if (r > UDPS_MAX_ELEMENTS || c > UDPS_MAX_ELEMENTS) { + return UDPS_MAX_ELEMENTS; + } + n = r * c; + return (n > UDPS_MAX_ELEMENTS) ? UDPS_MAX_ELEMENTS : n; +} + +const char *udps_type_name(uint8_t tc) { + switch (tc) { + case UDPS_T_UINT8: return "uint8"; + case UDPS_T_INT8: return "int8"; + case UDPS_T_UINT16: return "uint16"; + case UDPS_T_INT16: return "int16"; + case UDPS_T_UINT32: return "uint32"; + case UDPS_T_INT32: return "int32"; + case UDPS_T_UINT64: return "uint64"; + case UDPS_T_INT64: return "int64"; + case UDPS_T_FLOAT32: return "float32"; + case UDPS_T_FLOAT64: return "float64"; + default: return "unknown"; + } +} + +int udps_parse_header(const void *buf, size_t len, udps_header_t *out) { + const uint8_t *b = (const uint8_t *)buf; + if (b == NULL || out == NULL || len < UDPS_HEADER_SIZE) { + return -1; + } + out->magic = rd_u32(b); + if (out->magic != UDPS_MAGIC) { + return -1; + } + out->type = b[4]; + out->counter = rd_u32(b + 5); + out->fragment_idx = rd_u16(b + 9); + out->total_fragments = rd_u16(b + 11); + out->payload_bytes = rd_u32(b + 13); + return 0; +} + +int udps_parse_config(const void *payload, size_t len, udps_signal_t *sigs, + uint32_t max_signals, uint32_t *num_signals, + uint8_t *publish_mode) { + const uint8_t *p = (const uint8_t *)payload; + uint32_t n, i; + size_t off; + + if (p == NULL || sigs == NULL || num_signals == NULL || len < 4u) { + return -1; + } + n = rd_u32(p); + /* A CONFIG cannot describe more signals than its payload can hold. */ + if (n > (len - 4u) / UDPS_SIGNAL_DESC_SIZE || n > max_signals) { + return -1; + } + off = 4u; + for (i = 0u; i < n; i++) { + const uint8_t *d = p + off; + udps_signal_t *s = &sigs[i]; + memcpy(s->name, d, 64); + s->name[64] = '\0'; + s->type_code = d[64]; + s->quant_type = d[65]; + s->num_dimensions = d[66]; + s->num_rows = rd_u32(d + 67); + s->num_cols = rd_u32(d + 71); + s->range_min = rd_f64(d + 75); + s->range_max = rd_f64(d + 83); + s->time_mode = d[91]; + s->sampling_rate = rd_f64(d + 92); + s->time_signal_idx = rd_u32(d + 100); + memcpy(s->unit, d + 104, 32); + s->unit[32] = '\0'; + off += UDPS_SIGNAL_DESC_SIZE; + } + *num_signals = n; + if (publish_mode != NULL) { + /* Trailing byte, absent in streams from older producers. */ + *publish_mode = (off < len) ? p[off] : (uint8_t)UDPS_PUBLISH_STRICT; + } + return 0; +} + +double udps_frame_value(const udps_frame_t *f, uint32_t sig, uint32_t sample, + uint32_t elem) { + const udps_signal_values_t *v; + uint32_t idx; + if (f == NULL || sig >= f->num_signals) { + return 0.0; + } + v = &f->values[sig]; + /* Accumulated scalars hold one value per batch slot; everything else holds + * one value per element and repeats across slots. */ + idx = (v->count == f->num_samples && udps_signal_num_elements(&f->signals[sig]) == 1u) + ? sample + : elem; + return (idx < v->count) ? v->values[idx] : 0.0; +} + +double udps_frame_element_time(const udps_frame_t *f, uint32_t sig, + uint32_t elem) { + const udps_signal_t *s; + uint32_t n; + double dt; + + if (f == NULL || sig >= f->num_signals) { + return 0.0; + } + s = &f->signals[sig]; + n = udps_signal_num_elements(s); + if (n <= 1u || s->sampling_rate <= 0.0 || + (s->time_mode != UDPS_TIME_FIRST_SAMPLE && + s->time_mode != UDPS_TIME_LAST_SAMPLE)) { + return f->recv_time; + } + if (elem >= n) { + elem = n - 1u; + } + dt = 1.0 / s->sampling_rate; + /* Anchor on arrival == the newest element, and walk backwards. */ + return f->recv_time - (double)(n - 1u - elem) * dt; +} + +/*---------------------------------------------------------------------------*/ +/* Payload decoding */ +/*---------------------------------------------------------------------------*/ + +static size_t raw_type_size(uint8_t tc) { + switch (tc) { + case UDPS_T_UINT8: case UDPS_T_INT8: return 1u; + case UDPS_T_UINT16: case UDPS_T_INT16: return 2u; + case UDPS_T_UINT32: case UDPS_T_INT32: case UDPS_T_FLOAT32: return 4u; + case UDPS_T_UINT64: case UDPS_T_INT64: case UDPS_T_FLOAT64: return 8u; + default: return 0u; + } +} + +static size_t quant_size(uint8_t qt) { + switch (qt) { + case UDPS_QUANT_UINT8: case UDPS_QUANT_INT8: return 1u; + case UDPS_QUANT_UINT16: case UDPS_QUANT_INT16: return 2u; + default: return 0u; + } +} + +static double read_raw(const uint8_t *p, uint8_t tc) { + switch (tc) { + case UDPS_T_UINT8: return (double)p[0]; + case UDPS_T_INT8: return (double)(int8_t)p[0]; + case UDPS_T_UINT16: return (double)rd_u16(p); + case UDPS_T_INT16: return (double)(int16_t)rd_u16(p); + case UDPS_T_UINT32: return (double)rd_u32(p); + case UDPS_T_INT32: return (double)(int32_t)rd_u32(p); + case UDPS_T_UINT64: return (double)rd_u64(p); + case UDPS_T_INT64: return (double)(int64_t)rd_u64(p); + case UDPS_T_FLOAT32: return (double)rd_f32(p); + case UDPS_T_FLOAT64: return rd_f64(p); + default: return 0.0; + } +} + +/** Expands a quantised integer back onto the signal's physical range. */ +static double dequantise(uint8_t qt, uint16_t raw, double lo, double hi) { + double span = hi - lo; + switch (qt) { + case UDPS_QUANT_UINT8: + return lo + ((double)(uint8_t)raw / 255.0) * span; + case UDPS_QUANT_INT8: + return lo + (((double)(int8_t)(uint8_t)raw + 127.0) / 254.0) * span; + case UDPS_QUANT_UINT16: + return lo + ((double)raw / 65535.0) * span; + case UDPS_QUANT_INT16: + return lo + (((double)(int16_t)raw + 32767.0) / 65534.0) * span; + default: + return 0.0; + } +} + +/** Reads @p n consecutive elements of @p s, advancing @p off. */ +static int parse_elems(const uint8_t *pl, size_t len, size_t *off, uint32_t n, + const udps_signal_t *s, double *out) { + int quantised = (s->quant_type != UDPS_QUANT_NONE); + size_t sz = quantised ? quant_size(s->quant_type) + : raw_type_size(s->type_code); + size_t base = *off; + uint32_t i; + + if (sz == 0u || base > len) { + return -1; + } + /* Division rather than multiplication: no overflow on a crafted count. */ + if ((size_t)n > (len - base) / sz) { + return -1; + } + for (i = 0u; i < n; i++) { + const uint8_t *p = pl + base + (size_t)i * sz; + if (quantised) { + uint16_t raw = (sz == 1u) ? (uint16_t)p[0] : rd_u16(p); + out[i] = dequantise(s->quant_type, raw, s->range_min, s->range_max); + } else { + out[i] = read_raw(p, s->type_code); + } + } + *off = base + (size_t)n * sz; + return 0; +} + +static int decode_config(udps_client_t *c, const uint8_t *pl, size_t len) { + uint32_t claimed; + + if (len < 4u) { + return fail(c, "CONFIG payload too short (%lu bytes)", (unsigned long)len); + } + claimed = rd_u32(pl); + if (claimed > (len - 4u) / UDPS_SIGNAL_DESC_SIZE || claimed > UDPS_MAX_SIGNALS) { + return fail(c, "CONFIG claims %lu signals, payload holds %lu", + (unsigned long)claimed, + (unsigned long)((len - 4u) / UDPS_SIGNAL_DESC_SIZE)); + } + if (ensure_cap((void **)&c->sigs, &c->sigs_cap, + (claimed ? claimed : 1u) * sizeof(udps_signal_t)) != 0) { + return fail(c, "out of memory for %lu signals", (unsigned long)claimed); + } + if (udps_parse_config(pl, len, c->sigs, claimed, &c->num_sigs, + &c->publish_mode) != 0) { + c->num_sigs = 0u; + return fail(c, "malformed CONFIG payload"); + } + if (ensure_cap((void **)&c->vals, &c->vals_cap, + (c->num_sigs ? c->num_sigs : 1u) * sizeof(udps_signal_values_t)) != 0) { + c->num_sigs = 0u; + return fail(c, "out of memory for signal value table"); + } + c->stats.config_updates++; + if (c->on_config != NULL) { + c->on_config(c->sigs, c->num_sigs, c->publish_mode, c->user); + } + return 0; +} + +static int decode_data(udps_client_t *c, const uint8_t *pl, size_t len, + uint32_t counter, double recv_time) { + uint32_t nsamples = 1u; + size_t off = 8u; + size_t total = 0u; + size_t written = 0u; + uint32_t i; + udps_frame_t frame; + + if (c->num_sigs == 0u) { + return 0; /* DATA before CONFIG: nothing to decode against. */ + } + if (len < 8u) { + return fail(c, "DATA payload too short (%lu bytes)", (unsigned long)len); + } + if (c->publish_mode == UDPS_PUBLISH_ACCUMULATE) { + if (len < 12u) { + return fail(c, "accumulate DATA payload missing sample count"); + } + nsamples = rd_u32(pl + 8); + off = 12u; + if (nsamples == 0u) { + return 0; + } + if (nsamples > UDPS_MAX_ELEMENTS) { + return fail(c, "accumulate sample count %lu out of range", + (unsigned long)nsamples); + } + } + + for (i = 0u; i < c->num_sigs; i++) { + uint32_t n = udps_signal_num_elements(&c->sigs[i]); + total += (n == 1u) ? nsamples : n; + } + if (ensure_cap((void **)&c->valbuf, &c->valbuf_cap, + (total ? total : 1u) * sizeof(double)) != 0) { + return fail(c, "out of memory for %lu decoded values", + (unsigned long)total); + } + + for (i = 0u; i < c->num_sigs; i++) { + uint32_t n = udps_signal_num_elements(&c->sigs[i]); + uint32_t count = (n == 1u) ? nsamples : n; + if (parse_elems(pl, len, &off, count, &c->sigs[i], + c->valbuf + written) != 0) { + return fail(c, "DATA payload truncated at signal '%s'", + c->sigs[i].name); + } + c->vals[i].values = c->valbuf + written; + c->vals[i].count = count; + written += count; + } + + if (c->have_counter && counter > c->last_counter + 1u) { + c->stats.counter_gaps += counter - c->last_counter - 1u; + } + c->last_counter = counter; + c->have_counter = 1; + c->stats.frames_delivered++; + + if (c->on_data != NULL) { + frame.counter = counter; + frame.hrt = rd_u64(pl); + frame.recv_time = recv_time; + frame.publish_mode = c->publish_mode; + frame.num_samples = nsamples; + frame.num_signals = c->num_sigs; + frame.signals = c->sigs; + frame.values = c->vals; + c->on_data(&frame, c->user); + } + return 0; +} + +/*---------------------------------------------------------------------------*/ +/* Fragment reassembly */ +/*---------------------------------------------------------------------------*/ + +static void slot_reset_all(udps_client_t *c) { + int i; + for (i = 0; i < UDPS_MAX_SLOTS; i++) { + c->slots[i].active = 0; + } +} + +static void gc_slots(udps_client_t *c, double now) { + int i; + for (i = 0; i < UDPS_MAX_SLOTS; i++) { + if (c->slots[i].active && (now - c->slots[i].first_seen) > UDPS_SLOT_STALE_S) { + c->slots[i].active = 0; + c->stats.fragments_dropped += c->slots[i].received_fragments; + } + } +} + +static void deliver_payload(udps_client_t *c, uint8_t type, const uint8_t *pl, + size_t len, uint32_t counter, double recv_time) { + if (type == UDPS_PKT_CONFIG) { + (void)decode_config(c, pl, len); + } else { + (void)decode_data(c, pl, len, counter, recv_time); + } +} + +/** + * @brief Files one fragment of a multi-fragment update. + * + * Fragments carry no offset, only an index, so placement relies on every + * fragment but the last being the same size — learned from fragment 0. + */ +static void place_fragment(udps_client_t *c, const udps_header_t *h, + const uint8_t *payload, size_t payload_bytes, + double recv_time) { + int slot = -1; + int i; + udps_slot_t *s; + size_t byte_idx; + uint8_t bit; + size_t offset; + + if (h->fragment_idx >= h->total_fragments || + h->total_fragments > UDPS_MAX_FRAGMENTS) { + c->stats.fragments_dropped++; + return; + } + + for (i = 0; i < UDPS_MAX_SLOTS; i++) { + if (c->slots[i].active && c->slots[i].counter == h->counter && + c->slots[i].type == h->type) { + slot = i; + break; + } + } + if (slot < 0) { + double oldest = 0.0; + for (i = 0; i < UDPS_MAX_SLOTS; i++) { + if (!c->slots[i].active) { + slot = i; + break; + } + if (slot < 0 || c->slots[i].first_seen < oldest) { + oldest = c->slots[i].first_seen; + slot = i; + } + } + s = &c->slots[slot]; + if (s->active) { + /* All slots busy: the oldest update is never going to complete. */ + c->stats.fragments_dropped += s->received_fragments; + } + if (ensure_cap((void **)&s->payload, &s->payload_cap, + c->cfg.max_packet_bytes) != 0) { + (void)fail(c, "out of memory for a %lu byte reassembly buffer", + (unsigned long)c->cfg.max_packet_bytes); + s->active = 0; + return; + } + s->active = 1; + s->counter = h->counter; + s->type = h->type; + s->total_fragments = h->total_fragments; + s->received_fragments = 0u; + s->chunk_size = 0u; + s->assembled_bytes = 0u; + s->first_seen = recv_time; + memset(s->mask, 0, sizeof s->mask); + } + s = &c->slots[slot]; + + byte_idx = (size_t)h->fragment_idx / 8u; + bit = (uint8_t)(1u << (h->fragment_idx % 8u)); + if ((s->mask[byte_idx] & bit) != 0u) { + c->stats.fragments_dropped++; /* duplicate */ + return; + } + + if (s->chunk_size == 0u) { + if (h->fragment_idx != 0u) { + /* Fragment 0 was lost, so no offset can be computed for this one. */ + c->stats.fragments_dropped++; + return; + } + s->chunk_size = (uint32_t)payload_bytes; + } + offset = (size_t)h->fragment_idx * s->chunk_size; + if (offset + payload_bytes > s->payload_cap) { + c->stats.fragments_dropped++; + return; + } + if (payload_bytes > 0u) { + memcpy(s->payload + offset, payload, payload_bytes); + } + if (offset + payload_bytes > s->assembled_bytes) { + s->assembled_bytes = (uint32_t)(offset + payload_bytes); + } + s->mask[byte_idx] |= bit; + s->received_fragments++; + + if (s->received_fragments >= s->total_fragments) { + s->active = 0; + deliver_payload(c, s->type, s->payload, s->assembled_bytes, s->counter, + recv_time); + } +} + +/** Validates one datagram (or TCP frame) and routes it. */ +static void handle_packet(udps_client_t *c, const uint8_t *buf, size_t len) { + udps_header_t h; + double recv_time = now_wall(); + + if (udps_parse_header(buf, len, &h) != 0) { + return; /* Not ours: stray traffic on a shared multicast port. */ + } + if (h.type != UDPS_PKT_DATA && h.type != UDPS_PKT_CONFIG) { + return; + } + if ((size_t)h.payload_bytes + UDPS_HEADER_SIZE > len) { + return; /* truncated */ + } + + c->stats.packets_received++; + c->stats.bytes_received += len; + + if (h.total_fragments <= 1u) { + deliver_payload(c, h.type, buf + UDPS_HEADER_SIZE, h.payload_bytes, + h.counter, recv_time); + } else { + place_fragment(c, &h, buf + UDPS_HEADER_SIZE, h.payload_bytes, recv_time); + } +} + +/*---------------------------------------------------------------------------*/ +/* Transport */ +/*---------------------------------------------------------------------------*/ + +/** Resolves a dotted quad, falling back to a name lookup. */ +static int resolve_ipv4(const char *host, struct in_addr *out) { + struct addrinfo hints; + struct addrinfo *res = NULL; + + if (inet_pton(AF_INET, host, out) == 1) { + return 0; + } + memset(&hints, 0, sizeof hints); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_DGRAM; + if (getaddrinfo(host, NULL, &hints, &res) != 0 || res == NULL) { + return -1; + } + *out = ((struct sockaddr_in *)(void *)res->ai_addr)->sin_addr; + freeaddrinfo(res); + return 0; +} + +static void set_nonblocking(int fd) { + int flags = fcntl(fd, F_GETFL, 0); + if (flags >= 0) { + (void)fcntl(fd, F_SETFL, flags | O_NONBLOCK); + } +} + +/** Raises SO_RCVBUF; a small kernel buffer silently drops burst traffic. */ +static void set_recv_buffer(int fd, uint32_t bytes) { + int v = (int)bytes; + if (v > 0) { + (void)setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &v, (socklen_t)sizeof v); + } +} + +static int send_control(udps_client_t *c, uint8_t type) { + uint8_t pkt[UDPS_HEADER_SIZE]; + ssize_t n; + + build_header(pkt, type, 0u, 0u, 1u, 0u); + if (c->tcp_fd >= 0) { + n = send(c->tcp_fd, pkt, sizeof pkt, 0); + } else if (c->udp_fd >= 0) { + n = sendto(c->udp_fd, pkt, sizeof pkt, 0, + (struct sockaddr *)&c->server_sa, (socklen_t)sizeof c->server_sa); + } else { + return -1; + } + return (n == (ssize_t)sizeof pkt) ? 0 : -1; +} + +static void close_sockets(udps_client_t *c) { + if (c->udp_fd >= 0) { + (void)close(c->udp_fd); + c->udp_fd = -1; + } + if (c->tcp_fd >= 0) { + (void)close(c->tcp_fd); + c->tcp_fd = -1; + } +} + +static int connect_unicast(udps_client_t *c) { + struct sockaddr_in local; + + c->udp_fd = socket(AF_INET, SOCK_DGRAM, 0); + if (c->udp_fd < 0) { + return fail(c, "socket() failed: %s", strerror(errno)); + } + set_recv_buffer(c->udp_fd, c->cfg.recv_buffer_bytes); + + memset(&local, 0, sizeof local); + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(INADDR_ANY); + local.sin_port = 0; /* ephemeral: the server replies to this port */ + if (bind(c->udp_fd, (struct sockaddr *)&local, (socklen_t)sizeof local) != 0) { + int e = errno; + close_sockets(c); + return fail(c, "bind() failed: %s", strerror(e)); + } + set_nonblocking(c->udp_fd); + + if (send_control(c, UDPS_PKT_CONNECT) != 0) { + int e = errno; + close_sockets(c); + return fail(c, "CONNECT to %s:%u failed: %s", c->server_addr, + (unsigned)c->cfg.server_port, strerror(e)); + } + return 0; +} + +static int connect_multicast(udps_client_t *c) { + struct sockaddr_in local; + struct ip_mreq mreq; + struct timeval tv; + int on = 1; + + /* Join before announcing: the server multicasts CONFIG the moment it sees + * CONNECT, and a group we have not joined yet drops it in the kernel. */ + c->udp_fd = socket(AF_INET, SOCK_DGRAM, 0); + if (c->udp_fd < 0) { + return fail(c, "socket() failed: %s", strerror(errno)); + } + (void)setsockopt(c->udp_fd, SOL_SOCKET, SO_REUSEADDR, &on, (socklen_t)sizeof on); + set_recv_buffer(c->udp_fd, c->cfg.recv_buffer_bytes); + + memset(&local, 0, sizeof local); + local.sin_family = AF_INET; + local.sin_addr.s_addr = htonl(INADDR_ANY); + local.sin_port = htons(c->cfg.data_port); + if (bind(c->udp_fd, (struct sockaddr *)&local, (socklen_t)sizeof local) != 0) { + int e = errno; + close_sockets(c); + return fail(c, "bind() on data port %u failed: %s", + (unsigned)c->cfg.data_port, strerror(e)); + } + + memset(&mreq, 0, sizeof mreq); + if (resolve_ipv4(c->mcast_group, &mreq.imr_multiaddr) != 0) { + close_sockets(c); + return fail(c, "bad multicast group '%s'", c->mcast_group); + } + if (c->iface_addr[0] != '\0') { + if (resolve_ipv4(c->iface_addr, &mreq.imr_interface) != 0) { + close_sockets(c); + return fail(c, "bad interface address '%s'", c->iface_addr); + } + } else { + mreq.imr_interface.s_addr = htonl(INADDR_ANY); + } + if (setsockopt(c->udp_fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, + (socklen_t)sizeof mreq) != 0) { + int e = errno; + close_sockets(c); + return fail(c, "joining %s failed: %s", c->mcast_group, strerror(e)); + } + set_nonblocking(c->udp_fd); + + /* Control channel: CONNECT out, CONFIG back. */ + c->tcp_fd = socket(AF_INET, SOCK_STREAM, 0); + if (c->tcp_fd < 0) { + int e = errno; + close_sockets(c); + return fail(c, "TCP socket() failed: %s", strerror(e)); + } + if (connect(c->tcp_fd, (struct sockaddr *)&c->server_sa, + (socklen_t)sizeof c->server_sa) != 0) { + int e = errno; + close_sockets(c); + return fail(c, "TCP connect to %s:%u failed: %s", c->server_addr, + (unsigned)c->cfg.server_port, strerror(e)); + } + /* Bound wait so a half-received control frame cannot wedge the poll loop. */ + tv.tv_sec = 1; + tv.tv_usec = 0; + (void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, (socklen_t)sizeof tv); + + if (send_control(c, UDPS_PKT_CONNECT) != 0) { + int e = errno; + close_sockets(c); + return fail(c, "CONNECT over TCP failed: %s", strerror(e)); + } + return 0; +} + +static int do_connect(udps_client_t *c) { + int rc = (c->mcast_group[0] != '\0') ? connect_multicast(c) + : connect_unicast(c); + if (rc != 0) { + c->disconnect_t = now_mono(); + return rc; + } + c->connected = 1; + c->last_data = now_mono(); + c->last_keepalive = c->last_data; + c->have_counter = 0; + slot_reset_all(c); + if (c->ever_connected) { + c->stats.reconnects++; + } + c->ever_connected = 1; + emit_event(c, UDPS_EVENT_CONNECTED, c->server_addr); + return 0; +} + +static void do_disconnect(udps_client_t *c, const char *why) { + if (!c->connected) { + return; + } + (void)send_control(c, UDPS_PKT_DISCONNECT); + close_sockets(c); + c->connected = 0; + c->disconnect_t = now_mono(); + slot_reset_all(c); + emit_event(c, UDPS_EVENT_DISCONNECTED, why); +} + +/** Reads exactly @p n bytes from the control connection. */ +static int read_exact_tcp(udps_client_t *c, uint8_t *dst, size_t n) { + size_t got = 0u; + while (got < n) { + ssize_t r = recv(c->tcp_fd, dst + got, n - got, 0); + if (r == 0) { + return -1; /* orderly close */ + } + if (r < 0) { + if (errno == EINTR) { + continue; + } + return -1; + } + got += (size_t)r; + } + return 0; +} + +/** Reads one framed UDPS packet off the TCP control connection. */ +static int read_tcp_frame(udps_client_t *c) { + udps_header_t h; + + if (read_exact_tcp(c, c->rxbuf, UDPS_HEADER_SIZE) != 0) { + return -1; + } + if (udps_parse_header(c->rxbuf, UDPS_HEADER_SIZE, &h) != 0) { + return -1; /* desynchronised: the stream cannot be resynced */ + } + if (h.payload_bytes > (uint32_t)(UDPS_RX_BUF_BYTES - UDPS_HEADER_SIZE)) { + return -1; + } + if (h.payload_bytes > 0u && + read_exact_tcp(c, c->rxbuf + UDPS_HEADER_SIZE, h.payload_bytes) != 0) { + return -1; + } + handle_packet(c, c->rxbuf, UDPS_HEADER_SIZE + h.payload_bytes); + return 0; +} + +/*---------------------------------------------------------------------------*/ +/* Public client API */ +/*---------------------------------------------------------------------------*/ + +void udps_client_config_init(udps_client_config_t *cfg) { + if (cfg == NULL) { + return; + } + memset(cfg, 0, sizeof *cfg); + cfg->silence_timeout_s = 1.0; + cfg->reconnect_delay_s = 2.0; + cfg->keepalive_interval_s = 15.0; /* server evicts silent clients at 30 s */ + cfg->recv_buffer_bytes = 4u * 1024u * 1024u; + cfg->max_packet_bytes = 1u * 1024u * 1024u; +} + +static void copy_str(char *dst, size_t cap, const char *src) { + if (src == NULL) { + dst[0] = '\0'; + return; + } + strncpy(dst, src, cap - 1u); + dst[cap - 1u] = '\0'; +} + +udps_client_t *udps_client_create(const udps_client_config_t *cfg) { + udps_client_t *c; + + if (cfg == NULL || cfg->server_addr == NULL || cfg->server_port == 0u) { + return NULL; + } + c = (udps_client_t *)calloc(1u, sizeof *c); + if (c == NULL) { + return NULL; + } + c->cfg = *cfg; + c->udp_fd = -1; + c->tcp_fd = -1; + copy_str(c->server_addr, sizeof c->server_addr, cfg->server_addr); + copy_str(c->mcast_group, sizeof c->mcast_group, cfg->multicast_group); + copy_str(c->iface_addr, sizeof c->iface_addr, cfg->interface_addr); + /* The config's string pointers must not be used again: they belong to the + * caller and the copies above are what the client works from. */ + c->cfg.server_addr = c->server_addr; + c->cfg.multicast_group = (c->mcast_group[0] != '\0') ? c->mcast_group : NULL; + c->cfg.interface_addr = (c->iface_addr[0] != '\0') ? c->iface_addr : NULL; + + if (c->cfg.max_packet_bytes == 0u) { + c->cfg.max_packet_bytes = 1u * 1024u * 1024u; + } + if (c->cfg.data_port == 0u) { + c->cfg.data_port = (uint16_t)(cfg->server_port + 1u); + } + + memset(&c->server_sa, 0, sizeof c->server_sa); + c->server_sa.sin_family = AF_INET; + c->server_sa.sin_port = htons(cfg->server_port); + if (resolve_ipv4(c->server_addr, &c->server_sa.sin_addr) != 0) { + free(c); + return NULL; + } + + c->rxbuf = (uint8_t *)malloc(UDPS_RX_BUF_BYTES); + if (c->rxbuf == NULL) { + free(c); + return NULL; + } + strcpy(c->err, "no error"); + return c; +} + +void udps_client_destroy(udps_client_t *c) { + int i; + if (c == NULL) { + return; + } + do_disconnect(c, "closed by application"); + close_sockets(c); + for (i = 0; i < UDPS_MAX_SLOTS; i++) { + free(c->slots[i].payload); + } + free(c->sigs); + free(c->valbuf); + free(c->vals); + free(c->rxbuf); + free(c); +} + +void udps_client_set_callbacks(udps_client_t *c, udps_config_cb on_config, + udps_data_cb on_data, udps_event_cb on_event, + void *user) { + if (c == NULL) { + return; + } + c->on_config = on_config; + c->on_data = on_data; + c->on_event = on_event; + c->user = user; +} + +int udps_client_is_connected(const udps_client_t *c) { + return (c != NULL) && c->connected; +} + +const udps_signal_t *udps_client_signals(const udps_client_t *c, + uint32_t *num_signals) { + if (c == NULL) { + if (num_signals != NULL) { + *num_signals = 0u; + } + return NULL; + } + if (num_signals != NULL) { + *num_signals = c->num_sigs; + } + return (c->num_sigs > 0u) ? c->sigs : NULL; +} + +uint8_t udps_client_publish_mode(const udps_client_t *c) { + return (c != NULL) ? c->publish_mode : (uint8_t)UDPS_PUBLISH_STRICT; +} + +void udps_client_stats(const udps_client_t *c, udps_stats_t *out) { + if (c != NULL && out != NULL) { + *out = c->stats; + } +} + +const char *udps_client_last_error(const udps_client_t *c) { + return (c != NULL) ? c->err : "invalid client"; +} + +/** Clamps @p budget so a pending deadline is not slept through. */ +static double clamp_deadline(double budget, double interval, double elapsed) { + double left; + if (interval <= 0.0) { + return budget; + } + left = interval - elapsed; + if (left < 0.0) { + left = 0.0; + } + return (left < budget) ? left : budget; +} + +int udps_client_poll(udps_client_t *c, int timeout_ms) { + double now; + double budget; + fd_set rset; + struct timeval tv; + int maxfd; + int nready; + int processed = 0; + + if (c == NULL) { + return -1; + } + + if (!c->connected) { + now = now_mono(); + if (c->disconnect_t > 0.0) { + double wait = c->cfg.reconnect_delay_s - (now - c->disconnect_t); + if (wait > 0.0) { + /* Idle out the retry delay rather than hammering the server. */ + double cap = (timeout_ms < 0) ? wait : (double)timeout_ms / 1000.0; + sleep_s((wait < cap) ? wait : cap); + return 0; + } + } + if (do_connect(c) != 0) { + return -1; + } + } + + now = now_mono(); + budget = (timeout_ms < 0) ? 1.0 : (double)timeout_ms / 1000.0; + /* Only unicast sends keepalives; clamping on them in multicast mode would + * spin on a deadline that never gets refreshed. */ + if (c->tcp_fd < 0) { + budget = clamp_deadline(budget, c->cfg.keepalive_interval_s, + now - c->last_keepalive); + } + budget = clamp_deadline(budget, c->cfg.silence_timeout_s, now - c->last_data); + + FD_ZERO(&rset); + maxfd = -1; + if (c->udp_fd >= 0 && c->udp_fd < FD_SETSIZE) { + FD_SET(c->udp_fd, &rset); + maxfd = c->udp_fd; + } + if (c->tcp_fd >= 0 && c->tcp_fd < FD_SETSIZE) { + FD_SET(c->tcp_fd, &rset); + if (c->tcp_fd > maxfd) { + maxfd = c->tcp_fd; + } + } + if (maxfd < 0) { + do_disconnect(c, "no usable socket"); + return fail(c, "socket descriptor outside FD_SETSIZE"); + } + + tv.tv_sec = (time_t)budget; + tv.tv_usec = (suseconds_t)((budget - (double)tv.tv_sec) * 1e6); + nready = select(maxfd + 1, &rset, NULL, NULL, &tv); + if (nready < 0) { + if (errno == EINTR) { + return 0; + } + do_disconnect(c, "select failed"); + return fail(c, "select() failed: %s", strerror(errno)); + } + + if (nready > 0 && c->tcp_fd >= 0 && FD_ISSET(c->tcp_fd, &rset)) { + if (read_tcp_frame(c) != 0) { + do_disconnect(c, "control connection lost"); + return fail(c, "TCP control connection lost"); + } + c->last_data = now_mono(); + processed++; + } + + if (nready > 0 && c->udp_fd >= 0 && FD_ISSET(c->udp_fd, &rset)) { + int drained = 0; + while (drained < UDPS_DRAIN_LIMIT) { + ssize_t n = recv(c->udp_fd, c->rxbuf, UDPS_RX_BUF_BYTES, 0); + if (n < 0) { + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + break; /* socket drained */ + } + do_disconnect(c, "receive failed"); + return fail(c, "recv() failed: %s", strerror(errno)); + } + drained++; + c->last_data = now_mono(); + if ((size_t)n >= UDPS_HEADER_SIZE) { + handle_packet(c, c->rxbuf, (size_t)n); + } + } + processed += drained; + } + + now = now_mono(); + if (c->cfg.keepalive_interval_s > 0.0 && c->tcp_fd < 0 && + (now - c->last_keepalive) >= c->cfg.keepalive_interval_s) { + /* ACK, not CONNECT: it refreshes the server's last-seen without making + * it resend CONFIG. */ + (void)send_control(c, UDPS_PKT_ACK); + c->last_keepalive = now; + } + gc_slots(c, now_wall()); + + if (c->cfg.silence_timeout_s > 0.0 && + (now - c->last_data) >= c->cfg.silence_timeout_s) { + do_disconnect(c, "server went silent"); + return fail(c, "no data for %.3f s", c->cfg.silence_timeout_s); + } + return processed; +} diff --git a/Common/Client/c/udps_client.h b/Common/Client/c/udps_client.h new file mode 100644 index 0000000..d23a09c --- /dev/null +++ b/Common/Client/c/udps_client.h @@ -0,0 +1,324 @@ +#ifndef UDPS_CLIENT_H +#define UDPS_CLIENT_H + +/** + * @file udps_client.h + * @brief Standalone UDPS (UDPStreamer) receiver library — C99, no MARTe2. + * + * Depends only on libc and BSD sockets, so it can be dropped into any C or C++ + * program that needs to consume a UDPStreamer / DebugService stream. The wire + * format is specified in Docs/Protocol.md; the library reference (and a worked + * example) is Docs/UDPS-C-Client.md. + * + * Usage in one paragraph: fill a udps_client_config_t, create a client, install + * callbacks, then call udps_client_poll() in a loop. The client owns the + * connection state machine — it sends CONNECT, reassembles fragmented packets, + * decodes CONFIG and DATA, sends keepalives, and reconnects when the server + * goes silent. Nothing is done behind your back: no threads are created and + * every callback runs inside your call to udps_client_poll(). + * + * Threading: a udps_client_t must be used from one thread at a time. + */ + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/*---------------------------------------------------------------------------*/ +/* Protocol constants */ +/*---------------------------------------------------------------------------*/ + +/** Magic number: ASCII 'UDPS' stored little-endian. */ +#define UDPS_MAGIC 0x53504455u + +/** Size of the packed packet header on the wire. */ +#define UDPS_HEADER_SIZE 17u + +/** Size of one serialised signal descriptor in a CONFIG payload. */ +#define UDPS_SIGNAL_DESC_SIZE 136u + +/** Value of udps_signal_t::time_signal_idx when the signal has no time reference. */ +#define UDPS_NO_TIME_SIGNAL 0xFFFFFFFFu + +/** Upper bound on elements per signal; larger descriptors are rejected. */ +#define UDPS_MAX_ELEMENTS (1u << 20) + +/** Packet types (udps_header_t::type). */ +enum { + UDPS_PKT_DATA = 0, /**< Server -> client: signal samples. */ + UDPS_PKT_CONFIG = 1, /**< Server -> client: signal metadata. */ + UDPS_PKT_ACK = 2, /**< Client -> server: keepalive. */ + UDPS_PKT_CONNECT = 3, /**< Client -> server: open a session. */ + UDPS_PKT_DISCONNECT = 4 /**< Either direction: close a session. */ +}; + +/** Sample type codes (udps_signal_t::type_code). */ +enum { + UDPS_T_UINT8 = 0, + UDPS_T_INT8 = 1, + UDPS_T_UINT16 = 2, + UDPS_T_INT16 = 3, + UDPS_T_UINT32 = 4, + UDPS_T_INT32 = 5, + UDPS_T_UINT64 = 6, + UDPS_T_INT64 = 7, + UDPS_T_FLOAT32 = 8, + UDPS_T_FLOAT64 = 9, + UDPS_T_UNKNOWN = 255 +}; + +/** Quantisation codes (udps_signal_t::quant_type). */ +enum { + UDPS_QUANT_NONE = 0, /**< Raw values in the signal's own type. */ + UDPS_QUANT_UINT8 = 1, /**< [range_min, range_max] mapped onto uint8. */ + UDPS_QUANT_INT8 = 2, + UDPS_QUANT_UINT16 = 3, + UDPS_QUANT_INT16 = 4 +}; + +/** Time-reference modes (udps_signal_t::time_mode). */ +enum { + UDPS_TIME_PACKET = 0, /**< No per-element time; use packet arrival. */ + UDPS_TIME_FULL_ARRAY = 1, /**< The time signal carries one stamp per element. */ + UDPS_TIME_FIRST_SAMPLE = 2, /**< Time signal (scalar) stamps element 0. */ + UDPS_TIME_LAST_SAMPLE = 3 /**< Time signal (scalar) stamps element N-1. */ +}; + +/** Publishing modes (udps_frame_t::publish_mode). */ +enum { + UDPS_PUBLISH_STRICT = 0, /**< One packet per RT cycle. */ + UDPS_PUBLISH_ACCUMULATE = 1, /**< A batch of cycles per packet. */ + UDPS_PUBLISH_DECIMATE = 2 /**< One packet every N cycles. */ +}; + +/*---------------------------------------------------------------------------*/ +/* Data model */ +/*---------------------------------------------------------------------------*/ + +/** Decoded 17-byte packet header. */ +typedef struct { + uint32_t magic; + uint8_t type; + uint32_t counter; /**< Same for every fragment of one update. */ + uint16_t fragment_idx; + uint16_t total_fragments; /**< 1 when the update fits in one datagram. */ + uint32_t payload_bytes; +} udps_header_t; + +/** Metadata for one streamed signal, as carried by the CONFIG payload. */ +typedef struct { + char name[65]; /**< NUL-terminated. */ + uint8_t type_code; /**< UDPS_T_*. */ + uint8_t quant_type; /**< UDPS_QUANT_*. */ + uint8_t num_dimensions; /**< 0 scalar, 1 vector, 2 matrix. */ + uint32_t num_rows; + uint32_t num_cols; + double range_min; /**< Physical range, used to dequantise. */ + double range_max; + uint8_t time_mode; /**< UDPS_TIME_*. */ + double sampling_rate; /**< Hz; 0 when unknown. */ + uint32_t time_signal_idx;/**< Index into the signal list, or UDPS_NO_TIME_SIGNAL. */ + char unit[33]; /**< NUL-terminated. */ +} udps_signal_t; + +/** + * @brief Decoded values of one signal within a frame. + * + * Values are always physical doubles: quantised signals are already expanded + * back onto [range_min, range_max]. @c count is @c num_samples for a scalar + * signal in Accumulate mode (one value per batched cycle) and the signal's + * element count in every other case. + */ +typedef struct { + const double *values; + uint32_t count; +} udps_signal_values_t; + +/** One fully decoded DATA packet. */ +typedef struct { + uint32_t counter; /**< Packet counter; gaps mean lost datagrams. */ + uint64_t hrt; /**< Producer's high-resolution timer at send. */ + double recv_time; /**< Wall-clock seconds (CLOCK_REALTIME) at arrival. */ + uint8_t publish_mode; /**< UDPS_PUBLISH_*. */ + uint32_t num_samples; /**< Batched cycles; 1 unless Accumulate. */ + uint32_t num_signals; + const udps_signal_t *signals; /**< num_signals entries, CONFIG order. */ + const udps_signal_values_t *values; /**< num_signals entries, same order. */ +} udps_frame_t; + +/** Connection lifecycle events reported through udps_event_cb. */ +typedef enum { + UDPS_EVENT_CONNECTED, /**< Sockets are up and CONNECT was sent. */ + UDPS_EVENT_DISCONNECTED, /**< Session dropped; the client will retry. */ + UDPS_EVENT_ERROR /**< Recoverable problem; detail says what. */ +} udps_event_t; + +/** Cumulative counters, never reset. */ +typedef struct { + uint64_t packets_received; /**< Datagrams (and TCP frames) accepted. */ + uint64_t bytes_received; + uint64_t frames_delivered; /**< DATA packets decoded and handed to you. */ + uint64_t config_updates; + uint64_t fragments_dropped; /**< Duplicate, stale or unplaceable fragments. */ + uint64_t counter_gaps; /**< DATA packets missing from the sequence. */ + uint64_t reconnects; +} udps_stats_t; + +/*---------------------------------------------------------------------------*/ +/* Client */ +/*---------------------------------------------------------------------------*/ + +typedef struct udps_client udps_client_t; + +/** Called whenever a CONFIG packet redefines the signal set. */ +typedef void (*udps_config_cb)(const udps_signal_t *signals, + uint32_t num_signals, + uint8_t publish_mode, + void *user); + +/** + * @brief Called for every decoded DATA packet. + * + * The frame and everything it points at are owned by the client and are only + * valid until the callback returns — copy anything you need to keep. + */ +typedef void (*udps_data_cb)(const udps_frame_t *frame, void *user); + +/** Called on connection state changes and on recoverable errors. */ +typedef void (*udps_event_cb)(udps_event_t event, const char *detail, void *user); + +/** + * @brief Transport configuration. + * + * Zero-initialise with udps_client_config_init(), then override what you need. + * Set @c multicast_group to switch from unicast to multicast: in unicast the + * client sends CONNECT over UDP and receives everything on its ephemeral port; + * in multicast it joins the group for DATA and opens a TCP control connection + * to @c server_port for CONNECT and CONFIG. + */ +typedef struct { + const char *server_addr; /**< IPv4 dotted quad. Required. */ + uint16_t server_port; /**< UDP port (unicast) or TCP port (multicast). Required. */ + const char *multicast_group;/**< IPv4 group; NULL selects unicast. */ + const char *interface_addr; /**< Local IPv4 of the interface to join on. NULL = default route. */ + uint16_t data_port; /**< Multicast data port; 0 means server_port + 1. */ + double silence_timeout_s; /**< Reconnect after this long without data. 0 disables. */ + double reconnect_delay_s; /**< Wait between reconnect attempts. */ + double keepalive_interval_s;/**< Unicast ACK period. 0 disables. */ + uint32_t recv_buffer_bytes; /**< SO_RCVBUF; large bursts need a large value. */ + uint32_t max_packet_bytes; /**< Ceiling on one reassembled payload. */ +} udps_client_config_t; + +/** Fills @p cfg with the defaults documented in Docs/UDPS-C-Client.md. */ +void udps_client_config_init(udps_client_config_t *cfg); + +/** + * @brief Creates a client. No socket is opened until the first poll. + * @return NULL if @p cfg is invalid or memory ran out. + */ +udps_client_t *udps_client_create(const udps_client_config_t *cfg); + +/** Closes the session (sending DISCONNECT if connected) and frees the client. */ +void udps_client_destroy(udps_client_t *client); + +/** Installs the callbacks. Any of them may be NULL. */ +void udps_client_set_callbacks(udps_client_t *client, + udps_config_cb on_config, + udps_data_cb on_data, + udps_event_cb on_event, + void *user); + +/** + * @brief Drives the client: connects if needed, then waits for and processes + * packets for at most @p timeout_ms milliseconds. + * + * Callbacks fire from inside this call. A negative @p timeout_ms blocks until + * something happens. Call it in a loop; it is the only function that does work. + * + * @return the number of packets processed (0 on timeout), or -1 if the session + * broke. -1 is not fatal: the next call retries after reconnect_delay_s. + */ +int udps_client_poll(udps_client_t *client, int timeout_ms); + +/** Non-zero once the sockets are up (which does not yet imply CONFIG arrived). */ +int udps_client_is_connected(const udps_client_t *client); + +/** + * @brief The current signal set, or NULL before the first CONFIG. + * @param num_signals Out; may be NULL. + */ +const udps_signal_t *udps_client_signals(const udps_client_t *client, + uint32_t *num_signals); + +/** The publishing mode from the last CONFIG (UDPS_PUBLISH_*). */ +uint8_t udps_client_publish_mode(const udps_client_t *client); + +/** Copies the counters into @p out. */ +void udps_client_stats(const udps_client_t *client, udps_stats_t *out); + +/** Human-readable description of the last failure. Never NULL. */ +const char *udps_client_last_error(const udps_client_t *client); + +/*---------------------------------------------------------------------------*/ +/* Stateless helpers */ +/*---------------------------------------------------------------------------*/ + +/** Elements in one sample of @p signal (rows x cols, at least 1). */ +uint32_t udps_signal_num_elements(const udps_signal_t *signal); + +/** Short name of a type code, e.g. "float32". Never NULL. */ +const char *udps_type_name(uint8_t type_code); + +/** + * @brief Decodes a packet header. + * @return 0 on success, -1 if @p len is too small or the magic is wrong. + */ +int udps_parse_header(const void *buf, size_t len, udps_header_t *out); + +/** + * @brief Decodes a reassembled CONFIG payload. + * @param signals Out array of at most @p max_signals entries. + * @param num_signals Out; the number actually written. + * @param publish_mode Out; may be NULL. + * @return 0 on success, -1 if the payload is malformed or does not fit. + */ +int udps_parse_config(const void *payload, + size_t len, + udps_signal_t *signals, + uint32_t max_signals, + uint32_t *num_signals, + uint8_t *publish_mode); + +/** + * @brief One value out of a frame. + * @param sample Accumulate batch slot; ignored for non-scalar signals. + * @param elem Element within the sample; ignored for accumulated scalars. + * @return the value, or 0.0 if any index is out of range. + */ +double udps_frame_value(const udps_frame_t *frame, + uint32_t signal_idx, + uint32_t sample, + uint32_t elem); + +/** + * @brief Arrival-anchored estimate of the wall-clock time of one element. + * + * Exact only for streams that declare a sampling rate: the packet is assumed to + * have arrived as its last element was produced, and earlier elements are dated + * backwards by 1/sampling_rate. Signals with UDPS_TIME_PACKET, or without a + * sampling rate, all report the arrival time. When the stream carries a time + * signal (time_signal_idx != UDPS_NO_TIME_SIGNAL) that signal is the accurate + * source — read it like any other signal instead of using this helper. + */ +double udps_frame_element_time(const udps_frame_t *frame, + uint32_t signal_idx, + uint32_t elem); + +#ifdef __cplusplus +} /* extern "C" */ +#endif + +#endif /* UDPS_CLIENT_H */ diff --git a/Docs/UDPS-C-Client.md b/Docs/UDPS-C-Client.md new file mode 100644 index 0000000..139d7f3 --- /dev/null +++ b/Docs/UDPS-C-Client.md @@ -0,0 +1,287 @@ +# UDPS C Client Library + +`Common/Client/c/` is a standalone receiver for the UDPS streaming protocol: it connects to a +`UDPStreamer` DataSource (or any other UDPS producer, such as `DebugService`), decodes the +signals, and hands them to your callbacks as plain `double`s. + +It has **no MARTe2 dependency** and no third-party dependencies at all — just libc and BSD +sockets. Two files, `udps_client.h` and `udps_client.c`, drop into any C or C++ project. + +The wire format itself is specified in [Protocol.md](Protocol.md); this document covers the +library. The producer side is documented in [UDPStreamer.md](UDPStreamer.md). + +--- + +## Build + +```bash +cd Common/Client/c +make # libudpsclient.a + the udps_dump example +make cxxcheck # verifies the header compiles and links from C++ +make clean +``` + +Or just add the two files to your own build: + +```bash +cc -std=c99 -O2 -c udps_client.c +``` + +Requirements: a C99 compiler and POSIX sockets. On glibc older than 2.17 add `-lrt` +(`clock_gettime` lived in librt back then). The header is wrapped in `extern "C"`, so C++ +callers include it directly. + +--- + +## Quick start + +```c +#include "udps_client.h" +#include + +static void on_data(const udps_frame_t *f, void *user) { + (void)user; + /* Signals are in CONFIG order; values are already physical doubles. */ + printf("#%u %s = %g\n", f->counter, f->signals[0].name, f->values[0].values[0]); +} + +int main(void) { + udps_client_config_t cfg; + udps_client_t *cli; + + udps_client_config_init(&cfg); + cfg.server_addr = "127.0.0.1"; + cfg.server_port = 44500; + + cli = udps_client_create(&cfg); + udps_client_set_callbacks(cli, NULL, on_data, NULL, NULL); + + for (;;) { + udps_client_poll(cli, 200); /* connects, receives, decodes, reconnects */ + } + udps_client_destroy(cli); + return 0; +} +``` + +`udps_client_poll()` is the only function that does work. It never spawns a thread, and every +callback runs inside it — so if your program already has an event loop, call it from there and +you need no synchronisation at all. A client must be used from one thread at a time. + +--- + +## Connection model + +The library implements both transports of the protocol and picks one from the configuration: + +| | Unicast (`multicast_group == NULL`) | Multicast (`multicast_group` set) | +|---|---|---| +| CONNECT | UDP datagram to `server_addr:server_port` | over a TCP connection to `server_addr:server_port` | +| CONFIG | UDP, back to the client's ephemeral port | over the same TCP connection | +| DATA | UDP, same ephemeral port | UDP multicast on `data_port` | +| Keepalive | ACK every `keepalive_interval_s` | not needed (the TCP session is the liveness signal) | + +In multicast mode the group is joined *before* CONNECT is sent, because the server multicasts +CONFIG as soon as it sees a client — a group joined afterwards would miss it. + +The client reconnects on its own: if nothing arrives for `silence_timeout_s` it sends +DISCONNECT, closes the sockets, waits `reconnect_delay_s`, and starts over. `udps_client_poll()` +returns `-1` when that happens, which is informational, not fatal. + +--- + +## Configuration + +Always start from `udps_client_config_init()` — it fills in the defaults below — then override +what you need. Strings are copied into the client, so they need not outlive `udps_client_create()`. + +| Field | Default | Meaning | +|---|---|---| +| `server_addr` | — (required) | Server IPv4 address; a hostname is resolved if it is not a dotted quad. | +| `server_port` | — (required) | Server UDP port, or the TCP control port in multicast mode. | +| `multicast_group` | `NULL` | IPv4 group to join. Non-`NULL` selects the multicast transport. | +| `interface_addr` | `NULL` | Local IPv4 **address** (not a name, e.g. `"127.0.0.1"`) of the interface to join on. Defaults to the default route, which silently receives nothing if the server sends elsewhere. | +| `data_port` | `server_port + 1` | Multicast data port. Must match the producer's `DataPort`. | +| `silence_timeout_s` | `1.0` | Reconnect after this long without data. `0` disables the check — use it for streams that are idle by design. | +| `reconnect_delay_s` | `2.0` | Wait between reconnection attempts. | +| `keepalive_interval_s` | `15.0` | Unicast ACK period. The server evicts silent clients after its `ClientTimeout` (30 s by default). `0` disables. | +| `recv_buffer_bytes` | 4 MiB | `SO_RCVBUF`. The Linux default (~208 KiB) is overrun by fast producers and the kernel drops datagrams silently. | +| `max_packet_bytes` | 1 MiB | Ceiling on one reassembled payload; a reassembly buffer of this size is allocated per in-flight update (4 at most). | + +--- + +## API + +### Lifecycle + +```c +void udps_client_config_init(udps_client_config_t *cfg); +udps_client_t *udps_client_create(const udps_client_config_t *cfg); +void udps_client_set_callbacks(udps_client_t *c, udps_config_cb, udps_data_cb, + udps_event_cb, void *user); +int udps_client_poll(udps_client_t *c, int timeout_ms); +void udps_client_destroy(udps_client_t *c); +``` + +`udps_client_create()` returns `NULL` on a bad address or an invalid configuration; no socket is +opened until the first poll. `udps_client_poll()` returns the number of packets processed, `0` on +timeout, or `-1` if the session broke — pass a negative `timeout_ms` to block. `destroy` sends +DISCONNECT before closing. + +### Callbacks + +```c +void on_config(const udps_signal_t *signals, uint32_t n, uint8_t publish_mode, void *user); +void on_data (const udps_frame_t *frame, void *user); +void on_event (udps_event_t event, const char *detail, void *user); +``` + +`on_config` fires on every CONFIG packet: the signal set can change at runtime, so treat it as a +reset of everything you cached. `on_event` reports `UDPS_EVENT_CONNECTED`, +`UDPS_EVENT_DISCONNECTED` and `UDPS_EVENT_ERROR` with a human-readable `detail`. + +> **The frame and everything it points at are owned by the client and are valid only until +> `on_data` returns.** The decode buffers are reused by the next packet. Copy what you keep. + +### Inspection + +```c +int udps_client_is_connected(const udps_client_t *c); +const udps_signal_t *udps_client_signals(const udps_client_t *c, uint32_t *n); +uint8_t udps_client_publish_mode(const udps_client_t *c); +void udps_client_stats(const udps_client_t *c, udps_stats_t *out); +const char *udps_client_last_error(const udps_client_t *c); +``` + +### Helpers + +```c +uint32_t udps_signal_num_elements(const udps_signal_t *s); +const char *udps_type_name(uint8_t type_code); +int udps_parse_header(const void *buf, size_t len, udps_header_t *out); +int udps_parse_config(const void *payload, size_t len, udps_signal_t *sigs, + uint32_t max_signals, uint32_t *n, uint8_t *publish_mode); +double udps_frame_value(const udps_frame_t *f, uint32_t sig, uint32_t sample, uint32_t elem); +double udps_frame_element_time(const udps_frame_t *f, uint32_t sig, uint32_t elem); +``` + +`udps_parse_header` and `udps_parse_config` are stateless and socket-free, so captured or +replayed traffic can be decoded without a client. + +--- + +## Reading a frame + +```c +typedef struct { + uint32_t counter; /* gaps in this sequence are lost datagrams */ + uint64_t hrt; /* producer's high-resolution timer at send */ + double recv_time; /* CLOCK_REALTIME seconds at arrival */ + uint8_t publish_mode; + uint32_t num_samples; /* batched RT cycles; 1 unless Accumulate */ + uint32_t num_signals; + const udps_signal_t *signals; /* CONFIG order */ + const udps_signal_values_t *values; /* same order */ +} udps_frame_t; +``` + +`values[i].values` is an array of `values[i].count` physical `double`s. Quantised signals are +already expanded back onto `[range_min, range_max]`, and integer types are widened — the decoded +form does not depend on the wire type, so a consumer need not branch on `type_code` at all. + +**Element count.** `count` is the signal's element count (`num_rows × num_cols`), *except* for a +scalar signal in Accumulate mode, where the producer batches several RT cycles into one packet +and `count == num_samples` — one value per cycle. Arrays are not batched: they appear once and +apply to the whole packet. `udps_frame_value(f, sig, sample, elem)` applies that rule for you. + +**Timestamps.** The protocol does not put a timestamp on every element; how to date them depends +on the signal's `time_mode` (see [Protocol.md](Protocol.md#time-mode-codes)): + +| `time_mode` | Where the time comes from | +|---|---| +| `UDPS_TIME_PACKET` | No per-element time. Use `recv_time`. | +| `UDPS_TIME_FULL_ARRAY` | The signal at `time_signal_idx` holds one timestamp per element — read it like any other signal. | +| `UDPS_TIME_FIRST_SAMPLE` / `UDPS_TIME_LAST_SAMPLE` | The signal at `time_signal_idx` is a scalar stamping element 0 (or N−1); the rest follow at `1/sampling_rate`. | + +The time signal is a raw producer-side counter (µs, or ns when it is a `uint64`), not wall clock, +so plotting it against real time needs a one-off calibration against `recv_time` — that is what +the Go hub does. `udps_frame_element_time()` skips all that and returns an arrival-anchored +estimate: good enough for a quick plot, but when a time signal exists, it is the accurate source. + +--- + +## Diagnosing loss + +```c +udps_stats_t s; +udps_client_stats(cli, &s); +``` + +| Counter | Meaning | +|---|---| +| `packets_received`, `bytes_received` | Accepted datagrams and TCP frames. | +| `frames_delivered` | DATA packets decoded and passed to `on_data`. | +| `config_updates` | CONFIG packets applied. | +| `counter_gaps` | Missing packet counters — datagrams lost on the wire or in the kernel. | +| `fragments_dropped` | Duplicate, stale or unplaceable fragments; a non-zero value with `counter_gaps` means fragmented updates are arriving incomplete. | +| `reconnects` | Sessions re-established after a silence timeout. | + +Persistent loss on a fast stream is almost always the receive buffer: raise `recv_buffer_bytes` +(and `net.core.rmem_max`, which caps it). A fragmented producer is more fragile than one sending +whole cycles, because losing any fragment discards the whole update — if you control the +producer, sizing `MaxPayloadSize` above one cycle removes that failure mode entirely. + +--- + +## Example program + +`example/udps_dump.c` connects, prints the signal table on CONFIG, then a throttled summary of +each frame, and a receive-statistics report on Ctrl-C. + +```bash +# unicast +./udps_dump --host 127.0.0.1 --port 44500 + +# multicast +./udps_dump --host 127.0.0.1 --port 44500 --multicast 239.0.0.1 --iface 127.0.0.1 + +# quieter, and stop after 500 frames +./udps_dump --host 127.0.0.1 --port 44500 --interval 5 --frames 500 +``` + +| Flag | Meaning | +|---|---| +| `--host ADDR` | Server address (default `127.0.0.1`). | +| `--port N` | Server UDP port, or TCP control port in multicast mode (default 44500). | +| `--multicast GROUP` | Join `GROUP` for data instead of using unicast. | +| `--iface ADDR` | Local interface address for the multicast join. | +| `--data-port N` | Multicast data port (default `--port + 1`). | +| `--silence SEC` | Reconnect after `SEC` without data; `0` disables. | +| `--interval SEC` | Seconds between printouts (default 1). | +| `--frames N` | Exit after `N` frames. | + +Against the repository's own producer (`./run_udp_producer.sh -n 2`, two 1 Msps channels of +1000-element `float32` arrays at 1 kHz) the output looks like: + +``` +CONFIG: 3 signal(s), publish mode strict + # name type shape unit rate[Hz] time-mode + 0 TimeArray uint64 1x1000 ns 0 packet + 1 Ch1 float32 1x1000 V 0 full-array + 2 Ch2 float32 1x1000 V 0 full-array + +frame #2289897 t=1787409851.723466 samples=1 + Ch1 n=1000 first=-6.9e-10 last=-0.00628 min=-1 max=1 V + Ch2 n=1000 first=0.5 last=0.49975 min=-0.5 max=0.5 V +``` + +--- + +## Limitations + +- IPv4 only, matching the protocol and the producer. +- One thread per client; there is no internal locking. +- The receive path allocates only when a CONFIG grows the signal set or a frame grows the decode + arena, so a steady stream is allocation-free — but this is not a hard real-time component. +- DATA arriving before the first CONFIG is dropped: without descriptors it cannot be decoded. + This is normal for a few packets after joining a multicast group. diff --git a/README.md b/README.md index 9039f7f..ca9d933 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,7 @@ Open `http://localhost:9090`, explore the object tree, trace signals, force valu | ----------------------------- | -------------------------------------------------------------- | | `Docs/Protocol.md` | UDPS binary wire protocol specification | | `Docs/UDPStreamer.md` | UDPStreamer DataSource configuration reference | +| `Docs/UDPS-C-Client.md` | Standalone C/C++ UDPS receiver library (`Common/Client/c`) | | `Docs/SineArrayGAM.md` | SineArrayGAM configuration reference | | `Docs/DebugService.md` | DebugService TCP API and architecture | | `Docs/Tutorial.md` | Step-by-step tutorial covering both components |