diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index cb4557e..22c25ca 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -117,9 +117,16 @@ Sent when the signal set changes or a client connects: ``` [uint32 numSigs] numSigs × UDPSSignalDescriptor (136 bytes each, packed) -[uint8 publishMode] 0=Strict/Decimate, 1=Accumulate +[uint8 publishMode] 0=Strict, 1=Accumulate, 2=Decimate +[uint64 hrtFrequency] producer's HRT ticks per second; 0 = unknown ``` +Everything after the descriptors is an optional trailer: a receiver accepts a +payload that stops early and ignores bytes it does not know. `hrtFrequency` is +what lets a receiver on another host turn the raw counter in DATA into seconds +— without it the only option is the receiver's own timer, which agrees with the +producer only when the two share a machine. + ### DATA Payload (Strict / Decimate modes) ``` @@ -130,9 +137,9 @@ per-signal data in CONFIG order (quantised or raw, no inter-signal padding) ### DATA Payload (Accumulate mode) ``` -[uint64 HRT timestamp] -[uint32 numSamples] -for each signal: if scalar → numSamples elements; else → NumElements once +[uint64 HRT timestamp of the first slot in the batch] +[uint32 numSamples] RT cycles accumulated into this packet +for each signal, in CONFIG order: numSamples × NumElements values ``` ### Quantization / Dequantization diff --git a/Common/UDP/UDPSProtocol.h b/Common/UDP/UDPSProtocol.h index 2d09c52..eb9f1a6 100644 --- a/Common/UDP/UDPSProtocol.h +++ b/Common/UDP/UDPSProtocol.h @@ -22,16 +22,23 @@ * CONFIG payload: * [uint32 numSigs] * numSigs × UDPSSignalDescriptor (136 bytes each, packed) - * [uint8 publishMode] (PublishModeStrict / Accumulate / Decimate) + * [uint8 publishMode] (PublishModeStrict / Accumulate / Decimate) + * [uint64 hrtFrequency] ticks per second of the producer's HRT + * + * Everything after the descriptors is an optional trailer: a receiver must + * accept a payload that stops early and must ignore bytes it does not know. + * publishMode defaults to Strict when absent, hrtFrequency to + * UDPS_HRT_FREQUENCY_UNKNOWN. * * DATA payload (Strict / Decimate): * [uint64 HRT timestamp] * per-signal data in CONFIG order (quantised or raw, no padding) * * DATA payload (Accumulate): - * [uint64 HRT timestamp] - * [uint32 numSamples] - * for each signal: if scalar → numSamples elements; else → NumElements once + * [uint64 HRT timestamp of the first slot in the batch] + * [uint32 numSamples] RT cycles accumulated into this packet + * for each signal, in CONFIG order: numSamples × NumElements values + * (signal-major, one full snapshot per accumulated cycle) */ #ifndef UDPS_PROTOCOL_H_ @@ -123,6 +130,20 @@ static const uint8 UDPS_PUBLISH_STRICT = 0u; ///< One packet per Synchronise static const uint8 UDPS_PUBLISH_ACCUMULATE = 1u; ///< Variable batch; flush on size or time static const uint8 UDPS_PUBLISH_DECIMATE = 2u; ///< One packet per Ratio calls +/*---------------------------------------------------------------------------*/ +/* HRT frequency (CONFIG trailing uint64) */ +/*---------------------------------------------------------------------------*/ + +/** + * Sentinel for a CONFIG that carries no HRT frequency, either because the + * trailer is absent (producer older than this field) or because the producer + * could not determine it. DATA timestamps are raw ticks of the producer's + * high-resolution timer, so without this a receiver on another host has no + * way to turn them into seconds and can only fall back to its own timer's + * frequency — which is right only while the two happen to agree. + */ +static const uint64 UDPS_HRT_FREQUENCY_UNKNOWN = 0u; + /*---------------------------------------------------------------------------*/ /* CONFIG payload — per-signal descriptor */ /*---------------------------------------------------------------------------*/ diff --git a/Docs/Protocol.md b/Docs/Protocol.md index 2c02be5..5e361a9 100644 --- a/Docs/Protocol.md +++ b/Docs/Protocol.md @@ -84,8 +84,28 @@ Offset Size Type Field 0xFFFFFFFF = PacketTime (no reference) 104 32 char[32] unit null-terminated physical unit string ── (total per signal: 136 bytes) ──────────────────────────── +── trailer, immediately after the last descriptor ─────────── +0 1 uint8 publishMode 0 = Strict, 1 = Accumulate, 2 = Decimate +1 8 uint64 hrtFrequency producer's HRT ticks per second; + 0 = unknown ``` +### CONFIG trailer + +Everything after the descriptors is a trailer that grew field by field, so a +receiver must accept a payload that stops early and must ignore bytes it does +not recognise. An absent `publishMode` means Strict; an absent or zero +`hrtFrequency` means the producer did not publish its tick rate. + +`hrtFrequency` is what makes DATA timestamps interpretable off-box. DATA +carries the raw value of the producer's high-resolution counter, and on x86 +that counter runs at the TSC frequency — a different number on every model. A +receiver that divides by its own timer's frequency instead is right only while +producer and consumer sit on the same host; anywhere else every batch is laid +out over the wrong span of time. Fall back to the local frequency only when the +field is missing, and reject implausible values (nothing below 1 kHz is a +high-resolution timer). + ### Type Codes | Code | C type | Bytes/element | @@ -129,7 +149,9 @@ After reassembly, the DATA payload layout is: ``` Offset Size Type Field ────── ──── ────── ──────────────────────────────────────────────────── -0 8 uint64 hrtTimestamp hardware reference timer count at Synchronise() +0 8 uint64 hrtTimestamp producer's high-resolution counter at + Synchronise(); divide by the CONFIG + hrtFrequency to get seconds ── for each signal (in config order) ──────────────────────────────────── varies N×sz — signal data N = numRows×numCols, sz = element size (wire size if quantized, raw size otherwise) diff --git a/Source/Applications/StreamHub/UDPSourceSession.cpp b/Source/Applications/StreamHub/UDPSourceSession.cpp index a0a5029..e5b8dc9 100644 --- a/Source/Applications/StreamHub/UDPSourceSession.cpp +++ b/Source/Applications/StreamHub/UDPSourceSession.cpp @@ -228,6 +228,9 @@ void UDPSourceSession::ParseConfigPayload(const uint8 *payload, uint32 size) { sigDescs_[i].unit[sizeof(sigDescs_[i].unit) - 1u] = '\0'; } publishMode_ = payload[4u + numSigs * UDPS_SIGNAL_DESC_SIZE]; + hrtFreq_ = UDPSConfigHrtFrequency( + payload, size, numSigs, + static_cast(MARTe::HighResolutionTimer::Frequency())); numSignals_ = numSigs; configured_ = true; @@ -276,8 +279,9 @@ void UDPSourceSession::ParseConfigPayload(const uint8 *payload, uint32 size) { } REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information, - "UDPSourceSession[%s]: CONFIG received — %u signals.", - id_.Buffer(), numSigs); + "UDPSourceSession[%s]: CONFIG received — %u signals, " + "producer HRT %.0f Hz.", + id_.Buffer(), numSigs, hrtFreq_); } void UDPSourceSession::AllocateRingBuffers() { @@ -572,9 +576,9 @@ void UDPSourceSession::ParseDataPayload(const uint8 *payload, uint32 size, * immune to this because it is sampled at acquisition. * * hrtTimestamp is the HRT counter of sample 0; hrtFreq_ (the - * local HRT frequency, identical to the sender on the same - * host) converts it to seconds, then a one-time calibration - * maps the sender clock onto wall-clock. */ + * producer's tick rate, taken from the CONFIG trailer) converts + * it to seconds, then a one-time calibration maps the sender + * clock onto wall-clock. */ const float64 hrt0Sec = static_cast(hrtTimestamp) / hrtFreq_; if ((!timeSigCalibValid_[s]) || diff --git a/Source/Applications/StreamHub/UDPSourceSession.h b/Source/Applications/StreamHub/UDPSourceSession.h index f0f1ec3..4f99bcf 100644 --- a/Source/Applications/StreamHub/UDPSourceSession.h +++ b/Source/Applications/StreamHub/UDPSourceSession.h @@ -102,6 +102,41 @@ inline float64 UDPSEstimateAccumDt(const float64 gap, const uint32 prevN, return dtEMA; } +/** + * @brief Pick the tick rate to divide a producer's DATA timestamps by. + * + * DATA packets carry the raw value of the producer's high-resolution counter, + * which is meaningless without the rate it runs at. The rate is published in + * the CONFIG trailer, after the descriptors and the publish-mode byte. When it + * is missing — an older producer, or one that could not determine it — the + * only remaining option is this host's own timer, which is right only while + * the two machines agree; on x86 that is the TSC frequency, so it is a + * different number on every model. + * + * @param payload Reassembled CONFIG payload. + * @param size Bytes in @p payload. + * @param numSigs Signal count already read from the payload, capped to what + * the receiver will store. + * @param localFreq This host's HRT frequency, used as the fallback. + * @return Ticks per second to convert DATA timestamps with; never 0. + */ +inline float64 UDPSConfigHrtFrequency(const uint8 *payload, const uint32 size, + const uint32 numSigs, + const float64 localFreq) { + const uint32 offset = 4u + (numSigs * MARTe::UDPS_SIGNAL_DESC_SIZE) + 1u; + if ((payload != NULL_PTR(const uint8 *)) && (size >= (offset + 8u))) { + uint64 wireFreq = 0u; + memcpy(&wireFreq, payload + offset, 8u); + /* Anything below 1 kHz is not a high-resolution timer; the field is + * either absent, unset, or the payload was mis-parsed, and adopting it + * would stretch every timestamp far enough to make the trace useless. */ + if (wireFreq >= 1000u) { + return static_cast(wireFreq); + } + } + return localFreq; +} + /** * @brief One connected UDPStreamer source. * @@ -463,10 +498,11 @@ private: float64 lastPktWallS_[UDPSS_MAX_SIGNALS]; bool lastPktWallValid_[UDPSS_MAX_SIGNALS]; - /* Accumulated-scalar timing: HRT counter frequency (local == sender on the - * same host) and the previous packet's sample count, used to reconstruct - * per-sample timestamps from the embedded sender HRT instead of the (UDP - * burst-sensitive) packet arrival time. */ + /* Accumulated-scalar timing: the producer's HRT counter frequency and the + * previous packet's sample count, used to reconstruct per-sample timestamps + * from the embedded sender HRT instead of the (UDP burst-sensitive) packet + * arrival time. Seeded from this host's timer and replaced by the rate the + * producer publishes in CONFIG; see UDPSConfigHrtFrequency. */ float64 hrtFreq_; uint32 accScalarPrevN_[UDPSS_MAX_SIGNALS]; diff --git a/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp b/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp index 42c0359..8d807c8 100644 --- a/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp +++ b/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp @@ -810,7 +810,9 @@ bool UDPStreamer::PrepareNextState(const char8 *const currentStateName, * receives it immediately. The config is static for the lifetime of this * state. */ if (ok) { - uint32 configBufSize = 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 32u + 1u; + /* numSigs + descriptors + publishMode + hrtFrequency (+ slack). */ + uint32 configBufSize = + 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 1u + 8u + 32u; HeapI *heap = GlobalObjectsDatabase::Instance()->GetStandardHeap(); uint8 *cfgBuf = reinterpret_cast(heap->Malloc(configBufSize)); if (cfgBuf != NULL_PTR(uint8 *)) { @@ -1244,6 +1246,16 @@ bool UDPStreamer::BuildConfigPayload(uint8 *buf, uint32 bufSize, buf[payloadSize] = static_cast(publishMode); payloadSize += 1u; + /* 8 bytes: this host's HRT tick rate. DATA packets carry raw counter + * values, so a receiver on another machine cannot turn them into seconds + * without it. */ + if ((payloadSize + 8u) > bufSize) { + return false; + } + uint64 hrtFrequency = HighResolutionTimer::Frequency(); + (void)MemoryOperationsHelper::Copy(buf + payloadSize, &hrtFrequency, 8u); + payloadSize += 8u; + return true; } diff --git a/Source/Components/Interfaces/DebugService/DebugService.cpp b/Source/Components/Interfaces/DebugService/DebugService.cpp index 1db8235..8d9ce7a 100644 --- a/Source/Components/Interfaces/DebugService/DebugService.cpp +++ b/Source/Components/Interfaces/DebugService/DebugService.cpp @@ -674,6 +674,14 @@ bool DebugService::SendUDPSConfig() { payloadOffset++; } + // Write this host's HRT tick rate: DATA packets carry raw counter values, + // which a receiver on another machine cannot convert to seconds without it. + if ((payloadOffset + 8u) <= CFG_BUF_SIZE) { + uint64 hrtFrequency = HighResolutionTimer::Frequency(); + memcpy(payload + payloadOffset, &hrtFrequency, 8u); + payloadOffset += 8u; + } + udpsNumSlots = newNumSlots; mutex.FastUnLock(); diff --git a/Test/Applications/StreamHub/ConfigHrtFreqGTest.cpp b/Test/Applications/StreamHub/ConfigHrtFreqGTest.cpp new file mode 100644 index 0000000..7741ee8 --- /dev/null +++ b/Test/Applications/StreamHub/ConfigHrtFreqGTest.cpp @@ -0,0 +1,151 @@ +/** + * @file ConfigHrtFreqGTest.cpp + * @brief Tests UDPSConfigHrtFrequency, which picks the tick rate used to turn + * a producer's DATA timestamps into seconds. + * + * DATA packets carry the raw value of the producer's high-resolution counter. + * Until the rate was published in CONFIG the hub divided by its own timer's + * frequency, which is only right while the producer runs on the same host — + * on x86 that number is the TSC frequency and differs from model to model, so + * off-box every accumulated batch was laid out over the wrong span of time. + * + * The field is a trailer, so these tests pin both directions: a payload that + * carries it must be believed, and one that stops early — an older producer — + * must still decode against the local fallback rather than against zero. + * + * @copyright Copyright 2015 F4E | European Joint Undertaking for ITER and + * the Development of Fusion Energy ('Fusion for Energy'). + * Licensed under the EUPL, Version 1.1 or - as soon they will be approved + * by the European Commission - subsequent versions of the EUPL (the "Licence") + * You may not use this work except in compliance with the Licence. + * You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl + * + * @warning Unless required by applicable law or agreed to in writing, + * software distributed under the Licence is distributed on an "AS IS" + * basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the Licence permissions and limitations under the Licence. + */ + +#include +#include + +#include "UDPSourceSession.h" + +using MARTe::uint8; +using MARTe::uint32; +using MARTe::uint64; +using MARTe::float64; +using MARTe::UDPS_SIGNAL_DESC_SIZE; +using StreamHub::UDPSConfigHrtFrequency; + +namespace { + +/** This hub's own timer rate, i.e. what the code must fall back to. */ +const float64 kLocalFreq = 1.0e9; + +/** A plausible producer rate that is deliberately not kLocalFreq. */ +const uint64 kWireFreq = 2400000000ULL; + +const uint32 kNumSigs = 2u; + +/** Offset of the CONFIG trailer that follows the publish-mode byte. */ +uint32 TrailerOffset(uint32 numSigs) { + return 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 1u; +} + +/** + * Builds a CONFIG payload for kNumSigs signals. + * @param withFreq Append the 8-byte HRT frequency trailer. + * @param freq Value to append when @p withFreq. + * @param[out] size Bytes written. + */ +const uint8 *BuildConfig(bool withFreq, uint64 freq, uint32 &size) { + static uint8 buf[4u + (kNumSigs * UDPS_SIGNAL_DESC_SIZE) + 1u + 8u]; + (void) memset(buf, 0, sizeof(buf)); + (void) memcpy(buf, &kNumSigs, 4u); + size = TrailerOffset(kNumSigs); + if (withFreq) { + (void) memcpy(buf + size, &freq, 8u); + size += 8u; + } + return buf; +} + +} // namespace + +/* The whole point of the field: a producer that publishes its rate is believed + * even when the hub's own timer runs at a different one. */ +TEST(ConfigHrtFreqGTest, AdoptsThePublishedRate) { + uint32 size = 0u; + const uint8 *cfg = BuildConfig(true, kWireFreq, size); + + EXPECT_DOUBLE_EQ(static_cast(kWireFreq), + UDPSConfigHrtFrequency(cfg, size, kNumSigs, kLocalFreq)); +} + +/* A producer older than the field stops after the publish-mode byte. Reading + * past it would take whatever follows in the receive buffer as a frequency. */ +TEST(ConfigHrtFreqGTest, FallsBackWhenTheTrailerIsAbsent) { + uint32 size = 0u; + const uint8 *cfg = BuildConfig(false, 0u, size); + + EXPECT_DOUBLE_EQ(kLocalFreq, + UDPSConfigHrtFrequency(cfg, size, kNumSigs, kLocalFreq)); +} + +/* A trailer cut short mid-field is not a frequency either; taking the bytes + * that are there would assemble one out of whatever the rest of the buffer + * holds. */ +TEST(ConfigHrtFreqGTest, FallsBackOnATruncatedTrailer) { + uint32 size = 0u; + const uint8 *cfg = BuildConfig(true, kWireFreq, size); + + EXPECT_DOUBLE_EQ(kLocalFreq, + UDPSConfigHrtFrequency(cfg, size - 1u, kNumSigs, + kLocalFreq)); +} + +/* Zero is the protocol's "I do not know my own rate". Dividing by it yields + * infinities that propagate into every timestamp. */ +TEST(ConfigHrtFreqGTest, FallsBackOnTheUnknownSentinel) { + uint32 size = 0u; + const uint8 *cfg = BuildConfig(true, MARTe::UDPS_HRT_FREQUENCY_UNKNOWN, + size); + + EXPECT_DOUBLE_EQ(kLocalFreq, + UDPSConfigHrtFrequency(cfg, size, kNumSigs, kLocalFreq)); +} + +/* No high-resolution timer ticks slower than 1 kHz, so a value that low means + * the payload was misread. Adopting it would stretch a millisecond batch + * across whole seconds. */ +TEST(ConfigHrtFreqGTest, FallsBackOnAnImplausiblyLowRate) { + uint32 size = 0u; + const uint8 *cfg = BuildConfig(true, 999u, size); + + EXPECT_DOUBLE_EQ(kLocalFreq, + UDPSConfigHrtFrequency(cfg, size, kNumSigs, kLocalFreq)); +} + +/* The trailer sits after the descriptors, so its offset moves with the signal + * count; a fixed offset would read descriptor bytes on any other config. */ +TEST(ConfigHrtFreqGTest, LocatesTheTrailerAfterTheDescriptors) { + const uint32 numSigs = 7u; + const uint32 size = TrailerOffset(numSigs) + 8u; + uint8 buf[4u + (7u * UDPS_SIGNAL_DESC_SIZE) + 1u + 8u]; + /* Fill the descriptor area with a byte pattern that would decode as a + * plausible frequency if the offset were wrong. */ + (void) memset(buf, 0x11, sizeof(buf)); + (void) memcpy(buf, &numSigs, 4u); + (void) memcpy(buf + TrailerOffset(numSigs), &kWireFreq, 8u); + + EXPECT_DOUBLE_EQ(static_cast(kWireFreq), + UDPSConfigHrtFrequency(buf, size, numSigs, kLocalFreq)); +} + +/* A null payload must not be dereferenced: CONFIG arrives from the network. */ +TEST(ConfigHrtFreqGTest, FallsBackOnANullPayload) { + EXPECT_DOUBLE_EQ(kLocalFreq, + UDPSConfigHrtFrequency(NULL_PTR(const uint8 *), 1024u, + kNumSigs, kLocalFreq)); +} diff --git a/Test/Applications/StreamHub/Makefile.inc b/Test/Applications/StreamHub/Makefile.inc index b840f16..fe6be0b 100644 --- a/Test/Applications/StreamHub/Makefile.inc +++ b/Test/Applications/StreamHub/Makefile.inc @@ -22,7 +22,7 @@ # ############################################################# -OBJSX = TriggerEngineSrc.x BinaryRecorderSrc.x SignalRingBufferGTest.x TriggerEngineGTest.x LTTBGTest.x BinaryRecorderGTest.x BoundsCheckTest.x WSServerBufferTest.x AccumDtGTest.x +OBJSX = TriggerEngineSrc.x BinaryRecorderSrc.x SignalRingBufferGTest.x TriggerEngineGTest.x LTTBGTest.x BinaryRecorderGTest.x BoundsCheckTest.x WSServerBufferTest.x AccumDtGTest.x ConfigHrtFreqGTest.x PACKAGE=Applications ROOT_DIR=../../.. diff --git a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp index 461e41c..379f5a4 100644 --- a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp +++ b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp @@ -36,6 +36,7 @@ #include "ConfigurationDatabase.h" #include "GAM.h" #include "GAMScheduler.h" +#include "HighResolutionTimer.h" #include "MemoryOperationsHelper.h" #include "ObjectRegistryDatabase.h" #include "RealTimeApplication.h" @@ -907,6 +908,23 @@ bool UDPStreamerTest::TestExecute_ConnectDataDisconnect() { reinterpret_cast(recvBuf); ok &= (hdr->magic == UDPS_MAGIC); ok &= (hdr->type == UDPS_TYPE_CONFIG); + + /* The CONFIG trailer must carry this host's HRT tick rate: DATA + * packets timestamp with the raw counter, so a receiver on another + * machine has nothing to convert it with otherwise. */ + const uint8 *payload = recvBuf + UDPS_HEADER_SIZE; + uint32 numSigs = 0u; + if (ok && (hdr->payloadBytes >= 4u)) { + (void) memcpy(&numSigs, payload, 4u); + } + const uint32 freqOff = + 4u + (numSigs * UDPS_SIGNAL_DESC_SIZE) + 1u; + ok &= (hdr->payloadBytes >= (freqOff + 8u)); + if (ok) { + uint64 wireFreq = 0u; + (void) memcpy(&wireFreq, payload + freqOff, 8u); + ok &= (wireFreq == HighResolutionTimer::Frequency()); + } } }