myxce/src/lib/netio.c

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <poll.h>
#include <sched.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

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

static int iomgr_update_poll(iomgr_t*, endpoint_t*, int);
static int endpoint_update_epoll(endpoint_t*);

static int io_fsm_init(io_fsm_t* fsm, endpoint_t* ep, io_fsm_op_t op)
{
    if ((fsm == NULL) || (ep == NULL)
        || ((op != FSM_IO_READ) && (op != FSM_IO_WRITE))) {
        return -EINVAL;
    }

    fsm->ep      = ep;
    fsm->op      = op;
    fsm->state   = FSM_STATE_INIT;
    fsm->pkt     = NULL;
    fsm->iocount = 0;

    return 0;
}

int io_fsm_start(io_fsm_t* fsm, io_packet_t* pkt)
{
    endpoint_t* ep;

    if ((fsm == NULL) || (pkt == NULL)) {
        return -EINVAL;
    }

    ep = (endpoint_t*)fsm->ep;
    if ((ep == NULL) || (ep->ssl == NULL)) {
        return -EINVAL;
    }

    if ((fsm->state != FSM_STATE_INIT) && (fsm->state != FSM_STATE_DONE)) {
        return -EINVAL;
    }

    switch (fsm->op) {
    case FSM_IO_READ:
        fsm->state = FSM_STATE_READ_PKT_HEAD;
        break;
    case FSM_IO_WRITE:
        fsm->state = FSM_STATE_WRITE_PKT_HEAD;
        break;
    }

    fsm->pkt     = pkt;
    fsm->iocount = 0;

    return 0;
}

/*
 * io_classify_io_error: map a failed SSL_read/SSL_write (io_rc <= 0) to an FSM
 * step result, given SSL_get_error()'s value and the errno captured right after
 * the SSL call.
 *
 *   returns 0  → retriable would-block (WANT_READ/WANT_WRITE, or an
 *                EAGAIN/EWOULDBLOCK syscall): the FSM stays put and retries on
 *                the next readiness event.
 *   returns <0 → the peer closed or reset the connection: the caller MUST move
 *                the FSM to FSM_STATE_ERROR and propagate this so a blocked
 *                reply-waiter is woken.  A clean EOF reports errno==0, and the
 *                old code returned `-errno == 0` here — making the close look
 *                like success, so handle_input never reached handle_error and a
 *                blocked cbuf_get() hung forever (bug-648).  We therefore return
 *                -ECONNRESET whenever errno is 0.
 *
 * Pure (no SSL object, no endpoint state) so it is unit-testable.
 */
int io_classify_io_error(int ssl_err, int ecode, int io_rc)
{
    switch (ssl_err) {
    case SSL_ERROR_WANT_READ:
    case SSL_ERROR_WANT_WRITE:
        return 0;
    case SSL_ERROR_SYSCALL:
        if ((io_rc < 0) && ((ecode == EAGAIN) || (ecode == EWOULDBLOCK)))
            return 0;
        return (ecode != 0) ? -ecode : -ECONNRESET;
    default:
        /* SSL_ERROR_ZERO_RETURN (clean TLS close_notify) and any other SSL
         * error: the session is closed — never report success. */
        return (ecode != 0) ? -ecode : -ECONNRESET;
    }
}

static int io_fsm_do(io_fsm_t* fsm, char* data, size_t len)
{
    int         rc = 0;
    size_t      count;
    endpoint_t* ep = (endpoint_t*)fsm->ep;

    count = len - fsm->iocount;

    switch (fsm->op) {
    case FSM_IO_READ:
        rc = SSL_read(ep->ssl, data + fsm->iocount, count);
        break;
    case FSM_IO_WRITE:
        rc = SSL_write(ep->ssl, data + fsm->iocount, count);
        break;
    }

    if (rc <= 0) {
        int ecode = errno;
        int err   = SSL_get_error(ep->ssl, rc);

        /* Classify the failure.  A negative result means the connection is
         * closed/errored — move to ERROR and propagate (never return 0 on a
         * close, or a blocked reply-waiter is never woken: bug-648). */
        rc = io_classify_io_error(err, ecode, rc);
        if (rc < 0) {
            fsm->state = FSM_STATE_ERROR;
        }

        return rc;
    }

    fsm->iocount += rc;
    count -= rc;
    if (count == 0) {
        rc = fsm->iocount;

        switch (fsm->state) {
        case FSM_STATE_READ_PKT_HEAD:
            fsm->state   = FSM_STATE_READ_PKT;
            fsm->iocount = 0;
            if (alloc_packet(fsm->pkt) < 0) {
                fsm->state = FSM_STATE_ERROR;
                rc         = -ENOMEM;
            }
            break;
        case FSM_STATE_READ_PKT:
        case FSM_STATE_WRITE_PKT:
            fsm->state   = FSM_STATE_DONE;
            fsm->iocount = 0;
            break;
        case FSM_STATE_WRITE_PKT_HEAD:
            fsm->state   = FSM_STATE_WRITE_PKT;
            fsm->iocount = 0;
            break;
        case FSM_STATE_DONE:
            break;
        case FSM_STATE_INIT:
        case FSM_STATE_ERROR:
            /* should never go here and make compiler happy */
            rc = -EINVAL;
            break;
        }
    }

    return rc;
}

static int io_fsm_perform(io_fsm_t* fsm)
{
    char*  data;
    size_t len;

    if ((fsm == NULL) || (fsm->state == FSM_STATE_INIT)
        || (fsm->state == FSM_STATE_DONE) || (fsm->state == FSM_STATE_ERROR)) {
        return -EINVAL;
    }

    switch (fsm->state) {
    case FSM_STATE_READ_PKT_HEAD:
    case FSM_STATE_WRITE_PKT_HEAD:
        data = (char*)&(fsm->pkt->sz);
        len  = sizeof(fsm->pkt->sz);
        break;
    case FSM_STATE_READ_PKT:
    case FSM_STATE_WRITE_PKT:
        data = fsm->pkt->data;
        len  = fsm->pkt->sz;
        break;
    default:
        return -EINVAL;
    }

    return io_fsm_do(fsm, data, len);
}

static int set_fd_nonblocking(int fd)
{
    int flags = 0;
    int rc    = 0;

    flags = fcntl(fd, F_GETFL, 0);
    if (flags < 0) {
        return -errno;
    }

    rc = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    if (rc < 0) {
        return -errno;
    }

    return rc;
}

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

/*
 * socket_connect — bounded, non-blocking TCP connect (raft-drive-async R2.3).
 *
 * Sets O_NONBLOCK *before* connect() and, on EINPROGRESS, bounds the wait with
 * poll() up to connect_timeout_ms (0 => DEFAULT_CONNECT_TIMEOUT_MS). This stops
 * an unreachable peer from pinning a caller (notably the delaymgr heartbeat
 * thread) for the multi-second OS connect timeout. The fd is LEFT non-blocking
 * on success — endpoint_connect relies on that for the SSL handshake.
 *
 * Mirrors the xce_probe_config_tag bounded-connect pattern (config_repl.c).
 *
 * Returns:
 *   0            — connected (fd left non-blocking)
 *   -ETIMEDOUT   — poll bound expired before the connect completed
 *   -<errno>     — name resolution / socket / connect error (e.g. -ECONNREFUSED)
 */
int socket_connect(int sock, char* host, int port, int connect_timeout_ms)
{
    struct sockaddr_in server;
    struct hostent*    hp;
    int                rc;
    int                flags;
    int                timeout_ms;

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

        return -errno;
    }

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

    server.sin_port = htons(port);

    timeout_ms = (connect_timeout_ms > 0) ? connect_timeout_ms
                                          : DEFAULT_CONNECT_TIMEOUT_MS;

    /* Set non-blocking BEFORE connect so an unreachable peer returns
     * EINPROGRESS immediately instead of blocking. */
    flags = fcntl(sock, F_GETFL, 0);
    if (flags < 0) {
        SET_ERROR(errno);

        return -errno;
    }
    if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) < 0) {
        SET_ERROR(errno);

        return -errno;
    }

    rc = connect(sock, (struct sockaddr*)&server, sizeof(server));
    if (rc == 0) {
        /* Immediate completion (e.g. loopback). Leave fd non-blocking. */
        return 0;
    }
    if (errno != EINPROGRESS) {
        SET_ERROR(errno);

        return -errno;
    }

    /* Connect in progress — bound the wait with poll(). */
    {
        struct pollfd pfd;
        int           pr;

        pfd.fd      = sock;
        pfd.events  = POLLOUT | POLLERR | POLLHUP;
        pfd.revents = 0;

        do {
            pr = poll(&pfd, 1, timeout_ms);
        } while (pr < 0 && errno == EINTR);

        if (pr == 0) {
            /* Bound expired: report timeout instead of waiting for the OS. */
            SET_ERROR(ETIMEDOUT);

            return -ETIMEDOUT;
        }
        if (pr < 0) {
            SET_ERROR(errno);

            return -errno;
        }

        /* poll returned ready — confirm the connect actually succeeded. */
        {
            int       so_err = 0;
            socklen_t elen   = sizeof(so_err);

            if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &so_err, &elen) < 0) {
                SET_ERROR(errno);

                return -errno;
            }
            if (so_err != 0) {
                SET_ERROR(so_err);

                return -so_err;
            }
        }
    }

    /* Connected; fd left non-blocking for the caller's SSL handshake. */
    return 0;
}

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

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

#define MAX_EPOLL_EVENTS (5)

static int handle_input(endpoint_t* ep)
{
    int rc = 0;

    if ((ep->fsm_in.state == FSM_STATE_INIT)
        || (ep->fsm_in.state == FSM_STATE_DONE)) {
        io_packet_t* pkt;

        pkt = ep->fsm_in.pkt;
        if (pkt == NULL) {
            pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
            if (pkt == NULL) {
                return -ENOMEM;
            }
        } else {
            free_packet(pkt);
            ep->fsm_in.pkt = NULL;
        }

        io_fsm_start(&(ep->fsm_in), pkt);
        ep->ts_last_in = time(NULL);
    }

    rc = io_fsm_perform(&(ep->fsm_in));
    if (rc < 0) {
        return rc;
    }

    if (ep->fsm_in.state == FSM_STATE_DONE) {
        INVOKE_MSG_HANDLER(ep);
    }

    return 0;
}

static int handle_output(endpoint_t* ep)
{
    int rc = 0;

    if ((ep->fsm_out.state == FSM_STATE_INIT)
        || (ep->fsm_out.state == FSM_STATE_DONE)) {
        data_item_t* it  = NULL;
        io_packet_t* pkt = NULL;

        it = endpoint_deq(ep);
        if (it == NULL) {
            return 0;
        }

        pkt = ITPTR_TO_PACKET(it);
        rc  = io_fsm_start(&(ep->fsm_out), pkt);
        if (rc < 0) {
            return rc;
        }

        ep->ts_last_out = time(NULL);
    }

    rc = io_fsm_perform(&(ep->fsm_out));
    if (rc < 0) {
        return rc;
    }

    if (ep->fsm_out.state == FSM_STATE_DONE) {
        /* handle output done */
        io_packet_t* pkt = ep->fsm_out.pkt;

        if (pkt != NULL) {
            free_packet(pkt);
            FREEUP(pkt);
        }

        ep->fsm_out.pkt = NULL;

        /* Re-arm (drop EPOLLOUT now the queue is drained).  A failure here means
         * the fd is no longer registered — surface it so io_handler runs the
         * error path rather than leaving a silently-dead endpoint (bug-646). */
        rc = endpoint_update_epoll(ep);
        if (rc < 0) {
            return rc;
        }
    }

    return 0;
}

static int handle_error(endpoint_t* ep)
{
    iomgr_t* iomgr = (iomgr_t*)ep->iomgr;

    if (iomgr != NULL) {
        iomgr_unmonitor(iomgr, ep);
    }

    INVOKE_ERR_HANDLER(ep);

    return 0;
}

static int handle_ssl_handshake(endpoint_t* ep)
{
    int rc;
    int err;
    int ecode = 0;

    rc = SSL_do_handshake(ep->ssl);
    if (rc == 1) {
        ep->ssl_handshake_done = 1;
        if (ep->on_connected) {
            ep->on_connected(ep);
            ep->on_connected = NULL;
        }
        return 1;
    }

    ecode = errno;
    err   = SSL_get_error(ep->ssl, rc);
    switch (err) {
    case SSL_ERROR_WANT_READ:
    case SSL_ERROR_WANT_WRITE:
        rc = 0;
        break;
    default:
        /* Only log if it's a genuine error (not errno=0 after success) */
        if (ecode != 0) {
            errno = ecode;
            perror("SSL_do_handshake");
        }
        rc = handle_error(ep);
        break;
    }

    return rc;
}

static int io_handler(struct epoll_event* p)
{
    endpoint_t* ep = (endpoint_t*)p->data.ptr;
    int         rc = 0;

    if (ep == NULL) {
        return -EINVAL;
    }

    if (!ep->ssl_handshake_done) {
        return handle_ssl_handshake(ep);
    }

    if ((p->events & EPOLLIN) != 0) {
        if (ep->msg_handler != NULL) {
            rc = handle_input(ep);
        } else if (ep->conn_handler != NULL) {
            rc = ep->conn_handler(p);
        }
    }

    if (rc >= 0) {
        if ((p->events & EPOLLOUT) != 0) {
            rc = handle_output(ep);
        }
    }

    if (rc < 0) {
        rc = handle_error(ep);
    }

    return rc;
}

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++) {
            io_handler(&events[i]);
        }

        pthread_testcancel();
    }

    return NULL;
}

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

    if (!ep->ssl_handshake_done) {
        flags = flags | EPOLLOUT;
    } else if (ep->outq.head != NULL) {
        flags = flags | EPOLLOUT;
    } else {
        io_fsm_t* fsm = &(ep->fsm_out);

        switch (fsm->state) {
            /* we are in the middle stages of output */
        case FSM_STATE_WRITE_PKT_HEAD:
        case FSM_STATE_WRITE_PKT:
            flags = flags | EPOLLOUT;
            break;
        default:
            break;
        }
    }

    return flags;
}

static int endpoint_update_epoll(endpoint_t* ep)
{
    int flags;
    int rc = 0;

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

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

            if (rc1 >= 0) {
                ep->flags = flags;
            } else {
                /* Do NOT swallow this: an unarmed poll interest silently
                 * strands all I/O on the endpoint — outbound bytes never flush
                 * and inbound replies never wake a blocked cbuf_get(), so a
                 * synchronous caller hangs forever while io_worker sleeps in
                 * epoll_wait().  EPOLL_CTL_MOD fails with ENOENT when the fd is
                 * no longer registered (e.g. after a failed reconnect).
                 * Surface the failure so the caller tears the endpoint down and
                 * the failover loop rotates instead of hanging (bug-646). */
                rc = (saved_errno != 0) ? -saved_errno : -EIO;
                WARN("endpoint_update_epoll: iomgr_update_poll failed "
                     "(rc1=%d errno=%d) — endpoint marked dead",
                     rc1, saved_errno);
            }
        }
    }

    mylock_unlock(&(ep->lock));
    return rc;
}

static void outq_item_killer(data_item_t* it)
{
    if (it != NULL) {
        io_packet_t* pkt = ITPTR_TO_PACKET(it);

        free_packet(pkt);
        FREEUP(pkt);
    }
}

static int do_endpoint_init(endpoint_t* ep)
{
    if (ep == NULL) {
        return -EINVAL;
    }

    ep->flags              = 0;
    ep->fd                 = -1;
    ep->owner              = NULL;
    ep->data               = NULL;
    ep->iomgr              = NULL;
    ep->ssl                = NULL;
    ep->busy               = 0;
    ep->tearing_down       = 0;
    ep->dispatch_inflight  = 0;
    ep->ssl_handshake_done = 0;
    ep->msg_handler        = NULL;
    ep->err_handler        = NULL;

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

    io_fsm_init(&(ep->fsm_in), ep, FSM_IO_READ);
    io_fsm_init(&(ep->fsm_out), ep, FSM_IO_WRITE);

    return 1;
}
endpoint_t* endpoint_get()
{
    endpoint_t* ep;

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

        return ep;
    }

    do_endpoint_init(ep);

    /* bug-1298: a heap ep starts with the single creator/owner reference.  Set
     * AFTER do_endpoint_init (which never touches refcnt/heap_alloc, so a later
     * endpoint_init() re-init on this same heap ep preserves them). */
    ep->heap_alloc = 1;
    atomic_init(&ep->refcnt, 1);

    return ep;
}

/* bug-1298: take/drop an endpoint reference.  The last drop fini's the ep and,
 * for a heap ep, frees it.  Embedded eps (heap_alloc==0) are never fini'd/freed
 * here — their owner manages the struct — so the count is purely advisory for
 * them.  Both are NULL-safe and lockless (the count is _Atomic). */
void endpoint_ref(endpoint_t* ep)
{
    if (ep == NULL) {
        return;
    }
    atomic_fetch_add_explicit(&ep->refcnt, 1, memory_order_relaxed);
}

void endpoint_unref(endpoint_t* ep)
{
    if (ep == NULL) {
        return;
    }

    /* acq_rel: the decrement that observes the last reference must happen-after
     * every other holder's release so the fini/free sees no concurrent access.
     * fini+free ONLY for heap eps: an embedded ep (heap_alloc==0) is owned by its
     * container (e.g. client ctx->ep), which fini's it directly — a vtask-balanced
     * ref/unref pair on it must NOT fini or free the live struct. */
    if (atomic_fetch_sub_explicit(&ep->refcnt, 1, memory_order_acq_rel) == 1 &&
        ep->heap_alloc) {
        endpoint_fini(ep);
        free(ep);
    }
}

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

    do_endpoint_init(ep);
    ep->fd = fd;
    /* no need to ssl handle shake in server endpoint just for init connection
     */
    ep->ssl_handshake_done = 1;

    return fd;
}

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

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

    /* If ep->ssl is already set (e.g. pre-created with client context for
     * outgoing connections), use it; otherwise create from iomgr->ssl_ctx. */
    if (ep->ssl == NULL) {
        ep->ssl = SSL_new(iomgr->ssl_ctx);
        SSL_set_connect_state(ep->ssl);
    }

    SSL_set_fd(ep->ssl, ep->fd);

    set_fd_nonblocking(ep->fd);

    ep->flags              = 0;
    ep->ssl_handshake_done = 0;

    handle_ssl_handshake(ep);

    return rc;
}

int endpoint_accept(endpoint_t* ep, endpoint_t* ep2)
{
    int      sock  = -1;
    int      rc    = 0;
    iomgr_t* iomgr = (iomgr_t*)ep->iomgr;

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

    ep2->ssl = SSL_new(iomgr->ssl_ctx);

    rc = SSL_set_fd(ep2->ssl, sock);
    if (rc <= 0) {
        int err = SSL_get_error(ep2->ssl, rc);

        ERROR("SSL_set_fd returns error: %d", err);
        close(sock);
        return -errno;
    }

    SSL_set_accept_state(ep2->ssl);

    set_fd_nonblocking(sock);

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

    return sock;
}

int endpoint_set_owner(endpoint_t* ep, void* owner)
{
    if ((ep == NULL) || (owner == NULL)) {
        return -EINVAL;
    }

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

    return 1;
}

int endpoint_set_peer(endpoint_t* ep, xce_peerinfo_t* peer)
{
    mylock_lock(&(ep->lock));
    ep->peer = peer;
    mylock_unlock(&(ep->lock));

    return 0;
}

int endpoint_set_conn_handler(endpoint_t* ep, conn_handler_t h)
{
    if (ep->fd < 0) {
        SET_ERROR(EINVAL);

        return -EINVAL;
    }

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

    return 1;
}

int endpoint_set_msg_handler(endpoint_t* ep, msg_handler_t h)
{
    if ((ep == NULL) || (h == NULL)) {
        return -EINVAL;
    }

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

    return 1;
}

int endpoint_set_err_handler(endpoint_t* ep, err_handler_t h)
{
    if ((ep == NULL) || (h == NULL)) {
        return -EINVAL;
    }

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

    return 1;
}

int endpoint_set_on_connected(endpoint_t* ep, void (*cb)(endpoint_t*))
{
    if (ep == NULL) {
        return -EINVAL;
    }

    mylock_lock(&(ep->lock));
    ep->on_connected = cb;
    mylock_unlock(&(ep->lock));

    return 0;
}

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

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

        return -EINVAL;
    }

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

    /* Arm the output interest.  If the poll registration cannot be updated
     * (e.g. EPOLL_CTL_MOD on an fd no longer registered after a failed
     * reconnect), the enqueued bytes will never flush and the endpoint is dead.
     * Surface a RETRIABLE transport error so the caller propagates it (the
     * request.c senders return it; the client failover loop classifies
     * -ECONNRESET as retriable and rotates — bug-650).
     *
     * Do NOT invoke handle_error()/the err_handler here.  endpoint_enq() runs on
     * the *caller's* thread, which may already hold a lock the err_handler needs
     * — e.g. the controller's primary_heartbeat_fn holds ha->lock and
     * vote_ep_err_handler re-acquires it, self-deadlocking the controller
     * (bug-652).  The real teardown stays on the io_worker thread, which drives
     * handle_error() off the next epoll error event holding no caller locks. */
    if (endpoint_update_epoll(ep) < 0) {
        return -ECONNRESET;
    }

    return rc;
}

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

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

        return NULL;
    }

    mylock_lock(&(ep->lock));
    it = dataq_deq(&(ep->outq), NULL);
    mylock_unlock(&(ep->lock));

    return it;
}

uint64_t endpoint_getseq(endpoint_t* ep)
{
    uint64_t seq;

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

    return seq;
}

/* bug-1282: `busy` is a self-contained eligibility flag — the ep->lock that used
 * to wrap each access protected nothing else (it only gates the vtask queue).
 * Reading it under ep->lock from the vtask-eligibility path
 * (endpoint_is_task_blocked, called inside dataq_deq while holding the dataq
 * q->lock) inverted io_worker's ep->lock -> q->lock order and deadlocked the
 * iomgr (ABBA).  `busy` is now an _Atomic int, so all three sites use lockless
 * release/acquire and never touch ep->lock.  The inherent set-then-check TOCTOU
 * (busy may change right after the read) is unchanged — the lock never prevented
 * it; the eligibility decision was always advisory. */
void endpoint_block_tasks(endpoint_t* ep)
{
    TRACE("%s[%ld] block tasks for endpoint %p", __func__, pthread_self(), ep);
    atomic_store_explicit(&ep->busy, 1, memory_order_release);
}

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

int endpoint_is_task_blocked(endpoint_t* ep)
{
    int busy = atomic_load_explicit(&ep->busy, memory_order_acquire);

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

    return (busy != 0);
}

/* bug-1197: set the tearing-down flag, then DRAIN in-flight dispatches.
 *
 * Once set, tearing_down is never cleared (the ep is about to be endpoint_fini'd
 * + freed).  The flag is set under ep->lock so any late-event dispatch decision
 * (which reads it under ep->lock via endpoint_is_tearing_down) observes it.
 *
 * The drain closes the TOCTOU window: a dispatch increments dispatch_inflight
 * (seq_cst) BEFORE reading the flag and decrements it AFTER its pool enqueue
 * completes.  We release ep->lock before spinning (the dispatch's flag-read
 * takes ep->lock, so holding it here would deadlock the very dispatch we wait
 * on), then spin until the counter is 0.  When this returns, every dispatch
 * that read the flag as false has finished its enqueue; since the teardown
 * caller enqueues the teardown vtask only AFTER this returns, no same-ep
 * peer-control vtask can be enqueued after the teardown vtask.  The drained
 * window is just a calloc + non-blocking enqueue on the single io_worker, so
 * the spin is bounded and short. */
void endpoint_mark_tearing_down(endpoint_t* ep)
{
    mylock_lock(&(ep->lock));
    ep->tearing_down = 1;
    mylock_unlock(&(ep->lock));

    while (atomic_load_explicit(&ep->dispatch_inflight, memory_order_seq_cst)
           > 0) {
        sched_yield();
    }
}

int endpoint_is_tearing_down(endpoint_t* ep)
{
    int tearing;

    mylock_lock(&(ep->lock));
    tearing = ep->tearing_down;
    mylock_unlock(&(ep->lock));

    return (tearing != 0);
}

void endpoint_fini(endpoint_t* ep)
{
    dataq_fini(&(ep->outq));
    FREEUP(ep->data);

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

    if (ep->ssl != NULL) {
        SSL_free(ep->ssl);
        ep->ssl = NULL;
    }

    if (ep->fsm_in.pkt != NULL) {
        free_packet(ep->fsm_in.pkt);

        FREEUP(ep->fsm_in.pkt);
    }

    if (ep->fsm_out.pkt != NULL) {
        free_packet(ep->fsm_out.pkt);

        FREEUP(ep->fsm_out.pkt);
    }

    ep->flags       = 0;
    ep->msg_handler = NULL;
    ep->err_handler = NULL;
    ep->owner       = NULL;
}

void initialize_openssl()
{
    SSL_load_error_strings();
    OpenSSL_add_ssl_algorithms();
}

void cleanup_openssl() { EVP_cleanup(); }

SSL_CTX* create_ssl_context(char* cert_file, char* key_file)
{
    const SSL_METHOD* method;
    SSL_CTX*          ctx;

    /* any of cert_file or key_file is NULL, just as client */
    if ((cert_file == NULL) || (key_file == NULL)) {
        method = TLS_client_method();
        ctx    = SSL_CTX_new(method);
        return ctx;
    }

    method = TLS_server_method();

    ctx = SSL_CTX_new(method);
    if (ctx == NULL) {
        return NULL;
    }

    if (SSL_CTX_use_certificate_chain_file(ctx, cert_file) <= 0) {
        SSL_CTX_free(ctx);
        return NULL;
    }

    if (SSL_CTX_use_PrivateKey_file(ctx, key_file, SSL_FILETYPE_PEM) <= 0) {
        SSL_CTX_free(ctx);
        return NULL;
    }

    if (!SSL_CTX_check_private_key(ctx)) {
        SSL_CTX_free(ctx);
        return NULL;
    }

    return ctx;
}

int iomgr_init(iomgr_t* iomgr)
{
    /* Ignore SIGPIPE — writing to a closed peer socket returns EPIPE
     * instead of killing the daemon. All daemons call iomgr_init before
     * any socket I/O, so this protects controller, worker, and scheduler. */
    signal(SIGPIPE, SIG_IGN);

    /* bug-1406: establish every iomgr_t invariant here rather than assuming
     * the caller already zeroed the struct (do_endpoint_init follows this
     * same fully-explicit pattern for endpoint_t, a few lines above in this
     * file). Every production caller embeds iomgr_t inside a calloc'd/static
     * context struct, which previously masked this gap, but a caller that
     * stack-allocates an iomgr_t without a memset (test_ha_vep_teardown.c,
     * 4 call sites) left ssl_ctx as garbage; iomgr_fini's `!= NULL` guard then
     * passed on that garbage pointer and SSL_CTX_free() segfaulted on a wild
     * address. */
    iomgr->ssl_ctx = NULL;
    iomgr->worker  = 0;

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

        return -errno;
    }

    return 0;
}

void iomgr_set_ssl_ctx(iomgr_t* iomgr, SSL_CTX* ssl_ctx)
{
    if ((iomgr == NULL) || (ssl_ctx == NULL)) {
        return;
    }

    iomgr->ssl_ctx = ssl_ctx;
}

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(endpoint_t* ep)
{
    struct epoll_event ev;
    int                rc;
    int                flags;
    iomgr_t*           iomgr = (iomgr_t*)ep->iomgr;

    if (iomgr->ssl_ctx == NULL) {
        return -EINVAL;
    }

    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)
{
    if (iomgr->ssl_ctx != NULL) {
        SSL_CTX_free(iomgr->ssl_ctx);
        iomgr->ssl_ctx = NULL;
    }

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