martino/MARTe-Integrated-Components
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
MARTe-Integrated-Components/Source/Components/Interfaces/TCPStreamService/TCPStreamService.cpp
T
Martino FerrariandClaude Opus 4.6 50d85363a3 feat(tcp): TCPStreamer DataSources, in-process TCPStreamService, stream multiplexing and TCPStreamJoin 
Add TCPStreamer/TCPStreamerClient (TCP transport behind the UDPSServerI/UDPSClientI
seam), an optional single in-process TCPStreamService serving many streams over
worker threads, multiplexed SUBSCRIBE (comma list, stream index in fragmentIdx),
and TCPStreamJoin, which merges several streams into one sample-and-hold stream
paced by a master. StreamHub takes a comma list of streams over one connection.
Includes GTests, E2E scenarios s58-s73 and docs.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-11-06 13:41:27 +01:00

503 lines
15 KiB
C++
Raw Permalink Blame History

/**
* @file TCPStreamService.cpp
* @brief Source file for class TCPStreamService
* @date 06/11/2026
* @author Martino Ferrari
*
* @copyright Copyright 2015 F4E | European Joint Undertaking for ITER and
* the Development of Fusion Energy ('Fusion for Energy').
* Licensed under the EUPL, Version 1.1 or - as soon they will be approved
* by the European Commission - subsequent versions of the EUPL (the "Licence")
* You may not use this work except in compliance with the Licence.
* You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl
*
* @warning Unless required by applicable law or agreed to in writing,
* software distributed under the Licence is distributed on an "AS IS"
* basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the Licence permissions and limitations under the Licence.
*/
#define DLL_API
#include "TCPStreamService.h"
#include "AdvancedErrorManagement.h"
#include "Atomic.h"
#include "HighResolutionTimer.h"
#include "TCPSConn.h"
#include "TCPSServiceStream.h"
#include "TCPSWorker.h"
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
namespace MARTe {
static const int TCPS_ACCEPT_SLICE_MS = 50;
TCPStreamService::TCPStreamService()
: Object(), EmbeddedServiceMethodBinderI(), port(44500u), boundPort(0u),
numWorkers(2u), maxClients(64u), maxStreams(32u), queueDepth(8u),
stallTimeoutS(5u), handshakeTimeoutS(5u), sendBufferSize(0u),
cpuMask(0xFFFFFFFFu), stackSize(65536u), stallTicks(0u), handshakeTicks(0u),
listenFd(-1), acceptThread(*this), acceptorStarted(false),
workers(NULL_PTR(TCPSWorker **)), workersCreated(0u),
numObservers(0u),
streams(NULL_PTR(TCPSServiceStream **)), numStreams(0u),
connCount(0), evictions(0), rejected(0) {
(void)notifyMutex.Create(true);
for (uint32 i = 0u; i < TCPS_MAX_OBSERVERS; i++) {
observers[i] = NULL_PTR(TCPSStreamObserverI *);
}
}
TCPStreamService::~TCPStreamService() {
Shutdown();
}
void TCPStreamService::Shutdown() {
if (acceptorStarted) {
(void)acceptThread.Stop();
acceptorStarted = false;
}
for (uint32 i = 0u; i < workersCreated; i++) {
workers[i]->Stop();
}
if (listenFd >= 0) {
(void)close(listenFd);
listenFd = -1;
}
for (uint32 i = 0u; i < workersCreated; i++) {
delete workers[i];
}
workersCreated = 0u;
if (workers != NULL_PTR(TCPSWorker **)) {
delete[] workers;
workers = NULL_PTR(TCPSWorker **);
}
if (streams != NULL_PTR(TCPSServiceStream **)) {
delete[] streams;
streams = NULL_PTR(TCPSServiceStream **);
}
numStreams = 0u;
}
bool TCPStreamService::Initialise(StructuredDataI &data) {
bool ok = Object::Initialise(data);
port = 44500u;
(void)data.Read("Port", port);
if (port > 65535u) {
REPORT_ERROR(ErrorManagement::ParametersError, "Port %u out of range.", port);
ok = false;
}
numWorkers = 2u;
(void)data.Read("Workers", numWorkers);
if ((numWorkers == 0u) || (numWorkers > 32u)) {
REPORT_ERROR(ErrorManagement::ParametersError, "Workers %u must be in 1..32.", numWorkers);
ok = false;
}
maxClients = 64u;
(void)data.Read("MaxClients", maxClients);
if ((maxClients == 0u) || (maxClients > 1024u)) {
REPORT_ERROR(ErrorManagement::ParametersError, "MaxClients %u must be in 1..1024.", maxClients);
ok = false;
}
maxStreams = 32u;
(void)data.Read("MaxStreams", maxStreams);
if ((maxStreams == 0u) || (maxStreams > 256u)) {
REPORT_ERROR(ErrorManagement::ParametersError, "MaxStreams %u must be in 1..256.", maxStreams);
ok = false;
}
queueDepth = 8u;
(void)data.Read("ClientQueueDepth", queueDepth);
if ((queueDepth == 0u) || (queueDepth > 4096u)) {
REPORT_ERROR(ErrorManagement::ParametersError, "ClientQueueDepth %u must be in 1..4096.", queueDepth);
ok = false;
}
stallTimeoutS = 5u;
(void)data.Read("ClientStallTimeout", stallTimeoutS);
handshakeTimeoutS = 5u;
(void)data.Read("HandshakeTimeout", handshakeTimeoutS);
if (handshakeTimeoutS == 0u) {
REPORT_ERROR(ErrorManagement::ParametersError, "HandshakeTimeout must be at least 1 second.");
ok = false;
}
sendBufferSize = 0u;
(void)data.Read("SendBufferSize", sendBufferSize);
(void)data.Read("CPUMask", cpuMask);
(void)data.Read("StackSize", stackSize);
stallTicks = static_cast<uint64>(stallTimeoutS) * HighResolutionTimer::Frequency();
handshakeTicks = static_cast<uint64>(handshakeTimeoutS) * HighResolutionTimer::Frequency();
if (!ok) {
return false;
}
/* Listening socket. */
int fd = socket(AF_INET, SOCK_STREAM, 0);
ok = (fd >= 0);
if (!ok) {
REPORT_ERROR(ErrorManagement::FatalError, "socket() failed: %s.", strerror(errno));
}
if (ok) {
int one = 1;
(void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_ANY);
addr.sin_port = htons(static_cast<uint16>(port));
if (bind(fd, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr)) != 0) {
REPORT_ERROR(ErrorManagement::FatalError, "bind(%u) failed: %s.", port, strerror(errno));
ok = false;
}
}
if (ok && (listen(fd, 32) != 0)) {
REPORT_ERROR(ErrorManagement::FatalError, "listen() failed: %s.", strerror(errno));
ok = false;
}
if (ok) {
int flags = fcntl(fd, F_GETFL, 0);
ok = (flags >= 0) && (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
}
if (ok) {
struct sockaddr_in addr;
socklen_t len = static_cast<socklen_t>(sizeof(addr));
memset(&addr, 0, sizeof(addr));
boundPort = static_cast<uint16>(port);
if (getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &len) == 0) {
boundPort = ntohs(addr.sin_port);
}
}
if (!ok) {
if (fd >= 0) {
(void)close(fd);
}
return false;
}
listenFd = fd;
streams = new TCPSServiceStream *[maxStreams];
workers = new TCPSWorker *[numWorkers];
for (uint32 i = 0u; (i < numWorkers) && ok; i++) {
workers[i] = new TCPSWorker(*this, maxClients);
workersCreated = i + 1u;
ok = workers[i]->Start(cpuMask, stackSize);
if (!ok) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not start worker %u.", i);
}
}
if (ok) {
acceptThread.SetCPUMask(cpuMask);
acceptThread.SetStackSize(stackSize);
ok = (acceptThread.Start() == ErrorManagement::NoError);
acceptorStarted = ok;
if (!ok) {
REPORT_ERROR(ErrorManagement::FatalError, "Could not start the acceptor thread.");
}
}
if (!ok) {
Shutdown();
}
return ok;
}
ErrorManagement::ErrorType TCPStreamService::Execute(ExecutionInfo &info) {
if (info.GetStage() != ExecutionInfo::MainStage) {
return ErrorManagement::NoError;
}
struct pollfd p;
p.fd = listenFd;
p.events = POLLIN;
p.revents = 0;
if (poll(&p, 1, TCPS_ACCEPT_SLICE_MS) <= 0) {
return ErrorManagement::NoError;
}
for (;;) {
int fd = accept(listenFd, NULL_PTR(struct sockaddr *), NULL_PTR(socklen_t *));
if (fd < 0) {
break;
}
if (static_cast<uint32>(connCount) >= maxClients) {
REPORT_ERROR(ErrorManagement::Warning,
"MaxClients (%u) reached; connection refused.", maxClients);
(void)close(fd);
continue;
}
int flags = fcntl(fd, F_GETFL, 0);
if ((flags < 0) || (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)) {
(void)close(fd);
continue;
}
int one = 1;
(void)setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
if (sendBufferSize > 0u) {
int sz = static_cast<int>(sendBufferSize);
(void)setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz));
}
TCPSWorker *best = workers[0];
for (uint32 i = 1u; i < workersCreated; i++) {
if (workers[i]->Load() < best->Load()) {
best = workers[i];
}
}
TCPSConn *c = new TCPSConn(fd, best, queueDepth,
HighResolutionTimer::Counter() + handshakeTicks);
Atomic::Increment(&connCount);
if (!best->AddIncoming(c)) {
delete c;
Atomic::Decrement(&connCount);
}
}
return ErrorManagement::NoError;
}
uint16 TCPStreamService::GetBoundPort() const {
return boundPort;
}
uint32 TCPStreamService::GetClientCount() const {
uint32 n = 0u;
topology.FastLock();
for (uint32 i = 0u; i < numStreams; i++) {
n += streams[i]->GetClientCount();
}
topology.FastUnLock();
return n;
}
uint32 TCPStreamService::GetStreamCount() const {
return numStreams;
}
uint32 TCPStreamService::GetWorkerCount() const {
return workersCreated;
}
uint64 TCPStreamService::GetClientDrops() const {
uint64 n = 0u;
topology.FastLock();
for (uint32 i = 0u; i < numStreams; i++) {
n += streams[i]->GetClientDrops();
}
topology.FastUnLock();
return n;
}
uint32 TCPStreamService::GetStallEvictions() const {
return static_cast<uint32>(evictions);
}
uint32 TCPStreamService::GetRejectedSubscriptions() const {
return static_cast<uint32>(rejected);
}
bool TCPStreamService::RegisterStream(TCPSServiceStream *s) {
bool ok = true;
// Held across the registration so observers see (un)registrations in order.
(void)notifyMutex.Lock();
topology.FastLock();
if (numStreams >= maxStreams) {
REPORT_ERROR(ErrorManagement::ParametersError,
"Cannot register stream '%s': MaxStreams (%u) reached.",
s->GetStreamName(), maxStreams);
ok = false;
}
for (uint32 i = 0u; (i < numStreams) && ok; i++) {
if (strcmp(streams[i]->GetStreamName(), s->GetStreamName()) == 0) {
REPORT_ERROR(ErrorManagement::ParametersError,
"A stream named '%s' is already registered.", s->GetStreamName());
ok = false;
}
}
if (ok) {
streams[numStreams] = s;
numStreams++;
}
topology.FastUnLock();
if (ok) {
for (uint32 i = 0u; i < numObservers; i++) {
observers[i]->OnStreamRegistered(s);
}
}
(void)notifyMutex.UnLock();
return ok;
}
void TCPStreamService::UnregisterStream(TCPSServiceStream *s) {
(void)notifyMutex.Lock();
bool wasRegistered = false;
topology.FastLock();
for (uint32 i = 0u; i < numStreams; i++) {
if (streams[i] == s) {
streams[i] = streams[numStreams - 1u];
numStreams--;
wasRegistered = true;
break;
}
}
s->DropAllConns();
topology.FastUnLock();
if (wasRegistered) {
for (uint32 i = 0u; i < numObservers; i++) {
observers[i]->OnStreamUnregistered(s);
}
}
(void)notifyMutex.UnLock();
}
bool TCPStreamService::AddObserver(TCPSStreamObserverI *obs) {
bool ok = false;
(void)notifyMutex.Lock();
if (numObservers < TCPS_MAX_OBSERVERS) {
observers[numObservers] = obs;
numObservers++;
ok = true;
// Replay: streams cannot (un)register while notifyMutex is held.
topology.FastLock();
const uint32 n = numStreams;
TCPSServiceStream **snap = new TCPSServiceStream *[(n > 0u) ? n : 1u];
for (uint32 i = 0u; i < n; i++) {
snap[i] = streams[i];
}
topology.FastUnLock();
for (uint32 i = 0u; i < n; i++) {
obs->OnStreamRegistered(snap[i]);
}
delete[] snap;
}
(void)notifyMutex.UnLock();
return ok;
}
void TCPStreamService::RemoveObserver(TCPSStreamObserverI *obs) {
(void)notifyMutex.Lock();
for (uint32 i = 0u; i < numObservers; i++) {
if (observers[i] == obs) {
observers[i] = observers[numObservers - 1u];
numObservers--;
break;
}
}
(void)notifyMutex.UnLock();
}
bool TCPStreamService::Attach(TCPSConn *c, char8 *streamList) {
TCPSServiceStream *found[TCPS_MAX_CONN_STREAMS];
uint32 n = 0u;
bool ok = true;
topology.FastLock();
char8 *p = streamList;
while (ok) {
char8 *comma = strchr(p, ',');
if (comma != NULL_PTR(char8 *)) {
*comma = '\0';
}
const uint32 len = static_cast<uint32>(strlen(p));
TCPSServiceStream *match = NULL_PTR(TCPSServiceStream *);
if ((len == 0u) || (len > TCPS_MAX_STREAM_NAME) || (n >= TCPS_MAX_CONN_STREAMS)) {
ok = false;
}
for (uint32 i = 0u; ok && (i < numStreams) && (match == NULL_PTR(TCPSServiceStream *)); i++) {
if (strcmp(streams[i]->GetStreamName(), p) == 0) {
match = streams[i];
}
}
for (uint32 i = 0u; ok && (i < n); i++) {
if (found[i] == match) {
ok = false; // the same stream listed twice
}
}
if (ok && (match == NULL_PTR(TCPSServiceStream *))) {
ok = false;
}
if (ok) {
found[n] = match;
n++;
if (comma == NULL_PTR(char8 *)) {
break;
}
p = comma + 1;
}
}
if (ok) {
c->qMutex.FastLock();
c->q.Reserve(n);
c->qMutex.FastUnLock();
uint32 attached = 0u;
for (uint32 i = 0u; ok && (i < n); i++) {
ok = found[i]->AttachConn(c, static_cast<uint16>(i));
if (ok) {
attached++;
}
}
if (!ok) {
for (uint32 i = 0u; i < attached; i++) {
found[i]->DetachConn(c);
}
} else {
for (uint32 i = 0u; i < n; i++) {
c->streams[i] = found[i];
}
c->numStreams = n;
}
}
topology.FastUnLock();
return ok;
}
void TCPStreamService::Detach(TCPSConn *c) {
topology.FastLock();
for (uint32 i = 0u; i < c->numStreams; i++) {
if (c->streams[i] != NULL_PTR(TCPSServiceStream *)) {
c->streams[i]->DetachConn(c);
c->streams[i] = NULL_PTR(TCPSServiceStream *);
}
}
topology.FastUnLock();
}
void TCPStreamService::ConnClosed() {
Atomic::Decrement(&connCount);
}
void TCPStreamService::NoteEviction() {
Atomic::Increment(&evictions);
}
void TCPStreamService::NoteRejected() {
Atomic::Increment(&rejected);
}
uint32 TCPStreamService::QueueDepth() const {
return queueDepth;
}
uint64 TCPStreamService::StallTicks() const {
return stallTicks;
}
uint64 TCPStreamService::HandshakeTicks() const {
return handshakeTicks;
}
uint32 TCPStreamService::MaxClients() const {
return maxClients;
}
CLASS_REGISTER(TCPStreamService, "1.0")
} // namespace MARTe
Reference in new issue View Git Blame Copy Permalink