stellar-stellar/src/stellar/stellar.cpp

#include <stdio.h>
#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <sys/prctl.h>
#include <signal.h>

#include "packet.h"
#include "timestamp.h"
#include "session_manager.h"

#define STELLAR_LOG_ERROR(format, ...) LOG_ERROR("stellar", format, ##__VA_ARGS__)
#ifndef STELLAR_LOG_ERROR
#define STELLAR_LOG_ERROR(format, ...) \
    fprintf(stderr, "ERROR (stellar), " format "\n", ##__VA_ARGS__);
#endif
#define STELLAR_LOG_DEBUG(format, ...) LOG_DEBUG("stellar", format, ##__VA_ARGS__)
#ifndef STELLAR_LOG_DEBUG
#define STELLAR_LOG_DEBUG(format, ...) \
    fprintf(stderr, "DEBUG (stellar), " format "\n", ##__VA_ARGS__);
#endif

#define ATOMIC_SET(x, y) __atomic_store_n(x, y, __ATOMIC_RELAXED)
#define ATOMIC_READ(x) __atomic_load_n(x, __ATOMIC_RELAXED)

struct thread_context
{
    pthread_t tid;
    uint16_t index;
    uint64_t need_exit;
    uint64_t is_runing;
    struct session_manager *sess_mgr;
};

struct stellar_ctx
{
    uint64_t need_exit;
    uint16_t thread_num;

    struct session_manager_config sess_mgr_cfg;

    struct thread_context thread_ctx[128];
} g_stellar_ctx = {
    .need_exit = 0,
    .thread_num = 1,
    .sess_mgr_cfg = {
        // max session number
        .max_tcp_session_num = 3,
        .max_udp_session_num = 3,

        // session overload
        .tcp_overload_evict_old_sess = 1, // 1: evict old session, 0: bypass new session
        .udp_overload_evict_old_sess = 1, // 1: evict old session, 0: bypass new session

        // tcp timeout
        .tcp_timeout_init = 2,
        .tcp_timeout_handshake = 2,
        .tcp_timeout_data = 2,
        .tcp_timeout_half_closed = 2,
        .tcp_timeout_time_wait = 2,
        .tcp_timeout_discard = 2,

        // udp timeout
        .udp_timeout_data = 1,

        // tcp duplicate packet filter
        .tcp_dupkt_filter_enable = 1,
        .tcp_dupkt_filter_capacity = 1000,
        .tcp_dupkt_filter_timeout = 10,
        .tcp_dupkt_filter_error_rate = 0.0001,

        // udp eviction filter
        .udp_eviction_filter_enable = 1,
        .udp_eviction_filter_capacity = 1000,
        .udp_eviction_filter_timeout = 10,
        .udp_eviction_filter_error_rate = 0.0001,
    }};

static int recv_packet(const char **data)
{
    static unsigned char packet_data[] = {
        0x5c, 0x5e, 0xab, 0x2a, 0xa2, 0x00, 0x2c, 0x6b, 0xf5, 0x45, 0x88, 0x29, 0x08, 0x00, 0x45, 0x00, 0x00, 0x5c, 0x0b, 0x4d, 0x00, 0x00, 0x3b, 0x29, 0x09, 0xc8,
        0xd2, 0x4d, 0x58, 0xa3, 0x3b, 0x42, 0x04, 0x32, 0x60, 0x00, 0x00, 0x00, 0x00, 0x20, 0x06, 0x40, 0x20, 0x01, 0x0d, 0xa8, 0x02, 0x00, 0x90, 0x0e, 0x02, 0x00,
        0x5e, 0xfe, 0xd2, 0x4d, 0x58, 0xa3, 0x26, 0x00, 0x14, 0x0e, 0x00, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x17, 0x02, 0x10, 0x58, 0xcd, 0x4c, 0x00, 0x50,
        0x81, 0x80, 0x5c, 0x76, 0x00, 0x00, 0x00, 0x00, 0x80, 0x02, 0x20, 0x00, 0xf7, 0x57, 0x00, 0x00, 0x02, 0x04, 0x04, 0xc4, 0x01, 0x03, 0x03, 0x08, 0x01, 0x01,
        0x04, 0x02};

    *data = (const char *)packet_data;
    return sizeof(packet_data);
}

static void send_packet(const char *data, uint16_t len)
{
}

static void signal_handler(int signo)
{
    if (signo == SIGINT)
    {
        STELLAR_LOG_DEBUG("recv SIGINT, exit !!!");
        ATOMIC_SET(&g_stellar_ctx.need_exit, 1);
    }

    if (signo == SIGQUIT)
    {
        STELLAR_LOG_DEBUG("recv SIGQUIT, exit !!!");
        ATOMIC_SET(&g_stellar_ctx.need_exit, 1);
    }

    if (signo == SIGTERM)
    {
        STELLAR_LOG_DEBUG("recv SIGTERM, exit !!!");
        ATOMIC_SET(&g_stellar_ctx.need_exit, 1);
    }
}

static void __packet_dispatch(const struct packet *pkt)
{
    if (pkt == NULL)
    {
        return;
    }

    printf("\n");
    printf("=> packet dispatch: %p\n", pkt);
    printf("<= packet dispatch\n\n");
}

static void __session_dispatch(struct session *sess)
{
    if (sess == NULL)
    {
        return;
    }

    printf("\n");
    printf("=> session dispatch: %p\n", sess);
    session_dump(sess);
    printf("<= session dispatch\n\n");

    session_set0_cur_pkt(sess, NULL);
    session_set_cur_dir(sess, SESSION_DIR_NONE);
}

static int thread_context_init(struct stellar_ctx *ctx)
{
    for (uint16_t i = 0; i < ctx->thread_num; i++)
    {
        struct thread_context *thread_ctx = &ctx->thread_ctx[i];
        thread_ctx->index = i;
        thread_ctx->need_exit = 0;
        thread_ctx->is_runing = 0;

        thread_ctx->sess_mgr = session_manager_create(&ctx->sess_mgr_cfg);
        if (thread_ctx->sess_mgr == NULL)
        {
            STELLAR_LOG_ERROR("unable to create session manager");
            return -1;
        }
    }

    return 0;
}

static void thread_context_free(struct stellar_ctx *ctx)
{
    for (uint16_t i = 0; i < ctx->thread_num; i++)
    {
        struct thread_context *thread_ctx = &ctx->thread_ctx[i];
        if (ATOMIC_READ(&thread_ctx->is_runing) == 0)
        {
            session_manager_destroy(thread_ctx->sess_mgr);
        }
    }
}

static void *thread_cycle(void *arg)
{
    uint16_t len = 0;
    const char *data = NULL;
    struct packet pkt;
    struct session *sess = NULL;
    struct thread_context *thread_ctx = (struct thread_context *)arg;
    struct session_manager *sess_mgr = thread_ctx->sess_mgr;
    char thd_name[16];

    ATOMIC_SET(&thread_ctx->is_runing, 1);
    snprintf(thd_name, sizeof(thd_name), "stellar:%d", thread_ctx->index);
    prctl(PR_SET_NAME, (unsigned long long)thd_name, NULL, NULL, NULL);
    STELLAR_LOG_DEBUG("worker thread %s runing\n", thd_name);

    while (ATOMIC_READ(&thread_ctx->need_exit) == 0)
    {
        len = recv_packet(&data);
        if (data == NULL)
        {
            goto poll_wait;
        }

        packet_parse(&pkt, data, len);
        __packet_dispatch(&pkt);

        sess = session_manager_update_session(sess_mgr, &pkt);
        __session_dispatch(sess);

        send_packet(data, len);

        sess = session_manager_get_evicted_session(sess_mgr);
        __session_dispatch(sess);

    poll_wait:
        sess = session_manager_get_expired_session(sess_mgr);
        __session_dispatch(sess);

        usleep(1000); // session_manager_get_expire_interval(sess_mgr); (seconds)
    }

    ATOMIC_SET(&thread_ctx->is_runing, 0);
    STELLAR_LOG_DEBUG("worker thread %s exit\n", thd_name);

    return NULL;
}

int main(int argc, char **argv)
{
    // TODO parse command line

    // TODO init log

    // TODO init plugin

    signal(SIGINT, signal_handler);
    signal(SIGQUIT, signal_handler);
    signal(SIGTERM, signal_handler);

    timestamp_update();

    if (thread_context_init(&g_stellar_ctx) != 0)
    {
        STELLAR_LOG_ERROR("unable to init thread context");
        goto error_out;
    }

    for (uint16_t i = 0; i < g_stellar_ctx.thread_num; i++)
    {
        struct thread_context *thread_ctx = &g_stellar_ctx.thread_ctx[i];
        if (pthread_create(&thread_ctx->tid, NULL, thread_cycle, (void *)thread_ctx) < 0)
        {
            STELLAR_LOG_ERROR("unable to create worker thread, error %d: %s", errno, strerror(errno));
            goto error_out;
        }
    }

    while (!g_stellar_ctx.need_exit)
    {
        timestamp_update();
        sleep(1);
    }

    for (uint16_t i = 0; i < g_stellar_ctx.thread_num; i++)
    {
        struct thread_context *thread_ctx = &g_stellar_ctx.thread_ctx[i];
        ATOMIC_SET(&thread_ctx->need_exit, 1);
        while (ATOMIC_READ(&thread_ctx->is_runing) == 1)
        {
            sleep(1);
        }
    }

error_out:
    thread_context_free(&g_stellar_ctx);

    // TODO free plugin

    // TODO free log

    return 0;
}