martino/MARTe-Integrated-Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implemented and fixed many issues

Browse Source
This commit is contained in:
Martino Ferrari
2026-08-21 23:24:48 +02:00
parent 14d5351a81
commit e03c60db25
52 changed files with 7726 additions and 769 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Applications/StreamHub/SignalRingBuffer.h
+96
View File
@@ -78,6 +78,32 @@ public:
/** @return Current number of stored points (≤ capacity). */
uint32 Count() const;
/** @return Allocated capacity in points. */
uint32 Capacity() const;
/**
* @brief Enlarge the buffer to @p newCap points, keeping the stored data
* and the TotalWritten() counter (unlike Allocate(), which resets both so
* every reader cursor and every retained sample is lost).
* @return true if the buffer now holds at least @p newCap points.
*/
bool Grow(uint32 newCap);
/**
* @brief Wall-clock span currently retained, i.e. newest minus oldest
* timestamp. 0 when fewer than two points are stored.
*/
float64 TimeSpan() const;
/**
* @brief Timestamp of the most recently stored point, 0 when empty.
*
* This is the source's own time base, which is *not* the hub's wall clock:
* use it, never clock_gettime(), whenever a decision depends on how far
* the data itself has advanced.
*/
float64 NewestTime() const;
/** @brief Discard all stored points. */
void Clear();
@@ -138,6 +164,69 @@ inline bool SignalRingBuffer::Allocate(uint32 maxPts) {
return true;
}
inline bool SignalRingBuffer::Grow(uint32 newCap) {
if (newCap <= capacity) { return true; }
/* Allocate outside the lock; readers may be active. */
float64 *newT = new float64[newCap];
float64 *newV = new float64[newCap];
if ((newT == static_cast<float64 *>(0)) ||
(newV == static_cast<float64 *>(0))) {
delete[] newT;
delete[] newV;
return false;
}
(void) mutex.FastLock();
if (newCap > capacity) {
/* Copy oldest-to-newest so the new buffer starts unwrapped. */
const uint32 avail = count;
for (uint32 i = 0u; i < avail; i++) {
const uint32 idx = (head + capacity - avail + i) % capacity;
newT[i] = tBuf[idx];
newV[i] = vBuf[idx];
}
float64 *oldT = tBuf;
float64 *oldV = vBuf;
tBuf = newT;
vBuf = newV;
capacity = newCap;
head = avail;
/* count and totalWritten are unchanged: no sample is gained or lost,
* so push cursors stay valid across the resize. */
mutex.FastUnLock();
delete[] oldT;
delete[] oldV;
return true;
}
mutex.FastUnLock();
delete[] newT;
delete[] newV;
return true;
}
inline float64 SignalRingBuffer::TimeSpan() const {
(void) mutex.FastLock();
float64 span = 0.0;
if ((capacity > 0u) && (count > 1u)) {
const uint32 oldest = (head + capacity - count) % capacity;
const uint32 newest = (head + capacity - 1u) % capacity;
span = tBuf[newest] - tBuf[oldest];
}
mutex.FastUnLock();
return (span > 0.0) ? span : 0.0;
}
inline float64 SignalRingBuffer::NewestTime() const {
(void) mutex.FastLock();
float64 t = 0.0;
if ((capacity > 0u) && (count > 0u)) {
t = tBuf[(head + capacity - 1u) % capacity];
}
mutex.FastUnLock();
return t;
}
inline void SignalRingBuffer::Write(float64 t, float64 v) {
(void) mutex.FastLock();
if (capacity > 0u) {
@@ -288,6 +377,13 @@ inline MARTe::uint64 SignalRingBuffer::TotalWritten() const {
return tw;
}
inline uint32 SignalRingBuffer::Capacity() const {
(void) mutex.FastLock();
const uint32 c = capacity;
mutex.FastUnLock();
return c;
}
inline uint32 SignalRingBuffer::Count() const {
(void) mutex.FastLock();
uint32 c = count;
Source/Applications/StreamHub/StreamHub.cpp
+283 -112
View File
@@ -65,6 +65,8 @@ StreamHub::StreamHub()
statsRateHz_(1u),
ringTemporal_(1000000u),
ringScalar_(100000u),
ringMaxPts_(8388608u),
trigRetentionSec_(0.0),
nextSourceId_(1u),
calibration_(static_cast<CalibrationEntry *>(0)),
numCalibration_(0u),
@@ -79,13 +81,19 @@ StreamHub::StreamHub()
pushV_(static_cast<float64 *>(0)),
lastTrigState_(kTrigIdle),
rearmPending_(false),
rearmAtWallS_(0.0) {
rearmAtWallS_(0.0),
collectStartWallS_(0.0),
capBuf_(static_cast<uint8 *>(0)),
capCap_(0u),
capOff_(0u),
capNSig_(0u) {
memset(&recorderCfg_, 0, sizeof(recorderCfg_));
calibration_ = new CalibrationEntry[kMaxCalibration];
memset(calibration_, 0, sizeof(CalibrationEntry) * kMaxCalibration);
for (uint32 i = 0u; i < kMaxSessions; i++) {
sessionActive_[i] = false;
configBroadcast_[i] = false;
capHarvested_[i] = false;
for (uint32 s = 0u; s < UDPSS_MAX_SIGNALS; s++) {
pushCursor_[i][s] = 0u;
}
@@ -104,6 +112,10 @@ StreamHub::~StreamHub() {
delete[] pushBuf_;
pushBuf_ = static_cast<uint8 *>(0);
}
if (capBuf_ != static_cast<uint8 *>(0)) {
delete[] capBuf_;
capBuf_ = static_cast<uint8 *>(0);
}
if (lttbT_ != static_cast<float64 *>(0)) {
delete[] lttbT_;
lttbT_ = static_cast<float64 *>(0);
@@ -138,11 +150,38 @@ bool StreamHub::Initialise(StructuredDataI &cfg) {
if (cfg.Read("StatsRate", tmp)) { statsRateHz_ = (tmp > 0u) ? tmp : 1u; }
if (cfg.Read("RingTemporal", tmp)) { ringTemporal_ = (tmp > 0u) ? tmp : 1000000u; }
if (cfg.Read("RingScalar", tmp)) { ringScalar_ = (tmp > 0u) ? tmp : 100000u; }
/* Per-signal ceiling when a trigger window forces a ring to grow.
* 128 MiB / (2 × float64) = 8388608 points — ~8 s at 1 Msps, ~1.7 s at
* 5 Msps. Raise it if you need longer windows on very fast sources. */
if (cfg.Read("RingMaxMB", tmp) && (tmp > 0u)) {
ringMaxPts_ = tmp * (1048576u / 16u);
}
if (ringMaxPts_ < ringTemporal_) { ringMaxPts_ = ringTemporal_; }
sourcesFile_ = "streamhub_sources.json";
StreamString sf;
if (cfg.Read("SourcesFile", sf)) { sourcesFile_ = sf; }
/* Origins allowed to open the WebSocket, comma-separated
* ("http://localhost:8080,http://box.lan:8080"). Without this only
* same-origin upgrades pass, which rejects every browser that loaded the
* SPA from a separate web server (the usual deployment). */
StreamString origins;
if (cfg.Read("AllowedOrigins", origins)) {
char list[1024];
strncpy(list, origins.Buffer(), sizeof(list) - 1u);
list[sizeof(list) - 1u] = '\0';
char *tok = strtok(list, ", \t");
while (tok != static_cast<char *>(0)) {
if (!wsServer_.AddAllowedOrigin(tok)) {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning,
"StreamHub: rejected allowed-origin '%s' (list full or too long).",
tok);
}
tok = strtok(static_cast<char *>(0), ", \t");
}
}
/* Parse +History block (optional).
* StandardParser stores the '+' prefix in the node name, so we try both. */
if (cfg.MoveRelative("+History") || cfg.MoveRelative("History")) {
@@ -194,8 +233,8 @@ bool StreamHub::Initialise(StructuredDataI &cfg) {
/* Allocate scratch buffers */
pushBuf_ = new uint8[kPushBufSize];
lttbT_ = new float64[maxPushPoints_];
lttbV_ = new float64[maxPushPoints_];
lttbT_ = new float64[kPushScratchPts];
lttbV_ = new float64[kPushScratchPts];
pushT_ = new float64[kPushScratchPts];
pushV_ = new float64[kPushScratchPts];
@@ -309,6 +348,7 @@ bool StreamHub::Run() {
if (statsDivisor == 0u) { statsDivisor = 1u; }
if ((tickCount_ % statsDivisor) == 0u) {
PushStats();
GrowRingsForTrigger();
}
/* History: flush headers at the configured interval, then re-broadcast
@@ -337,11 +377,16 @@ bool StreamHub::Run() {
tickCount_++;
/* Sleep for remainder of period */
/* Sleep for remainder of period. Both operands are unsigned, so the
* comparison must be done additively: a tick that overruns the period
* (easy at multi-Msps ingest, and guaranteed on the first tick, which
* drains the whole ring) would otherwise wrap periodUs - elapsedUs to
* ~2^64 and park the push thread for weeks — no data frames, no stats
* and no trigger captures for the rest of the run. */
uint64 t1 = MARTe::HighResolutionTimer::Counter();
uint64 freq = MARTe::HighResolutionTimer::Frequency();
uint64 elapsedUs = ((t1 - t0) * 1000000u) / freq;
if (periodUs - elapsedUs > 1000) {
if ((elapsedUs + 1000u) < periodUs) {
Sleep::MSec(static_cast<uint32>((periodUs - elapsedUs) / 1000u));
}
}
@@ -468,19 +513,35 @@ uint32 StreamHub::SerializeBinaryFrame(uint32 sessionIdx,
pushT_, pushV_, kPushScratchPts);
if (nRaw == 0u) { continue; }
/* LTTB decimation only for temporal (multi-element, sample-timed)
* signals — Go hub policy. Scalars and PACKET-timed arrays are
* pushed verbatim (their per-tick batches are small). */
/* LTTB decimation for the live push, bounded for every signal.
*
* A PACKET-timed array is a snapshot waveform, so its floor is one
* packet's worth of elements: below that LTTB would flatten the very
* waveform the operator is looking at, above it each extra point is
* just backlog from packets that piled up during the tick. Exempting
* those arrays altogether (the old rule, on the assumption that their
* per-tick batches are small) does not survive a fast producer: the
* 5 kHz x 1000-element time array in the demo pushes ~65k points per
* tick, 31 MB/s — 30x the channel it timestamps — and the per-client
* push queue never drains.
*
* Decimating here costs no fidelity downstream: LTTB picks real
* samples (it never interpolates), the rings keep every sample, and
* zoom/history/trigger all re-read the rings at full resolution. */
const uint32 nElems = desc.numRows * ((desc.numCols > 0u) ? desc.numCols : 1u);
const bool temporal = (nElems > 1u) &&
(desc.timeMode != MARTe::UDPS_TIMEMODE_PACKET);
uint32 threshold = maxPushPoints_;
if ((desc.timeMode == MARTe::UDPS_TIMEMODE_PACKET) &&
(nElems > threshold)) {
threshold = nElems;
}
if (threshold > kPushScratchPts) { threshold = kPushScratchPts; }
const float64 *tOut;
const float64 *vOut;
uint32 nOut;
if (temporal && (nRaw > maxPushPoints_)) {
if (nRaw > threshold) {
nOut = LTTBDecimate(pushT_, pushV_, nRaw,
lttbT_, lttbV_, maxPushPoints_);
lttbT_, lttbV_, threshold);
tOut = lttbT_;
vOut = lttbV_;
} else {
@@ -1556,7 +1617,8 @@ void StreamHub::HandleTrigStop(const char *json) {
void StreamHub::HandleSetTrigger(const char *json) {
/* Web client shape:
* {"type":"setTrigger","signal":"src:sig[i]","edge":"rising|falling|both",
* "threshold":F,"windowSec":F,"prePercent":F,"mode":"normal|single"} */
* "threshold":F,"windowSec":F,"prePercent":F,"mode":"normal|single",
* "holdoffSec":F} */
TriggerConfig cfg = trigger_.GetConfig();
char key[160] = "";
@@ -1565,6 +1627,7 @@ void StreamHub::HandleSetTrigger(const char *json) {
float64 thr = cfg.threshold;
float64 winSec = cfg.windowSec;
float64 prePct = cfg.prePercent;
float64 holdoff = cfg.holdoffSec;
if (JsonGetString(json, "signal", key, sizeof(key))) {
cfg.signalKey = key;
@@ -1580,8 +1643,20 @@ void StreamHub::HandleSetTrigger(const char *json) {
if (JsonGetFloat(json, "threshold", thr)) { cfg.threshold = thr; }
if (JsonGetFloat(json, "windowSec", winSec)) { cfg.windowSec = winSec; }
if (JsonGetFloat(json, "prePercent", prePct)) { cfg.prePercent = prePct; }
if (JsonGetFloat(json, "holdoffSec", holdoff)) { cfg.holdoffSec = holdoff; }
trigger_.SetConfig(cfg);
/* A capture can only contain what the rings still hold: the default
* capacity is a point count, so at 1 Msps it covers ~1 s and every longer
* window came back with only its tail populated. Publish the requested
* retention so the push thread can size the rings to the actual measured
* sample rate. */
trigRetentionSec_ = cfg.windowSec;
/* Grow now, not on the next stats tick: clients send setTrigger and arm
* back to back, and a trigger that fires before the rings are resized
* still loses its pre-trigger data. The periodic call stays as the catch-up
* path for sources that connect later. */
GrowRingsForTrigger();
BroadcastTriggerState();
}
@@ -1593,22 +1668,67 @@ void StreamHub::TriggerTick(float64 wallNowS) {
/* Capture-margin: wait a little past the post window so the rings have
* received the last post-trigger samples (web client used 120 ms). */
static const float64 kCaptureMarginS = 0.15;
static const float64 kAutoRearmDelayS = 0.2;
const TrigState st = trigger_.GetState();
/* Wall-clock grace on top of the post window before giving up on a source
* that stopped advancing; the capture is then broadcast with whatever the
* rings hold. */
static const float64 kCaptureWatchdogS = 2.0;
if (st == kTrigCollecting) {
const bool justEntered = (lastTrigState_ != kTrigCollecting);
if (justEntered) { collectStartWallS_ = wallNowS; }
float64 trigTime = 0.0;
float64 preSec = 0.0;
float64 postSec = 0.0;
if (trigger_.GetFiredWindow(trigTime, preSec, postSec) &&
(wallNowS >= (trigTime + postSec + kCaptureMarginS))) {
BroadcastTriggerCapture(trigTime, preSec, postSec);
trigger_.MarkTriggered();
TriggerConfig cfg = trigger_.GetConfig();
if ((cfg.mode == kTrigNormal) && !trigger_.GetStopped()) {
rearmPending_ = true;
rearmAtWallS_ = wallNowS + kAutoRearmDelayS;
if (trigger_.GetFiredWindow(trigTime, preSec, postSec)) {
/* Always restart the frame on entry: a capture abandoned by a
* disarm would otherwise be resumed with the previous trigTime. */
if (justEntered || (capBuf_ == static_cast<MARTe::uint8 *>(0))) {
BeginTriggerCapture(trigTime, preSec, postSec);
}
/* Harvest on the *data's* clock. trigTime comes from the sample
* timestamps, and a source's time base is offset from — and drifts
* against — CLOCK_REALTIME, so a wall-clock deadline chops the tail
* off every capture by exactly that offset. The wall clock is only
* a watchdog for a source that went quiet. */
const bool timedOut =
(wallNowS >= (collectStartWallS_ + postSec + kCaptureWatchdogS));
const float64 deadline = trigTime + postSec + kCaptureMarginS;
bool allDone = true;
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i] || capHarvested_[i]) { continue; }
const float64 frontier = SourceFrontierTime(i, wallNowS);
if (frontier >= deadline) {
HarvestTriggerCapture(i, trigTime - preSec,
trigTime + postSec);
}
else if (timedOut) {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning,
"StreamHub: source %s timed out at %.3f s of the %.3f s "
"trigger window; its traces will be short.",
sessions_[i].GetId().Buffer(), frontier - trigTime,
postSec);
HarvestTriggerCapture(i, trigTime - preSec,
trigTime + postSec);
}
else {
allDone = false;
}
}
if (allDone || timedOut) {
FinishTriggerCapture();
trigger_.MarkTriggered();
TriggerConfig cfg = trigger_.GetConfig();
if ((cfg.mode == kTrigNormal) && !trigger_.GetStopped()) {
rearmPending_ = true;
rearmAtWallS_ = wallNowS + cfg.holdoffSec;
}
}
}
}
@@ -1627,6 +1747,45 @@ void StreamHub::TriggerTick(float64 wallNowS) {
}
}
float64 StreamHub::SourceFrontierTime(uint32 i, float64 wallNowS) const {
const float64 t = sessions_[i].ProducerNewestTime();
/* No producer clock means every sample was stamped on arrival, so this
* source and the trigger both live in the hub's wall-clock domain. */
return (t > 0.0) ? t : wallNowS;
}
uint32 StreamHub::CurrentMaxRingCapacity() const {
/* Rings grow at runtime for long trigger windows, so read the live
* capacities rather than the configured starting size. */
uint32 maxCap = (ringTemporal_ > ringScalar_) ? ringTemporal_ : ringScalar_;
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
const uint32 c = sessions_[i].GetMaxRingCapacity();
if (c > maxCap) { maxCap = c; }
}
return maxCap;
}
void StreamHub::GrowRingsForTrigger() {
const float64 want = trigRetentionSec_;
if (want <= 0.0) { return; }
/* Retain the whole window plus the capture margin and one push period, so
* the tail of the window is still in the ring when TriggerTick reads it. */
const float64 target = want + 0.5;
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
if (!sessions_[i].IsConfigured()) { continue; }
if (sessions_[i].GrowRingsForSeconds(target, ringMaxPts_)) {
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: grew rings of source %s to hold %.2f s "
"(trigger window %.2f s, cap %u pts/signal).",
sessions_[i].GetId().Buffer(), target, want, ringMaxPts_);
}
}
}
void StreamHub::BroadcastTriggerState() {
const TrigState st = trigger_.GetState();
TriggerConfig cfg = trigger_.GetConfig();
@@ -1639,17 +1798,23 @@ void StreamHub::BroadcastTriggerState() {
(st == kTrigTriggered) ? "triggered" : "idle";
const char *modeStr = (cfg.mode == kTrigSingle) ? "single" : "normal";
char buf[256];
char buf[512];
int n;
float64 trigTime = 0.0;
float64 preSec = 0.0;
float64 postSec = 0.0;
if (((st == kTrigCollecting) || (st == kTrigTriggered)) &&
trigger_.GetFiredWindow(trigTime, preSec, postSec)) {
/* The window latched at fire time. Clients draw the filling capture on
* this axis before the v2 frame arrives, and config edits between arm
* and fire would otherwise leave them inferring the wrong window from
* their own copy of the config. */
n = snprintf(buf, sizeof(buf),
"{\"type\":\"triggerState\",\"state\":\"%s\",\"mode\":\"%s\","
"\"stopped\":%s,\"trigTime\":%.17g}",
stateStr, modeStr, (stopped ? "true" : "false"), trigTime);
"\"stopped\":%s,\"trigTime\":%.17g,\"preSec\":%.17g,"
"\"postSec\":%.17g}",
stateStr, modeStr, (stopped ? "true" : "false"), trigTime,
preSec, postSec);
} else {
n = snprintf(buf, sizeof(buf),
"{\"type\":\"triggerState\",\"state\":\"%s\",\"mode\":\"%s\","
@@ -1661,109 +1826,116 @@ void StreamHub::BroadcastTriggerState() {
}
}
void StreamHub::BroadcastTriggerCapture(float64 trigTime, float64 preSec,
float64 postSec) {
const float64 t0 = trigTime - preSec;
const float64 t1 = trigTime + postSec;
void StreamHub::BeginTriggerCapture(float64 trigTime, float64 preSec,
float64 postSec) {
delete[] capBuf_;
capCap_ = 1u << 20;
capBuf_ = new uint8[capCap_];
capOff_ = 0u;
capNSig_ = 0u;
for (uint32 i = 0u; i < kMaxSessions; i++) { capHarvested_[i] = false; }
/* Header: [u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig] */
capBuf_[capOff_++] = 2u;
memcpy(capBuf_ + capOff_, &trigTime, 8u); capOff_ += 8u;
memcpy(capBuf_ + capOff_, &preSec, 8u); capOff_ += 8u;
memcpy(capBuf_ + capOff_, &postSec, 8u); capOff_ += 8u;
capOff_ += 4u; /* nSig, patched in FinishTriggerCapture */
}
void StreamHub::HarvestTriggerCapture(uint32 i, float64 t0, float64 t1) {
capHarvested_[i] = true;
UDPSourceSession &sess = sessions_[i];
if ((capBuf_ == static_cast<uint8 *>(0)) || !sess.IsConfigured()) { return; }
/* Read scratch sized for the largest ring; LTTB scratch for the cap. */
const uint32 scratchCap = (ringTemporal_ > ringScalar_) ? ringTemporal_
: ringScalar_;
const uint32 scratchCap = CurrentMaxRingCapacity();
float64 *tRaw = new float64[scratchCap];
float64 *vRaw = new float64[scratchCap];
float64 *tDec = new float64[kTrigCapturePts];
float64 *vDec = new float64[kTrigCapturePts];
uint32 cap = 1u << 20;
uint8 *buf = new uint8[cap];
uint32 off = 0u;
StreamString sid = sess.GetId();
const uint32 numSigs = sess.GetNumSignals();
/* Header: [u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig] */
buf[off++] = 2u;
memcpy(buf + off, &trigTime, 8u); off += 8u;
memcpy(buf + off, &preSec, 8u); off += 8u;
memcpy(buf + off, &postSec, 8u); off += 8u;
const uint32 nSigOff = off;
uint32 nSigWritten = 0u;
off += 4u;
for (uint32 s = 0u; s < numSigs; s++) {
MARTe::UDPSSignalDescriptor desc;
if (!sess.GetSignalDescriptor(s, desc)) { continue; }
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
UDPSourceSession &sess = sessions_[i];
if (!sess.IsConfigured()) { continue; }
const uint32 nRaw = sess.ReadSignalRange(s, t0, t1,
tRaw, vRaw, scratchCap);
if (nRaw == 0u) { continue; }
StreamString sid = sess.GetId();
const uint32 numSigs = sess.GetNumSignals();
for (uint32 s = 0u; s < numSigs; s++) {
MARTe::UDPSSignalDescriptor desc;
if (!sess.GetSignalDescriptor(s, desc)) { continue; }
const uint32 nRaw = sess.ReadSignalRange(s, t0, t1,
tRaw, vRaw, scratchCap);
if (nRaw == 0u) { continue; }
const float64 *tOut = tRaw;
const float64 *vOut = vRaw;
uint32 nOut = nRaw;
if (nRaw > kTrigCapturePts) {
nOut = LTTBDecimate(tRaw, vRaw, nRaw, tDec, vDec,
kTrigCapturePts);
tOut = tDec;
vOut = vDec;
}
char fullKey[192];
const int kn = snprintf(fullKey, sizeof(fullKey), "%s:%s",
sid.Buffer(), desc.name);
if (kn <= 0) { continue; }
const uint32 keyLen = static_cast<uint32>(kn);
const uint32 need = 2u + keyLen + 4u + nOut * 16u;
if ((off + need) > cap) {
uint32 newCap = cap * 2u;
while ((off + need) > newCap) { newCap *= 2u; }
uint8 *nb = new uint8[newCap];
memcpy(nb, buf, off);
delete[] buf;
buf = nb;
cap = newCap;
}
buf[off++] = static_cast<uint8>( keyLen & 0xFFu);
buf[off++] = static_cast<uint8>((keyLen >> 8) & 0xFFu);
memcpy(buf + off, fullKey, keyLen);
off += keyLen;
buf[off++] = static_cast<uint8>( nOut & 0xFFu);
buf[off++] = static_cast<uint8>((nOut >> 8) & 0xFFu);
buf[off++] = static_cast<uint8>((nOut >> 16) & 0xFFu);
buf[off++] = static_cast<uint8>((nOut >> 24) & 0xFFu);
memcpy(buf + off, tOut, nOut * sizeof(float64));
off += nOut * 8u;
memcpy(buf + off, vOut, nOut * sizeof(float64));
off += nOut * 8u;
nSigWritten++;
const float64 *tOut = tRaw;
const float64 *vOut = vRaw;
uint32 nOut = nRaw;
if (nRaw > kTrigCapturePts) {
nOut = LTTBDecimate(tRaw, vRaw, nRaw, tDec, vDec, kTrigCapturePts);
tOut = tDec;
vOut = vDec;
}
char fullKey[192];
const int kn = snprintf(fullKey, sizeof(fullKey), "%s:%s",
sid.Buffer(), desc.name);
if (kn <= 0) { continue; }
const uint32 keyLen = static_cast<uint32>(kn);
const uint32 need = 2u + keyLen + 4u + nOut * 16u;
if ((capOff_ + need) > capCap_) {
uint32 newCap = capCap_ * 2u;
while ((capOff_ + need) > newCap) { newCap *= 2u; }
uint8 *nb = new uint8[newCap];
memcpy(nb, capBuf_, capOff_);
delete[] capBuf_;
capBuf_ = nb;
capCap_ = newCap;
}
capBuf_[capOff_++] = static_cast<uint8>( keyLen & 0xFFu);
capBuf_[capOff_++] = static_cast<uint8>((keyLen >> 8) & 0xFFu);
memcpy(capBuf_ + capOff_, fullKey, keyLen);
capOff_ += keyLen;
capBuf_[capOff_++] = static_cast<uint8>( nOut & 0xFFu);
capBuf_[capOff_++] = static_cast<uint8>((nOut >> 8) & 0xFFu);
capBuf_[capOff_++] = static_cast<uint8>((nOut >> 16) & 0xFFu);
capBuf_[capOff_++] = static_cast<uint8>((nOut >> 24) & 0xFFu);
memcpy(capBuf_ + capOff_, tOut, nOut * sizeof(float64));
capOff_ += nOut * 8u;
memcpy(capBuf_ + capOff_, vOut, nOut * sizeof(float64));
capOff_ += nOut * 8u;
capNSig_++;
}
/* Patch nSig */
buf[nSigOff] = static_cast<uint8>( nSigWritten & 0xFFu);
buf[nSigOff + 1u] = static_cast<uint8>((nSigWritten >> 8) & 0xFFu);
buf[nSigOff + 2u] = static_cast<uint8>((nSigWritten >> 16) & 0xFFu);
buf[nSigOff + 3u] = static_cast<uint8>((nSigWritten >> 24) & 0xFFu);
delete[] tRaw; delete[] vRaw;
delete[] tDec; delete[] vDec;
}
wsServer_.BroadcastBinary(buf, off);
void StreamHub::FinishTriggerCapture() {
if (capBuf_ == static_cast<uint8 *>(0)) { return; }
/* Patch nSig (immediately after the [u8 2] + 3×f64 header). */
const uint32 nSigOff = 25u;
capBuf_[nSigOff] = static_cast<uint8>( capNSig_ & 0xFFu);
capBuf_[nSigOff + 1u] = static_cast<uint8>((capNSig_ >> 8) & 0xFFu);
capBuf_[nSigOff + 2u] = static_cast<uint8>((capNSig_ >> 16) & 0xFFu);
capBuf_[nSigOff + 3u] = static_cast<uint8>((capNSig_ >> 24) & 0xFFu);
wsServer_.BroadcastBinary(capBuf_, capOff_);
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: trigger capture broadcast (%u signal(s), %u bytes).",
nSigWritten, off);
capNSig_, capOff_);
delete[] buf;
delete[] tRaw; delete[] vRaw;
delete[] tDec; delete[] vDec;
delete[] capBuf_;
capBuf_ = static_cast<uint8 *>(0);
capCap_ = 0u;
capOff_ = 0u;
capNSig_ = 0u;
}
void StreamHub::HandleZoom(const char *json, uint32 slotIdx) {
@@ -1797,8 +1969,7 @@ void StreamHub::HandleZoom(const char *json, uint32 slotIdx) {
/* Read scratch sized for the largest possible ring (no double decimation:
* the whole [t0,t1] slice is read, then LTTB'd once to maxOut). */
const uint32 scratchCap = (ringTemporal_ > ringScalar_) ? ringTemporal_
: ringScalar_;
const uint32 scratchCap = CurrentMaxRingCapacity();
float64 *tRaw = new float64[scratchCap];
float64 *vRaw = new float64[scratchCap];
float64 *tDec = (maxOut > 0u) ? new float64[maxOut] : static_cast<float64 *>(0);
Source/Applications/StreamHub/StreamHub.h
+48 -8
View File
@@ -91,7 +91,7 @@ public:
* WSPort (uint32, default 8090)
* MaxPoints (uint32, default 20000) — ring buffer capacity per signal
* PushRate (uint32, default 30) — push loop rate in Hz
* MaxPushPoints (uint32, default 500) — LTTB threshold for live push
* MaxPushPoints (uint32, default 50) — LTTB threshold for live push
* StatsRate (uint32, default 1) — stats broadcast rate in Hz
* +Sources { +<id> { Label=...; Addr=...; Port=... } }
*
@@ -152,12 +152,40 @@ private:
void BroadcastTriggerState();
/**
* @brief Build and broadcast the version=2 binary capture frame:
* [u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig]
* {[u16 keyLen][fullKey][u32 N][t f64×N][v f64×N]}
* @brief Size every ring so it retains the current trigger window.
* Called from the push loop once per stats tick; a no-op once the rings
* are large enough. Rates are measured from the rings themselves because
* most sources advertise samplingRate = 0.
*/
void BroadcastTriggerCapture(float64 trigTime, float64 preSec,
float64 postSec);
void GrowRingsForTrigger();
/** @return Largest ring capacity currently allocated across all sessions. */
uint32 CurrentMaxRingCapacity() const;
/**
* @brief How far source @p i has produced, in the trigger's time base;
* @p wallNowS when it publishes no producer clock (its samples are then
* stamped on arrival, so they share the hub's wall clock).
*/
float64 SourceFrontierTime(uint32 i, float64 wallNowS) const;
/* ---- Trigger capture assembly ---------------------------------------
* Sources are harvested one at a time, each as soon as *it* has produced
* past the end of the window, rather than all together once the slowest
* has. Sources free-run on their own clocks and can lag each other by
* seconds; making every source wait for the slowest lets the leaders' ring
* buffers roll past the pre-trigger region before it is ever read. */
/** @brief Start a version=2 capture frame:
* [u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig]. */
void BeginTriggerCapture(float64 trigTime, float64 preSec, float64 postSec);
/** @brief Append session @p i's signals to the pending frame, each as
* {[u16 keyLen][fullKey][u32 N][t f64×N][v f64×N]}. */
void HarvestTriggerCapture(uint32 i, float64 t0, float64 t1);
/** @brief Patch nSig, broadcast the pending frame and release it. */
void FinishTriggerCapture();
/* ---- Command handlers (called from OnWSCommand) ---------------------- */
@@ -271,8 +299,10 @@ private:
uint32 pushRateHz_;
uint32 maxPushPoints_;
uint32 statsRateHz_;
uint32 ringTemporal_; ///< Ring capacity for multi-element (waveform) signals
uint32 ringTemporal_; ///< Initial ring capacity for multi-element (waveform) signals
uint32 ringScalar_; ///< Ring capacity for scalar signals
uint32 ringMaxPts_; ///< Ceiling a ring may be grown to for a trigger window
volatile float64 trigRetentionSec_; ///< Retention the current trigger window needs
StreamString sourcesFile_; ///< Persistent dynamic source list (JSON)
uint32 nextSourceId_; ///< Counter for generated session ids ("sN")
@@ -292,7 +322,9 @@ private:
static const uint32 kPushBufSize = 8u * 1024u * 1024u;
uint8 *pushBuf_;
/* Decimated output scratch (LTTB): maxPushPoints × 2 arrays per signal */
/* Decimated output scratch (LTTB). Sized like the read scratch rather
* than maxPushPoints_: a PACKET-timed array raises its own threshold to
* one packet's worth of elements, which can exceed maxPushPoints_. */
float64 *lttbT_;
float64 *lttbV_;
@@ -311,6 +343,14 @@ private:
TrigState lastTrigState_; ///< Last broadcast FSM state
bool rearmPending_; ///< Normal-mode auto-rearm scheduled
float64 rearmAtWallS_; ///< Wall time of the scheduled auto-rearm
float64 collectStartWallS_; ///< Wall time COLLECTING began (watchdog only)
/* Capture frame under assembly across ticks (push thread only) */
MARTe::uint8 *capBuf_; ///< Pending frame, NULL when idle
uint32 capCap_; ///< Allocated size of capBuf_
uint32 capOff_; ///< Bytes written so far
uint32 capNSig_; ///< Signals appended so far
bool capHarvested_[kMaxSessions]; ///< Session already appended
};
} /* namespace StreamHub */
Source/Applications/StreamHub/TriggerEngine.cpp
+8 -1
View File
@@ -27,9 +27,16 @@ void TriggerEngine::SetConfig(const TriggerConfig &cfg) {
config_ = cfg;
/* Clamp to web UI bounds */
if (config_.windowSec < 1.0e-4) { config_.windowSec = 1.0e-4; }
if (config_.windowSec > 10.0) { config_.windowSec = 10.0; }
/* 60 s where the Go hub allows 600. Deliberate: these rings are
* fixed-capacity and store every sample, so a window they cannot hold is
* harvested truncated and silently decimated to kTrigCapturePts. The Go
* hub stores min/max pairs instead once a window outgrows its budget, so
* there a long window costs resolution rather than coverage. */
if (config_.windowSec > 60.0) { config_.windowSec = 60.0; }
if (config_.prePercent < 0.0) { config_.prePercent = 0.0; }
if (config_.prePercent > 100.0) { config_.prePercent = 100.0; }
if (config_.holdoffSec < 0.0) { config_.holdoffSec = 0.0; }
if (config_.holdoffSec > 60.0) { config_.holdoffSec = 60.0; }
epoch_++;
prevValid_ = false;
prevValue_ = 0.0;
Source/Applications/StreamHub/TriggerEngine.h
+4 -2
View File
@@ -62,9 +62,10 @@ struct TriggerConfig {
StreamString signalKey; ///< Full key: "src:sig" or "src:sig[i]"
TrigEdge edge; ///< Rising / falling / both
float64 threshold; ///< Trigger threshold (physical units)
float64 windowSec; ///< Capture window length [1e-4 .. 10] s
float64 windowSec; ///< Capture window length [1e-4 .. 60] s
float64 prePercent; ///< Pre-trigger part of the window [0 .. 100] %
TrigAcqMode mode; ///< Normal (auto-rearm) or single
float64 holdoffSec; ///< Rearm delay after a capture [0 .. 60] s
};
/**
@@ -149,7 +150,8 @@ inline TriggerConfig::TriggerConfig()
threshold(0.0),
windowSec(1.0),
prePercent(20.0),
mode(kTrigNormal) {
mode(kTrigNormal),
holdoffSec(0.2) {
}
} /* namespace StreamHub */
Source/Applications/StreamHub/UDPSourceSession.cpp
+77
View File
@@ -295,6 +295,83 @@ void UDPSourceSession::AllocateRingBuffers() {
}
}
bool UDPSourceSession::GrowRingsForSeconds(float64 seconds, uint32 maxPts) {
if ((seconds <= 0.0) || (maxPts == 0u)) { return false; }
(void) metaMutex_.FastLock();
const uint32 nSigs = numSignals_;
metaMutex_.FastUnLock();
bool grew = false;
for (uint32 i = 0u; i < nSigs; i++) {
const uint32 count = rings_[i].Count();
const float64 span = rings_[i].TimeSpan();
/* Need a decent sample of the stream before extrapolating a rate;
* a couple of packets' worth of span is enough at any rate. */
if ((count < 2u) || (span <= 0.0)) { continue; }
const float64 rate = static_cast<float64>(count) / span;
/* 20 % headroom absorbs rate jitter and the capture margin. */
float64 need = rate * seconds * 1.2;
if (need > static_cast<float64>(maxPts)) {
need = static_cast<float64>(maxPts);
}
const uint32 needPts = static_cast<uint32>(need);
if (needPts > rings_[i].Capacity()) {
if (rings_[i].Grow(needPts)) { grew = true; }
}
}
return grew;
}
uint32 UDPSourceSession::GetMaxRingCapacity() const {
(void) metaMutex_.FastLock();
const uint32 nSigs = numSignals_;
metaMutex_.FastUnLock();
uint32 maxCap = 0u;
for (uint32 i = 0u; i < nSigs; i++) {
const uint32 c = rings_[i].Capacity();
if (c > maxCap) { maxCap = c; }
}
return maxCap;
}
float64 UDPSourceSession::ProducerNewestTime() const {
/* Mirror exactly the ParseDataPayload branches that timestamp from the
* referenced time signal; every other branch stamps on arrival and so
* would report "now" in the hub's clock, not the producer's. The time
* signal itself is one of those — it is PACKET-timed. */
(void) metaMutex_.FastLock();
const uint32 nSigs = numSignals_;
bool producerTimed[UDPSS_MAX_SIGNALS];
for (uint32 i = 0u; i < nSigs; i++) {
const UDPSSignalDescriptor &d = sigDescs_[i];
uint64 ne = static_cast<uint64>(d.numRows) *
static_cast<uint64>(d.numCols);
if (ne == 0u) { ne = 1u; }
const bool hasTimeSig = (d.timeSignalIdx != UDPS_NO_TIME_SIGNAL) &&
(d.timeSignalIdx < nSigs);
const bool isFirstLast = (ne > 1u) &&
((d.timeMode == UDPS_TIMEMODE_FIRST_SAMPLE) ||
(d.timeMode == UDPS_TIMEMODE_LAST_SAMPLE));
const bool isFullArray = (d.timeMode == UDPS_TIMEMODE_FULL_ARRAY);
producerTimed[i] = hasTimeSig && (isFullArray || isFirstLast);
}
metaMutex_.FastUnLock();
/* Signals of one source share a packet, so they advance together; the max
* is "how far this source has produced" without stalling on a signal that
* simply is not being sent. */
float64 newest = 0.0;
for (uint32 i = 0u; i < nSigs; i++) {
if (!producerTimed[i]) { continue; }
const float64 t = rings_[i].NewestTime();
if (t > newest) { newest = t; }
}
return newest;
}
/*---------------------------------------------------------------------------*/
/* DATA parsing */
/*---------------------------------------------------------------------------*/
Source/Applications/StreamHub/UDPSourceSession.h
+57 -8
View File
@@ -154,6 +154,37 @@ public:
*/
void SetRingCapacities(uint32 temporal, uint32 scalar);
/**
* @brief Grow every ring so it can retain at least @p seconds of history.
*
* The required capacity is seconds × the rate measured from the ring
* itself (count / time span), because most sources advertise
* samplingRate = 0. Signals whose ring has not filled enough to measure a
* rate are left alone; the caller is expected to retry.
*
* @param seconds Retention target.
* @param maxPts Per-signal ceiling, so a multi-Msps source cannot be
* asked to allocate an unbounded amount of memory.
* @return true if at least one ring was enlarged.
*/
bool GrowRingsForSeconds(float64 seconds, uint32 maxPts);
/** @return Largest ring capacity currently allocated in this session. */
uint32 GetMaxRingCapacity() const;
/**
* @brief Newest timestamp this source has produced on its *own* clock, or
* 0 when it publishes no producer-timed signal (or has no data yet).
*
* Only signals that reference a time signal count: PACKET-timed signals
* are stamped on arrival and so live in the hub's wall-clock domain, not
* the producer's, even when they come from the very same source. A source
* free-running on its own clock sits seconds away from wall time and drifts,
* so anything waiting for a capture window to fill must compare against
* this, never clock_gettime().
*/
float64 ProducerNewestTime() const;
/**
* @brief Attach the (shared) hub trigger engine.
* Every decoded sample of the trigger's configured signal — resolved
@@ -241,24 +272,42 @@ private:
* signal @p tIdx given the first decoded timer value @p timer0S of the
* current packet and the arrival wall time @p wallNowS.
*
* Re-anchors the offset (offset = wallNowS − timer0S) when (a) it is the
* first packet, (b) the source clock jumped backward versus the previous
* packet (a looping/rewinding producer), or (c) the computed wall time has
* drifted past kRecalibThresholdS from the true arrival wall time.
* Snaps the offset to wallNowS − timer0S only when there is a genuine
* discontinuity in the source: the first packet, or a backward jump of the
* source clock (a looping/rewinding producer).
*
* A source that free-runs on its own clock also *drifts* against wall time,
* without any discontinuity. Snapping that away would shift the whole
* published timeline in one step and so tear a hole of exactly the drift
* into a stream that is in fact continuous, which is worse than the drift
* itself. Past kRecalibThresholdS the offset is therefore slewed instead:
* nudged towards wall time by at most kMaxSlewFraction of the packet's own
* duration, so the seam can never exceed a fraction of one packet.
*
* @return the calibration offset to add to timer-seconds for this signal.
*/
inline float64 CalibrateTimeSignal(uint32 tIdx, float64 timer0S,
float64 wallNowS) {
static const float64 kRecalibThresholdS = 2.0;
static const float64 kMaxSlewFraction = 0.1;
const bool reset = timeSigLastValid_[tIdx] &&
(timer0S < timeSigLastTimerS_[tIdx]);
const float64 drift = (timeSigCalib_[tIdx] + timer0S) - wallNowS;
const float64 absDrift = (drift < 0.0) ? -drift : drift;
if ((!timeSigCalibValid_[tIdx]) || reset ||
(absDrift > kRecalibThresholdS)) {
if ((!timeSigCalibValid_[tIdx]) || reset) {
timeSigCalib_[tIdx] = wallNowS - timer0S;
timeSigCalibValid_[tIdx] = true;
}
else {
const float64 drift = (timeSigCalib_[tIdx] + timer0S) - wallNowS;
const float64 absDrift = (drift < 0.0) ? -drift : drift;
if (absDrift > kRecalibThresholdS) {
const float64 pktSpan = timer0S - timeSigLastTimerS_[tIdx];
const float64 maxStep = pktSpan * kMaxSlewFraction;
float64 step = -drift;
if (step > maxStep) { step = maxStep; }
if (step < -maxStep) { step = -maxStep; }
timeSigCalib_[tIdx] += step;
}
}
timeSigLastTimerS_[tIdx] = timer0S;
timeSigLastValid_[tIdx] = true;
return timeSigCalib_[tIdx];
Source/Applications/StreamHub/WSServer.cpp
+79 -18
View File
@@ -8,6 +8,7 @@
#include "SHA1.h"
#include "Base64.h"
#include "AdvancedErrorManagement.h"
#include "Select.h"
#include "Sleep.h"
#include "Threads.h"
#include "TimeoutType.h"
@@ -57,8 +58,10 @@ static const char *FindSubstr(const char *s, const char *pattern) {
WSServer::WSServer()
: numClients(0u),
liveReadThreads(0u),
callback(static_cast<WSCommandCallback *>(0)),
running(false),
numAllowedOrigins(0u),
acceptTid(MARTe::InvalidThreadIdentifier) {
for (uint32 i = 0u; i < WS_MAX_CLIENTS; i++) {
@@ -66,6 +69,20 @@ WSServer::WSServer()
clients[i].active = false;
clients[i].readTid = MARTe::InvalidThreadIdentifier;
}
for (uint32 i = 0u; i < WS_MAX_ORIGINS; i++) {
allowedOrigins[i][0] = '\0';
}
}
bool WSServer::AddAllowedOrigin(const char *origin) {
if ((origin == static_cast<const char *>(0)) || (origin[0] == '\0')) {
return false;
}
if (numAllowedOrigins >= WS_MAX_ORIGINS) { return false; }
if (strlen(origin) >= WS_MAX_ORIGIN_LEN) { return false; }
strcpy(allowedOrigins[numAllowedOrigins], origin);
numAllowedOrigins++;
return true;
}
WSServer::~WSServer() {
@@ -104,9 +121,9 @@ bool WSServer::Start(uint16 port, WSCommandCallback *cb) {
bool WSServer::Stop() {
if (!running) { return true; }
running = false;
Sleep::MSec(200u);
/* Close all client connections — their read threads will exit on error */
/* Close all client connections — their read threads wake out of select()
* and unwind through FreeSlot. */
(void) clientsMutex.FastLock();
for (uint32 i = 0u; i < WS_MAX_CLIENTS; i++) {
if (clients[i].active && (clients[i].sock != static_cast<BasicTCPSocket *>(0))) {
@@ -114,10 +131,23 @@ bool WSServer::Stop() {
}
}
clientsMutex.FastUnLock();
Sleep::MSec(200u);
/* The accept loop polls WaitConnection with a 500 ms timeout, so it is out
* of the listener by now. */
Sleep::MSec(600u);
tcpListener.Close();
Sleep::MSec(100u);
/* Wait for the read threads: they hold pointers to the sockets freed
* below. Bounded — leaking a socket at exit beats deleting one that a
* wedged thread is still reading from. */
static const uint32 kReadJoinMs = 3000u;
for (uint32 waited = 0u; waited < kReadJoinMs; waited += 20u) {
(void) clientsMutex.FastLock();
const uint32 live = liveReadThreads;
clientsMutex.FastUnLock();
if (live == 0u) { break; }
Sleep::MSec(20u);
}
/* Free any remaining slots */
(void) clientsMutex.FastLock();
@@ -170,6 +200,10 @@ void WSServer::AcceptLoop() {
}
/* Start per-client read thread */
(void) clientsMutex.FastLock();
liveReadThreads++;
clientsMutex.FastUnLock();
ClientThreadArg *arg = new ClientThreadArg();
arg->srv = this;
arg->slot = slot;
@@ -200,12 +234,28 @@ bool WSServer::UpgradeHTTP(BasicTCPSocket *sock) {
}
/* Origin validation (CSWSH / CSRF defence, RFC 6455 §10.2).
* If an Origin header is present, its host must match the Host header
* (same-origin). Non-browser clients (no Origin) are allowed. */
* If an Origin header is present it must either be on the configured
* allowlist or its host must match the Host header (same-origin).
* Non-browser clients (no Origin) are allowed. */
const char *originHdr = FindSubstr(hdrBuf, "Origin:");
if (originHdr != static_cast<const char *>(0)) {
originHdr += 7; /* skip "Origin:" */
while (*originHdr == ' ') { originHdr++; }
/* Full origin value "scheme://host[:port]", for the allowlist. */
char originFull[WS_MAX_ORIGIN_LEN];
uint32 ofLen = 0u;
while ((originHdr[ofLen] != '\r') && (originHdr[ofLen] != '\n') &&
(originHdr[ofLen] != '\0') && (ofLen < (WS_MAX_ORIGIN_LEN - 1u))) {
originFull[ofLen] = originHdr[ofLen];
ofLen++;
}
originFull[ofLen] = '\0';
bool allowed = false;
for (uint32 i = 0u; (i < numAllowedOrigins) && !allowed; i++) {
if (strcmp(originFull, allowedOrigins[i]) == 0) { allowed = true; }
}
/* Extract the host part of Origin: "scheme://host[:port]" */
char originHost[256];
uint32 ohLen = 0u;
@@ -221,7 +271,7 @@ bool WSServer::UpgradeHTTP(BasicTCPSocket *sock) {
/* Extract Host header value */
const char *hostHdr = FindSubstr(hdrBuf, "Host:");
if (hostHdr != static_cast<const char *>(0)) {
if (!allowed && (hostHdr != static_cast<const char *>(0))) {
hostHdr += 5; /* skip "Host:" */
while (*hostHdr == ' ') { hostHdr++; }
char hostVal[256];
@@ -299,23 +349,30 @@ void WSServer::ClientReadLoop(uint32 slotIdx) {
uint32 filled = 0u;
while (running && slot.active) {
/* Read more bytes (with short timeout so we can check running) */
uint32 want = kRecvBuf - filled;
if (want == 0u) {
/* Buffer full — discard old frame (shouldn't happen with reasonable clients) */
filled = 0u;
continue;
}
bool ok = sock->Read(reinterpret_cast<char *>(buf + filled), want,
TimeoutType(500u));
if (!ok) {
/* Timeout or error — check running and retry */
if (!running) { break; }
if (want == kRecvBuf) {
/* Zero bytes read — connection likely closed */
break;
}
continue;
/* Wait for readability before reading. BasicTCPSocket::Read reports a
* timeout and a closed peer identically (false, zero bytes), so polling
* it on its own cannot end the loop: once the client goes away recv
* returns immediately and forever, and the thread spins at 100% CPU
* until it starves the rest of the hub. select() tells the two apart —
* readable followed by no data is end of stream. A wait consumes the
* handle set, hence a fresh Select each pass. */
MARTe::Select sel;
if (!sel.AddReadHandle(*sock)) { break; }
const MARTe::int32 ready = sel.WaitUntil(TimeoutType(500u));
if (ready == 0) { continue; } /* idle client — recheck running */
if (ready < 0) { break; } /* socket closed or errored */
/* Readable: this returns at once, and only fails at end of stream. */
if (!sock->Read(reinterpret_cast<char *>(buf + filled), want,
TimeoutType(500u))) {
break;
}
filled += want;
@@ -383,6 +440,10 @@ client_done:
callback->OnWSClientDisconnected();
}
FreeSlot(slotIdx);
(void) clientsMutex.FastLock();
if (liveReadThreads > 0u) { liveReadThreads--; }
clientsMutex.FastUnLock();
}
/*---------------------------------------------------------------------------*/
Source/Applications/StreamHub/WSServer.h
+25 -1
View File
@@ -34,6 +34,12 @@ static const uint32 WS_MAX_RECV_PAYLOAD = 65536u;
/** Maximum WebSocket frame payload we will send (data frames can be large). */
static const uint32 WS_MAX_SEND_PAYLOAD = 4u * 1024u * 1024u; /* 4 MiB */
/** Maximum entries in the Origin allowlist. */
static const uint32 WS_MAX_ORIGINS = 8u;
/** Maximum length of one allowlisted Origin ("scheme://host[:port]"). */
static const uint32 WS_MAX_ORIGIN_LEN = 128u;
/**
* @brief Callback interface — implemented by StreamHub.
*/
@@ -77,6 +83,20 @@ public:
*/
bool Start(uint16 port, WSCommandCallback *cb);
/**
* @brief Add an Origin that is accepted for the WebSocket upgrade.
*
* With an empty allowlist (the default) only same-origin requests pass:
* the Origin's host must equal the Host header, which excludes the usual
* deployment where the SPA is served by a separate web server on another
* port. Add that server's origin (e.g. "http://localhost:8080") to allow
* it. Requests without an Origin header (non-browser clients) always pass.
*
* @param origin "scheme://host[:port]", compared verbatim.
* @return false if the allowlist is full or the string is too long.
*/
bool AddAllowedOrigin(const char *origin);
/**
* @brief Stop accept thread; close all client connections; close listener.
*/
@@ -119,11 +139,15 @@ private:
BasicTCPSocket tcpListener;
WSClientSlot clients[WS_MAX_CLIENTS];
uint32 numClients;
mutable FastPollingMutexSem clientsMutex; ///< Protects numClients and clients[] array
uint32 liveReadThreads; ///< Read threads not yet unwound; Stop() waits on it
mutable FastPollingMutexSem clientsMutex; ///< Protects numClients, liveReadThreads and clients[] array
WSCommandCallback *callback;
volatile bool running;
char allowedOrigins[WS_MAX_ORIGINS][WS_MAX_ORIGIN_LEN];
uint32 numAllowedOrigins;
MARTe::ThreadIdentifier acceptTid;
};
Source/Components/GAMs/TimeArrayGAM/TimeArrayGAM.cpp
+23 -2
View File
@@ -16,6 +16,10 @@ TimeArrayGAM::TimeArrayGAM() :
GAM(),
samplingRate(1000000.0),
anchorIsFirst(true),
anchorIsCont(false),
contStarted(false),
contOriginNs(0u),
contSamples(0u),
nElements(0u),
inputTime(NULL_PTR(uint32 *)),
outputBuf(NULL_PTR(uint64 *)) {
@@ -42,9 +46,12 @@ bool TimeArrayGAM::Initialise(StructuredDataI &data) {
else if (anchor == "LastSample") {
anchorIsFirst = false;
}
else if (anchor == "Continuous") {
anchorIsCont = true;
}
else {
REPORT_ERROR(ErrorManagement::InitialisationError,
"TimeArrayGAM: Anchor must be 'FirstSample' or 'LastSample'.");
"TimeArrayGAM: Anchor must be 'FirstSample', 'LastSample' or 'Continuous'.");
ok = false;
}
}
@@ -88,7 +95,21 @@ bool TimeArrayGAM::Execute() {
/* Input is uint32 microseconds (LinuxTimer); convert to nanoseconds. */
uint64 anchorNs = static_cast<uint64>(*inputTime) * 1000u;
if (anchorIsFirst) {
if (anchorIsCont) {
/* Latch the timer once, then run off an internal sample counter so a
* lost RT cycle (LinuxTimer re-phases with counter += nCycles) cannot
* punch a hole into an otherwise contiguous sample stream. */
if (!contStarted) {
contOriginNs = anchorNs;
contStarted = true;
}
for (uint32 k = 0u; k < nElements; k++) {
outputBuf[k] = contOriginNs +
(contSamples + static_cast<uint64>(k)) * periodNs;
}
contSamples += static_cast<uint64>(nElements);
}
else if (anchorIsFirst) {
/* out[k] = anchorNs + k * periodNs */
for (uint32 k = 0u; k < nElements; k++) {
outputBuf[k] = anchorNs + static_cast<uint64>(k) * periodNs;
Source/Components/GAMs/TimeArrayGAM/TimeArrayGAM.h
+14 -1
View File
@@ -10,6 +10,15 @@
*
* Anchor = FirstSample: out[k] = input + k * period_us
* Anchor = LastSample: out[k] = input - (N-1-k) * period_us
* Anchor = Continuous: out[k] = input(first cycle) + (n + k) * period_us
*
* FirstSample/LastSample re-read the timer every cycle, so they propagate any
* cycle the RT thread loses: LinuxTimer re-phases (counter += nCycles) and the
* emitted time base jumps by a whole period while only one array of samples is
* produced, leaving a hole. Continuous anchors once and then advances an
* internal sample counter by N per cycle, which is what an acquisition card
* with its own clock does — use it when the data signal is itself contiguous
* (SineArrayGAM, for instance, never skips phase on a lost cycle).
*
* The resulting time array is suitable as the TimeSignal for a UDPStreamer signal
* configured with TimeMode = FullArray, providing exact per-sample timestamps.
@@ -19,7 +28,7 @@
* +TimeArrayGAM1 = {
* Class = TimeArrayGAM
* SamplingRate = 1000000.0 // Sample rate in Hz (must match data signal)
* Anchor = FirstSample // FirstSample (default) or LastSample
* Anchor = FirstSample // FirstSample (default), LastSample or Continuous
* InputSignals = {
* Time = { DataSource = DDB; Type = uint32 }
* }
@@ -54,6 +63,10 @@ public:
private:
float64 samplingRate; /**< Sample rate [Hz] */
bool anchorIsFirst; /**< true = FirstSample anchor, false = LastSample */
bool anchorIsCont; /**< true = Continuous anchor (internal sample counter) */
bool contStarted; /**< Continuous: origin has been latched */
uint64 contOriginNs; /**< Continuous: timer value latched on the first cycle */
uint64 contSamples; /**< Continuous: samples emitted so far */
uint32 nElements; /**< Number of output elements */
uint32 *inputTime; /**< Pointer to scalar input (microseconds, uint32 from LinuxTimer) */
uint64 *outputBuf; /**< Pointer to output array (nanoseconds, uint64) */