gateway_c_common.c

Go to the documentation of this file.

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
 
#include "rte/app/appmanager/rte_appmanager.h"
#include "rte/core/rte_core_version.h"
#include "rte/oal/log/rte_log.h"
#include "rte/oal/memory/rte_mem_util.h"
#include "rte/posix_osadapter/rte_posix_osadapter_channel_service.h"
#include "rte/utils/safestate/rte_safestate.h"
 
#include "common/app_main_common.h"
#include "common/common_config.h"
#include "common/rbc_util_frame.h"
#include "gateway_c_common.h"
#include "gateway_c_selfstatus.h"
#include "gateway_c_config.h"
#include "gateway_c_types.h"
#include "gateway_c_util.h"
#include "rte_rbc2oo2_version.h"
 
static gateway_c_context_t          g_c_ctx;
static const gateway_c_service_def_t *g_svc;
 
static rte_status_t gateway_c_shutdown(void *context);
 
static rte_status_t resolve_channel_c(const char *name, rte_netlink_config_t *config, void *context)
{
    gateway_c_context_t *ctx = (gateway_c_context_t *)context;
 
    if ((name == NULL) || (config == NULL) || (ctx == NULL))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    if (gateway_c_config_find_channel(&ctx->config, name, config) == RTE_STATUS_OK)
    {
        /* The JSON lists every c-broker-* channel once, as the status service's LISTEN side. A train/il/ctc
         * gateway must instead DIAL its own broker channel (see the fallback below); without this override both
         * sides listened, no gateway ever reached the broker and the SUMMARY showed every gateway and sim down. */
        if ((g_svc != NULL) && (g_svc->broker_channel != NULL) && (strcmp(name, g_svc->broker_channel) == 0))
        {
            config->role = RTE_NETLINK_ROLE_CONNECT;
            config->host = "127.0.0.1";
        }
        return RTE_STATUS_OK;
    }
 
    rte_mem_set(config, 0, sizeof(*config));
    config->role = RTE_NETLINK_ROLE_LISTEN;
    config->host = NULL;
    config->message_size = RBC_ENVELOPE_WIRE_SIZE;
    config->connect_timeout_ms = ctx->config.timers.reconnect_attempt_timeout_ms;
 
    /* ADR-036: one link per {kind} - the Train and IL sim connections
     * carry every train, demuxed by the envelope's train_id (in A/B, not
     * here). No per-instance -%u suffix any more.
     * RBC_MSG_ERTMS_ENVELOPE (rbc_wire_types.h): the Train sim link's
     * three channels (this LISTEN one plus its two DIAL relay legs to
     * A/B below) all now carry the wider RBC_TRAIN_ENVELOPE_WIRE_SIZE
     * frame, not the default RBC_ENVELOPE_WIRE_SIZE set above - same
     * override pattern the UTIL/broker channels already use for
     * RBC_UTIL_FRAME_SIZE. gateway_c_train_handler.c is a pure byte-
     * blind relay either way (never inspects message content), but the
     * transport's own per-channel message_size still has to match what
     * actually gets sent on it. */
    if (strcmp(name, "c-sim-train") == 0)
    {
        config->message_size = RBC_TRAIN_ENVELOPE_WIRE_SIZE;
        config->port = site_config_c_port_for_train(ctx->site);
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "c-relay-train-a") == 0)
    {
        config->message_size = RBC_TRAIN_ENVELOPE_WIRE_SIZE;
        config->port = site_config_c_port_for_a(ctx->site, RTE_EXAMPLE_RELAY_KIND_TRAIN);
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "c-relay-train-b") == 0)
    {
        config->message_size = RBC_TRAIN_ENVELOPE_WIRE_SIZE;
        config->port = site_config_c_port_for_b(ctx->site, RTE_EXAMPLE_RELAY_KIND_TRAIN);
        return RTE_STATUS_OK;
    }
 
    if (strcmp(name, "c-sim-il") == 0)
    {
        config->port = site_config_c_port_for_il(ctx->site);
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "c-relay-il-a") == 0)
    {
        config->port = site_config_c_port_for_a(ctx->site, RTE_EXAMPLE_RELAY_KIND_IL);
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "c-relay-il-b") == 0)
    {
        config->port = site_config_c_port_for_b(ctx->site, RTE_EXAMPLE_RELAY_KIND_IL);
        return RTE_STATUS_OK;
    }
 
    if (strcmp(name, "c-sim-ctc") == 0)
    {
        config->port = site_config_c_port_for_ctc(ctx->site);
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "c-relay-ctc-a") == 0)
    {
        config->port = site_config_c_port_for_a(ctx->site, RTE_EXAMPLE_RELAY_KIND_CTC);
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "c-relay-ctc-b") == 0)
    {
        config->port = site_config_c_port_for_b(ctx->site, RTE_EXAMPLE_RELAY_KIND_CTC);
        return RTE_STATUS_OK;
    }
 
    /* ADR-038: the status/broker service's LISTEN links - A's/B's util
     * side channel + one broker port per local gateway service. The util
     * links carry rbc_util_frame_t (256 B), not rbc_envelope_t. */
    if ((strcmp(name, "c-relay-util-a") == 0) || (strcmp(name, "c-relay-util-b") == 0)
        || (strcmp(name, "c-broker-train") == 0) || (strcmp(name, "c-broker-il") == 0)
        || (strcmp(name, "c-broker-ctc") == 0))
    {
        config->message_size = RBC_UTIL_FRAME_SIZE;
        if (strcmp(name, "c-relay-util-a") == 0)
        {
            config->port = site_config_c_port_for_a(ctx->site, RTE_EXAMPLE_RELAY_KIND_UTIL);
        }
        else if (strcmp(name, "c-relay-util-b") == 0)
        {
            config->port = site_config_c_port_for_b(ctx->site, RTE_EXAMPLE_RELAY_KIND_UTIL);
        }
        else if (strcmp(name, "c-broker-train") == 0)
        {
            config->port = site_config_c_port_for_broker(ctx->site, RTE_EXAMPLE_RELAY_KIND_TRAIN);
        }
        else if (strcmp(name, "c-broker-il") == 0)
        {
            config->port = site_config_c_port_for_broker(ctx->site, RTE_EXAMPLE_RELAY_KIND_IL);
        }
        else
        {
            config->port = site_config_c_port_for_broker(ctx->site, RTE_EXAMPLE_RELAY_KIND_CTC);
        }
        /* ADR-038: the status service LISTENs on every c-broker-* (default
         * role above); a train/il/ctc gateway DIALS the one broker channel
         * named in its gateway_c_service_def_t. All Role C services share
         * one c-<site> container, so the broker link is pure loopback. */
        if ((g_svc != NULL) && (g_svc->broker_channel != NULL)
            && (strcmp(name, g_svc->broker_channel) == 0))
        {
            config->role = RTE_NETLINK_ROLE_CONNECT;
            config->host = "127.0.0.1";
        }
        return RTE_STATUS_OK;
    }
 
    return RTE_STATUS_INVALID_PARAM;
}
 
static void cycle_tick_callback(rte_timer_handle_t handle, void *user_ctx)
{
    gateway_c_context_t *ctx = (gateway_c_context_t *)user_ctx;
    (void)handle;
    ctx->cycle_ticks++;
}
 
static rte_status_t gateway_c_init(void *context)
{
    gateway_c_context_t *ctx = (gateway_c_context_t *)context;
    rte_status_t        status;
    rte_timer_config_t  timer_cfg;
 
    RTE_EXAMPLE_LOG("[GW-%s %s] starting\n", g_svc->log_tag, site_config_name(ctx->site));
    rte_log_write_event(RTE_LOG_LEVEL_INFO, site_config_name(ctx->site), 0U, g_svc->log_tag, "-", "LIFECYCLE",
                          "gateway service starting", NULL);
 
    ctx->stop_requested = false;
 
    (void)gateway_c_config_init_defaults(ctx->site, &ctx->config);
    {
        const char *env_cfg = getenv("RTE_CONFIG_FILE");
        char default_cfg_path[128];
        const char *site_str = (ctx->site == RTE_EXAMPLE_SITE_WEST) ? "west" : "east";
 
        if (env_cfg != NULL)
        {
            (void)gateway_c_config_load_file(env_cfg, &ctx->config);
        }
        else
        {
            (void)snprintf(default_cfg_path, sizeof(default_cfg_path), "etc/config/c_%s.json", site_str);
            if (gateway_c_config_load_file(default_cfg_path, &ctx->config) != RTE_STATUS_OK)
            {
                (void)snprintf(default_cfg_path, sizeof(default_cfg_path), "/etc/sapi/c_%s.json", site_str);
                (void)gateway_c_config_load_file(default_cfg_path, &ctx->config);
            }
        }
    }
 
    status = rte_posix_osadapter_channel_service_register_resolver(resolve_channel_c, ctx);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
 
    /* REQ-RBC-009: handler_init() sets up the service's links without waiting for an external Train/IL/CTC sim
     * (each sim channel is listen-role), so the service starts whether or not its sim is running yet. */
    status = g_svc->handler_init(ctx);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
 
    /* ADR-038: opt-in per-service status heartbeat to the C broker. */
    gateway_c_selfstatus_init(g_svc->broker_channel);
 
    ctx->cycle = 1U;
    ctx->cycle_ticks = 0U;
    rte_mem_set(&timer_cfg, 0, sizeof(timer_cfg));
    timer_cfg.mode = RTE_TIMER_MODE_PERIODIC;
    timer_cfg.period_ms = (rte_duration_ms_t)ctx->config.timers.report_period_ms;
    timer_cfg.callback = cycle_tick_callback;
    timer_cfg.user_ctx = ctx;
    status = rte_timer_create(&ctx->cycle_timer_storage, &timer_cfg, &ctx->cycle_timer_handle);
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW-%s %s] cycle timer_create failed (%s)\n", g_svc->log_tag,
                             site_config_name(ctx->site), rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, g_svc->log_tag, "-", "TIMER",
                              "cycle timer_create failed", NULL);
        return status;
    }
    status = rte_timer_start(ctx->cycle_timer_handle);
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW-%s %s] cycle timer_start failed (%s)\n", g_svc->log_tag,
                             site_config_name(ctx->site), rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, g_svc->log_tag, "-", "TIMER",
                              "cycle timer_start failed", NULL);
        return status;
    }
 
    RTE_EXAMPLE_LOG("[GW-%s %s] RBC relay armed (period=%ums)\n", g_svc->log_tag, site_config_name(ctx->site),
                         (unsigned int)ctx->config.timers.report_period_ms);
    return RTE_STATUS_OK;
}
 
static void gateway_c_check_channel_down_reboot(gateway_c_context_t *ctx, rte_timestamp_ms_t now_ms)
{
    const rte_timestamp_ms_t threshold_ms = (rte_timestamp_ms_t)RTE_EXAMPLE_CHANNEL_DOWN_REBOOT_MS;
    char                     which[64];
 
    which[0] = '\0';
    if (g_svc->handler_check_link_down(threshold_ms, now_ms, which, sizeof(which)))
    {
        RTE_EXAMPLE_LOG("[GW-%s %s] %s down >= %ums: rebooting Gateway service\n",
                             g_svc->log_tag, site_config_name(ctx->site), which, (unsigned int)threshold_ms);
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, g_svc->log_tag, "-",
                              "CHANNEL_REBOOT", which, NULL);
        (void)gateway_c_shutdown(ctx);
        RTE_REBOOT(0U);
    }
}
 
static rte_status_t gateway_c_execute(void *context)
{
    gateway_c_context_t *ctx = (gateway_c_context_t *)context;
    rte_timestamp_ms_t  now_ms = 0U;
 
    (void)rte_timer_now(&now_ms);
 
    /* Optional per-service pre-hook (IL: gateway_c_logctl_poll - the
     * non-safety diagnostic log-level control, applied to this process
     * and fanned to A/B; before the handler so a level bump takes effect
     * for this same cycle's own logging). */
    if (g_svc->pre_execute_hook != NULL)
    {
        g_svc->pre_execute_hook(ctx);
    }
 
    g_svc->handler_execute(ctx, ctx->cycle);
 
    if (g_svc->broker_channel != NULL)
    {
        gateway_c_selfstatus_tick(ctx, ctx->cycle, g_svc->log_tag, g_svc->status_src_id);
    }
 
    gateway_c_check_channel_down_reboot(ctx, now_ms);
 
    while ((ctx->cycle_ticks == 0U) && !ctx->stop_requested)
    {
        if (g_svc->idle_hook != NULL)
        {
            g_svc->idle_hook(ctx);
        }
        gateway_c_sleep_ms(5U);
    }
    if (ctx->cycle_ticks > 0U)
    {
        ctx->cycle_ticks--;
    }
 
    ctx->cycle++;
    return RTE_STATUS_OK;
}
 
static rte_status_t gateway_c_shutdown(void *context)
{
    gateway_c_context_t *ctx = (gateway_c_context_t *)context;
 
    RTE_EXAMPLE_LOG("[GW-%s %s] shutting down\n", g_svc->log_tag, site_config_name(ctx->site));
    rte_log_write_event(RTE_LOG_LEVEL_INFO, site_config_name(ctx->site), 0U, g_svc->log_tag, "-", "LIFECYCLE",
                          "gateway service shutting down", NULL);
 
    ctx->stop_requested = true;
    (void)rte_timer_stop(ctx->cycle_timer_handle);
    (void)rte_timer_destroy(ctx->cycle_timer_handle);
 
    g_svc->handler_shutdown();
    gateway_c_selfstatus_shutdown();
 
    RTE_EXAMPLE_LOG("[GW-%s %s] shutdown complete\n", g_svc->log_tag, site_config_name(ctx->site));
    return RTE_STATUS_OK;
}
 
static const char *gateway_c_get_name(void)
{
    return g_svc->app_name;
}
 
static const char *gateway_c_get_version(void)
{
    return "3.0.0";
}
 
static rte_appmanager_config_t gateway_c_common_make_config(rte_example_site_t site,
                                                             const gateway_c_service_def_t *svc)
{
    static const rte_appmanager_operations_t ops = {
        .init = gateway_c_init,
        .pre_execute = NULL,
        .execute = gateway_c_execute,
        .post_execute = NULL,
        .shutdown = gateway_c_shutdown,
        .get_name = gateway_c_get_name,
        .get_version = gateway_c_get_version
    };
    rte_appmanager_config_t config;
 
    g_svc = svc;
 
    rte_mem_set(&g_c_ctx, 0, sizeof(g_c_ctx));
    g_c_ctx.site = site;
 
    rte_mem_set(&config, 0, sizeof(config));
    config.ops = &ops;
    config.context = &g_c_ctx;
    config.max_iterations = 0U;
    return config;
}
 
int gateway_c_main(int argc, char **argv, const gateway_c_service_def_t *svc)
{
    rte_status_t            status;
    rte_example_site_t      site;
    rte_appmanager_config_t config;
    const char             *framework_actual_version;
 
    /* Printed unconditionally, before arg validation - see main.c's own
     * identical block (AB/GP/main/main.c) for the full rationale, shared
     * by every one of this project's binaries. */
    framework_actual_version = rte_core_get_version();
    (void)fprintf(stdout, "%s version %s (framework %s, expected %s)\n", argv[0], RTE_RBC2OO2_VERSION,
                  framework_actual_version, RTE_FRAMEWORK_EXPECTED_VERSION);
    if (strcmp(framework_actual_version, RTE_FRAMEWORK_EXPECTED_VERSION) != 0)
    {
        (void)fprintf(stderr,
                       "WARNING: linked safeAPIFreamwork version (%s) does not match what this binary was "
                       "built against (%s) - a stale library archive or wrong framework dist/ may be linked in\n",
                       framework_actual_version, RTE_FRAMEWORK_EXPECTED_VERSION);
    }
 
    if (argc < 2)
    {
        (void)fprintf(stderr, "usage: %s {WEST|EAST}\n", argv[0]);
        return 1;
    }
    if (!site_config_parse(argv[1], &site))
    {
        (void)fprintf(stderr, "unknown site '%s' - expected WEST or EAST\n", argv[1]);
        return 1;
    }
 
    status = app_main_common_init(argc, argv);
    if (status != RTE_STATUS_OK)
    {
        return 1;
    }
 
    config = gateway_c_common_make_config(site, svc);
    return rte_appmanager_run(&config);
}
 
static uint64_t gateway_c_hash_bytes(const uint8_t *bytes, size_t size)
{
    uint64_t h = 14695981039346656037ULL;
    size_t i;
    for (i = 0U; i < size; i++)
    {
        h ^= (uint64_t)bytes[i];
        h *= 1099511628211ULL;
    }
    return h;
}
 
void gateway_c_dedup_init(gateway_c_dedup_t *dedup)
{
    if (dedup != NULL)
    {
        rte_mem_set(dedup, 0, sizeof(*dedup));
    }
}
 
bool gateway_c_dedup_check_and_record(gateway_c_dedup_t *dedup, const uint8_t *bytes, size_t size)
{
    uint64_t h;
    rte_timestamp_ms_t now_ms = 0U;
    rte_timestamp_ms_t expiry_ms;
    uint32_t i;
 
    if ((dedup == NULL) || (bytes == NULL) || (size == 0U))
    {
        return false;
    }
 
    (void)rte_timer_now(&now_ms);
    h = gateway_c_hash_bytes(bytes, size);
    expiry_ms = (bytes[0] == (uint8_t)RBC_MSG_SAF_ENVELOPE) ? (rte_timestamp_ms_t)GATEWAY_C_DEDUP_SAF_EXPIRY_MS
                                                            : (rte_timestamp_ms_t)GATEWAY_C_DEDUP_EXPIRY_MS;
 
    /* Check if duplicate within expiry window */
    for (i = 0U; i < GATEWAY_C_DEDUP_WINDOW_SIZE; i++)
    {
        if (dedup->entries[i].hash == h)
        {
            if ((now_ms >= dedup->entries[i].timestamp_ms) &&
                ((now_ms - dedup->entries[i].timestamp_ms) <= expiry_ms))
            {
                return true; /* duplicate frame! */
            }
        }
    }
 
    /* Record first arrival */
    dedup->entries[dedup->head].hash = h;
    dedup->entries[dedup->head].timestamp_ms = now_ms;
    dedup->head = (dedup->head + 1U) % GATEWAY_C_DEDUP_WINDOW_SIZE;
 
    return false; /* first arrival, forward to sim */
}