ut_rpc_svr.c

/*
 * Description:
 *     History: yang@haipo.me, 2016/03/29, create
 */

# include <assert.h>

# include "ut_rpc_svr.h"
# include "ut_misc.h"
# include "ut_pack.h"
# include "ut_log.h"
# include "nw_sock.h"

struct clt_info {
    double last_heartbeat;
    double heartbeat_timeout;
};

static void on_error_msg(nw_ses *ses, const char *msg)
{
    log_error("peer: %s: %s", nw_sock_human_addr(&ses->peer_addr), msg);
}

static void on_new_connection(nw_ses *ses)
{
    struct clt_info *info = ses->privdata;
    info->last_heartbeat = current_timestamp();
    info->heartbeat_timeout = RPC_HEARTBEAT_TIMEOUT_DEFAULT;
    rpc_svr *svr = rpc_svr_from_ses(ses);
    if (svr->on_new_connection)
        svr->on_new_connection(ses);
}

static int on_heartbeat(nw_ses *ses, rpc_pkg *pkg)
{
    struct clt_info *info = ses->privdata;
    info->last_heartbeat = current_timestamp();

    void *p = pkg->body;
    size_t left = pkg->body_size;
    while (left > 0) {
        uint16_t type;
        uint16_t len;
        ERR_RET_LN(unpack_uint16_le(&p, &left, &type));
        ERR_RET_LN(unpack_uint16_le(&p, &left, &len));
        if (left < len) {
            return -__LINE__;
        }
        switch (type) {
        case RPC_HEARTBEAT_TYPE_TIMEOUT:
            {
                if (len != sizeof(uint32_t)) {
                    return -__LINE__;
                }
                double timeout = le32toh(*((uint32_t *)p));
                timeout /= 1000;
                if (timeout > RPC_HEARTBEAT_TIMEOUT_MAX) {
                    timeout = RPC_HEARTBEAT_TIMEOUT_MAX;
                } else if (timeout < RPC_HEARTBEAT_TIMEOUT_MIN) {
                    timeout = RPC_HEARTBEAT_TIMEOUT_MIN;
                }
                if (info->heartbeat_timeout != timeout) {
                    log_info("peer: %s update heartbeat timeout from: %f to: %f",
                            nw_sock_human_addr(&ses->peer_addr), info->heartbeat_timeout, timeout);
                    info->heartbeat_timeout = timeout;
                }
            }
            break;
        }
        p += len;
        left -= len;
    }

    pkg->pkg_type = RPC_PKG_TYPE_REPLY;
    pkg->body_size = 0;
    rpc_send(ses, pkg);

    return 0;
}

static void on_recv_pkg(nw_ses *ses, void *data, size_t size)
{
    struct rpc_pkg pkg;
    memcpy(&pkg, data, RPC_PKG_HEAD_SIZE);
    pkg.ext = data + RPC_PKG_HEAD_SIZE;
    pkg.body = pkg.ext + pkg.ext_size;

    if (pkg.command == RPC_CMD_HEARTBEAT) {
        int ret = on_heartbeat(ses, &pkg);
        if (ret < 0) {
            sds hex = bin2hex(data, size);
            log_error("peer: %s, on_heartbeat fail: %d, data: %s", nw_sock_human_addr(&ses->peer_addr), ret, hex);
            nw_svr_close_clt(ses->svr, ses);
            sdsfree(hex);
        }
        return;
    }

    rpc_svr *svr = rpc_svr_from_ses(ses);
    svr->on_recv_pkg(ses, &pkg);
}

static void *on_privdata_alloc(void *svr)
{
    rpc_svr *r_svr = ((nw_svr *)svr)->privdata;
    void *privdata = nw_cache_alloc(r_svr->privdata_cache);
    return privdata;
}

static void on_privdata_free(void *svr, void *privdata)
{
    rpc_svr *r_svr = ((nw_svr *)svr)->privdata;
    return nw_cache_free(r_svr->privdata_cache, privdata);
}

static void on_timer(nw_timer *timer, void *privdata)
{
   rpc_svr *svr = privdata;
   double now = current_timestamp();

   nw_ses *curr = svr->raw_svr->clt_list_head;
   nw_ses *next;
   while (curr) {
       next = curr->next;
       struct clt_info *info = curr->privdata;
       if (now - info->last_heartbeat > info->heartbeat_timeout) {
           log_error("peer: %s: heartbeat timeout, last_heartbeat: %f, timeout: %f",
                   nw_sock_human_addr(&curr->peer_addr), info->last_heartbeat, info->heartbeat_timeout);
           nw_svr_close_clt(svr->raw_svr, curr);
       }
       curr = next;
   }
}

rpc_svr *rpc_svr_create(rpc_svr_cfg *cfg, rpc_svr_type *type)
{
    if (type->on_recv_pkg == NULL)
        return NULL;

    nw_svr_cfg raw_cfg;
    memset(&raw_cfg, 0, sizeof(raw_cfg));
    raw_cfg.bind_count = cfg->bind_count;
    raw_cfg.bind_arr = cfg->bind_arr;
    raw_cfg.max_pkg_size = cfg->max_pkg_size;
    raw_cfg.buf_limit = cfg->buf_limit;
    raw_cfg.read_mem = cfg->read_mem;
    raw_cfg.write_mem = cfg->write_mem;

    nw_svr_type raw_type;
    memset(&raw_type, 0, sizeof(raw_type));
    raw_type.decode_pkg = rpc_decode;
    raw_type.on_connection_close = type->on_connection_close;
    raw_type.on_recv_fd = type->on_recv_fd;
    raw_type.on_error_msg = on_error_msg;
    raw_type.on_new_connection = on_new_connection;
    raw_type.on_recv_pkg = on_recv_pkg;
    raw_type.on_privdata_alloc = on_privdata_alloc;
    raw_type.on_privdata_free = on_privdata_free;

    rpc_svr *svr = malloc(sizeof(rpc_svr));
    assert(svr != NULL);
    memset(svr, 0, sizeof(rpc_svr));
    svr->raw_svr = nw_svr_create(&raw_cfg, &raw_type, svr);
    if (svr->raw_svr == NULL) {
        free(svr);
        return NULL;
    }
    if (cfg->heartbeat_check) {
        nw_timer_set(&svr->timer, RPC_HEARTBEAT_INTERVAL, true, on_timer, svr);
    }
    svr->privdata_cache = nw_cache_create(sizeof(struct clt_info));
    assert(svr->privdata_cache != NULL);
    svr->heartbeat_check = cfg->heartbeat_check;
    svr->on_recv_pkg = type->on_recv_pkg;
    svr->on_new_connection = type->on_new_connection;

    return svr;
}

int rpc_svr_start(rpc_svr *svr)
{
    int ret = nw_svr_start(svr->raw_svr);
    if (ret < 0)
        return ret;
    if (svr->heartbeat_check) {
        nw_timer_start(&svr->timer);
    }
    return 0;
}

int rpc_svr_stop(rpc_svr *svr)
{
    int ret = nw_svr_stop(svr->raw_svr);
    if (ret < 0)
        return ret;
    nw_timer_stop(&svr->timer);
    return 0;
}

void rpc_svr_release(rpc_svr *svr)
{
    nw_svr_release(svr->raw_svr);
    nw_timer_stop(&svr->timer);
    nw_cache_release(svr->privdata_cache);
    free(svr);
}

void rpc_svr_close_clt(rpc_svr *svr, nw_ses *ses)
{
    nw_svr_close_clt(svr->raw_svr, ses);
}

rpc_svr *rpc_svr_from_ses(nw_ses *ses)
{
    return ((nw_svr *)ses->svr)->privdata;
}