119 lines
3.3 KiB
C++
119 lines
3.3 KiB
C++
#include <errno.h>
|
|
#include <unistd.h>
|
|
#include <signal.h>
|
|
#include <pthread.h>
|
|
|
|
#include "sce.h"
|
|
#include "log.h"
|
|
#include "utils.h"
|
|
#include "health_check.h"
|
|
#include "global_metrics.h"
|
|
|
|
static void sig_handler(int signo)
|
|
{
|
|
if (signo == SIGHUP)
|
|
{
|
|
LOG_INFO("%s: recv SIGHUP, reload zlog.conf", LOG_TAG_SCE);
|
|
LOG_RELOAD();
|
|
}
|
|
}
|
|
|
|
static void *worker_thread_cycle(void *arg)
|
|
{
|
|
struct thread_ctx *thread_ctx = (struct thread_ctx *)arg;
|
|
struct packet_io *handle = thread_ctx->ref_io;
|
|
int n_packet_recv;
|
|
|
|
LOG_INFO("%s: worker thread %d running", LOG_TAG_SCE, thread_ctx->thread_index);
|
|
|
|
while (1)
|
|
{
|
|
n_packet_recv = packet_io_polling_nf_interface(handle, thread_ctx->thread_index, thread_ctx);
|
|
if (n_packet_recv)
|
|
{
|
|
LOG_INFO("%s: worker thread %d recv %03d packets from nf_interface", LOG_TAG_SCE, thread_ctx->thread_index, n_packet_recv);
|
|
}
|
|
|
|
n_packet_recv = packet_io_polling_endpoint(handle, thread_ctx->thread_index, thread_ctx);
|
|
if (n_packet_recv)
|
|
{
|
|
LOG_INFO("%s: worker thread %d recv %03d packets from endpoint", LOG_TAG_SCE, thread_ctx->thread_index, n_packet_recv);
|
|
}
|
|
|
|
if (__atomic_fetch_add(&thread_ctx->session_table_need_reset, 0, __ATOMIC_RELAXED) > 0)
|
|
{
|
|
session_table_reset(thread_ctx->session_table);
|
|
__atomic_fetch_and(&thread_ctx->session_table_need_reset, 0, __ATOMIC_RELAXED);
|
|
}
|
|
}
|
|
|
|
LOG_ERROR("%s: worker thread %d exiting", LOG_TAG_SCE, thread_ctx->thread_index);
|
|
return (void *)NULL;
|
|
}
|
|
|
|
int main(int argc, char **argv)
|
|
{
|
|
const char *profile = "./conf/sce.conf";
|
|
|
|
if (LOG_INIT("./conf/zlog.conf") == -1)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
if (signal(SIGHUP, sig_handler) == SIG_ERR)
|
|
{
|
|
LOG_ERROR("%s: unable to register SIGHUP signal handler, error %d: %s", LOG_TAG_SCE, errno, strerror(errno));
|
|
LOG_CLOSE();
|
|
return -1;
|
|
}
|
|
|
|
health_check_session_init(profile);
|
|
|
|
struct sce_ctx *ctx = sce_ctx_create(profile);
|
|
if (ctx == NULL)
|
|
{
|
|
LOG_CLOSE();
|
|
return -1;
|
|
}
|
|
|
|
for (int i = 0; i < ctx->nr_worker_threads; i++)
|
|
{
|
|
ctx->work_threads[i].tid = 0;
|
|
ctx->work_threads[i].thread_index = i;
|
|
ctx->work_threads[i].session_table = session_table_create();
|
|
ctx->work_threads[i].ref_io = ctx->io;
|
|
ctx->work_threads[i].ref_metrics = ctx->metrics;
|
|
ctx->work_threads[i].ref_enforcer = ctx->enforcer;
|
|
ctx->work_threads[i].ref_sce_ctx = ctx;
|
|
ctx->work_threads[i].session_table_need_reset = 0;
|
|
}
|
|
|
|
for (int i = 0; i < ctx->nr_worker_threads; i++)
|
|
{
|
|
struct thread_ctx *thread_ctx = &ctx->work_threads[i];
|
|
if (pthread_create(&thread_ctx->tid, NULL, worker_thread_cycle, (void *)thread_ctx) < 0)
|
|
{
|
|
LOG_ERROR("%s: unable to create worker thread %d, error %d: %s", LOG_TAG_SCE, i, errno, strerror(errno));
|
|
goto error_out;
|
|
}
|
|
}
|
|
|
|
while (1)
|
|
{
|
|
global_metrics_dump(ctx->metrics);
|
|
sleep(ctx->metrics->config.statsd_cycle);
|
|
}
|
|
|
|
error_out:
|
|
for (int i = 0; i < ctx->nr_worker_threads; i++)
|
|
{
|
|
struct thread_ctx *thread_ctx = &ctx->work_threads[i];
|
|
session_table_destory(thread_ctx->session_table);
|
|
}
|
|
sce_ctx_destory(ctx);
|
|
|
|
LOG_CLOSE();
|
|
|
|
return 0;
|
|
}
|