myxce/src/include/netio.h

#if !defined(_NETIO_H_)

#define _NETIO_H_ 1

#include <openssl/err.h>
#include <openssl/ssl.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <sys/socket.h>

#include "dataq.h"
#include "mylock.h"

typedef int (*conn_handler_t)(struct epoll_event*);
typedef int (*msg_handler_t)(void*);
typedef int (*err_handler_t)(void*);

typedef struct {
    char* name;
    char* key;
} xce_peerinfo_t;

#define IO_MON_INPUT  (0x01)
#define IO_MON_OUTPUT (0x02)

typedef struct {
    data_item_t link;
    uint64_t    sz;
    char*       data;
} io_packet_t;

#define ITPTR_TO_PACKET(it)                                                    \
    ((io_packet_t*)(((char*)it) - offsetof(io_packet_t, link)))

typedef struct {
    mylock_t           lock;
    int                epoll_fd;
    struct epoll_event ev;
    pthread_t          worker;
    SSL_CTX*           ssl_ctx;
} iomgr_t;

typedef enum {
    FSM_IO_READ = 0,
    FSM_IO_WRITE
} io_fsm_op_t;

typedef enum {
    FSM_STATE_INIT = 0,
    FSM_STATE_READ_PKT_HEAD,
    FSM_STATE_READ_PKT,
    FSM_STATE_WRITE_PKT_HEAD,
    FSM_STATE_WRITE_PKT,
    FSM_STATE_DONE,
    FSM_STATE_ERROR
} io_fsm_state_t;

typedef struct {
    void*          ep;
    io_fsm_op_t    op;
    io_fsm_state_t state;
    io_packet_t*   pkt;
    size_t         iocount;
} io_fsm_t;

typedef struct endpoint_t {
    mylock_t        lock;
    iomgr_t*        iomgr;
    SSL*            ssl;
    int             fd;
    int             idx;
    int             flags;
    /* bug-1282: `busy` is read by endpoint_is_task_blocked from the vtask-
     * eligibility match callback (vtask_eligible), which runs UNDER the dataq
     * q->lock.  Taking ep->lock there inverted io_worker's ep->lock -> q->lock
     * order (io_worker holds ep->lock and enqueues via dataq_enq), an ABBA
     * deadlock that wedged the iomgr under chaotic failover churn.  Promoted from
     * a bitfield to a full _Atomic int (bitfields cannot be _Atomic — same reason
     * as dispatch_inflight) so the eligibility check reads it locklessly. */
    _Atomic int     busy;
    unsigned int    tearing_down       : 1;
    unsigned int    ssl_handshake_done : 1;
    /* bug-1197: count of peer-control dispatch attempts currently between the
     * tearing_down flag-read and the completion of their pool enqueue.  Bitfields
     * cannot be _Atomic, so this is a full atomic int.  endpoint_mark_tearing_down
     * drains this to 0 (after setting the flag) so no dispatch can enqueue a
     * same-ep vtask after teardown begins. */
    _Atomic int     dispatch_inflight;
    dataq_t         outq;
    uint64_t        seq;
    conn_handler_t  conn_handler;
    void*           owner;
    void*           data;
    io_fsm_t        fsm_in;
    io_fsm_t        fsm_out;
    xce_peerinfo_t* peer;
    msg_handler_t   msg_handler;
    err_handler_t   err_handler;
    time_t          ts_last_in;
    time_t          ts_last_out;
    void (*on_connected)(struct endpoint_t *ep); /* fire-once; called in iomgr thread after ssl_handshake_done=1 */
    /* bug-1298: reference count governing the lifetime of a HEAP-allocated
     * endpoint (endpoint_get).  An ep is fini'd + freed exactly when refcnt
     * drops to 0.  Holders: the creator/owner (endpoint_get sets refcnt=1, the
     * teardown path drops it via endpoint_unref) and EVERY ep-bound VTASK_IO
     * vtask (vtask_enq +1, run_task / taskq_item_killer -1) — so the vtask
     * dispatcher can read ep->busy (endpoint_is_task_blocked) on a queued vtask
     * without the io_worker freeing the ep out from under it.  Embedded eps
     * (struct members like client ctx->ep — heap_alloc==0) participate in the
     * count but are NEVER fini'd/freed by endpoint_unref; their owner manages
     * their lifetime directly. */
    _Atomic int     refcnt;
    int             heap_alloc;   /* 1 iff allocated by endpoint_get (unref may free) */
} endpoint_t;

#define INVOKE_MSG_HANDLER(ep)                                                 \
    do {                                                                       \
        TRACE("%s msg handler %p", __func__, (ep)->msg_handler);               \
        if ((ep)->msg_handler != NULL) {                                       \
            (ep)->msg_handler(ep);                                             \
            /* passing ownership of packet to upper layer */                   \
            (ep)->fsm_in.pkt = NULL;                                           \
        }                                                                      \
    } while (0)

#define INVOKE_ERR_HANDLER(ep)                                                 \
    do {                                                                       \
        if ((ep)->err_handler != NULL) {                                       \
            (ep)->err_handler(ep);                                             \
        }                                                                      \
    } while (0)

typedef struct {
    endpoint_t*  ep;
    io_packet_t* pkt;
} io_task_data_t;

extern int  alloc_packet(io_packet_t*);
extern void free_packet(io_packet_t*);

extern endpoint_t*  endpoint_get(void);
extern int          endpoint_init(endpoint_t*, int, int, int, int);
extern int          endpoint_connect(endpoint_t*, char*, int, int);
/* socket_connect: bounded, non-blocking TCP connect on an existing fd.
 * connect_timeout_ms == 0 uses DEFAULT_CONNECT_TIMEOUT_MS (config.h).
 * Returns 0 on success (fd left non-blocking), -ETIMEDOUT on bound expiry,
 * or -<errno> on error. raft-drive-async R2.3. */
extern int          socket_connect(int sock, char* host, int port,
                                   int connect_timeout_ms);
extern int          endpoint_accept(endpoint_t*, endpoint_t*);
extern int          endpoint_set_conn_handler(endpoint_t*, conn_handler_t);
extern int          endpoint_enq(endpoint_t*, data_item_t*);
extern data_item_t* endpoint_deq(endpoint_t*);
extern uint64_t     endpoint_getseq(endpoint_t*);
extern void         endpoint_block_tasks(endpoint_t*);
extern void         endpoint_unblock_tasks(endpoint_t*);
extern int          endpoint_is_task_blocked(endpoint_t*);
/* bug-1197: mark an endpoint as undergoing teardown so a late same-batch event
 * delivered by the io_worker after iomgr_unmonitor is dropped instead of
 * dispatched onto the vtask pool (which could outlive the ep). */
extern void         endpoint_mark_tearing_down(endpoint_t*);
extern int          endpoint_is_tearing_down(endpoint_t*);
extern int          endpoint_set_owner(endpoint_t*, void*);
extern int          endpoint_set_msg_handler(endpoint_t*, msg_handler_t);
extern int          endpoint_set_err_handler(endpoint_t*, err_handler_t);
extern int          endpoint_set_on_connected(endpoint_t*, void (*cb)(endpoint_t *));
extern void         endpoint_fini(endpoint_t*);
/* bug-1298: endpoint refcount. endpoint_ref takes a reference; endpoint_unref
 * drops one and, when the last reference is released, fini's the ep and (for
 * heap_alloc eps) frees it.  Use endpoint_unref in place of endpoint_fini+free
 * on every HEAP-ep teardown path so a free never races a vtask still reading
 * ep->busy.  Safe on NULL. */
extern void         endpoint_ref(endpoint_t*);
extern void         endpoint_unref(endpoint_t*);
extern int          endpoint_set_peer(endpoint_t*, xce_peerinfo_t*);

extern int io_fsm_start(io_fsm_t*, io_packet_t*);

extern void     initialize_openssl(void);
extern void     cleanup_openssl(void);
extern SSL_CTX* create_ssl_context(char*, char*);
extern void     iomgr_set_ssl_ctx(iomgr_t*, SSL_CTX*);

extern int iomgr_init(iomgr_t*);
extern int iomgr_monitor(endpoint_t*);
extern int iomgr_unmonitor(iomgr_t*, endpoint_t*);
extern int iomgr_start(iomgr_t*);
extern int iomgr_fini(iomgr_t*);
extern int iomgr_stop(iomgr_t*);
#endif