MARTe-Integrated-Components/Source/Components/Interfaces/TCPStreamService/TCPStreamService.h

#ifndef TCPSTREAMSERVICE_H_
#define TCPSTREAMSERVICE_H_

/**
 * @file TCPStreamService.h
 * @brief One in-process TCP streaming server shared by any number of
 * TCPStreamer DataSources, served by a pool of worker threads.
 *
 * @details Declared once, at the root of the configuration:
 * <pre>
 * +StreamService = {
 *     Class   = TCPStreamService
 *     Port    = 44500
 *     Workers = 2
 * }
 * </pre>
 * and referenced from any number of DataSources:
 * <pre>
 * +Fast = { Class = TCPStreamer  Service = StreamService  StreamName = Fast ... }
 * +Slow = { Class = TCPStreamer  Service = StreamService  StreamName = Slow ... }
 * </pre>
 *
 * Wire protocol: plain UDPS framing over TCP, exactly like TCPSServer, plus one
 * client-to-server frame. Right after connecting, a client sends a SUBSCRIBE
 * frame (UDPS_TYPE_SUBSCRIBE, payload = stream name, 1..255 bytes). The service
 * then sends that stream's CONFIG followed by its DATA frames. One connection =
 * one stream. A connection that does not subscribe within HandshakeTimeout
 * seconds, or that names an unknown stream, is closed (clients retry).
 *
 * Threads: one acceptor plus `Workers` I/O threads. A connection lives on one
 * worker. Producer threads never touch sockets: they copy each frame into every
 * subscriber's bounded queue and poke the owning worker through a self-pipe, so
 * a slow client only loses its own frames and never delays a producer or
 * another client.
 *
 * @par Parameters (all optional)
 * | Parameter          | Type   | Default | Description |
 * |--------------------|--------|---------|-------------|
 * | Port               | uint32 | 44500   | TCP listen port; 0 = ephemeral (see GetBoundPort()). |
 * | Workers            | uint32 | 2       | I/O threads (1..32). |
 * | MaxClients         | uint32 | 64      | Simultaneous connections (1..1024). |
 * | MaxStreams         | uint32 | 32      | Registered streams (1..256). |
 * | ClientQueueDepth   | uint32 | 8       | DATA frames queued per client (1..4096). |
 * | ClientStallTimeout | uint32 | 5       | Seconds without progress before eviction; 0 disables. |
 * | HandshakeTimeout   | uint32 | 5       | Seconds allowed to send SUBSCRIBE. |
 * | SendBufferSize     | uint32 | 0       | SO_SNDBUF for client sockets; 0 keeps the OS default. |
 * | CPUMask            | uint32 | 0xFFFFFFFF | Affinity of the service threads. |
 * | StackSize          | uint32 | 65536   | Stack of the service threads. |
 */

#include "EmbeddedServiceMethodBinderI.h"
#include "ExecutionInfo.h"
#include "FastPollingMutexSem.h"
#include "MutexSem.h"
#include "Object.h"
#include "SingleThreadService.h"
#include "StructuredDataI.h"
#include "TCPSStreamTapI.h"
#include "UDPSProtocol.h"

namespace MARTe {

class TCPSWorker;
class TCPSServiceStream;
struct TCPSConn;

/** Longest stream name accepted in a SUBSCRIBE frame. */
static const uint32 TCPS_MAX_STREAM_NAME = 255u;
/** Most streams one connection may multiplex. */
static const uint32 TCPS_MAX_CONN_STREAMS = UDPS_TCP_MAX_STREAMS;
/** Longest SUBSCRIBE payload: a comma separated list of stream names. */
static const uint32 TCPS_MAX_SUBSCRIBE_PAYLOAD = UDPS_TCP_MAX_SUBSCRIBE_PAYLOAD;

class TCPStreamService : public Object, public EmbeddedServiceMethodBinderI {
public:
    CLASS_REGISTER_DECLARATION()

    TCPStreamService();
    virtual ~TCPStreamService();

    /** Reads the parameters, opens the listening socket and starts the threads. */
    virtual bool Initialise(StructuredDataI &data);

    /** Acceptor thread entry point; do not call directly. */
    virtual ErrorManagement::ErrorType Execute(ExecutionInfo &info);

    /** @return the port actually bound. */
    uint16 GetBoundPort() const;
    /** @return subscribed connections across all streams. */
    uint32 GetClientCount() const;
    uint32 GetStreamCount() const;
    uint32 GetWorkerCount() const;
    /** @return DATA frames dropped for slow clients, summed over registered streams. */
    uint64 GetClientDrops() const;
    /** @return connections closed by the stall timeout (all time). */
    uint32 GetStallEvictions() const;
    /** @return connections closed for a bad / unknown SUBSCRIBE (all time). */
    uint32 GetRejectedSubscriptions() const;

    /* ---- stream side ---- */
    /** @return false if a stream with this name is already registered or the table is full. */
    bool RegisterStream(TCPSServiceStream *s);
    /** Remove the stream and make its subscribers close. */
    void UnregisterStream(TCPSServiceStream *s);

    /**
     * Be told when streams register / unregister. The streams already registered are
     * replayed to the observer before AddObserver returns. Notifications are serialised
     * (and may nest on one thread), they never run under the topology lock, and once
     * RemoveObserver returns no callback is running or will start.
     * @return false if the observer table is full.
     */
    bool AddObserver(TCPSStreamObserverI *obs);
    void RemoveObserver(TCPSStreamObserverI *obs);

    /* ---- worker side (internal) ---- */
    /**
     * Bind a connection to the streams named in @p streamList (comma separated,
     * modified in place). The position in the list is the stream index stamped
     * on every frame. All-or-nothing: @return false if any name is unknown,
     * duplicated or malformed.
     */
    bool Attach(TCPSConn *c, char8 *streamList);
    void Detach(TCPSConn *c);
    void ConnClosed();
    void NoteEviction();
    void NoteRejected();
    uint32 QueueDepth() const;
    uint64 StallTicks() const;
    uint64 HandshakeTicks() const;
    /** Max subscribers a stream may hold. */
    uint32 MaxClients() const;

private:
    void Shutdown();

    uint32 port;
    uint16 boundPort;
    uint32 numWorkers;
    uint32 maxClients;
    uint32 maxStreams;
    uint32 queueDepth;
    uint32 stallTimeoutS;
    uint32 handshakeTimeoutS;
    uint32 sendBufferSize;
    uint32 cpuMask;
    uint32 stackSize;
    uint64 stallTicks;
    uint64 handshakeTicks;

    int listenFd;
    SingleThreadService acceptThread;
    bool acceptorStarted;
    TCPSWorker **workers;
    uint32 workersCreated;

    /** Most observers (TCPStreamJoin objects) one service can feed. */
    static const uint32 TCPS_MAX_OBSERVERS = 16u;
    MutexSem notifyMutex; ///< recursive; serialises observer calls with (un)registration
    TCPSStreamObserverI *observers[TCPS_MAX_OBSERVERS];
    uint32 numObservers;

    mutable FastPollingMutexSem topology; ///< guards streams[] and every TCPSConn::stream
    TCPSServiceStream **streams;
    uint32 numStreams;

    volatile int32 connCount;
    volatile int32 evictions;
    volatile int32 rejected;
};

} // namespace MARTe

#endif // TCPSTREAMSERVICE_H_