martino/marte-debug
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
marte-debug/Source/Components/Interfaces/DebugService/DebugService.cpp
T
Martino Ferrari 74816028a8 major improvements for big apps
2026-05-20 17:21:33 +02:00

478 lines
19 KiB
C++
Raw Permalink Blame History

#include "BasicTCPSocket.h"
#include "ConfigurationDatabase.h"
#include "DebugService.h"
#include "GlobalObjectsDatabase.h"
#include "HighResolutionTimer.h"
#include "LoggerService.h"
#include "Message.h"
#include "ObjectRegistryDatabase.h"
#include "ReferenceT.h"
#include "Sleep.h"
#include "StreamString.h"
#include "TcpLogger.h"
#include "Threads.h"
#include "TimeoutType.h"
namespace MARTe {
CLASS_REGISTER(DebugService, "1.0")
// C++98 ODR definitions for static constants
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::INPUT_BUFFER_MAX;
// Maximum data bytes that fit in one sample entry inside a datagram:
// STREAMER_BUFFER_SIZE(65535) - TraceHeader(20) - entry_header(16) = 65499
static const uint32 MAX_SAMPLE_DATA_SIZE = 65499u;
// ---------------------------------------------------------------------------
// Constructor / Destructor
// ---------------------------------------------------------------------------
DebugService::DebugService()
: DebugServiceBase(),
EmbeddedServiceMethodBinderI(),
binderServer(this, ServiceBinder::ServerType),
binderStreamer(this, ServiceBinder::StreamerType),
threadService(binderServer),
streamerService(binderStreamer) {
controlPort = 0u;
streamPort = 8081u;
logPort = 8082u;
streamIP = "127.0.0.1";
isServer = false;
suppressTimeoutLogs = true;
activeClient = NULL_PTR(BasicTCPSocket *);
streamerPacketOffset = 0u;
streamerSequenceNumber = 0u;
cmdCountInWindow = 0u;
cmdWindowStartMs = 0u;
lastDataTimeMs = 0u;
inputBuffer = "";
}
DebugService::~DebugService() {
if (DebugServiceI::GetInstance() == this) {
DebugServiceI::SetInstance(NULL_PTR(DebugServiceI *));
}
threadService.Stop();
streamerService.Stop();
tcpServer.Close();
udpSocket.Close();
if (activeClient != NULL_PTR(BasicTCPSocket *)) {
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
}
// signals owned by DebugServiceBase destructor
}
// ---------------------------------------------------------------------------
// Initialise
// ---------------------------------------------------------------------------
bool DebugService::Initialise(StructuredDataI &data) {
if (!ReferenceContainer::Initialise(data)) return false;
uint32 port = 0u;
if (data.Read("ControlPort", port)) {
controlPort = (uint16)port;
} else {
(void)data.Read("TcpPort", port);
controlPort = (uint16)port;
}
if (controlPort > 0u) {
isServer = true;
DebugServiceI::SetInstance(this);
}
port = 8081u;
if (data.Read("StreamPort", port)) {
streamPort = (uint16)port;
} else {
(void)data.Read("UdpPort", port);
streamPort = (uint16)port;
}
port = 8082u;
if (data.Read("LogPort", port)) {
logPort = (uint16)port;
} else {
(void)data.Read("TcpLogPort", port);
logPort = (uint16)port;
}
StreamString tempIP;
if (data.Read("StreamIP", tempIP)) {
streamIP = tempIP;
} else {
streamIP = "127.0.0.1";
}
uint32 suppress = 1u;
if (data.Read("SuppressTimeoutLogs", suppress)) {
suppressTimeoutLogs = (suppress == 1u);
}
// Capture only the local subtree — do NOT call MoveToRoot() on the shared CDB.
(void)data.Copy(fullConfig);
if (isServer) {
if (!traceBuffer.Init(8 * 1024 * 1024)) return false;
PatchRegistry();
ConfigurationDatabase threadData;
threadData.Write("Timeout", (uint32)1000);
threadService.Initialise(threadData);
streamerService.Initialise(threadData);
if (!tcpServer.Open()) return false;
if (!tcpServer.Listen(controlPort)) return false;
if (!udpSocket.Open()) return false;
if (threadService.Start() != ErrorManagement::NoError) return false;
if (streamerService.Start() != ErrorManagement::NoError) return false;
}
return true;
}
// ---------------------------------------------------------------------------
// Transport config hook (called by RebuildConfigFromRegistry in base)
// ---------------------------------------------------------------------------
void DebugService::RebuildTransportConfig() {
const char8 *myName = GetName();
if (myName != NULL_PTR(const char8 *)) {
if (fullConfig.MoveRelative(myName)) {
(void)fullConfig.Write("ControlPort", static_cast<uint32>(controlPort));
(void)fullConfig.Write("UdpPort", static_cast<uint32>(streamPort));
(void)fullConfig.Write("LogPort", static_cast<uint32>(logPort));
if (streamIP.Size() > 0u)
(void)fullConfig.Write("StreamIP", streamIP.Buffer());
(void)fullConfig.MoveToAncestor(1u);
}
}
}
// ---------------------------------------------------------------------------
// SERVICE_INFO hook
// ---------------------------------------------------------------------------
void DebugService::GetServiceInfo(StreamString &out) {
out.Printf("OK SERVICE_INFO TCP_CTRL:%u UDP_STREAM:%u TCP_LOG:%u STATE:%s\n",
controlPort, streamPort, logPort,
isPaused ? "PAUSED" : "RUNNING");
}
// ---------------------------------------------------------------------------
// Execute / HandleMessage (framework boilerplate)
// ---------------------------------------------------------------------------
ErrorManagement::ErrorType DebugService::Execute(ExecutionInfo &info) {
(void)info;
return ErrorManagement::FatalError;
}
ErrorManagement::ErrorType DebugService::HandleMessage(ReferenceT<Message> &data) {
(void)data;
return ErrorManagement::NoError;
}
// ---------------------------------------------------------------------------
// InjectTcpLoggerIfNeeded
// ---------------------------------------------------------------------------
void DebugService::InjectTcpLoggerIfNeeded() {
if (logPort == 0u) return;
Reference existing = ObjectRegistryDatabase::Instance()->Find("LoggerService");
if (existing.IsValid()) {
ReferenceContainer *rc =
dynamic_cast<ReferenceContainer *>(existing.operator->());
if (rc != NULL_PTR(ReferenceContainer *)) {
for (uint32 i = 0u; i < rc->Size(); i++) {
ReferenceT<TcpLogger> child = rc->Get(i);
if (child.IsValid()) return; // already has a TcpLogger
}
}
}
ConfigurationDatabase lsCdb;
(void)lsCdb.Write("Class", "LoggerService");
uint32 cpus = 1u;
(void)lsCdb.Write("CPUs", cpus);
if (lsCdb.CreateRelative("+DebugConsumer")) {
(void)lsCdb.Write("Class", "TcpLogger");
uint32 p = static_cast<uint32>(logPort);
(void)lsCdb.Write("Port", p);
(void)lsCdb.MoveToAncestor(1u);
}
(void)lsCdb.MoveToRoot();
ReferenceT<LoggerService> ls(
"LoggerService", GlobalObjectsDatabase::Instance()->GetStandardHeap());
if (!ls.IsValid()) return;
ls->SetName("LoggerService");
if (!ls->Initialise(lsCdb)) return;
(void)ObjectRegistryDatabase::Instance()->Insert(ls);
}
// ---------------------------------------------------------------------------
// Server thread
// ---------------------------------------------------------------------------
ErrorManagement::ErrorType DebugService::Server(ExecutionInfo &info) {
if (info.GetStage() == ExecutionInfo::TerminationStage)
return ErrorManagement::NoError;
if (info.GetStage() == ExecutionInfo::StartupStage) {
serverThreadId = Threads::Id();
Sleep::MSec(500u);
InjectTcpLoggerIfNeeded();
return ErrorManagement::NoError;
}
uint64 nowMs = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1000.0);
if (activeClient == NULL_PTR(BasicTCPSocket *)) {
BasicTCPSocket *newClient = tcpServer.WaitConnection(TimeoutType(100));
if (newClient != NULL_PTR(BasicTCPSocket *)) {
clientMutex.FastLock();
activeClient = newClient;
clientMutex.FastUnLock();
cmdCountInWindow = 0u;
cmdWindowStartMs = nowMs;
lastDataTimeMs = nowMs;
}
} else {
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 = "";
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
cmdCountInWindow = 0u;
} else if (!activeClient->IsConnected()) {
inputBuffer = "";
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
} else {
char buffer[1024];
uint32 size = 1024u;
if (activeClient->Read(buffer, size)) {
if (size > 0u) {
lastDataTimeMs = nowMs;
if (nowMs - cmdWindowStartMs >= 1000u) {
cmdWindowStartMs = nowMs;
cmdCountInWindow = 0u;
}
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 {
(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;
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);
// Dispatch via base HandleCommand, write response to socket.
// Loop to handle partial writes for large responses.
StreamString out;
HandleCommand(command, out);
if (out.Size() > 0u) {
const char8 *wPtr = out.Buffer();
uint32 remaining = (uint32)out.Size();
lastDataTimeMs = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1000.0);
while (remaining > 0u) {
uint32 wrote = remaining;
if (!activeClient->Write(wPtr, wrote) || wrote == 0u) {
break;
}
wPtr += wrote;
remaining -= wrote;
lastDataTimeMs = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1000.0);
}
}
}
lineStart = pos + 1u;
}
if (rateLimitExceeded) {
inputBuffer = "";
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
cmdCountInWindow = 0u;
} else {
StreamString newInputBuffer;
if (lineStart < total) {
uint32 remLen = total - lineStart;
newInputBuffer.Write(raw + lineStart, remLen);
}
inputBuffer = newInputBuffer;
}
}
}
} else {
inputBuffer = "";
clientMutex.FastLock();
activeClient->Close();
delete activeClient;
activeClient = NULL_PTR(BasicTCPSocket *);
clientMutex.FastUnLock();
}
}
}
return ErrorManagement::NoError;
}
// ---------------------------------------------------------------------------
// Streamer thread (UDP binary telemetry)
// ---------------------------------------------------------------------------
ErrorManagement::ErrorType DebugService::Streamer(ExecutionInfo &info) {
if (info.GetStage() == ExecutionInfo::TerminationStage)
return ErrorManagement::NoError;
if (info.GetStage() == ExecutionInfo::StartupStage) {
streamerThreadId = Threads::Id();
return ErrorManagement::NoError;
}
InternetHost dest(streamPort, streamIP.Buffer());
(void)udpSocket.SetDestination(dest);
// Poll monitored signals and push to trace ring buffer
uint64 currentTimeMs = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1000.0);
mutex.FastLock();
for (uint32 i = 0u; i < monitoredSignals.GetNumberOfElements(); i++) {
if (currentTimeMs >= (monitoredSignals[i].lastPollTime +
monitoredSignals[i].periodMs)) {
monitoredSignals[i].lastPollTime = currentTimeMs;
uint64 ts = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1000000.0);
void *address = NULL_PTR(void *);
if (monitoredSignals[i].dataSource->GetSignalMemoryBuffer(
monitoredSignals[i].signalIdx, 0u, address)) {
tracePushMutex.FastLock();
traceBuffer.Push(monitoredSignals[i].internalID, ts,
(uint8 *)address, monitoredSignals[i].size);
tracePushMutex.FastUnLock();
}
}
}
mutex.FastUnLock();
// Drain ring buffer into UDP packet(s).
//
// Batching strategy:
// - Normal samples (entry ≤ STREAMER_MTU): accumulate into a packet and
// flush when the next sample would push it past STREAMER_MTU.
// - Oversized samples (entry > STREAMER_MTU): flush any pending packet
// first, then send the oversized sample as its own datagram (up to
// STREAMER_BUFFER_SIZE bytes; handled transparently by IP fragmentation
// on loopback / internal networks).
// - MAX_SAMPLE_DATA_SIZE (65499 bytes) is the hard cap; larger entries
// in the ring buffer are skipped by Pop() and discarded.
bool hasData = false;
uint32 id, size;
uint64 ts;
while (traceBuffer.Pop(id, ts, streamerSampleBuffer, size, MAX_SAMPLE_DATA_SIZE)) {
hasData = true;
const uint32 entryBytes = 16u + size; // [id:4][ts:8][size:4][data:size]
// If this entry would push the current packet past the MTU target,
// flush what we have before appending (unless the buffer is empty).
if (streamerPacketOffset > 0u &&
streamerPacketOffset + entryBytes > STREAMER_MTU) {
uint32 toWrite = streamerPacketOffset;
(void)udpSocket.Write((char8 *)streamerPacketBuffer, toWrite);
streamerPacketOffset = 0u;
}
// Open a new packet header if the buffer is empty.
if (streamerPacketOffset == 0u) {
TraceHeader hdr;
hdr.magic = 0xDA7A57ADu;
hdr.seq = streamerSequenceNumber++;
hdr.timestamp = HighResolutionTimer::Counter();
hdr.count = 0u;
memcpy(streamerPacketBuffer, &hdr, sizeof(TraceHeader));
streamerPacketOffset = (uint32)sizeof(TraceHeader);
}
// Append the sample entry.
memcpy(&streamerPacketBuffer[streamerPacketOffset], &id, 4u);
memcpy(&streamerPacketBuffer[streamerPacketOffset + 4u], &ts, 8u);
memcpy(&streamerPacketBuffer[streamerPacketOffset + 12u], &size, 4u);
memcpy(&streamerPacketBuffer[streamerPacketOffset + 16u], streamerSampleBuffer, size);
streamerPacketOffset += entryBytes;
((TraceHeader *)streamerPacketBuffer)->count++;
// Oversized single sample: send immediately rather than accumulating.
if (streamerPacketOffset > STREAMER_MTU) {
uint32 toWrite = streamerPacketOffset;
(void)udpSocket.Write((char8 *)streamerPacketBuffer, toWrite);
streamerPacketOffset = 0u;
}
}
// Flush any remaining partial packet.
if (streamerPacketOffset > 0u) {
uint32 toWrite = streamerPacketOffset;
(void)udpSocket.Write((char8 *)streamerPacketBuffer, toWrite);
streamerPacketOffset = 0u;
}
if (!hasData) Sleep::MSec(1u);
return ErrorManagement::NoError;
}
} // namespace MARTe
Reference in New Issue View Git Blame Copy Permalink