path: root/app/src/marsio.c

#include <assert.h>
#include <rte_eal.h>
#include <rte_log.h>
#include <rte_malloc.h>
#include <rte_spinlock.h>
#include <rte_string_fns.h>
#include <rte_version.h>
#include <stdio.h>
#include <stdlib.h>

#include <MESA_prof_load.h>
#include <arp.h>
#include <arpa/inet.h>
#include <cJSON.h>
#include <ctrlmsg.h>
#include <ctrlmsg_define.h>
#include <libgen.h>
#include <marsio.h>
#include <mrapp.h>
#include <protect.h>
#include <rte_acl_osdep.h>
#include <rte_epoll.h>
#include <signal.h>
#include <sys/epoll.h>
#include <tap.h>
#include <unistd.h>

#define MR_LIB_MAX_EAL_ARGC 128

unsigned int g_logger_to_stdout = 1;
unsigned int g_logger_level = LOG_DEBUG;
unsigned int g_eal_inited = 0;
unsigned int g_in_protect_mode = 0;
rte_spinlock_t g_in_protect_lock = RTE_SPINLOCK_INITIALIZER;

__thread struct mr_thread_info thread_info;

struct mr_instance * _current_instance = NULL;

#define MRAPP_MONIT_FILE_PATH "/var/run/mrzcpd/mrmonit.app.%s"
#define MRAPP_SERVICE_MONIT_FILE_PATH "/var/run/mrzcpd/mrmonit.daemon"
#define MRAPP_GLOBAL_CONF_FILE_PATH "/opt/tsg/mrzcpd/etc/mrglobal.conf"
#define MRAPP_STATIC_NEIGH_FILE_PATH "/opt/tsg/mrzcpd/etc/mrneigh.table"

#ifndef MRAPP_DEFAULT_NEIGH_TABLE_MAX_ENTRIES
#define MRAPP_DEFAULT_NEIGH_TABLE_MAX_ENTRIES 4096
#endif

#ifndef MRAPP_DEFAULT_NEIGH_TABLE_TIMEOUT
#define MRAPP_DEFAULT_NEIGH_TABLE_TIMEOUT 0
#endif

#ifndef MRAPP_DEFAULT_NEIGH_ARP_SEND_INTERVAL
#define MRAPP_DEFAULT_NEIGH_ARP_SEND_INTERVAL 1
#endif

#ifndef MRAPP_DEFAULT_NEIGH_GRATUITOUS_ARP_SEND
#define MRAPP_DEFAULT_NEIGH_GRATUITOUS_ARP_SEND 3
#endif

#define MR_VDEV_BUFFER_SIZE 512

/* 写入Command参数 */
static void __write_arg(char * eal_argv[], unsigned int * eal_argc, unsigned int max_argc, const char * value)
{
    assert(max_argc >= *eal_argc);
    char * mem = (char *)malloc(MR_STRING_MAX * sizeof(char));
    assert(mem != NULL);

    snprintf(mem, MR_STRING_MAX * sizeof(char), "%s", value);
    eal_argv[(*eal_argc)++] = mem;
    return;
}

#define WRITE_ARG(x)                                                                                                   \
    do                                                                                                                 \
    {                                                                                                                  \
        __write_arg(eal_argv, &eal_argc, MR_LIB_MAX_EAL_ARGC, x);                                                      \
    } while (0)

void __mrapp_mem_protect_unlock_mempool_cb(struct rte_mempool * mp, void * arg)
{
    PROTECT_rte_mempool_unpoison(mp);
    MR_INFO("Unlock MEMPOOL %s: %p, local cache is %p", mp->name, mp, mp->local_cache);
}

#if 0
/* 基于ASAN的内存保护模式
 * 该保护模式开启后，共享大页内存将标记为不可达，对共享内存的访问将被ASAN探测并记录 */
static void mrapp_mem_protect_with_asan_init(struct ref_mr_instance * instance)
{
    MESA_load_profile_uint_def(instance->app_cfgfile_path, "protect", "enable",
        &instance->memory_protect_with_asan, 0);

    if(instance->memory_protect_with_asan)
    {
#ifndef __SANITIZE_ADDRESS__
        MR_WARNING("Memory Protection with ASAN is not supported for the APP is not compiled with address_sanitizer.");
        instance->memory_protect_with_asan = 0; return;
#endif
    }
    else { return; }

    MR_INFO("Memory Protection with ASAN is enabled.");
    g_in_protect_mode = 1;
    struct rte_config *cfg = rte_eal_get_configuration();
    const struct rte_mem_config * mcfg = cfg->mem_config;

    for(int i = 0; i < RTE_MAX_MEMSEG; i++)
    {
        if(mcfg->memseg[i].addr == NULL) break;
        void * memseg_addr = mcfg->memseg[i].addr;
        size_t memseg_len = mcfg->memseg[i].len;

        MR_ASAN_POISON_MEMORY_REGION(memseg_addr, memseg_len);
        MR_INFO("MEMSEG %d is protected by ASAN: addr = %p, len = %zu", i, memseg_addr, memseg_len);
    }

    /* 解锁所有的mempool */
    rte_mempool_walk(__mrapp_mem_protect_unlock_mempool_cb, NULL);
}
#endif

static void mrapp_rx_notify_init(struct mr_instance * instance)
{
    for (unsigned int i = 0; i < instance->nr_dataplane_thread; i++)
    {
        int epfd = epoll_create1(EPOLL_CLOEXEC);
        if (epfd < 0)
        {
            MR_ERROR("failed at create notify epoll epfd for thread %d", i);
        }

        MR_DEBUG("application: %s, thread: %d, rx_notify_epfd = %d", instance->appsym, i, epfd);
        instance->rx_notify_epfd[i] = epfd;
    }

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "poll_wait_throttle_usleep_threshold",
                               &instance->zero_recv_usleep_threshold, 32);

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "poll_wait_throttle_usleep_period",
                               &instance->zero_recv_usleep_period, 5);

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "poll_wait_throttle_notify_threshold",
                               &instance->zero_recv_notify_threshold, 256);

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "poll_wait_enable", &instance->en_notify, 1);
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "vdev_buffer_size", &instance->sz_vdev_buffer,
                               MR_VDEV_BUFFER_SIZE);
}

/* EAL环境初始化 */
static void mrapp_eal_init(struct mr_instance * instance)
{
    char * eal_argv[MR_LIB_MAX_EAL_ARGC];
    unsigned int eal_argc = 0;

    if (g_eal_inited > 0)
        return;

    WRITE_ARG(instance->appsym);
    WRITE_ARG("-c");
    WRITE_ARG("0x1");
    WRITE_ARG("--proc-type=secondary");

    char str_virtaddr[MR_STRING_MAX];
    int ret =
        MESA_load_profile_string_nodef(instance->g_cfgfile_path, "eal", "virtaddr", str_virtaddr, sizeof(str_virtaddr));

    if (ret >= 0)
    {
        WRITE_ARG("--base-virtaddr");
        WRITE_ARG(str_virtaddr);
    }

    unsigned int en_no_huge = 0;
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "eal", "nohuge", &en_no_huge, 0);

    if (en_no_huge > 0)
    {
        WRITE_ARG("--no-huge");
    }

    // DPDK和SYSTEMD的日志级别差1
    unsigned int loglevel = g_logger_level + 1;
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "eal", "loglevel", &loglevel, loglevel);

    /* 检查日志选项，必须在1~8之间 */
    if (!(loglevel >= RTE_LOG_EMERG && loglevel <= RTE_LOG_DEBUG))
    {
        MR_CFGERR_INVALID_VALUE(instance->g_cfgfile_path, "eal", "loglevel",
                                "Must between LOG_DEBUG(8) and LOG_EMERG(1)");
        exit(EXIT_FAILURE);
    }

#if RTE_VERSION >= RTE_VERSION_NUM(17, 5, 0, 0)
    rte_log_set_global_level(loglevel);
#else
    rte_set_log_level(loglevel);
#endif

    g_logger_level = loglevel - 1;

    char str_loglevel[MR_STRING_MAX];
    snprintf(str_loglevel, sizeof(str_loglevel), "%d", loglevel);

    WRITE_ARG("--log-level");
    WRITE_ARG(str_loglevel);

    char str_eal_cmdline[MR_STRING_MAX];
    unsigned int curser_str_eal_cmdline = 0;

    for (int i = 0; i < eal_argc; i++)
    {
        curser_str_eal_cmdline += snprintf(str_eal_cmdline + curser_str_eal_cmdline,
                                           sizeof(str_eal_cmdline) - curser_str_eal_cmdline, "%s ", eal_argv[i]);
    }

    MR_INFO("EAL Parameters: %s", str_eal_cmdline);

    /* 获得当前线程的亲和性设置，EAL初始化后恢复，避免从此线程派生的
       线程全部带有EAL设置的亲和性。 */

    cpu_set_t __cpu_set;
    ret = pthread_getaffinity_np(pthread_self(), sizeof(cpu_set_t), &__cpu_set);
    MR_VERIFY_2(ret >= 0, "Cannot get init thread affinity: %s", strerror(errno));

    rte_openlog_stream(stderr);
    ret = rte_eal_init(eal_argc, eal_argv);
    MR_VERIFY_2(ret >= 0, "Cannot init EAL Enviorment, Failed");

    ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &__cpu_set);
    MR_VERIFY_2(ret >= 0, "Cannot set init thread affinity: %s", strerror(errno));

    /* EAL环境初始化成功 */
    g_eal_inited = 1;
    return;
}

static int mrapp_ctrlmsg_init(struct mr_instance * instance)
{
    char str_ctrlmsg_addr[MR_STRING_MAX] = {0};
    unsigned int ctrlmsg_port = 0;

    /* 读控制面监听线程的监听地址、端口号 */
    // MESA_load_profile_string_def(instance->g_cfgfile_path, "ctrlmsg", "listen_addr",
    //     str_ctrlmsg_addr, sizeof(str_ctrlmsg_addr), CTRLMSG_DEFAULT_ADDR);

    const char * env_ctrlmsg_addr = getenv("MRZCPD_CTRLMSG_LISTEN_ADDR");
    if (env_ctrlmsg_addr != NULL)
    {
        MR_INFO("MRZCPD_CTRLMSG_LISTEN_ADDR is %s", env_ctrlmsg_addr);
        strncpy(str_ctrlmsg_addr, env_ctrlmsg_addr, sizeof(str_ctrlmsg_addr));
    }
    else
    {
        MR_WARNING("MRZCPD_CTRLMSG_LISTEN_ADDR is not set, default is 127.0.0.1.");
        strncpy(str_ctrlmsg_addr, "127.0.0.1", sizeof(str_ctrlmsg_addr));
    }

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "ctrlmsg", "listen_port", &ctrlmsg_port, CTRLMSG_DEFAULT_PORT);

    /* 地址转换 */
    struct sockaddr_in sockaddr_in;
    if (inet_pton(AF_INET, str_ctrlmsg_addr, &sockaddr_in.sin_addr) <= 0)
    {
        // MR_CFGERR_INVALID_FORMAT(instance->g_cfgfile_path, "ctrlmsg", "listen_addr");fd
        MR_ERROR("Mrapp ctrlmsg init error,The environment variable 'MRZCPD_CTRLMSG_LISTEN_ADDR=%s' is invalid.",
                 str_ctrlmsg_addr);
        return RT_ERR;
    }

    /* 端口 */
    sockaddr_in.sin_port = htons(ctrlmsg_port);
    sockaddr_in.sin_family = AF_INET;

    /* 创建消息处理框架句柄 */
    instance->ctrlmsg_handler = ctrlmsg_handler_create(CTRLMSG_HANDLER_MODE_CLIENT, sockaddr_in, NULL, -1);
    if (instance->ctrlmsg_handler == NULL)
        return RT_ERR;
    return RT_SUCCESS;
}

static int mrapp_distributer_init(struct mr_instance * instance)
{
    unsigned int distmode = LDBC_DIST_OUTER_TUPLE2;
    unsigned int hashmode = LDBC_HASH_SYM_CRC;

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "distmode", &distmode, LDBC_DIST_OUTER_TUPLE2);
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "hashmode", &hashmode, LDBC_HASH_SYM_CRC);

    if (distmode < 0 || distmode >= LDBC_DIST_MAX)
    {
        MR_CFGERR_INVALID_FORMAT(instance->g_cfgfile_path, "service", "distmode");
        return RT_ERR;
    }

    if (hashmode < 0 || distmode >= LDBC_HASH_MAX)
    {
        MR_CFGERR_INVALID_FORMAT(instance->g_cfgfile_path, "service", "hashmode");
        return RT_ERR;
    }

    instance->dist_object = distributer_create(distmode, hashmode, 0);
    if (instance->dist_object == NULL)
    {
        MR_ERROR("Create distributer handler failed. ");
        return RT_ERR;
    }

    return RT_SUCCESS;
}

static unsigned __table_strip(char * str, unsigned len)
{
    int newlen = len;
    if (len == 0)
        return 0;

    if (isspace(str[len - 1]))
    {
        /* strip trailing whitespace */
        while (newlen > 0 && isspace(str[newlen - 1]))
            str[--newlen] = '\0';
    }

    if (isspace(str[0]))
    {
        /* strip leading whitespace */
        int i, start = 1;
        while (isspace(str[start]) && start < newlen)
            start++; /* do nothing */
        newlen -= start;
        for (i = 0; i < newlen; i++)
            str[i] = str[i + start];
        str[i] = '\0';
    }

    return newlen;
}

static const char * __table_readline(FILE * fp, char * buffer, size_t sz_buffer, size_t * buffer_len,
                                     unsigned int * line_no)
{
    while (fgets(buffer, sz_buffer, fp) != NULL)
    {
        char * pos = NULL;
        size_t len = strnlen(buffer, sz_buffer);
        (*line_no)++;

        if ((len >= sizeof(buffer) - 1) && (buffer[len - 1] != '\n'))
        {
            continue;
        }

        pos = memchr(buffer, '#', sz_buffer);
        if (pos != NULL)
        {
            *pos = '\0';
            len = pos - buffer;
        }

        len = __table_strip(buffer, len);
        if (len == 0)
        {
            continue;
        }

        return buffer;
    }

    return NULL;
}

static int __table_split_line(const char * buffer, char str_tokens[MR_TOKENS_MAX][MR_STRING_MAX])
{
    char * __buffer = strdup(buffer);
    assert(__buffer != NULL);

    char * token_ptr;
    unsigned int total_nr_tokens = 0;

    while ((token_ptr = strsep(&__buffer, " \t")) != NULL)
    {
        if (strlen(token_ptr) == 0)
            continue;
        if (total_nr_tokens == MR_TOKENS_MAX)
            break;

        strncpy(str_tokens[total_nr_tokens], token_ptr, MR_STRING_MAX);

        int len = strnlen(str_tokens[total_nr_tokens], MR_STRING_MAX);
        __table_strip(str_tokens[total_nr_tokens], len);

        total_nr_tokens++;
    }

    free(__buffer);
    return total_nr_tokens;
}

/* 邻居子系统管理器 */
static int mrapp_neigh_init(struct mr_instance * instance)
{
    instance->neigh = malloc(sizeof(struct neighbour_manager));
    MR_VERIFY_MALLOC(instance->neigh);

    unsigned int neigh_max_entries;
    unsigned int neigh_timeout;
    unsigned int arp_send_interval;

    /* 临时静态邻居表初始化
       该表仅限初始化时使用，设备打开后，该表项读取邻居子系统。
    */
    TAILQ_INIT(&instance->static_neigh_list);

    MESA_load_profile_uint_def(instance->g_cfgfile_path, "neigh", "neigh_table_max_entries", &neigh_max_entries,
                               MRAPP_DEFAULT_NEIGH_TABLE_MAX_ENTRIES);
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "neigh", "neigh_table_timeout", &neigh_timeout,
                               MRAPP_DEFAULT_NEIGH_TABLE_TIMEOUT);
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "neigh", "arp_send_interval", &arp_send_interval,
                               MRAPP_DEFAULT_NEIGH_ARP_SEND_INTERVAL);
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "neigh", "gratuitous_arp_send",
                               &instance->nr_gratuitous_arp_send, MRAPP_DEFAULT_NEIGH_GRATUITOUS_ARP_SEND);

    /* 邻居子系统初始化 */
    if (neighbour_mamanger_init(instance->neigh, instance->appsym, neigh_max_entries, neigh_timeout,
                                arp_send_interval) != RT_SUCCESS)
    {
        goto err;
    }

    /* 读静态配置表 */
    FILE * fp_static_neigh_table = fopen(MRAPP_STATIC_NEIGH_FILE_PATH, "r");
    if (fp_static_neigh_table == NULL)
    {
        MR_DEBUG("Skip reading static neighbour table from file %s.", MRAPP_STATIC_NEIGH_FILE_PATH);
        goto success;
    }

    /* 按行读 */
    char __tb_line_buffer[MR_STRING_MAX];
    size_t __tb_buffer_len = 0;
    unsigned int __tb_line_no = 0;

    while (__table_readline(fp_static_neigh_table, __tb_line_buffer, sizeof(__tb_line_buffer), &__tb_buffer_len,
                            &__tb_line_no) != NULL)
    {
        /* 拆分，第一列为IP地址，第二列为MAC地址 */
        char __split_str[MR_TOKENS_MAX][MR_STRING_MAX];
        memset(__split_str, 0, sizeof(__split_str));

        int nr_tokens = __table_split_line(__tb_line_buffer, __split_str);
        if (nr_tokens != 3)
        {
            MR_WARNING("Table: %s, line: %u: Invalid line format, must have 3 column, ignore.",
                       MRAPP_STATIC_NEIGH_FILE_PATH, __tb_line_no);
            continue;
        }

        /* 读IP地址、MAC地址 */
        const char * str_in_addr = __split_str[0];
        const char * str_ether_addr = __split_str[1];
        const char * str_device = __split_str[2];

        /* 字符串转换 */
        struct in_addr in_addr;
        struct rte_ether_addr ether_addr;

        int ret = inet_pton(AF_INET, str_in_addr, &in_addr);
        if (ret < 0)
        {
            MR_WARNING("Table: %s, line: %u: Invaild IP address format, ignore. ", MRAPP_STATIC_NEIGH_FILE_PATH,
                       __tb_line_no);
            continue;
        }

        ret = sscanf(str_ether_addr, "%02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx", &ether_addr.addr_bytes[0],
                     &ether_addr.addr_bytes[1], &ether_addr.addr_bytes[2], &ether_addr.addr_bytes[3],
                     &ether_addr.addr_bytes[4], &ether_addr.addr_bytes[5]);

        if (ret != 6)
        {
            MR_WARNING("Table: %s, line: %u: Invaild MAC address format, ignore. ", MRAPP_STATIC_NEIGH_FILE_PATH,
                       __tb_line_no);
            continue;
        }

        /* 检查重复的表项 */
        struct mr_static_neigh_entry * __neigh_entry_iter;
        unsigned int is_dup_entry = 0;

        TAILQ_FOREACH(__neigh_entry_iter, &instance->static_neigh_list, next)
        {
            if (__neigh_entry_iter->in_addr.s_addr != in_addr.s_addr)
                continue;

            /* 重复表项，告警 */
            MR_WARNING("table: %s, line: %u: Duplicate entry, %s->%s, ignore.", MRAPP_STATIC_NEIGH_FILE_PATH,
                       __tb_line_no, str_in_addr, str_ether_addr);
            is_dup_entry = 1;
            break;
        }

        /* 重复表项跳过 */
        if (is_dup_entry)
            continue;

        /* 插入临时的静态邻居表 */
        struct mr_static_neigh_entry * __neigh_entry = malloc(sizeof(struct mr_static_neigh_entry));
        memset(__neigh_entry, 0, sizeof(struct mr_static_neigh_entry));

        __neigh_entry->in_addr = in_addr;
        __neigh_entry->ether_addr = ether_addr;
        strncpy(__neigh_entry->devsym, str_device, sizeof(__neigh_entry->devsym));

        /* 拷贝IP地址、MAC地址的字符串形式，便于后面显示日志信息 */
        strncpy(__neigh_entry->str_in_addr, str_in_addr, sizeof(__neigh_entry->str_in_addr));
        strncpy(__neigh_entry->str_ether_addr, str_ether_addr, sizeof(__neigh_entry->str_ether_addr));

        TAILQ_INSERT_TAIL(&instance->static_neigh_list, __neigh_entry, next);
    }

success:
    return RT_SUCCESS;

err:
    if (instance->neigh != NULL)
    {
        free(instance->neigh);
        instance->neigh = NULL;
    }

    return RT_ERR;
}

static void mp_cache_init_for_each_mp(struct rte_mempool * mp, void * arg)
{
    struct mr_instance * instance = (struct mr_instance *)arg;
    if (mp->name[0] != 'M' || mp->name[1] != 'Z')
    {
        return;
    }

    /* create a mp<->cache map */
    instance->mp_cache_map[instance->nr_mp_cache_map] = ZMALLOC(sizeof(struct mp_cache_map));
    struct mp_cache_map * mp_cache_map_ptr = instance->mp_cache_map[instance->nr_mp_cache_map];
    instance->nr_mp_cache_map++;

    mp_cache_map_ptr->mp = mp;
    for (unsigned int i = 0; i < instance->nr_dataplane_thread; i++)
    {
        cpu_id_t cpu_id = cpu_set_location(&instance->cpu_set, i);
        socket_id_t socket_id = (socket_id_t)rte_lcore_to_socket_id(cpu_id);

        struct rte_mempool_cache * mp_cache = rte_mempool_cache_create(512, socket_id);
        if (unlikely(mp_cache == NULL))
        {
            MR_ERROR("failed at create local mp cache, thread_id=%d, mp=%s", i, mp->name);
            return;
        }

        mp_cache_map_ptr->mp_cache[i] = mp_cache;
    }
}

static void mpapp_mp_cache_init(struct mr_instance * instance)
{
    rte_mempool_walk(mp_cache_init_for_each_mp, (void *)instance);
}

/* 注册应用 */
static int mrapp_app_register(struct mr_instance * instance)
{
    struct ctrl_msg_app_reg_request reg_cmd;
    memset(&reg_cmd, 0, sizeof(reg_cmd));

    ctrl_msg_header_construct(&reg_cmd.msg_header, sizeof(reg_cmd), CTRL_MSG_TYPE_REQUEST, CTRLMSG_TOPIC_APP_REGISTER);

    /* 应用标识符 */
    strncpy((char *)reg_cmd.symbol, instance->appsym, sizeof(reg_cmd.symbol));
    /* 状态监控文件，委托服务进程销毁 */
    strncpy((char *)reg_cmd.mntfile, instance->monit_file_path, sizeof(reg_cmd.mntfile));
    /* 进程号，便于应用跟踪，查找问题 */
    reg_cmd.pid = getpid();

    ctrlmsg_msg_send(instance->ctrlmsg_handler, NULL, (struct ctrl_msg_header *)(&reg_cmd));

    /* TODO: 抽象出单独的函数 */
    pthread_mutex_lock(&instance->lock_ctrlmsg_wait);
    while (instance->ctrlmsg_wait == 0)
    {
        pthread_cond_wait(&instance->cond_ctrlmsg_wait, &instance->lock_ctrlmsg_wait);
    }

    instance->ctrlmsg_wait = 0;
    pthread_mutex_unlock(&instance->lock_ctrlmsg_wait);
    return RT_SUCCESS;
}

/* 读全局配置文件路径等信息 */
static int mrapp_gconf_init(struct mr_instance * instance)
{
    /* 读JSON文件 */
    FILE * f = fopen(MRAPP_SERVICE_MONIT_FILE_PATH, "rb");
    if (f == NULL)
    {
        MR_ERROR("Cannot open mrzcpd monit file %s, perhaps mrzcpd program is not running : %s",
                 MRAPP_SERVICE_MONIT_FILE_PATH, strerror(errno));
        return RT_ERR;
    }

    fseek(f, 0, SEEK_END);
    long fsize = ftell(f);
    fseek(f, 0, SEEK_SET); // same as rewind(f);

    char * string = malloc(fsize + 1);
    MR_VERIFY_MALLOC(string);

    /* 读文件 */
    fread(string, fsize, 1, f);
    fclose(f);

    string[fsize] = 0;

    cJSON * j_global_monit = cJSON_Parse(string);
    if (j_global_monit == NULL)
        goto j_parse_error;

    cJSON * j_genernal = cJSON_GetObjectItem(j_global_monit, "general");
    if (j_genernal == NULL)
        goto j_parse_error;

    char * g_cfg_file = cJSON_GetObjectItem(j_genernal, "g_cfgfile")->valuestring;
    if (g_cfg_file == NULL)
        goto j_parse_error;

    /* 全局配置文件路径 */
    strncpy(instance->g_cfgfile_path, g_cfg_file, sizeof(instance->g_cfgfile_path));

    /* 本地配置文件路径 */
    char __str_cfgfile[MR_STRING_MAX] = {0};
    strncpy(__str_cfgfile, instance->g_cfgfile_path, sizeof(__str_cfgfile));

    snprintf(instance->app_cfgfile_path, sizeof(instance->app_cfgfile_path), "%s/mrapp.%s.conf", dirname(__str_cfgfile),
             instance->appsym);

    return RT_SUCCESS;

j_parse_error:
    MR_ERROR("Parse mrzcpd monit file %s failed", MRAPP_SERVICE_MONIT_FILE_PATH);
    return RT_ERR;
}

extern int mrapp_monit_loop(struct mr_instance * instance);

void * mrapp_ctrlplane_thread(void * args)
{
    struct mr_instance * _instance = (struct mr_instance *)args;
    pthread_detach(pthread_self());

    while (1)
    {
        mrapp_monit_loop(_instance);
        sleep(1);
    }
}

struct mr_vdev * marsio_device_lookup(struct mr_instance * instance, const char * devsym)
{
    for (int i = 0; i < instance->nr_vdevs; i++)
    {
        if (strncmp(devsym, instance->vdevs[i].devsym, sizeof(instance->vdevs[i].devsym)) == 0)
            return &instance->vdevs[i];
    }

    return NULL;
}

static int __ctrlplane_conn_close_handler(struct ctrlmsg_handler * ct_hand, struct ctrlmsg_conn * ct_conn,
                                          struct ctrl_msg_header * msg, void * arg)
{
    MR_INFO("Ctrlplane connection is terminated, Exit. ");
    exit(EXIT_FAILURE);
}

#if 0
static int __open_device_response_handler(struct ctrlmsg_handler * ct_hand,
    struct ctrlmsg_conn * ct_conn, struct ctrl_msg_header * msg, void * arg)
{
    struct ctrl_msg_vdev_open_response * rep_msg = (struct ctrl_msg_vdev_open_response *)msg;
    struct mr_instance * instance = (struct mr_instance *)arg;

    /* 打开失败 */
    if (rep_msg->msg_err.errcode != 0)
    {
        MR_ERROR("%s", rep_msg->msg_err.strerr);
        goto wake_up;
    }

    /* 打开成功 */
    struct mr_vdev * mr_vdev = &instance->vdevs[instance->nr_vdevs];

    mr_vdev->vdi = (struct vdev_instance *)rep_msg->ptr_vdi;
    mr_vdev->nr_rxstream = rep_msg->nr_rxstream;
    mr_vdev->nr_txstream = rep_msg->nr_txstream;
    mr_vdev->instance = instance;
    strncpy(mr_vdev->devsym, (char *)rep_msg->devsym, MR_SYMBOL_MAX);

    /* VDI使用了大页面共享内存，在保护模式下撤除对它的保护 */
    if (instance->memory_protect_with_asan)
    {
        __open_device_unposion(mr_vdev->vdi);
    }

    instance->nr_vdevs++;

wake_up:
    pthread_mutex_lock(&instance->lock_ctrlmsg_wait);
    instance->ctrlmsg_wait = 1;
    pthread_cond_broadcast(&instance->cond_ctrlmsg_wait);
    pthread_mutex_unlock(&instance->lock_ctrlmsg_wait);
    return 0;
}

#endif

static int __app_register_response_handler(struct ctrlmsg_handler * ct_hand, struct ctrlmsg_conn * ct_conn,
                                           struct ctrl_msg_header * msg, void * arg)
{
    struct ctrl_msg_app_reg_response * rep_msg = (struct ctrl_msg_app_reg_response *)msg;
    struct mr_instance * instance = (struct mr_instance *)arg;

    // 应用注册失败，退出。
    if (rep_msg->msg_err.errcode != 0)
    {
        MR_ERROR("%s", rep_msg->msg_err.strerr);
        exit(EXIT_FAILURE);
    }

    // 注册成功，唤醒等待的线程。
    pthread_mutex_lock(&instance->lock_ctrlmsg_wait);
    instance->ctrlmsg_wait = 1;
    pthread_cond_broadcast(&instance->cond_ctrlmsg_wait);
    pthread_mutex_unlock(&instance->lock_ctrlmsg_wait);
    return 0;
}

struct mr_vdev * marsio_open_device(struct mr_instance * instance, const char * devsym, unsigned int nr_rxstream,
                                    unsigned int nr_txstream)
{
    struct rte_mp_msg mp_req_msg = {};
    struct rte_mp_reply mp_reply = {};
    struct ctrl_msg_vdev_open_request * req_msg = (struct ctrl_msg_vdev_open_request *)mp_req_msg.param;
    struct mr_vdev * mr_vdev = NULL;

    snprintf((char *)req_msg->devsym, sizeof(req_msg->devsym) - 1, "%s", devsym);
    snprintf((char *)req_msg->appsym, sizeof(req_msg->appsym) - 1, "%s", instance->appsym);
    req_msg->nr_rxstream = nr_rxstream;
    req_msg->nr_txstream = nr_txstream;

    const struct timespec wait_timespec = {
        .tv_nsec = 0,
        .tv_sec = 30,
    };

    /* make request and wait the response */
    strncpy(mp_req_msg.name, "vdev_instance_request", sizeof(mp_req_msg.name) - 1);
    mp_req_msg.len_param = sizeof(struct ctrl_msg_vdev_open_request);

    int ret = rte_mp_request_sync(&mp_req_msg, &mp_reply, &wait_timespec);
    if (ret < 0)
    {
        MR_ERROR("failed at send device opening request for %s: %s", devsym, rte_strerror(rte_errno));
        goto errout;
    }

    /* response */
    struct ctrl_msg_vdev_open_response * msg_resp = (struct ctrl_msg_vdev_open_response *)mp_reply.msgs->param;
    assert(mp_reply.msgs->len_param == sizeof(struct ctrl_msg_vdev_open_response));

    if (msg_resp->errcode != 0)
    {
        MR_ERROR("failed at open device %s: errcode = %d", msg_resp->devsym, msg_resp->errcode);
        goto errout;
    }

    mr_vdev = &instance->vdevs[instance->nr_vdevs];
    mr_vdev->vdi = (struct vdev_instance *)msg_resp->ptr_vdi;
    mr_vdev->nr_rxstream = msg_resp->nr_rxstream;
    mr_vdev->nr_txstream = msg_resp->nr_txstream;
    mr_vdev->instance = instance;

    for (unsigned int i = 0; i < mp_reply.msgs->num_fds; i++)
    {
        mr_vdev->rx_notify_fds[i] = mp_reply.msgs->fds[i];
    }

    /* move the transferred fd */
    mr_vdev->nr_rx_notify_fds = mp_reply.msgs->num_fds;

    /* add notify event to epfd */
    for (unsigned int tid = 0; tid < mr_vdev->nr_rx_notify_fds; tid++)
    {
        struct rte_epoll_event * epoll_event = &mr_vdev->rx_notify_epoll_events[tid];
        epoll_event->epdata.event = EPOLLIN;
        epoll_event->epdata.data = NULL;

        if (instance->rx_notify_epfd[tid] <= 0)
        {
            continue;
        }

        ret = rte_epoll_ctl(instance->rx_notify_epfd[tid], EPOLL_CTL_ADD, mr_vdev->rx_notify_fds[tid], epoll_event);
        if (unlikely(ret < 0))
        {
            MR_ERROR("failed at add notify fd %d to thread %d's epoll fd: %s.", mr_vdev->rx_notify_fds[tid], tid,
                     strerror(errno));
            goto errout;
        }
    }

    for (unsigned int i = 0; i < mr_vdev->nr_rxstream; i++)
    {
        size_t sz_rx_buffer = sizeof(struct mr_vdev_rx_buffer) + sizeof(struct rte_mbuf *) * instance->sz_vdev_buffer;
        mr_vdev->rx_buffer[i] = rte_zmalloc(NULL, sz_rx_buffer, 0);
        mr_vdev->rx_buffer[i]->size = instance->sz_vdev_buffer;
    }

    for (unsigned int i = 0; i < mr_vdev->nr_txstream; i++)
    {
        size_t sz_tx_buffer = sizeof(struct mr_vdev_tx_buffer) + sizeof(struct rte_mbuf *) * instance->sz_vdev_buffer;
        mr_vdev->tx_buffer[i] = rte_zmalloc(NULL, sz_tx_buffer, 0);
        mr_vdev->tx_buffer[i]->size = instance->sz_vdev_buffer;
    }

    strncpy(mr_vdev->devsym, (char *)msg_resp->devsym, MR_SYMBOL_MAX);
    instance->nr_vdevs++;

    MR_INFO(" ");
    MR_INFO("Application %s, Device %s:", instance->appsym, mr_vdev->devsym);
    MR_INFO("   Rx Queue Count                 : %d", mr_vdev->nr_rxstream);
    MR_INFO("   Tx Queue Count                 : %d", mr_vdev->nr_txstream);

    if (mr_vdev->vdi->vdev->representor_config.enable > 0)
    {
        tap_representor_init(instance, mr_vdev);
    }

    return mr_vdev;

errout:
    /* close all transferred fds */
    if (mr_vdev != NULL)
    {
        for (unsigned int i = 0; i < mr_vdev->nr_rx_notify_fds; i++)
        {
            close(mr_vdev->rx_notify_fds[i]);
            mr_vdev->rx_notify_fds[i] = -1;
        }
    }

    /* the mr_vdev is not alloc from heap, should not free */
    return NULL;
}

void marsio_close_device(struct mr_vdev * vdev)
{
    return;
}

void marsio_get_device_ether_addr(struct mr_vdev * vdev, void * str_ether_addr, uint8_t size)
{
    rte_ether_format_addr(str_ether_addr, size, &vdev->vdi->vdev->ether_addr);
    return;
}

static void mask_to_cpuset(uint64_t mask, cpu_set_t * cpusetp)
{
    for (unsigned long bit_iter = 0; bit_iter < sizeof(mask) * 8; bit_iter++)
    {
        if ((mask & (1ULL << bit_iter)))
            CPU_SET(bit_iter, cpusetp);
    }
}

int marsio_option_set(struct mr_instance * instance, marsio_opt_type_t opt_type, void * opt, size_t sz_opt)
{

#define __CHECK_USER_PARAM(expect_type)                                                                                \
    do                                                                                                                 \
    {                                                                                                                  \
        if (sz_opt < sizeof(expect_type))                                                                              \
        {                                                                                                              \
            return -EINVAL;                                                                                            \
        }                                                                                                              \
    } while (0)

    int ret = 0;

    switch (opt_type)
    {
    case MARSIO_OPT_THREAD_NUM:
        __CHECK_USER_PARAM(unsigned int);
        instance->nr_dataplane_thread = *(unsigned int *)opt;
        ret = 0;
        break;

    case MARSIO_OPT_THREAD_MASK:
        __CHECK_USER_PARAM(uint64_t);
        uint64_t mask = *(uint64_t *)opt;
        mask_to_cpuset(mask, &instance->cpu_set);
        ret = 0;
        break;

    case MARSIO_OPT_THREAD_MASK_IN_CPUSET:
        __CHECK_USER_PARAM(cpu_set_t);
        instance->cpu_set = *(cpu_set_t *)opt;
        ret = 0;
        break;

    case MARSIO_OPT_EXIT_WHEN_ERR:
        __CHECK_USER_PARAM(unsigned int);
        instance->is_exit_when_err_raise = *(unsigned int *)opt;
        ret = 0;
        break;

    default:
        ret = -EINVAL;
        break;
    }

    return ret;
}

int marsio_option_get(struct mr_instance * instance, int opt_type, void * out_opt, size_t out_opt_buffer)
{
    return 0;
}

struct mr_instance * marsio_create()
{
    struct mr_instance * instance;
    instance = malloc(sizeof(struct mr_instance));
    memset(instance, 0, sizeof(struct mr_instance));

    instance->is_exit_when_err_raise = 1;
    _current_instance = instance;
    return instance;
}

struct mr_instance * marsio_current()
{
    return _current_instance;
}

int marsio_init(struct mr_instance * instance, const char * appsym)
{
    /* 写应用名称参数 */
    snprintf(instance->appsym, sizeof(instance->appsym), "%s", appsym);

    /* 状态监测路径 */
    snprintf(instance->monit_file_path, sizeof(instance->monit_file_path), MRAPP_MONIT_FILE_PATH, instance->appsym);

    pthread_mutex_init(&instance->lock_ctrlmsg_wait, NULL);
    pthread_cond_init(&instance->cond_ctrlmsg_wait, NULL);
    pthread_mutex_init(&instance->lock_thread_init, NULL);

    /* 根据CPU_MASK计算线程数 */
    if (instance->nr_dataplane_thread == 0 && CPU_COUNT(&instance->cpu_set) != 0)
    {
        instance->nr_dataplane_thread = CPU_COUNT(&instance->cpu_set);
    }

    if (mrapp_gconf_init(instance) != RT_SUCCESS)
    {
        MR_ERROR("Global configure initialization failed, recheck mrzcpd is running.");
        goto err;
    }

    /* 初始化消息通信框架
       在EAL环境启动之前启动消息通信框架，避免SERVICE启动之前启动APP
    */
    if (mrapp_ctrlmsg_init(instance) != RT_SUCCESS)
    {
        MR_ERROR("Ctrlmsg module initialization failed, recheck mrzcpd is running.");
        goto err;
    }

    /* 初始化EAL环境 */
    mrapp_eal_init(instance);
    mrapp_rx_notify_init(instance);

    /* 注册处理应用注册结果的回调函数 */
    ctrlmsg_msg_reciver_register(instance->ctrlmsg_handler, CTRLMSG_TOPIC_APP_REGISTER, CTRL_MSG_TYPE_RESPONSE,
                                 __app_register_response_handler, instance);

    /* 注册设备打开的回调函数 */

#if 0
    ctrlmsg_msg_reciver_register(instance->ctrlmsg_handler, CTRLMSG_TOPIC_VDEV_OPEN,
        CTRL_MSG_TYPE_RESPONSE, __open_device_response_handler, instance);
#endif

    /* 控制链接中断处理函数 */
    ctrlmsg_event_conn_close_register(instance->ctrlmsg_handler, __ctrlplane_conn_close_handler, instance);

    if (ctrlmsg_thread_launch(instance->ctrlmsg_handler) != RT_SUCCESS)
    {
        MR_ERROR("Launch ctrlmsg thread failed. ");
        goto err;
    }

    /* 应用注册 */
    if (mrapp_app_register(instance) != RT_SUCCESS)
    {
        MR_ERROR("App register failed. ");
        goto err;
    }

    /* 负载均衡器 */
    if (mrapp_distributer_init(instance) != RT_SUCCESS)
    {
        MR_ERROR("Distributer initialization failed.");
        goto err;
    }

    /* local mp cache */
    mpapp_mp_cache_init(instance);
    pthread_t pid_ctrlplane_thread;
    int ret = pthread_create(&pid_ctrlplane_thread, NULL, mrapp_ctrlplane_thread, instance);
    if (ret < 0)
    {
        MR_ERROR("Launch ctrlplane thread failed : %s", strerror(errno));
        goto err;
    }

    MR_INFO("Application %s in client mode register success. ", appsym);
    return RT_SUCCESS;

err:
    if (instance->is_exit_when_err_raise)
        exit(EXIT_FAILURE);
    return RT_ERR;
}

int marsio_thread_init(struct mr_instance * instance)
{
    if (thread_info.instance != NULL)
    {
        MR_ERROR("Duplicated marsio_thread_init() call happened, Failed. ");
        return RT_ERR;
    }

    if (CPU_COUNT(&instance->cpu_set) == 0)
    {
        MR_DEBUG("CPU mask is zero, thread affinity is not allowed.");
        return RT_SUCCESS;
    }

    int ret = 0;
    pthread_mutex_lock(&instance->lock_thread_init);

    /* 线程绑定 */
    cpu_id_t cpu_id = cpu_set_location(&instance->cpu_set, instance->to_suppose_tid);
    if (cpu_id < 0)
    {
        MR_ERROR("Too many threads call thread init, supposed tid is %d", instance->to_suppose_tid);
        ret = RT_ERR;
        goto out;
    }

    cpu_set_t _cpu_set;
    CPU_ZERO(&_cpu_set);
    CPU_SET(cpu_id, &_cpu_set);

    pthread_t ppid = pthread_self();
    ret = pthread_setaffinity_np(ppid, sizeof(_cpu_set), &_cpu_set);
    if (ret < 0)
    {
        MR_ERROR("Set thread affinity failed : %s. ", strerror(errno));
        ret = RT_ERR;
        goto out;
    }

    thread_info.is_dataplane_thread = 1;
    thread_info.instance = instance;
    thread_info.cpu_id = cpu_id;
    thread_info.thread_id = instance->to_suppose_tid++;

    int tid = rte_sys_gettid();
    MR_INFO("Thread %d(tid=%d) is affinity on lcore %d", thread_info.thread_id, tid, cpu_id);
    ret = RT_SUCCESS;

    goto out;

out:
    pthread_mutex_unlock(&instance->lock_thread_init);
    if (ret < 0 && instance->is_exit_when_err_raise)
        exit(EXIT_FAILURE);
    return ret;
}

int marsio_destory(struct mr_instance * instance)
{
    return 0;
}