#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(controlPort)); (void)fullConfig.Write("UdpPort", static_cast(streamPort)); (void)fullConfig.Write("LogPort", static_cast(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 &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(existing.operator->()); if (rc != NULL_PTR(ReferenceContainer *)) { for (uint32 i = 0u; i < rc->Size(); i++) { ReferenceT 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(logPort); (void)lsCdb.Write("Port", p); (void)lsCdb.MoveToAncestor(1u); } (void)lsCdb.MoveToRoot(); ReferenceT 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(); while (remaining > 0u) { uint32 wrote = remaining; if (!activeClient->Write(wPtr, wrote) || wrote == 0u) { break; } wPtr += wrote; remaining -= wrote; } } } 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