main.c

Go to the documentation of this file.

#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <poll.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
 
#include "safecomm/config/config.h"
#include "safecomm/ale/ale.h"
 
#include "rte/oal/netlink/rte_netlink.h"
#include "rte/posix_osadapter/rte_posix_osadapter.h"
 
#include "cs_config.h"

#define CS_DRAIN_MAX 256U
 
static volatile sig_atomic_t g_stop = 0;
static bool                  g_trace = false; /* COMMSERVER_TRACE in env */
 
static void on_signal(int sig)
{
    (void)sig;
    g_stop = 1;
}

static void trace_frame(const char *dir, const uint8_t *f, size_t n)
{
    if (!g_trace || (n < 4U))
    {
        return;
    }
    (void)fprintf(stderr, "commserver: TRACE %-12s kind=%3u train_id=%u nid_msg=%u\n",
                  dir, (unsigned int)f[0], (unsigned int)f[1], (unsigned int)f[2]);
}
 
static int tcp_listen(const char *host, uint16_t port)
{
    struct sockaddr_in addr;
    int                fd;
    int                one = 1;
 
    fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0)
    {
        return -1;
    }
    (void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
 
    (void)memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port   = htons(port);
    if (inet_pton(AF_INET, host, &addr.sin_addr) != 1)
    {
        addr.sin_addr.s_addr = htonl(INADDR_ANY);
    }
    if ((bind(fd, (const struct sockaddr *)&addr, sizeof(addr)) != 0) || (listen(fd, 1) != 0))
    {
        (void)close(fd);
        return -1;
    }
    return fd;
}
 
static rte_status_t netlink_connect(const char *host, uint16_t port, uint16_t msg_size,
                                     uint32_t timeout_ms, rte_netlink_storage_t *storage,
                                     rte_netlink_handle_t *out)
{
    rte_netlink_config_t nc;
 
    (void)memset(&nc, 0, sizeof(nc));
    nc.role               = RTE_NETLINK_ROLE_CONNECT;
    nc.host               = host;
    nc.port               = port;
    nc.message_size       = (size_t)msg_size;
    nc.connect_timeout_ms = (rte_duration_ms_t)timeout_ms;
    return rte_netlink_open(storage, &nc, out);
}

static void serve_client(int cfd, const cs_cfg_t *cfg)
{
    rte_netlink_storage_t west_storage;
    rte_netlink_storage_t east_storage;
    rte_netlink_handle_t  west = NULL;
    rte_netlink_handle_t  east = NULL;
    uint8_t                up[CS_MSG_MAX];
    uint8_t                dn[CS_MSG_MAX];
    size_t                 up_fill = 0U;
    const size_t           msz     = (size_t)cfg->message_size;
 
    if (netlink_connect(cfg->west_host, cfg->west_port, cfg->message_size,
                        cfg->connect_timeout_ms, &west_storage, &west) != RTE_STATUS_OK)
    {
        (void)fprintf(stderr, "commserver: WEST link (%s:%u) not established\n",
                      cfg->west_host, cfg->west_port);
        west = NULL;
    }
    if (netlink_connect(cfg->east_host, cfg->east_port, cfg->message_size,
                        cfg->connect_timeout_ms, &east_storage, &east) != RTE_STATUS_OK)
    {
        (void)fprintf(stderr, "commserver: EAST link (%s:%u) not established\n",
                      cfg->east_host, cfg->east_port);
        east = NULL;
    }
    if ((west == NULL) && (east == NULL))
    {
        (void)fprintf(stderr, "commserver: no RBC link — dropping train\n");
        return;
    }
    (void)fprintf(stderr, "commserver: train connected; relaying (west=%s east=%s)\n",
                  (west != NULL) ? "up" : "down", (east != NULL) ? "up" : "down");
 
    while (g_stop == 0)
    {
        struct pollfd pfd;
        int           pr;
        bool          closed = false;
        uint8_t       site;
 
        pfd.fd     = cfd;
        pfd.events = POLLIN;
        /* Short poll timeout: the loop must come back around quickly to
         * drain the RBC->train direction. A long wait here (an earlier
         * bug used poll.idle_ms) let RBC reply datagrams pile up in the
         * UDP socket buffer and get dropped, so MA / M24 / indication
         * traffic never reached the train - every MA-dependent scenario
         * test timed out. */
        pr = poll(&pfd, 1U, 1);
        if (pr < 0)
        {
            if (errno == EINTR)
            {
                continue;
            }
            break;
        }
 
        /* train -> RBC: drain EVERY whole frame available this pass, not
         * one partial read. recv(MSG_DONTWAIT) keeps write(cfd) blocking. */
        for (;;)
        {
            ssize_t n = recv(cfd, &up[up_fill], msz - up_fill, MSG_DONTWAIT);
 
            if (n == 0)
            {
                closed = true;
                break;
            }
            if (n < 0)
            {
                if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
                {
                    break;
                }
                if (errno == EINTR)
                {
                    continue;
                }
                closed = true;
                break;
            }
            up_fill += (size_t)n;
            if (up_fill == msz)
            {
                if (west != NULL)
                {
                    (void)rte_netlink_send(west, up, msz, 200U);
                    trace_frame("train->WEST", up, msz);
                }
                if (east != NULL)
                {
                    (void)rte_netlink_send(east, up, msz, 200U);
                    trace_frame("train->EAST", up, msz);
                }
                up_fill = 0U;
            }
        }
        if (closed)
        {
            break;
        }
 
        /* RBC -> train: drain each site until it has nothing left (bounded
         * so a flooding peer cannot starve the train read above). */
        for (site = 0U; (site < 2U) && !closed; site++)
        {
            rte_netlink_handle_t h     = (site == 0U) ? west : east;
            uint32_t              guard = 0U;
 
            while ((h != NULL) && (guard < CS_DRAIN_MAX))
            {
                guard++;
                /* 2 ms, not 0: a 0-timeout receive returns before a
                 * just-sent localhost reply datagram has landed, so the
                 * drain would keep missing it. With the drain loop the
                 * 2 ms is paid once per site per pass (when the queue is
                 * finally empty), not per datagram. */
                if (rte_netlink_receive(h, dn, msz, 2U) != RTE_STATUS_OK)
                {
                    break;
                }
                trace_frame((site == 0U) ? "WEST->train" : "EAST->train", dn, msz);
                if (write(cfd, dn, msz) != (ssize_t)msz)
                {
                    closed = true;
                    break;
                }
            }
        }
        if (closed)
        {
            break;
        }
    }
 
    if (west != NULL)
    {
        (void)rte_netlink_close(west);
    }
    if (east != NULL)
    {
        (void)rte_netlink_close(east);
    }
    (void)fprintf(stderr, "commserver: train disconnected\n");
}
 
int main(int argc, char **argv)
{
    cs_cfg_t cfg;
    int      lfd;
 
    if (argc < 2)
    {
        (void)fprintf(stderr, "usage: %s <commserver.conf>\n", argv[0]);
        return 2;
    }
    if (load_cfg(argv[1], &cfg) != 0)
    {
        return 1;
    }
 
    (void)signal(SIGINT, on_signal);
    (void)signal(SIGTERM, on_signal);
    (void)signal(SIGPIPE, SIG_IGN);
 
    g_trace = (getenv("COMMSERVER_TRACE") != NULL);
 
    if (rte_posix_osadapter_register_all() != RTE_STATUS_OK)
    {
        (void)fprintf(stderr, "commserver: rte_posix_osadapter_register_all failed\n");
        return 1;
    }
 
    lfd = tcp_listen(cfg.listen_host, cfg.listen_port);
    if (lfd < 0)
    {
        (void)fprintf(stderr, "commserver: cannot listen on %s:%u\n", cfg.listen_host, cfg.listen_port);
        return 1;
    }
    (void)fprintf(stderr, "commserver: listening on %s:%u — RBC west %s:%u / east %s:%u — msg %u\n",
                  cfg.listen_host, cfg.listen_port, cfg.west_host, cfg.west_port,
                  cfg.east_host, cfg.east_port, cfg.message_size);
 
    while (g_stop == 0)
    {
        int cfd = accept(lfd, NULL, NULL);
 
        if (cfd < 0)
        {
            if (errno == EINTR)
            {
                continue;
            }
            break;
        }
        serve_client(cfd, &cfg);
        (void)close(cfd);
    }
 
    (void)close(lfd);
    (void)fprintf(stderr, "commserver: stopped\n");
    return 0;
}