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

fix: three StreamHub C++ defects from final code review

Browse Source 
Finding 1 (HandleReloadConfig data loss): move ClearCalibration inside
LoadSourcesFile so the table is wiped only after a successful fread.
Adds a clearCalibration bool parameter (default false); the reload path
passes true, the startup path passes false.

Finding 2 (JSON injection via unit/source/signal): add JsonEscape()
static helper (escapes \", \\, \n \r \t, and \u00XX for other control
chars). Applied at all three emission sites: BroadcastCalibration,
HandleSaveSources, and BroadcastSources (label). Teach JsonGetString to
unescape the same set on read, so values round-trip correctly.

Finding 3 (%.17g verbosity): add ShortFloat() static helper that tries
%.15g then %.16g then %.17g, stopping at the first precision whose
strtod() output compares equal to the original. Applied at both float
emission sites. 0.1 now prints as "0.1", not "0.10000000000000001".

Minor: fix two inaccurate comments in StreamHub.h — the CalibrationEntry
rationale (not a 133 MB / address-limit issue; the real reason is no
per-entry heap churn, STL-free, trivially copyable) and "chars" to "bytes"
for the unit cap.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
2026-08-17 07:50:01 +02:00
co-authored by Claude Sonnet 4.6
parent 686fc2ce7d
commit 42f5a726af
2 changed files with 159 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Applications/StreamHub/StreamHub.cpp
+150 -20
View File
@@ -261,7 +261,7 @@ bool StreamHub::Initialise(StructuredDataI &cfg) {
}
/* Start any persisted dynamic sources (Go SourceConfig schema). */
(void) LoadSourcesFile(false);
(void) LoadSourcesFile(false, false);
REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information,
"StreamHub: initialised with %u session(s), WSPort=%u, MaxPoints=%u, PushRate=%u Hz.",
@@ -603,6 +603,71 @@ void StreamHub::PushStats() {
delete[] buf;
}
/*---------------------------------------------------------------------------*/
/* JSON string helpers */
/*---------------------------------------------------------------------------*/
/**
* JSON-escape a string: escapes '"' as '\"', '\' as '\\', and control
* characters below 0x20 (using '\n', '\r', '\t' for those three, and
* '\u00XX' for the rest). Always NUL-terminates; never writes past outSize.
* Worst case: 6 bytes output per input byte (for \u00XX form).
*/
static void JsonEscape(const MARTe::char8 *in, MARTe::char8 *out,
MARTe::uint32 outSize) {
if ((in == static_cast<const MARTe::char8 *>(0)) ||
(out == static_cast<MARTe::char8 *>(0)) ||
(outSize == 0u)) { return; }
MARTe::uint32 o = 0u;
for (MARTe::uint32 i = 0u; in[i] != '\0'; i++) {
unsigned char c = static_cast<unsigned char>(in[i]);
if (c == '"') {
if (o + 2u >= outSize) { break; }
out[o++] = '\\'; out[o++] = '"';
} else if (c == '\\') {
if (o + 2u >= outSize) { break; }
out[o++] = '\\'; out[o++] = '\\';
} else if (c == '\n') {
if (o + 2u >= outSize) { break; }
out[o++] = '\\'; out[o++] = 'n';
} else if (c == '\r') {
if (o + 2u >= outSize) { break; }
out[o++] = '\\'; out[o++] = 'r';
} else if (c == '\t') {
if (o + 2u >= outSize) { break; }
out[o++] = '\\'; out[o++] = 't';
} else if (c < 0x20u) {
if (o + 6u >= outSize) { break; }
out[o++] = '\\'; out[o++] = 'u';
out[o++] = '0'; out[o++] = '0';
out[o++] = static_cast<MARTe::char8>(
"0123456789abcdef"[(c >> 4u) & 0xFu]);
out[o++] = static_cast<MARTe::char8>(
"0123456789abcdef"[c & 0xFu]);
} else {
if (o + 1u >= outSize) { break; }
out[o++] = static_cast<MARTe::char8>(c);
}
}
out[o] = '\0';
}
/**
* Format a float64 with the shortest representation that round-trips.
* Tries %.15g, then %.16g, then %.17g; stops at the first precision where
* strtod(formatted) == original. 'out' must be at least 32 bytes.
*/
static void ShortFloat(MARTe::float64 v, MARTe::char8 *out,
MARTe::uint32 outSize) {
static const int kPrec[] = { 15, 16, 17 };
static const MARTe::uint32 kNPrec = 3u;
for (MARTe::uint32 p = 0u; p < kNPrec; p++) {
(void) snprintf(out, outSize, "%.*g", kPrec[p], v);
if (strtod(out, static_cast<char **>(0)) == v) { return; }
}
/* Fallback: %.17g is already stored in out from last iteration. */
}
/*---------------------------------------------------------------------------*/
/* Sources / Config broadcast */
/*---------------------------------------------------------------------------*/
@@ -625,12 +690,15 @@ void StreamHub::BroadcastSources() {
uint16 prt = sessions_[i].GetPort();
if (off >= kBuf - 256u) { break; }
/* Escape label (user-supplied) to guard against embedded quotes. */
char elbl[128u * 6u + 1u];
JsonEscape(lbl.Buffer(), elbl, sizeof(elbl));
/* Go hub shape: addr is the combined "host:port" string. */
off += static_cast<uint32>(snprintf(buf + off, kBuf - off,
"%s{\"id\":\"%s\",\"label\":\"%s\","
"\"addr\":\"%s:%u\",\"state\":\"%s\"}",
(first ? "" : ","),
sid.Buffer(), lbl.Buffer(),
sid.Buffer(), elbl,
adr.Buffer(), static_cast<uint32>(prt),
st.state.Buffer()));
first = false;
@@ -931,15 +999,26 @@ void StreamHub::BroadcastCalibration() {
delete[] idx;
for (uint32 i = 0u; i < n; i++) {
/* Worst-case escape: 6 bytes per input byte */
char esource[128u * 6u + 1u];
char esignal[128u * 6u + 1u];
char eunit[17u * 6u + 1u];
JsonEscape(snap[i].source, esource, sizeof(esource));
JsonEscape(snap[i].signal, esignal, sizeof(esignal));
JsonEscape(snap[i].unit, eunit, sizeof(eunit));
char sscale[32];
char soffset[32];
ShortFloat(snap[i].scale, sscale, sizeof(sscale));
ShortFloat(snap[i].offset, soffset, sizeof(soffset));
JsonAppendf(buf, off, cap,
"%s{\"source\":\"%s\",\"signal\":\"%s\","
"\"scale\":%.17g,\"offset\":%.17g,\"unit\":\"%s\"}",
"\"scale\":%s,\"offset\":%s,\"unit\":\"%s\"}",
(i > 0u) ? "," : "",
snap[i].source,
snap[i].signal,
snap[i].scale,
snap[i].offset,
snap[i].unit);
esource,
esignal,
sscale,
soffset,
eunit);
}
delete[] snap;
@@ -1147,7 +1226,7 @@ bool StreamHub::SourceIsActive(const char *addrPort) {
return false;
}
bool StreamHub::LoadSourcesFile(bool skipActive) {
bool StreamHub::LoadSourcesFile(bool skipActive, bool clearCalibration) {
if (sourcesFile_.Size() == 0u) { return false; }
FILE *f = fopen(sourcesFile_.Buffer(), "rb");
@@ -1165,6 +1244,10 @@ bool StreamHub::LoadSourcesFile(bool skipActive) {
data[nRead] = '\0';
(void) fclose(f);
/* Clear calibration only after a successful read so that a transient I/O
* failure (file deleted, renamed, etc.) does not silently wipe the table. */
if (clearCalibration) { ClearCalibration(); }
/* Flat JSON array of flat objects — iterate over each {...} block. A block
* with "addr" is a source, one with "signal" is a calibration. The array
* must stay flat: this scanner takes each "{" up to the next "}". */
@@ -1304,17 +1387,26 @@ void StreamHub::HandleSaveSources() {
delete[] cidx;
for (uint32 i = 0u; i < nCalTotal; i++) {
char esource[128u * 6u + 1u];
char esignal[128u * 6u + 1u];
char eunit[17u * 6u + 1u];
JsonEscape(csnap[i].source, esource, sizeof(esource));
JsonEscape(csnap[i].signal, esignal, sizeof(esignal));
JsonEscape(csnap[i].unit, eunit, sizeof(eunit));
char sscale[32];
char soffset[32];
ShortFloat(csnap[i].scale, sscale, sizeof(sscale));
ShortFloat(csnap[i].offset, soffset, sizeof(soffset));
(void) fprintf(f,
"%s {\n \"source\": \"%s\",\n \"signal\": \"%s\",\n"
" \"scale\": %.17g,\n \"offset\": %.17g",
" \"scale\": %s,\n \"offset\": %s",
((nSaved + nCal) > 0u) ? ",\n" : "",
csnap[i].source,
csnap[i].signal,
csnap[i].scale,
csnap[i].offset);
esource,
esignal,
sscale,
soffset);
if (csnap[i].unit[0] != '\0') {
(void) fprintf(f, ",\n \"unit\": \"%s\"",
csnap[i].unit);
(void) fprintf(f, ",\n \"unit\": \"%s\"", eunit);
}
(void) fprintf(f, "\n }");
nCal++;
@@ -1418,9 +1510,9 @@ void StreamHub::HandleReloadConfig() {
return;
}
/* Calibration is replaced wholesale; sources are only added. A reload must
* never interrupt a live UDP session. */
ClearCalibration();
if (!LoadSourcesFile(true)) {
* never interrupt a live UDP session. ClearCalibration is deferred inside
* LoadSourcesFile so the table is not wiped if the file cannot be read. */
if (!LoadSourcesFile(true, true)) {
BroadcastConfigAck("configReloaded", false, "cannot read sources file");
return;
}
@@ -2042,7 +2134,45 @@ bool StreamHub::JsonGetString(const char *json, const char *key,
p++;
uint32 i = 0u;
while ((*p != '\0') && (*p != '"') && (i < (outSize - 1u))) {
out[i++] = *p++;
if ((*p == '\\') && (*(p + 1) != '\0')) {
p++; /* skip backslash */
if (*p == '"') { out[i++] = '"'; p++; }
else if (*p == '\\') { out[i++] = '\\'; p++; }
else if (*p == 'n') { out[i++] = '\n'; p++; }
else if (*p == 'r') { out[i++] = '\r'; p++; }
else if (*p == 't') { out[i++] = '\t'; p++; }
else if (*p == 'u') {
/* \uXXXX — only handle the \u00XX subset we emit */
p++;
unsigned int code = 0u;
uint32 d = 0u;
while ((d < 4u) && (*p != '\0')) {
unsigned char ch = static_cast<unsigned char>(*p);
unsigned int nibble = 0u;
if ((ch >= '0') && (ch <= '9')) {
nibble = static_cast<unsigned int>(ch - '0');
} else if ((ch >= 'a') && (ch <= 'f')) {
nibble = static_cast<unsigned int>(ch - 'a') + 10u;
} else if ((ch >= 'A') && (ch <= 'F')) {
nibble = static_cast<unsigned int>(ch - 'A') + 10u;
} else {
break;
}
code = (code << 4u) | nibble;
p++;
d++;
}
if (i < (outSize - 1u)) {
out[i++] = static_cast<char>(code & 0xFFu);
}
} else {
/* Unknown escape: pass through literally */
if (i < (outSize - 1u)) { out[i++] = *p; }
p++;
}
} else {
out[i++] = *p++;
}
}
out[i] = '\0';
return true;
Source/Applications/StreamHub/StreamHub.h
+9 -6
View File
@@ -57,15 +57,15 @@ static const uint32 kMaxUnitLen = 16u;
* add-order and would rebind if the source list were reordered) and by the
* BASE signal name (no "[i]" suffix: one entry covers a whole array signal).
*
* Fixed-size char arrays are used deliberately: embedding 256 StreamString
* (each of which allocates its own heap buffer) into a 133 MB struct that is
* itself heap-allocated pushes offsets beyond the canonical x86-64 address
* limit and causes a SIGSEGV in the constructor.
* Fixed-size char arrays are used deliberately: they avoid per-entry heap
* churn (no StreamString allocation per calibration slot), keep the type free
* of STL, and make a CalibrationEntry snapshot trivially copyable under the
* calibration mutex lock.
*/
struct CalibrationEntry {
char source[128]; ///< Source label
char signal[128]; ///< Base signal name (no "[i]" suffix)
char unit[17]; ///< Unit override (max kMaxUnitLen chars + NUL)
char unit[17]; ///< Unit override (max kMaxUnitLen bytes + NUL)
MARTe::float64 scale;
MARTe::float64 offset;
};
@@ -211,9 +211,12 @@ private:
* {"source","signal","scale","offset","unit"} calibration blocks).
* @param skipActive when true, a source whose "host:port" is already
* streaming is left alone instead of being started a second time.
* @param clearCalibration when true, the calibration table is cleared
* after a successful fread (never before), so a transient I/O failure
* does not silently wipe user calibration data.
* @return true if the file was read.
*/
bool LoadSourcesFile(bool skipActive);
bool LoadSourcesFile(bool skipActive, bool clearCalibration = false);
/** @return true if a session for this "host:port" is already active. */
bool SourceIsActive(const char *addrPort);