path: root/infra/src/ctrlmsg.c

/* 控制面消息通信框架
 *
 * \author Qiuwen Lu
 * \date   2017-03-08
 */

#include <common.h>
#include <sys/queue.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <ctrlmsg.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <netinet/tcp.h>
#include <unistd.h>
#include <assert.h>
#include <fcntl.h>
#include <ctrlmsg.h>
#include <arpa/inet.h>
#include <errno.h>

#include <systemd/sd-daemon.h>

#define __EV_MAX_EVENTS					16
#define __EV_MAX_BUFFER					2048
#define __CONN_MAX_BUFFER				65536

static void __cleanup_unlock_mutex(pthread_mutex_t ** lock)
{
    pthread_mutex_unlock(*lock);
}

#define __MUTEX_LOCK(lock)												\
	pthread_mutex_lock(lock);											\
	pthread_mutex_t * __lock											\
	__attribute__((__cleanup__(__cleanup_unlock_mutex))) = lock			\

static int __tcp_setnonblock(int sockfd)
{
    int flags = fcntl(sockfd, F_GETFL, 0);
    if (flags == -1) return -1;
    int ret = fcntl(sockfd, F_SETFL, flags | O_NONBLOCK, 0);
    if (ret == -1)	return -1;
    return 0;
}

static int __update_epoll_event(int epoll_fd, int epoll_mode, int fd, int epoll_type)
{
    struct epoll_event event = { 0 };
    event.data.fd = fd;
    event.events = epoll_type;

    __tcp_setnonblock(fd);
    if (-1 == epoll_ctl(epoll_fd, epoll_mode, fd, &event))
    {
        MR_ERROR("update epoll event failed : %s\n", strerror(errno)); return -1;
    }

    return 0;
}

int __epoll_add_event(int epoll_fd, int sd, int epoll_type)
{
    return __update_epoll_event(epoll_fd, EPOLL_CTL_ADD, sd, epoll_type);
}

int __epoll_del_event(int epoll_fd, int sd, int epoll_type)
{
    return __update_epoll_event(epoll_fd, EPOLL_CTL_DEL, sd, epoll_type);
}

int __epoll_mod_event(int epoll_fd, int sd, int epoll_type)
{
    return __update_epoll_event(epoll_fd, EPOLL_CTL_MOD, sd, epoll_type);
}

struct ctrlmsg_conn * __lookup_conn_by_fd(struct ctrlmsg_handler * handler, int fd)
{
    struct ctrlmsg_conn * conn_iter = NULL;
    TAILQ_FOREACH(conn_iter, &handler->conn_list, next)
    {
        if (conn_iter->conn_fd == fd) return conn_iter;
    }

    return NULL;
}

enum parse_status
{
    PARSE_OK,
    PARSE_OPEN,
    PARSE_BAD,
    PARSE_MAX
};

/*
    控制报文解析：单行解析拆分
*/

#if 0
static int parse_line(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn)
{
    /* checked指向正在分析的字节，used为分析边界。*/
    unsigned int * m_checked_idx = &conn->rcv_buffer_checked;
    unsigned int * m_read_idx = &conn->rcv_buffer_used;
    char * m_read_buf = conn->rcv_buffer;

    for (; *m_checked_idx < *m_read_idx; (*m_checked_idx)++)
    {
        char temp = conn->rcv_buffer[conn->rcv_buffer_checked];

        /* 换行为\r\n或\n */
        if (temp == '\r')
        {
            if ((*m_checked_idx + 1) == *m_read_idx)
            {
                return PARSE_OPEN;
            }
            else if (m_read_buf[*m_checked_idx + 1] == '\n')
            {
                m_read_buf[(*m_checked_idx)++] = '\0';
                m_read_buf[(*m_checked_idx)++] = '\0';
                return PARSE_OK;
            }
            return PARSE_BAD;
        }
        else if (temp == '\n')
        {
            if ((*m_checked_idx > 1) && (m_read_buf[*m_checked_idx - 1] == '\r'))
            {
                m_read_buf[(*m_checked_idx) - 1] = '\0';
                m_read_buf[(*m_checked_idx)++] = '\0';
                return PARSE_OK;
            }
            return PARSE_BAD;
        }
    }

    return PARSE_OPEN;
}

static int parse_content_json(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn,
    int epoll_fd, int fd)
{
    return 0;
}

#endif

static int parse_content_binary(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn,
    int epoll_fd, int fd)
{
    /* 缓冲区的数据还不够头部，退出循环 */
    while (conn->rcv_buffer_used >= sizeof(struct ctrl_msg_header))
    {
        /* 缓冲区已经收到了头部，但数据长度不足报文声明的长度，数据不全 */
        struct ctrl_msg_header * p_msg_hdr = (struct ctrl_msg_header *)conn->rcv_buffer;
        if (conn->rcv_buffer_used < p_msg_hdr->msg_len) return 0;

        /* 查询调用链，精确匹配路由地址 */
        struct ctrlmsg_cb * cb_item = NULL;
        TAILQ_FOREACH(cb_item, &handler->cb_list, next)
        {
            if (p_msg_hdr->msg_type != cb_item->type)
                continue;

            if (strncmp(cb_item->topic, (const char *)p_msg_hdr->msg_topic,
                sizeof(cb_item->topic)) != 0) continue;

            cb_item->fn_ctrlmsg_cb(handler, conn, p_msg_hdr, cb_item->arg);

            MR_DEBUG("CtrlMessage Recieve Callback: topic = %s, type = %d, function = %p",
                cb_item->topic, cb_item->type, cb_item->fn_ctrlmsg_cb);
        }

        conn->rcv_buffer_used -= p_msg_hdr->msg_len;

        /* 处理完毕的消息，从缓冲区中移除 */
        memmove(conn->rcv_buffer, RTE_PTR_ADD(conn->rcv_buffer, p_msg_hdr->msg_len),
            conn->rcv_buffer_used);
    }

    return 0;
}

/* 处理新建的监听连接 */
void __handle_new_conn(struct ctrlmsg_handler * handler,
    int epoll_fd, int conn_fd, struct sockaddr_in remote_addr)
{
    /* 分配新的APP-CONN结构体，只保存链接信息，不创建应用信息 */
    struct ctrlmsg_conn * conn = LIBC_MALLOC(sizeof(struct ctrlmsg_conn));
    MR_VERIFY_MALLOC(conn);

    conn->conn_fd = conn_fd;
    conn->peer_sockaddr_in = remote_addr;
    conn->rcv_buffer = LIBC_MALLOC(__CONN_MAX_BUFFER);
    conn->rcv_buffer_max = __CONN_MAX_BUFFER;
    conn->rcv_buffer_used = 0;
    conn->snd_buffer = LIBC_MALLOC(__CONN_MAX_BUFFER);
    conn->snd_buffer_max = __CONN_MAX_BUFFER;
    conn->snd_buffer_used = 0;
    conn->snd_buffer_deal = 0;

    MR_VERIFY_MALLOC(conn->rcv_buffer);
    MR_VERIFY_MALLOC(conn->snd_buffer);

    /* 初始化锁，使用递归锁 */
    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_init(&mutex_attr);
    pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&conn->mutex, &mutex_attr);

    /* 加入链接表中 */
    TAILQ_INSERT_TAIL(&handler->conn_list, conn, next);

    /* 调用新链接回调函数 */
    struct ctrlmsg_cb * cb_object;
    TAILQ_FOREACH(cb_object, &handler->cb_conn_new_list, next)
    {
        cb_object->fn_ctrlmsg_cb(handler, conn, NULL, cb_object->arg);
    }

    __epoll_add_event(epoll_fd, conn_fd, EPOLLIN);
    return;
}

// 处理链接关闭信息
void __handle_close(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn,
    int epoll_fd, int fd)
{
    /* 调用链接终止回调函数 */
    struct ctrlmsg_cb * cb_object;
    TAILQ_FOREACH(cb_object, &handler->cb_conn_close_list, next)
    {
        cb_object->fn_ctrlmsg_cb(handler, conn, NULL, cb_object->arg);
    }

    __epoll_del_event(epoll_fd, fd, EPOLLIN);
    close(fd);

    TAILQ_REMOVE(&handler->conn_list, conn, next);
    free(conn->rcv_buffer);
    free(conn->snd_buffer);
    free(conn);
    return;
}

/* 处理监听连接到达的数据 */
void __handle_read(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn,
    int epoll_fd, int fd)
{
    __MUTEX_LOCK(&conn->mutex);
    for (;;)
    {
        int ret = recv(fd, RTE_PTR_ADD(conn->rcv_buffer, conn->rcv_buffer_used),
            conn->rcv_buffer_max - conn->rcv_buffer_used, 0);

        if (ret < 0 && (errno == EWOULDBLOCK || errno == EAGAIN))
        {
            break;
        }

        if (ret < 0)
        {
            MR_ERROR("File descripter recv error: %s", strerror(errno));
            __handle_close(handler, conn, epoll_fd, fd);
            return;
        }

        if (ret == 0)
        {
            __handle_close(handler, conn, epoll_fd, fd);
            return;
        }

        conn->rcv_buffer_used += ret;
    }

    parse_content_binary(handler, conn, epoll_fd, fd);
    return;
}

void __handle_write(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn,
    int epoll_fd, int fd)
{
    int ret = 0;
    __MUTEX_LOCK(&conn->mutex);

    do {
        ret = send(fd, conn->snd_buffer, conn->snd_buffer_used, MSG_NOSIGNAL);
    } while ((ret == -1 && (errno == EINTR)));

    conn->snd_buffer_deal += ret;

    /* 缓冲区内所有的数据都发送完了 */
    if (conn->snd_buffer_used == conn->snd_buffer_deal)
    {
        conn->snd_buffer_used = 0;
        conn->snd_buffer_deal = 0;
        __epoll_mod_event(handler->epoll_fd, conn->conn_fd, EPOLLIN);
    }

    return;
}

void __handle_timeout(struct ctrlmsg_handler * handler)
{
    /* 向systemd的watchdog喂狗 */
    /* TODO: 不应该在这里做这件事，应该在外面，通过回调函数调用 */

    if (sd_watchdog_enabled(0, NULL))
    {
        sd_notify(0, "WATCHDOG=1");
    }

    return;
}

static void * __epoll_event_loop(void * arg)
{
    struct ctrlmsg_handler * handler = (struct ctrlmsg_handler *)arg;
    struct epoll_event evlist[__EV_MAX_EVENTS];

    int epoll_fd = handler->epoll_fd;
    int listen_fd = handler->listen_fd;
    int timeout = handler->timeout;

    /* 对于客户端模式，没有监听的端口 */
    if (handler->mode == CTRLMSG_HANDLER_MODE_CLIENT)
        listen_fd = -1;

    while (1)
    {
        int ret = epoll_wait(epoll_fd, evlist, __EV_MAX_EVENTS, timeout);
        if (ret == -1 && errno == EINTR) continue;
        else if (ret == -1)
        {
            MR_ERROR("Ctrlmsg event loop thread failed :waiting on epoll fd : %s",
                strerror(errno)); goto errout;
        }

#if 0
        struct timespec __start_time;
        struct timespec __end_time;
        struct timespec __delay_time;

        /* 计时，时间处理开始时间 */
        clock_gettime(CLOCK_MONOTONIC, &__start_time);
#endif

        for (int i = 0; i < ret; i++)
        {
            int fd = evlist[i].data.fd;

            /* 新连接，仅对服务器模式有效 */
            if (fd == listen_fd && (evlist[i].events & EPOLLIN))
            {
                struct sockaddr_in remote_addr;
                socklen_t sz_remote_addr = sizeof(remote_addr);

                /* 使用accept4接口，在接收新链接时直接设置非阻塞选项、exec关闭选项 */
                int conn_fd = accept4(listen_fd, (struct sockaddr *)&remote_addr,
                    &sz_remote_addr, SOCK_CLOEXEC | SOCK_NONBLOCK);

                if (conn_fd < 0)
                {
                    MR_ERROR("Accept remote connnection failed : %s", strerror(errno));
                    continue;
                }

                __handle_new_conn(handler, epoll_fd, conn_fd, remote_addr);
                continue;
            }

            /* 已经存在的连接，先查找对应的conn */
            struct ctrlmsg_conn * conn = __lookup_conn_by_fd(handler, fd);
            if (conn == NULL)
            {
                MR_ERROR("Cannot locate conn structure for fd %d", fd);
                continue;
            }

            /* 异常链接 */
            if (evlist[i].events & (EPOLLHUP | EPOLLERR))
            {
                __handle_close(handler, conn, epoll_fd, fd);
                continue;
            }

            /* 数据到达 */
            if (evlist[i].events & EPOLLIN)
            {
                __handle_read(handler, conn, epoll_fd, fd);
                continue;
            }

            /* 描述符可写 */
            if (evlist[i].events & EPOLLOUT)
            {
                __handle_write(handler, conn, epoll_fd, fd);
                continue;
            }
        }

        /* 超时事件，喂狗 */
        if (ret == 0 && timeout >= 0)
            __handle_timeout(handler);

#if 0

        /* 计时，事件处理终止时间 */
        clock_gettime(CLOCK_MONOTONIC, &__end_time);
        timespec_diff(&__start_time, &__end_time, &__delay_time);

        MR_DEBUG("Eventloop: event handle delay: %lld.%09ld",
            (long long)__delay_time.tv_sec,  __delay_time.tv_nsec);

#endif
    }

errout:
    MR_ERROR("Ctrlmsg event loop thread terminated. ");
    return NULL;
}

static int __common_mode_create(struct ctrlmsg_handler * handle)
{
    int epoll_fd = epoll_create1(EPOLL_CLOEXEC);
    if (epoll_fd < 0)
    {
        MR_ERROR("Cannot create epoll fd: %s. ", strerror(errno));
        return RT_ERR;
    }

    int event_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (eventfd < 0)
    {
        MR_ERROR("Cannot create event fd: %s. ", strerror(errno));
        return RT_ERR;
    }

    handle->epoll_fd = epoll_fd;
    handle->event_fd = event_fd;
    return RT_SUCCESS;
}

static int __server_mode_create(struct ctrlmsg_handler * handle, struct sockaddr_in sockaddr_in)
{
    int ret = __common_mode_create(handle);
    if (ret < 0) return ret;

    // 监听本地端口，TCP连接
    int listen_fd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
    if (listen_fd < 0)
    {
        MR_ERROR("Create event listen fd failed : %s.", strerror(errno));
        goto out;
    }

    int reuse = 1;
    if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (const void*)&reuse, sizeof(int)) < 0)
    {
        MR_ERROR("Set crash event listen fd reuse failed : %s", strerror(errno));
        goto out;
    }

    int tcp_keepalive = 1;
    if (setsockopt(listen_fd, SOL_SOCKET, SO_KEEPALIVE, (const void*)&tcp_keepalive, sizeof(int)) < 0)
    {
        MR_ERROR("Set crash event listen fd keepalive failed : %s", strerror(errno));
        goto out;
    }

    int tcp_keepcnt = 4;
    if (setsockopt(listen_fd, IPPROTO_TCP, TCP_KEEPCNT, (const void*)&tcp_keepcnt, sizeof(int)) < 0)
    {
        MR_ERROR("Set crash event listen fd keepcnt failed : %s", strerror(errno));
        goto out;
    }

    int tcp_keepintvl = 1;
    if (setsockopt(listen_fd, IPPROTO_TCP, TCP_KEEPINTVL, (const void*)&tcp_keepintvl, sizeof(int)) < 0)
    {
        MR_ERROR("Set crash event listen fd keepintvl failed : %s", strerror(errno));
        goto out;
    }

    int tcp_keepidle = 5;
    if (setsockopt(listen_fd, IPPROTO_TCP, TCP_KEEPIDLE, (const void*)&tcp_keepidle, sizeof(int)) < 0)
    {
        MR_ERROR("Set crash event listen fd keepidle failed : %s", strerror(errno));
        goto out;
    }

    int tcp_user_timeout = 300;
    if (setsockopt(listen_fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (const void*)&tcp_user_timeout, sizeof(int)) < 0)
    {
        MR_ERROR("Set crash event listen fd user timeout failed : %s", strerror(errno));
        goto out;
    }

    ret = bind(listen_fd, (struct sockaddr *)&sockaddr_in, sizeof(struct sockaddr_in));
    if (ret < 0)
    {
        MR_ERROR("Bind ctrlmsg listen fd failed : %s.", strerror(errno));
        goto out;
    }

    ret = listen(listen_fd, 20);
    if (ret < 0)
    {
        MR_ERROR("Listen ctrlmsg listen fd failed : %s.", strerror(errno));
        goto out;
    }

    __epoll_add_event(handle->epoll_fd, listen_fd, EPOLLIN);

    char str_addr[MR_STRING_MAX];
    inet_ntop(AF_INET, &(sockaddr_in.sin_addr), str_addr, sizeof(str_addr));

    MR_INFO("Ctrlplane message module: Server mode");
    MR_INFO("Ctrlplane message server: %s:%d", str_addr, ntohs(sockaddr_in.sin_port));

    // 处理公共描述符
    handle->listen_fd = listen_fd;
    return RT_SUCCESS;

out:
    if (listen_fd > 0) close(listen_fd);
    return RT_ERR;
}

static int __client_mode_create(struct ctrlmsg_handler * handle, struct sockaddr_in sockaddr_in)
{
    int ret = __common_mode_create(handle);
    if (ret < 0) return ret;

    int conn_fd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
    if (conn_fd < 0)
    {
        MR_ERROR("Cannot create socket for ctrlplane message : %s.", strerror(errno));
        return RT_ERR;
    }

    ret = connect(conn_fd, (const struct sockaddr *)&sockaddr_in, sizeof(sockaddr_in));
    if (ret < 0)
    {
        MR_ERROR("Connect to ctrlplane message service failed : %s.", strerror(errno));
        return RT_ERR;
    }

    char str_server_addr[INET_ADDRSTRLEN] = { 0 };
    inet_ntop(AF_INET, &(sockaddr_in.sin_addr), str_server_addr, INET_ADDRSTRLEN);

    struct sockaddr local_sockaddr;
    socklen_t local_sockaddr_len = sizeof(struct sockaddr);
    if (getsockname(conn_fd, &local_sockaddr, &local_sockaddr_len) < 0)
    {
        MR_ERROR("Get local socket addr failed : %s", strerror(errno));
        return RT_ERR;
    }

    struct sockaddr_in * local_sockaddr_in = (struct sockaddr_in *)&local_sockaddr;

    char str_client_addr[INET_ADDRSTRLEN] = { 0 };
    inet_ntop(AF_INET, &(local_sockaddr_in->sin_addr), str_client_addr, INET_ADDRSTRLEN);

    MR_INFO("Ctrlplane message module: Client mode");
    MR_INFO("Ctrlplane message server: %s:%d", str_server_addr, ntohs(sockaddr_in.sin_port));
    MR_INFO("Ctrlplane message client: %s:%d", str_client_addr, ntohs(local_sockaddr_in->sin_port));

    __handle_new_conn(handle, handle->epoll_fd, conn_fd, sockaddr_in);
    return RT_SUCCESS;
}

struct ctrlmsg_handler * ctrlmsg_handler_create(enum ctrlmsg_handler_mode mode, struct sockaddr_in sockaddr,
    void * arg, int timeout)
{
    struct ctrlmsg_handler * handler;
    handler = LIBC_MALLOC(sizeof(struct ctrlmsg_handler));
    MR_VERIFY_MALLOC(handler);

    int ret = 0;
    handler->mode = mode;
    handler->timeout = timeout;
    handler->epoll_fd = 0;
    handler->event_fd = 0;
    handler->listen_fd = 0;

    TAILQ_INIT(&handler->cb_conn_close_list);
    TAILQ_INIT(&handler->cb_conn_new_list);
    TAILQ_INIT(&handler->cb_list);
    TAILQ_INIT(&handler->conn_list);

    if (handler->mode == CTRLMSG_HANDLER_MODE_SERVER)
    {
        ret = __server_mode_create(handler, sockaddr);
    }
    else if (handler->mode == CTRLMSG_HANDLER_MODE_CLIENT)
    {
        ret = __client_mode_create(handler, sockaddr);
    }

    if (ret < 0) goto errout;
    return handler;

errout:
    if (handler != NULL) free(handler);
    return NULL;
}

int ctrlmsg_thread_launch(struct ctrlmsg_handler * handler)
{
    assert(handler != NULL);
    pthread_t __pid = 0;

    int ret = pthread_create(&__pid, NULL, __epoll_event_loop, handler);
    if (ret < 0)
    {
        MR_ERROR("Launch ctrlmsg event dispatch thread failed : %s", strerror(errno));
        return ret;
    }

    return ret;
}

void ctrlmsg_msg_reciver_register(struct ctrlmsg_handler * handler, const char * msg_topic,
    unsigned int msg_type, ctrlmsg_cb_t cb, void * arg)
{
    struct ctrlmsg_cb * cb_object = LIBC_MALLOC(sizeof(struct ctrlmsg_cb));
    MR_VERIFY_MALLOC(cb_object);

    cb_object->fn_ctrlmsg_cb = cb;
    cb_object->arg = arg;
    cb_object->type = msg_type;
    strncpy(cb_object->topic, msg_topic, sizeof(cb_object->topic) - 1);
    TAILQ_INSERT_TAIL(&handler->cb_list, cb_object, next);
}

int ctrlmsg_msg_send(struct ctrlmsg_handler * handler, struct ctrlmsg_conn * conn,
    struct ctrl_msg_header * header)
{
    if (conn == NULL && handler->mode == CTRLMSG_HANDLER_MODE_CLIENT)
        conn = TAILQ_FIRST(&handler->conn_list);

    __MUTEX_LOCK(&conn->mutex);
    int msg_len = header->msg_len;
    if (conn->snd_buffer_used + msg_len >= conn->snd_buffer_max) return -ENOBUFS;
    memcpy(RTE_PTR_ADD(conn->snd_buffer, conn->snd_buffer_used), header, msg_len);
    conn->snd_buffer_used += msg_len;

    __epoll_mod_event(handler->epoll_fd, conn->conn_fd, EPOLLOUT | EPOLLIN);
    return 0;
}

void ctrlmsg_event_conn_new_register(struct ctrlmsg_handler * handler,
    ctrlmsg_cb_t cb, void * arg)
{
    struct ctrlmsg_cb * cb_object = LIBC_MALLOC(sizeof(struct ctrlmsg_cb));
    MR_VERIFY_MALLOC(cb_object);

    memset(cb_object, 0, sizeof(struct ctrlmsg_cb));
    cb_object->fn_ctrlmsg_cb = cb;
    cb_object->arg = arg;
    TAILQ_INSERT_TAIL(&handler->cb_conn_new_list, cb_object, next);
    return;
}

void ctrlmsg_event_conn_close_register(struct ctrlmsg_handler * handler,
    ctrlmsg_cb_t cb, void * arg)
{
    struct ctrlmsg_cb * cb_object = LIBC_MALLOC(sizeof(struct ctrlmsg_cb));
    MR_VERIFY_MALLOC(cb_object);

    memset(cb_object, 0, sizeof(struct ctrlmsg_cb));
    cb_object->fn_ctrlmsg_cb = cb;
    cb_object->arg = arg;
    TAILQ_INSERT_TAIL(&handler->cb_conn_close_list, cb_object, next);
    return;
}