myxce/src/lib/netio.c

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>

#include "dataq.h"
#include "netio.h"
#include "request.h"
#include "utils.h"

static int  iomgr_update_poll(iomgr_t*, endpoint_t*, int);
static void endpoint_fini(endpoint_t*);

static int socket_create(int domain,
                         int type,
                         int port,
                         int backlog)
{
    int sock;
    int opt = 1;

    if ((domain < 0) || (type < 0)) {
        return -EINVAL;
    }

    sock = socket(domain, type, 0);
    if (sock < 0) {
        SET_ERROR(errno);

        return -errno;
    }

    fcntl(sock, F_SETFD, (fcntl(sock, F_GETFD) | FD_CLOEXEC));

    if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
        close(sock);
        return -1;
    }

    if (port > 0) {
        struct sockaddr_in sin;

        memset((char*)&sin, 0, sizeof(sin));
        sin.sin_family      = domain;
        sin.sin_port        = htons(port);
        sin.sin_addr.s_addr = htonl(INADDR_ANY);

        if (bind(sock, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
            int rc;

            SET_ERROR(errno);

            rc = -errno;
            (void)close(sock);

            return rc;
        }

        if (backlog > 0) {
            if (listen(sock, backlog) < 0) {
                int rc;

                SET_ERROR(errno);

                rc = -errno;
                (void)close(sock);

                return rc;
            }
        }
    }

    return sock;
}

static int socket_connect(int   sock,
                          char* host,
                          int   port)
{
    struct sockaddr_in server;
    struct hostent*    hp;
    int                rc;

    server.sin_family = AF_INET;
    hp                = gethostbyname(host);
    if (hp == NULL) {
        SET_ERROR(errno);

        rc = -errno;

        return rc;
    }

    memcpy((char*)&server.sin_addr, (char*)hp->h_addr, (int)hp->h_length);

    server.sin_port = htons(port);

    rc = connect(sock, (struct sockaddr*)&server, sizeof(server));
    if (rc < 0) {
        SET_ERROR(errno);

        return -errno;
    }

    return rc;
}

static int socket_accept(int sock)
{
    socklen_t          len;
    int                s;
    struct sockaddr_in from;

    len = sizeof(from);
    s   = accept(sock, (struct sockaddr*)&from, &len);
    if (s < 0) {
        SET_ERROR(errno);

        return -errno;
    }

    fcntl(s, F_SETFD, (fcntl(s, F_GETFD) | FD_CLOEXEC));

    return s;
}

int netio_write(int   fd,
                char* data,
                int   sz)
{
    int nc = 0;

    while (sz > 0) {
        int rc;
        rc = write(fd, data + nc, sz);
        if (rc < 0) {
            SET_ERROR(errno);

            return -errno;
        }

        sz -= rc;
        nc += rc;
    }

    return nc;
}

int netio_read(int   fd,
               char* data,
               int   sz)
{
    int nc = 0;

    while (sz > 0) {
        int rc;

        rc = read(fd, data + nc, sz);
        if (rc < 0) {
            SET_ERROR(errno);

            return -errno;
        }

        if (rc == 0) {
            /* network read return 0, it seems remote
             * size has tear down the connection
             */
            return -EPIPE;
        }

        sz -= rc;
        nc += rc;
    }

    return nc;
}

int write_packet(int          fd,
                 io_packet_t* pkt)
{
    int rc;

    rc = netio_write(fd, (char*)&(pkt->sz),
                     sizeof(pkt->sz));
    if (rc < 0) {
        return -errno;
    }

    rc = netio_write(fd, pkt->data, pkt->sz);
    if (rc < 0) {
        return -errno;
    }
    return rc;
}

int alloc_packet(io_packet_t* pkt)
{
    if (pkt->sz <= 0) {
        SET_ERROR(EINVAL);

        return -EINVAL;
    }

    pkt->data = (char*)malloc(pkt->sz);
    if (pkt->data == NULL) {
        SET_ERROR(ENOMEM);

        return -ENOMEM;
    }

    return 1;
}
int read_packet(int          fd,
                io_packet_t* pkt)
{
    int rc;

    rc = netio_read(fd, (char*)&(pkt->sz),
                    sizeof(pkt->sz));
    if (rc < 0) {
        return rc;
    }

    rc = alloc_packet(pkt);
    if (rc < 0) {
        return rc;
    }

    rc = netio_read(fd, pkt->data, pkt->sz);
    if (rc < 0) {
        free_packet(pkt);
    }

    return rc;
}

void free_packet(io_packet_t* pkt)
{
    if (pkt->data != NULL) {
        free(pkt->data);
        pkt->data = NULL;
    }
}

#define MAX_EPOLL_EVENTS (5)

static void* io_worker(void* args)
{
    iomgr_t* iomgr = (iomgr_t*)args;

    while (1) {
        int                nfds = -1;
        int                i;
        struct epoll_event events[MAX_EPOLL_EVENTS];

        nfds = epoll_wait(iomgr->epoll_fd, events,
                          MAX_EPOLL_EVENTS, -1);
        if (nfds == -1) {
            if (errno == EINTR) {
                continue;
            }

            perror("epoll");

            return NULL;
        }

        for (i = 0; i < nfds; i++) {
            endpoint_t* p = (endpoint_t*)events[i].data.ptr;

            if (p->handler != NULL) {
                (p->handler)(&events[i]);
            }
        }

        pthread_testcancel();
    }

    return NULL;
}

static int endpoint_calc_flags(endpoint_t* ep)
{
    int flags = EPOLLIN;

    if (ep->outq.head != NULL) {
        flags = flags | EPOLLOUT;
    }

    return flags;
}

endpoint_t* endpoint_get()
{
    endpoint_t* ep;

    ep = (endpoint_t*)calloc(1, sizeof(*ep));
    if (ep == NULL) {
        SET_ERROR(errno);

        return ep;
    }

    ep->fd       = -1;
    ep->idx      = -1;
    ep->iomgr    = NULL;
    ep->busy     = 0;
    ep->needfree = 1;

    mylock_init(&(ep->lock));

    return endpoint_ref(ep);
}

static void outq_item_killer(data_item_t* it)
{
    if (it != NULL) {
        ioctx_t* ctx = ITPTR_TO_IOCTX(it);
        ioctx_deref(ctx, 1);
    }
}
int endpoint_init(endpoint_t* ep,
                  int         domain,
                  int         type,
                  int         port,
                  int         backlog)
{
    int fd = -1;

    if (ep == NULL) {
        SET_ERROR(EINVAL);

        return -EINVAL;
    }

    fd = socket_create(domain, type,
                       port, backlog);
    if (fd < 0) {
        return -errno;
    }

    ep->flags    = 0;
    ep->fd       = fd;
    ep->owner    = NULL;
    ep->data     = NULL;
    ep->iomgr    = NULL;
    ep->busy     = 0;
    ep->needfree = 0;

    dataq_init(&(ep->outq),
               outq_item_killer);
    mylock_init(&(ep->lock));

    endpoint_ref(ep);

    return fd;
}

int endpoint_connect(endpoint_t* ep,
                     char*       host,
                     int         port)
{
    int rc = -1;

    rc = socket_connect(ep->fd, host, port);
    if (rc < 0) {
        return rc;
    }

    ep->flags = 0;
    return rc;
}

int endpoint_accept(endpoint_t* ep,
                    endpoint_t* ep2)
{
    int sock = -1;

    sock = socket_accept(ep->fd);
    if (sock < 0) {
        return sock;
    }

    ep2->fd    = sock;
    ep2->flags = 0;

    return sock;
}

int endpoint_set_handler(endpoint_t*  ep,
                         msghandler_t h,
                         void*        owner)
{
    if (ep->fd < 0) {
        SET_ERROR(EINVAL);

        return -EINVAL;
    }

    mylock_lock(&(ep->lock));
    ep->handler = h;
    ep->owner   = owner;
    mylock_unlock(&(ep->lock));

    return 1;
}

int endpoint_enq(endpoint_t*  ep,
                 data_item_t* it)
{
    int rc = 0;
    int flags;

    if ((ep == NULL) || (it == NULL)) {
        SET_ERROR(EINVAL);

        return -EINVAL;
    }

    mylock_lock(&(ep->lock));
    rc = dataq_enq(&(ep->outq), it);

    flags = endpoint_calc_flags(ep);
    if (flags != ep->flags) {
        if (ep->iomgr != NULL) {
            int rc1;

            mylock_lock(&(ep->iomgr->lock));
            rc1 = iomgr_update_poll(ep->iomgr,
                                    ep, flags);
            mylock_unlock(&(ep->iomgr->lock));

            if (rc1 >= 0) {
                ep->flags = flags;
            } else {
                WARN("rc1=%d < 0", rc1);
            }
        }
    }
    mylock_unlock(&(ep->lock));

    return rc;
}

data_item_t* endpoint_deq(endpoint_t* ep)
{
    data_item_t* it = NULL;
    int          flags;

    if (ep == NULL) {
        SET_ERROR(EINVAL);

        return NULL;
    }

    mylock_lock(&(ep->lock));
    it    = dataq_deq(&(ep->outq), NULL);
    flags = endpoint_calc_flags(ep);
    if (flags != ep->flags) {
        if (ep->iomgr != NULL) {
            int rc;

            mylock_lock(&(ep->iomgr->lock));
            rc = iomgr_update_poll(ep->iomgr,
                                   ep, flags);
            mylock_unlock(&(ep->iomgr->lock));

            if (rc >= 0) {
                ep->flags = flags;
            }
        }
    }
    mylock_unlock(&(ep->lock));

    return it;
}

endpoint_t* endpoint_ref(endpoint_t* ep)
{
    mylock_lock(&(ep->lock));
    ep->refcount++;
    mylock_unlock(&(ep->lock));

    return ep;
}

void endpoint_deref(endpoint_t* ep)
{
    uint32_t refcount;

    mylock_lock(&(ep->lock));
    ep->refcount--;

    refcount = ep->refcount;
    mylock_unlock(&(ep->lock));

    if (refcount == 0) {
        endpoint_fini(ep);
    }
}

uint64_t endpoint_getseq(endpoint_t* ep)
{
    uint64_t seq;

    mylock_lock(&(ep->lock));
    seq = ep->seq++;
    mylock_unlock(&(ep->lock));

    return seq;
}

void endpoint_block_tasks(endpoint_t* ep)
{
    TRACE("%s[%ld] block tasks for endpoint %p",
          __func__, pthread_self(), ep);
    mylock_lock(&(ep->lock));
    ep->busy = 1;
    mylock_unlock(&(ep->lock));
}

void endpoint_unblock_tasks(endpoint_t* ep)
{
    TRACE("%s[%ld] unblock tasks for endpoint %p",
          __func__, pthread_self(), ep);
    mylock_lock(&(ep->lock));
    ep->busy = 0;
    mylock_unlock(&(ep->lock));
}

int endpoint_is_task_blocked(endpoint_t* ep)
{
    int busy;

    mylock_lock(&(ep->lock));
    busy = ep->busy;
    mylock_unlock(&(ep->lock));

    TRACE("%s[%ld] endpoint %p task %s",
          __func__, pthread_self(),
          ep, (busy != 0) ? "blocked" : "not blocked");

    return (busy != 0);
}

static void endpoint_fini(endpoint_t* ep)
{
    int needfree = ep->needfree;

    dataq_fini(&(ep->outq));
    FREEUP(ep->data);

    if (ep->fd >= 0) {
        close(ep->fd);
    }

    ep->flags   = 0;
    ep->handler = NULL;
    ep->owner   = NULL;

    if (needfree) {
        FREEUP(ep);
    }
}

int iomgr_init(iomgr_t* iomgr)
{
    mylock_init(&(iomgr->lock));
    iomgr->epoll_fd = epoll_create1(0);
    if (iomgr->epoll_fd < 0) {
        SET_ERROR(errno);

        return -errno;
    }

    return 0;
}

static int iomgr_update_poll(iomgr_t*    iomgr,
                             endpoint_t* ep,
                             int         flags)
{
    struct epoll_event ev;

    ev.events   = flags;
    ev.data.ptr = ep;

    return epoll_ctl(iomgr->epoll_fd,
                     EPOLL_CTL_MOD,
                     ep->fd, &ev);
}

int iomgr_monitor(iomgr_t*    iomgr,
                  endpoint_t* ep)
{
    struct epoll_event ev;
    int                rc;
    int                flags;

    flags       = EPOLLIN;
    ev.events   = flags;
    ev.data.ptr = ep;

    mylock_lock(&(ep->lock));
    ep->iomgr = iomgr;
    mylock_lock(&(iomgr->lock));
    rc = epoll_ctl(iomgr->epoll_fd, EPOLL_CTL_ADD,
                   ep->fd, &ev);
    mylock_unlock(&(iomgr->lock));

    if (rc >= 0) {
        ep->flags = flags;
    }

    mylock_unlock(&(ep->lock));

    return rc;
}

int iomgr_unmonitor(iomgr_t*    iomgr,
                    endpoint_t* ep)
{
    int rc;

    mylock_lock(&(iomgr->lock));
    rc = epoll_ctl(iomgr->epoll_fd, EPOLL_CTL_DEL,
                   ep->fd, NULL);
    if (rc < 0) {
        SET_ERROR(errno);

        rc = -errno;
    }
    mylock_unlock(&(iomgr->lock));

    return rc;
}

int iomgr_start(iomgr_t* iomgr)
{
    int rc;

    rc = pthread_create(&(iomgr->worker), NULL, &io_worker, iomgr);
    return rc;
}

int iomgr_fini(iomgr_t* iomgr)
{
    return 0;
}

int iomgr_stop(iomgr_t* iomgr)
{
    int   rc;
    void* res;

    if (iomgr->worker == 0) {
        return 0;
    }

    rc = pthread_cancel(iomgr->worker);
    if (rc != 0) {
        SET_ERROR(errno);

        return -errno;
    }

    rc = pthread_join(iomgr->worker, &res);
    if (rc != 0) {
        SET_ERROR(errno);

        return -errno;
    }

    if (res == PTHREAD_CANCELED) {
        iomgr->worker = 0;
        return 0;
    }

    return -1;
}