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 <libgen.h>
#include <marsio.h>
#include <mr_rte_msg.h>
#include <mrapp.h>
#include <mrdp_trace.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
#define MR_MP_CACHE_SIZE 256

/* 写入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);
    MESA_load_profile_uint_def(instance->g_cfgfile_path, "service", "mp_cache_size", &instance->sz_mp_cache,
                               MR_MP_CACHE_SIZE);
}

/* EAL环境初始化 */
static int 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 0;
    }

    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);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Cannot get init thread affinity: %s", strerror(errno));
        goto errout;
    }

    rte_openlog_stream(stderr);
    ret = rte_eal_init((int)eal_argc, eal_argv);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Cannot init EAL environment, Failed.");
        goto errout;
    }

    ret = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &_cpu_set);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Cannot set init thread affinity: %s", strerror(errno));
        goto errout;
    }

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

errout:
    return -1;
}

struct app_register_req * app_register_request_construct(struct mr_instance * instance)
{
    struct app_register_req * reg_req = ZMALLOC(sizeof(struct app_register_req));
    MR_VERIFY_MALLOC(reg_req);

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

    return reg_req;
}

void * marsio_survival_monitor(void * arg)
{
    int epoll_fd = epoll_create1(EPOLL_CLOEXEC);
    if (epoll_fd < 0)
    {
        MR_ERROR("Cannot create epoll fd: %s. ", strerror(errno));
        return NULL;
    }
    epoll_add_event(epoll_fd, _current_instance->marsio_fd, EPOLLIN);
    struct epoll_event evlist[__EV_MAX_EVENTS];
    while (1)
    {
        int ret = epoll_wait(epoll_fd, evlist, __EV_MAX_EVENTS, -1);
        if (ret == -1 && errno == EINTR)
            continue;
        else if (ret == -1)
        {
            MR_ERROR("loop thread failed : %s", strerror(errno));
            return NULL;
        }

        for (int i = 0; i < ret; i++)
        {
            /* 异常链接|数据到达 */
            if (evlist[i].events & (EPOLLHUP | EPOLLERR | EPOLLIN))
            {
                MR_INFO("marsio is terminated, Exit. ");
                _exit(EXIT_FAILURE);
            }
        }
    }
}

int send_register_request(struct app_register_req * reg_req)
{
    int socket_pair[2];
    int ret = socketpair(AF_UNIX, SOCK_STREAM, 0, socket_pair);
    if (ret < 0)
    {
        MR_WARNING("Failed to create socketpair");
        return RT_ERR;
    }

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

    uintptr_t ptr_address = (uintptr_t)reg_req;
    struct app_register_resp * rep_msg = NULL;

    struct rte_mp_msg reg_msg = {};
    strncpy(reg_msg.name, "instance_alive_register", sizeof(reg_msg.name) - 1);
    reg_msg.fds[0] = socket_pair[0];
    reg_msg.num_fds = 1;
    reg_msg.len_param = sizeof(ptr_address);
    memcpy(reg_msg.param, &ptr_address, sizeof(ptr_address));

    ret = rte_mp_request_sync(&reg_msg, &mp_reply, &wait_timespec);
    if (ret < 0)
    {
        MR_WARNING("Failed to execute rte_mp_request_sync:%s", rte_strerror(rte_errno));
        goto error;
    }

    uintptr_t stored_ptr_address;
    memcpy(&stored_ptr_address, mp_reply.msgs->param, sizeof(uintptr_t));
    rep_msg = (struct app_register_resp *)stored_ptr_address;

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

    // Make sure the socket_pair[0] of the main process is closed and the kernel sends EOF
    _current_instance->marsio_fd = socket_pair[1];
    close(socket_pair[0]);

    pthread_t __pid = 0;
    ret = pthread_create(&__pid, NULL, marsio_survival_monitor, NULL);
    if (ret != 0)
    {
        MR_ERROR("Launch marsio_survival_monitor thread failed : %s", strerror(ret));
        goto error;
    }

    if (mp_reply.msgs != NULL)
    {
        free(mp_reply.msgs);
    }

    rte_free(reg_req);
    rte_free(rep_msg);
    return RT_SUCCESS;

error:
    if (mp_reply.msgs != NULL)
    {
        free(mp_reply.msgs);
    }

    rte_free(reg_req);
    rte_free(rep_msg);
    close(socket_pair[0]);
    close(socket_pair[1]);
    return RT_ERR;
}

// Register application
static int mrapp_register(struct mr_instance * instance)
{
    struct app_register_req * reg_req = app_register_request_construct(instance);
    return send_register_request(reg_req);
}

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 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(instance->sz_mp_cache, 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_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;

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

    /* 本地配置文件路径 */
    char str_cfgfile[MR_STRING_MAX] = {0};
    snprintf(str_cfgfile, sizeof(str_cfgfile), "%s", g_cfg_file);

    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;
}

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);
    }

    free(mp_reply.msgs);
    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;
        }
    }
    free(mp_reply.msgs);
    /* 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);

    /* 根据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;
    }

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

    if (marsio_dp_trace_init(instance) != RT_SUCCESS)
    {
        MR_ERROR("data path instance get failed, recheck mrzcpd is running.");
        goto err;
    }

    if (mrapp_register(instance) != RT_SUCCESS)
    {
        MR_ERROR("app register failed, recheck mrzcpd is running.");
        goto err;
    }

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

    /* 负载均衡器 */
    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(ret));
        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;
}