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

feat(streamhub): wire BinaryRecorder into UDPSourceSession

Browse Source 
Each session owns a BinaryRecorder. The receive thread configures the
recorder layout at CONFIG time and captures each data packet; the push
thread performs all file I/O via RecorderFlushTick. WS subset overrides
are adopted through an epoch-deferred handoff mirroring the trigger path.
Arm/disarm/flush/info are exposed for StreamHub-level control.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
2026-06-25 01:10:27 +02:00
parent 0ae35d11ff
commit 93fc11c88f
3 changed files with 152 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Applications/StreamHub/Makefile.inc
+1 -1
View File
@@ -1,4 +1,4 @@
OBJSX = StreamHub.x UDPSourceSession.x WSServer.x TriggerEngine.x HistoryWriter.x
OBJSX = StreamHub.x UDPSourceSession.x WSServer.x TriggerEngine.x HistoryWriter.x BinaryRecorder.x
PACKAGE =
ROOT_DIR = ../../..
Source/Applications/StreamHub/UDPSourceSession.cpp
+109 -1
View File
@@ -71,8 +71,15 @@ UDPSourceSession::UDPSourceSession()
trigEngine_(static_cast<TriggerEngine *>(0)),
trigEpochSeen_(0xFFFFFFFFu),
trigSigIdx_(-1),
trigElemIdx_(-1) {
trigElemIdx_(-1),
recPendingEpoch_(0u),
recSeenEpoch_(0u) {
stateStr_ = "disconnected";
memset(&recCfg_, 0, sizeof(recCfg_));
recPendingSpec_[0] = '\0';
strncpy(recActiveSpec_, "all", sizeof(recActiveSpec_) - 1u);
recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0';
recSpecMutex_.Create();
ResetCalibration();
}
@@ -244,6 +251,19 @@ void UDPSourceSession::ParseConfigPayload(const uint8 *payload, uint32 size) {
/* Allocate ring buffers (outside metaMutex to avoid long hold) */
AllocateRingBuffers();
/* (Re)configure the recorder: a CONFIG change re-headers a fresh file and
* adopts any pending subset spec. */
if (recCfg_.enabled) {
bool mask[UDPSS_MAX_SIGNALS];
(void) recSpecMutex_.FastLock();
strncpy(recActiveSpec_, recPendingSpec_, sizeof(recActiveSpec_) - 1u);
recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0';
recSeenEpoch_ = recPendingEpoch_;
recSpecMutex_.FastUnLock();
ComputeIncludeMask(sigDescs_, numSigs, recActiveSpec_, mask);
recorder_.Configure(sigDescs_, numSigs, mask);
}
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"UDPSourceSession[%s]: CONFIG received — %u signals.",
id_.Buffer(), numSigs);
@@ -506,6 +526,21 @@ void UDPSourceSession::ParseDataPayload(const uint8 *payload, uint32 size) {
}
}
/* Binary recorder: adopt any pending subset spec, then capture the packet
* verbatim (native types, lossless) for disk recording. */
if (recCfg_.enabled) {
if (recPendingEpoch_ != recSeenEpoch_) {
bool mask[UDPSS_MAX_SIGNALS];
(void) recSpecMutex_.FastLock();
strncpy(recActiveSpec_, recPendingSpec_, sizeof(recActiveSpec_) - 1u);
recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0';
recSeenEpoch_ = recPendingEpoch_;
recSpecMutex_.FastUnLock();
ComputeIncludeMask(descs, nSigs, recActiveSpec_, mask);
recorder_.Configure(descs, nSigs, mask);
}
recorder_.CapturePacket(payload, sigOff, sigElems, pm, numSamples);
}
}
float64 UDPSourceSession::DecodeOneElem(const uint8 *payload, uint32 off,
@@ -808,6 +843,79 @@ void UDPSourceSession::SetTriggerEngine(TriggerEngine *engine) {
trigElemIdx_ = -1;
}
/*---------------------------------------------------------------------------*/
/* Binary recorder control */
/*---------------------------------------------------------------------------*/
void UDPSourceSession::SetRecorderConfig(const RecorderConfig &cfg) {
recCfg_ = cfg;
if (!cfg.enabled) { return; }
recorder_.Init(cfg, id_.Buffer());
(void) recSpecMutex_.FastLock();
strncpy(recPendingSpec_, cfg.signals, sizeof(recPendingSpec_) - 1u);
recPendingSpec_[sizeof(recPendingSpec_) - 1u] = '\0';
strncpy(recActiveSpec_, cfg.signals, sizeof(recActiveSpec_) - 1u);
recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0';
recPendingEpoch_++;
recSpecMutex_.FastUnLock();
}
void UDPSourceSession::SetRecorderSignals(const char *spec) {
if (spec == static_cast<const char *>(0)) { return; }
(void) recSpecMutex_.FastLock();
strncpy(recPendingSpec_, spec, sizeof(recPendingSpec_) - 1u);
recPendingSpec_[sizeof(recPendingSpec_) - 1u] = '\0';
recPendingEpoch_++;
recSpecMutex_.FastUnLock();
}
void UDPSourceSession::RequestRecArm() {
if (recCfg_.enabled) { recorder_.RequestArm(); }
}
void UDPSourceSession::RequestRecDisarm() {
if (recCfg_.enabled) { recorder_.RequestDisarm(); }
}
void UDPSourceSession::RecorderFlushTick(uint32 nowSec) {
if (recCfg_.enabled) { recorder_.FlushTick(nowSec); }
}
void UDPSourceSession::GetRecorderInfo(bool &recording, char *file, uint32 fileSz,
uint64 &bytesWritten, uint64 &rowsWritten,
uint64 &droppedRows, uint64 &freeMB) const {
recorder_.GetInfo(recording, file, fileSz, bytesWritten, rowsWritten,
droppedRows, freeMB);
}
void UDPSourceSession::ComputeIncludeMask(const UDPSSignalDescriptor *descs,
uint32 n, const char *spec,
bool *maskOut) {
if ((spec == static_cast<const char *>(0)) || (spec[0] == '\0') ||
(strcmp(spec, "all") == 0)) {
for (uint32 i = 0u; i < n; i++) { maskOut[i] = true; }
return;
}
for (uint32 i = 0u; i < n; i++) {
char key[160];
(void) snprintf(key, sizeof(key), "%s:%s", id_.Buffer(), descs[i].name);
const size_t klen = strlen(key);
maskOut[i] = false;
const char *p = spec;
while (*p != '\0') {
while ((*p == ',') || (*p == ' ')) { p++; }
const char *start = p;
while ((*p != '\0') && (*p != ',')) { p++; }
size_t tlen = static_cast<size_t>(p - start);
while ((tlen > 0u) && (start[tlen - 1u] == ' ')) { tlen--; }
if ((tlen == klen) && (strncmp(start, key, klen) == 0)) {
maskOut[i] = true;
break;
}
}
}
}
void UDPSourceSession::ResolveTriggerSignal(const UDPSSignalDescriptor *descs,
uint32 nSigs) {
trigSigIdx_ = -1;
Source/Applications/StreamHub/UDPSourceSession.h
+42
View File
@@ -19,6 +19,7 @@
#include "SignalRingBuffer.h"
#include "UDPSourceStats.h"
#include "TriggerEngine.h"
#include "BinaryRecorder.h"
#include <string.h>
namespace StreamHub {
@@ -161,8 +162,40 @@ public:
*/
void SetTriggerEngine(TriggerEngine *engine);
/* ---- Binary recorder control -------------------------------------- */
/** @brief Set the recorder configuration; initialises the recorder when
* enabled. Must be called before Start(). */
void SetRecorderConfig(const RecorderConfig &cfg);
/** @brief Override the recorded signal subset for this run (any thread).
* Adopted on the receive thread via an epoch check. */
void SetRecorderSignals(const char *spec);
/** @brief Arm the recorder (any thread). */
void RequestRecArm();
/** @brief Disarm the recorder (any thread). */
void RequestRecDisarm();
/** @brief Drive recorder disk I/O (push thread). */
void RecorderFlushTick(uint32 nowSec);
/** @brief Recorder status snapshot (push thread). */
void GetRecorderInfo(bool &recording, char *file, uint32 fileSz,
uint64 &bytesWritten, uint64 &rowsWritten,
uint64 &droppedRows, uint64 &freeMB) const;
/** @return true if the recorder is enabled for this session. */
bool IsRecorderEnabled() const { return recCfg_.enabled; }
private:
/** @brief Build the per-signal include mask from a subset spec
* ("all" or comma-separated "src:sig" keys). Receive thread only. */
void ComputeIncludeMask(const UDPSSignalDescriptor *descs, uint32 n,
const char *spec, bool *maskOut);
/* DATA payload parsing */
void ParseConfigPayload(const uint8 *payload, uint32 size);
void ParseDataPayload(const uint8 *payload, uint32 size);
@@ -293,6 +326,15 @@ private:
uint32 trigEpochSeen_; ///< Last resolved trigger config epoch
MARTe::int32 trigSigIdx_; ///< Watched signal index (-1 = none)
MARTe::int32 trigElemIdx_; ///< Watched element index (-1 = all)
/* Binary recorder (capture on receive thread, I/O on push thread). */
BinaryRecorder recorder_;
RecorderConfig recCfg_; ///< Recorder config (enabled flag gates use)
mutable FastPollingMutexSem recSpecMutex_;
char recPendingSpec_[1024]; ///< Subset spec awaiting adoption
char recActiveSpec_[1024]; ///< Currently applied subset spec
volatile uint32 recPendingEpoch_; ///< Bumped when the spec changes
uint32 recSeenEpoch_; ///< Last spec epoch adopted (receive thread)
};
} /* namespace StreamHub */