martino/marte-debug
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

interface added and web ui created

Browse Source
This commit is contained in:
Martino Ferrari
2026-05-07 10:49:24 +02:00
parent 340b024ae4
commit 774808a054
21 changed files with 4986 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Components/Interfaces/DebugService/DebugService.cpp
+286 -42
View File
@@ -1,3 +1,4 @@
#include "Atomic.h"
#include "BasicTCPSocket.h"
#include "ClassRegistryItem.h"
#include "ConfigurationDatabase.h"
@@ -21,7 +22,8 @@
namespace MARTe {
DebugService *DebugService::instance = (DebugService *)0;
// DebugServiceI static members — defined here so no extra .cpp is needed.
DebugServiceI *DebugServiceI::instance = NULL_PTR(DebugServiceI *);
static void EscapeJson(const char8 *src, StreamString &dst) {
if (src == NULL_PTR(const char8 *))
@@ -85,6 +87,15 @@ static bool FindPathInContainer(ReferenceContainer *container,
CLASS_REGISTER(DebugService, "1.0")
// Out-of-class definitions required by C++98 when the constants are odr-used
// (i.e. their address is taken or they appear in a context that needs linkage).
const uint32 DebugService::STREAMER_MTU;
const uint32 DebugService::STREAMER_BUFFER_SIZE;
const uint32 DebugService::CMD_RATE_LIMIT;
const uint32 DebugService::CLIENT_IDLE_TIMEOUT_MS;
const uint32 DebugService::GET_VALUE_MAX_ELEMENTS;
const uint32 DebugService::INPUT_BUFFER_MAX;
DebugService::DebugService()
: ReferenceContainer(), EmbeddedServiceMethodBinderI(),
binderServer(this, ServiceBinder::ServerType),
@@ -102,11 +113,15 @@ DebugService::DebugService()
activeClient = NULL_PTR(BasicTCPSocket *);
streamerPacketOffset = 0u;
streamerSequenceNumber = 0u;
cmdCountInWindow = 0u;
cmdWindowStartMs = 0u;
lastDataTimeMs = 0u;
inputBuffer = ""; // FIX #10: initialise carry-over buffer
}
DebugService::~DebugService() {
if (instance == this) {
instance = NULL_PTR(DebugService *);
if (DebugServiceI::GetInstance() == this) {
DebugServiceI::SetInstance(NULL_PTR(DebugServiceI *));
}
threadService.Stop();
streamerService.Stop();
@@ -135,7 +150,7 @@ bool DebugService::Initialise(StructuredDataI &data) {
if (controlPort > 0) {
isServer = true;
instance = this;
DebugServiceI::SetInstance(this);
}
port = 8081;
@@ -463,12 +478,28 @@ void DebugService::ProcessSignal(DebugSignalInfo *signalInfo, uint32 size,
memcpy(signalInfo->memoryAddress, signalInfo->forcedValue, size);
}
if (signalInfo->isTracing) {
if (signalInfo->decimationCounter == 0) {
traceBuffer.Push(signalInfo->internalID, timestamp,
(uint8 *)signalInfo->memoryAddress, size);
// FIX #4: decimationCounter read-modify-write is a data race when multiple
// RT threads call ProcessSignal() concurrently for signals with the same
// DebugSignalInfo (e.g. an output signal read by two GAMs).
//
// Pattern: atomically increment, then claim the "push token" by exchanging
// back to zero only when the counter reaches the decimation factor.
// Only the thread that observes old >= decimationFactor actually pushes;
// all others just increment and return. No separate lock is needed for
// the counter itself.
//
// tracePushMutex still serialises the Push() call (FIX #2) because
// TraceRingBuffer is SPSC and multiple RT threads can be pushing concurrently.
Atomic::Add((volatile int32 *)&signalInfo->decimationCounter, 1);
if ((uint32)signalInfo->decimationCounter >= signalInfo->decimationFactor) {
int32 old = Atomic::Exchange((volatile int32 *)&signalInfo->decimationCounter, 0);
if ((uint32)old >= signalInfo->decimationFactor) {
tracePushMutex.FastLock();
traceBuffer.Push(signalInfo->internalID, timestamp,
(uint8 *)signalInfo->memoryAddress, size);
tracePushMutex.FastUnLock();
}
}
signalInfo->decimationCounter =
(signalInfo->decimationCounter + 1) % signalInfo->decimationFactor;
}
}
@@ -501,6 +532,11 @@ void DebugService::RegisterBroker(DebugSignalInfo **signalPointers,
void DebugService::UpdateBrokersActiveStatus() {
for (uint32 i = 0; i < brokers.Size(); i++) {
// FIX #3: signalPointers may be NULL when numSignals > 0 if the broker was
// registered with a null array (e.g. from unit tests or misconfigured brokers).
// Dereferencing NULL would be UB; skip the broker entirely.
if (brokers[i].signalPointers == NULL_PTR(DebugSignalInfo **)) continue;
uint32 count = 0;
for (uint32 j = 0; j < brokers[i].numSignals; j++) {
DebugSignalInfo *s = brokers[i].signalPointers[j];
@@ -524,20 +560,28 @@ void DebugService::UpdateBrokersActiveStatus() {
if (brokers[i].activeMutex)
brokers[i].activeMutex->FastLock();
// FIX #2: Use O(1) Swap instead of operator= (heap alloc + copy) inside
// the critical section. The old arrays are carried out in tempInd/tempSizes
// and freed after the lock is released.
if (brokers[i].activeIndices)
*(brokers[i].activeIndices) = tempInd;
brokers[i].activeIndices->Swap(tempInd);
if (brokers[i].activeSizes)
*(brokers[i].activeSizes) = tempSizes;
brokers[i].activeSizes->Swap(tempSizes);
if (brokers[i].anyActiveFlag)
*(brokers[i].anyActiveFlag) = (count > 0);
if (brokers[i].activeMutex)
brokers[i].activeMutex->FastUnLock();
// tempInd and tempSizes now hold the old arrays and are freed here,
// outside the lock.
}
}
void DebugService::UpdateBrokersBreakStatus() {
for (uint32 i = 0; i < brokers.Size(); i++) {
// FIX #3: same null guard as UpdateBrokersActiveStatus.
if (brokers[i].signalPointers == NULL_PTR(DebugSignalInfo **)) continue;
Vec<uint32> tempBreak;
uint32 count = 0;
for (uint32 j = 0; j < brokers[i].numSignals; j++) {
@@ -549,12 +593,14 @@ void DebugService::UpdateBrokersBreakStatus() {
}
if (brokers[i].activeMutex)
brokers[i].activeMutex->FastLock();
// FIX #2: O(1) Swap — old array freed after the lock is released.
if (brokers[i].breakIndices)
*(brokers[i].breakIndices) = tempBreak;
brokers[i].breakIndices->Swap(tempBreak);
if (brokers[i].anyBreakFlag)
*(brokers[i].anyBreakFlag) = (count > 0);
if (brokers[i].activeMutex)
brokers[i].activeMutex->FastUnLock();
// tempBreak freed here, outside the lock.
}
}
@@ -582,50 +628,143 @@ ErrorManagement::ErrorType DebugService::Server(ExecutionInfo &info) {
// The MARTe2 framework calls Execute() in a loop; each call should do
// one unit of work and return so the framework can check for Stop().
// This replaces the old internal infinite-while pattern.
//
// FIX #2: activeClient is guarded by clientMutex.
// Rule: lock clientMutex only when ASSIGNING the pointer (including to NULL).
// Do NOT hold it across blocking I/O (Read/Write) — that would stall any future
// thread trying to grab the lock to check whether there is a live client.
// The command handler (HandleCommand) receives the pointer by value so it is
// safe to call without holding the lock.
// Helper: current time in milliseconds (used for rate limiting and idle timeout)
uint64 nowMs = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1000.0);
if (activeClient == NULL_PTR(BasicTCPSocket *)) {
// Wait briefly for a new connection; return so the framework loop can
// check if Stop() was requested between calls.
BasicTCPSocket *newClient = tcpServer.WaitConnection(TimeoutType(100));
if (newClient != NULL_PTR(BasicTCPSocket *)) {
activeClient = newClient;
clientMutex.FastLock();
activeClient = newClient; // publish pointer — visible to other threads after unlock
clientMutex.FastUnLock();
// FIX #6/#8: reset per-connection state when a new client connects
cmdCountInWindow = 0u;
cmdWindowStartMs = nowMs;
lastDataTimeMs = nowMs;
}
} else {
// Check if client is still connected
if (!activeClient->IsConnected()) {
// FIX #8: idle-timeout check — runs every Execute() invocation even when
// no data arrives, so a client that half-sends a command and goes silent
// cannot hold the slot open indefinitely.
if (nowMs - lastDataTimeMs > CLIENT_IDLE_TIMEOUT_MS) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"Server: TCP client idle for >%u ms — closing connection.", CLIENT_IDLE_TIMEOUT_MS);
inputBuffer = ""; // FIX #10: discard carry-over on disconnect
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
cmdCountInWindow = 0u;
} else if (!activeClient->IsConnected()) {
// Check if client is still connected
inputBuffer = ""; // FIX #10
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
} else {
char buffer[1024];
uint32 size = 1024;
if (activeClient->Read(buffer, size)) {
if (size > 0) {
// Process each line separately
char *ptr = buffer;
char *end = buffer + size;
while (ptr < end) {
char *newline = (char *)memchr(ptr, '\n', end - ptr);
if (!newline) {
break;
lastDataTimeMs = nowMs; // FIX #8: refresh idle timestamp
// FIX #6: slide rate-limit window
if (nowMs - cmdWindowStartMs >= 1000u) {
cmdWindowStartMs = nowMs;
cmdCountInWindow = 0u;
}
// FIX #10: guard against a client that never sends a newline,
// which would grow inputBuffer without bound.
if (inputBuffer.Size() + (uint32)size > INPUT_BUFFER_MAX) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"Server: input buffer overflow (>%u bytes without newline) "
"— disconnecting.", INPUT_BUFFER_MAX);
inputBuffer = "";
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
cmdCountInWindow = 0u;
} else {
// FIX #10: accumulate data; parse only complete (newline-terminated)
// commands so that commands split across TCP segments are assembled
// correctly before dispatch.
(void)inputBuffer.Seek(inputBuffer.Size());
uint32 writeSize = (uint32)size;
inputBuffer.Write(buffer, writeSize);
const char8 *raw = inputBuffer.Buffer();
uint32 total = (uint32)inputBuffer.Size();
uint32 lineStart = 0u;
bool rateLimitExceeded = false;
for (uint32 pos = 0u; pos < total && !rateLimitExceeded; pos++) {
if (raw[pos] != '\n') continue;
uint32 len = pos - lineStart;
// Trim trailing CR
if (len > 0u && raw[lineStart + len - 1u] == '\r') len--;
if (len > 0u) {
cmdCountInWindow++;
if (cmdCountInWindow > CMD_RATE_LIMIT) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"Server: client exceeded rate limit (%u cmd/s) — disconnecting.",
CMD_RATE_LIMIT);
rateLimitExceeded = true;
break;
}
StreamString command;
uint32 cmdLen = len;
command.Write(raw + lineStart, cmdLen);
HandleCommand(command, activeClient);
}
lineStart = pos + 1u;
}
*newline = '\0';
// Skip carriage return if present
if (newline > ptr && *(newline - 1) == '\r')
*(newline - 1) = '\0';
StreamString command;
uint32 len = (uint32)(newline - ptr);
command.Write(ptr, len);
if (command.Size() > 0) {
HandleCommand(command, activeClient);
if (rateLimitExceeded) {
inputBuffer = "";
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
cmdCountInWindow = 0u;
} else {
// Save the incomplete line (bytes after the last '\n') for the
// next Read() — they form the start of the next command.
StreamString newInputBuffer;
if (lineStart < total) {
uint32 remLen = total - lineStart;
newInputBuffer.Write(raw + lineStart, remLen);
}
inputBuffer = newInputBuffer;
}
ptr = newline + 1;
}
}
} else {
// Read failed (client disconnected or error), clean up
inputBuffer = ""; // FIX #10
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
}
}
}
@@ -654,19 +793,63 @@ ErrorManagement::ErrorType DebugService::Streamer(ExecutionInfo &info) {
HighResolutionTimer::Period() * 1000000.0);
void *address = NULL_PTR(void *);
if (monitoredSignals[i].dataSource->GetSignalMemoryBuffer(monitoredSignals[i].signalIdx, 0, address)) {
// FIX #11: ProcessSignal() (called from RT broker threads) serialises
// its traceBuffer.Push() under tracePushMutex. The Streamer's monitored-
// signal path previously skipped that lock, creating a multi-producer race
// between the RT threads and the Streamer thread.
//
// Lock order: mutex (always held here) → tracePushMutex.
// RT broker threads only ever acquire tracePushMutex (never mutex), so
// this ordering introduces no deadlock risk.
tracePushMutex.FastLock();
traceBuffer.Push(monitoredSignals[i].internalID, ts, (uint8 *)address, monitoredSignals[i].size);
tracePushMutex.FastUnLock();
}
}
}
mutex.FastUnLock();
// Drain ring buffer into UDP packet(s)
// Drain ring buffer into UDP packet(s).
//
// FIX #1: Two-level bounds checking to prevent buffer overflow.
//
// Level 1 — per-sample maximum size:
// sampleData is SAMPLE_BUF_SIZE bytes. Pop() enforces this via maxSize so
// the local buffer can never be overrun by Pop() itself.
//
// Level 2 — assembly-buffer hard limit:
// After a flush, streamerPacketOffset resets to sizeof(TraceHeader). The
// next sample occupies (sizeof(TraceHeader) + 16 + size) bytes. With
// SAMPLE_BUF_SIZE = 1024 that is at most ~1060 bytes — well within
// STREAMER_BUFFER_SIZE = 4096. The explicit guard below catches the case
// where SAMPLE_BUF_SIZE or STREAMER_MTU are later changed carelessly so
// that a single sample could still overflow the buffer.
static const uint32 SAMPLE_BUF_SIZE = 1024u;
// Compile-time sanity: one full header + per-sample header + max sample must fit
// inside the assembly buffer. If this fires, raise STREAMER_BUFFER_SIZE or
// lower SAMPLE_BUF_SIZE.
// (C++98-compatible static assert via sizeof a negative-size array)
typedef char StaticAssert_StreamerBufferTooSmall[
(sizeof(TraceHeader) + 16u + SAMPLE_BUF_SIZE <= STREAMER_BUFFER_SIZE) ? 1 : -1];
(void)sizeof(StaticAssert_StreamerBufferTooSmall); // suppress unused-typedef warning
uint32 id, size;
uint64 ts;
uint8 sampleData[1024];
uint8 sampleData[SAMPLE_BUF_SIZE];
bool hasData = false;
while (traceBuffer.Pop(id, ts, sampleData, size, 1024)) {
while (traceBuffer.Pop(id, ts, sampleData, size, SAMPLE_BUF_SIZE)) {
hasData = true;
// Level 2 guard — should never fire given the static assert above, but
// defends against future changes that widen SAMPLE_BUF_SIZE or shrink
// STREAMER_BUFFER_SIZE without updating the other.
if (size > SAMPLE_BUF_SIZE || streamerPacketOffset + 16u + size > STREAMER_BUFFER_SIZE) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"Streamer: sample size %u would overflow assembly buffer (%u bytes used of %u) "
"— sample dropped.", size, streamerPacketOffset, STREAMER_BUFFER_SIZE);
continue;
}
if (streamerPacketOffset == 0u) {
TraceHeader header;
header.magic = 0xDA7A57AD;
@@ -676,7 +859,8 @@ ErrorManagement::ErrorType DebugService::Streamer(ExecutionInfo &info) {
memcpy(streamerPacketBuffer, &header, sizeof(TraceHeader));
streamerPacketOffset = sizeof(TraceHeader);
}
if (streamerPacketOffset + 16u + size > 1400u) {
// Flush the current packet when adding this sample would exceed the MTU.
if (streamerPacketOffset + 16u + size > STREAMER_MTU) {
uint32 toWrite = streamerPacketOffset;
(void)udpSocket.Write((char8 *)streamerPacketBuffer, toWrite);
TraceHeader header;
@@ -705,8 +889,8 @@ ErrorManagement::ErrorType DebugService::Streamer(ExecutionInfo &info) {
return ErrorManagement::NoError;
}
bool DebugService::GetFullObjectName(const Object &obj,
StreamString &fullPath) {
bool DebugServiceI::GetFullObjectName(const Object &obj,
StreamString &fullPath) {
fullPath = "";
if (FindPathInContainer(ObjectRegistryDatabase::Instance(), &obj, fullPath)) {
return true;
@@ -893,15 +1077,42 @@ void DebugService::HandleCommand(StreamString cmd, BasicTCPSocket *client) {
const char8 *eq = StringHelper::SearchChar(line.Buffer(), '=');
if (eq != NULL_PTR(const char8 *)) {
uint32 eqPos = (uint32)(eq - line.Buffer());
(void)line.Seek(0u);
char8 keyBuf[256] = {'\0'};
// FIX #7: Enforce buffer bounds before reading key/value.
// A key longer than keyBuf would be silently truncated by
// Read(), producing a mismatched key (e.g. "LongKe" instead
// of "LongKey") that causes CDB.Write() to inject a garbled
// parameter. Skip the entire line if the key overflows —
// the CDB write would be nonsensical anyway.
// Values are safely capped: the worst case is a truncated
// value, which the target GAM can detect and reject.
static const uint32 KEY_BUF_SIZE = 256u;
static const uint32 VAL_BUF_SIZE = 1024u;
if (eqPos >= KEY_BUF_SIZE) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"MSG: key length %u exceeds buffer (%u) — line skipped.",
eqPos, KEY_BUF_SIZE);
line = "";
continue;
}
(void)line.Seek(0u);
char8 keyBuf[KEY_BUF_SIZE];
MemoryOperationsHelper::Set(keyBuf, '\0', KEY_BUF_SIZE);
uint32 keyReadSize = eqPos;
(void)line.Read(keyBuf, keyReadSize);
(void)line.Seek(eqPos + 1u);
uint32 valLen = (uint32)(line.Size() - eqPos - 1u);
char8 valBuf[1024] = {'\0'};
// Truncate silently to VAL_BUF_SIZE - 1 (leave room for '\0')
if (valLen >= VAL_BUF_SIZE) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"MSG: value length %u truncated to %u for key '%s'.",
valLen, VAL_BUF_SIZE - 1u, keyBuf);
valLen = VAL_BUF_SIZE - 1u;
}
char8 valBuf[VAL_BUF_SIZE];
MemoryOperationsHelper::Set(valBuf, '\0', VAL_BUF_SIZE);
(void)line.Read(valBuf, valLen);
// Trim trailing whitespace from value
@@ -1418,6 +1629,25 @@ uint32 DebugService::ForceSignal(const char8 *name, const char8 *valueStr) {
SuffixMatch(aliases[i].name.Buffer(), name)) {
DebugSignalInfo *s = signals[aliases[i].signalIndex];
// FIX #4: Guard against writing past the end of forcedValue[1024].
// TypeConvert writes (byteSize * numberOfElements) bytes into forcedValue.
// For large arrays (e.g. 512 float64s = 4096 bytes) this would silently
// overflow the 1024-byte buffer and corrupt adjacent struct members.
//
// TypeDescriptor::numberOfBits is the element size in bits; integer divide
// by 8 gives bytes. For structured/opaque types numberOfBits may be 0 —
// those are also skipped because we cannot determine the layout safely.
uint32 elemBytes = (uint32)(s->type.numberOfBits) / 8u;
uint32 totalBytes = elemBytes * s->numberOfElements;
if (elemBytes == 0u || totalBytes > (uint32)sizeof(s->forcedValue)) {
REPORT_ERROR_STATIC(ErrorManagement::Warning,
"ForceSignal: signal '%s' requires %u bytes but forcedValue buffer is "
"only %u bytes — force request ignored.",
s->name.Buffer(), totalBytes, (uint32)sizeof(s->forcedValue));
continue;
}
s->isForcing = true;
AnyType dest(s->type, 0u, s->forcedValue);
AnyType source(CharString, 0u, valueStr);
@@ -1536,7 +1766,18 @@ void DebugService::GetSignalValue(const char8 *name, BasicTCPSocket *client) {
TypeDescriptor td = sig->type;
uint32 nElem = sig->numberOfElements;
uint32 byteSize = (td.numberOfBits > 0u) ? (td.numberOfBits / 8u) : 1u;
// FIX #9: cap element count before computing byte totals.
// Without a cap, a 1M-element uint8 signal would produce a multi-MB
// comma-separated response, exhausting heap and blocking the Server thread.
// GET_VALUE_MAX_ELEMENTS = 256 gives a ~4 KB worst-case JSON string.
bool truncated = (nElem > GET_VALUE_MAX_ELEMENTS);
if (truncated) {
nElem = GET_VALUE_MAX_ELEMENTS;
}
uint32 totalBytes = byteSize * nElem;
// Secondary cap: even after the element limit, ensure we never read more
// than 1024 bytes from the RT memory region.
if (totalBytes > 1024u) { totalBytes = 1024u; nElem = totalBytes / byteSize; }
// Copy bytes while holding the mutex to avoid data races with the RT thread
uint8 localBuf[1024];
@@ -1582,7 +1823,10 @@ void DebugService::GetSignalValue(const char8 *name, BasicTCPSocket *client) {
vp++;
}
resp += "\", \"Elements\": ";
resp.Printf("%u}\nOK VALUE\n", nElem);
// Include the capped element count and a Truncated flag so the client can
// distinguish "this is the full signal" from "there are more elements".
resp.Printf("%u, \"Truncated\": %s}\nOK VALUE\n",
nElem, truncated ? "true" : "false");
uint32 s = resp.Size();
(void)client->Write(resp.Buffer(), s);
}