myxce/src/include/vtaskmgr.h

#if !defined(_VTASKMGR_H_)

#define _VTASKMGR_H_ 1

#include <pthread.h>
#include <stdatomic.h>
#include <stddef.h>

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

typedef int (*vtask_func_t)(void*);

#define VTASK_IO      0x01
#define VTASK_GENERIC 0x02

/* raft-drive-async task 1.6: LivenessAdmission task classes.
 *
 *   VTASK_LIVENESS   never parks — WAL-ack/heartbeat/vote receipt+send.
 *                    NEVER admission-gated, so >= VTASK_LIVENESS_RESERVED
 *                    workers are always free for it (deadlock-freedom).
 *   VTASK_LIVE_BLOCK bounded mandatory park — vote-grant term fsync (R4.6).
 *                    Runs in the reserved headroom; not gated by the
 *                    REQ_BLOCK cap (it is short and bounded).
 *   VTASK_REQ_BLOCK  may park seconds — commit-wait / config-ack wait.
 *                    Admission-gated: rejected with -E_XCE_NOT_READY when
 *                    admitting it would breach the reservation. */
typedef enum {
    VTASK_LIVENESS = 0,
    VTASK_LIVE_BLOCK,
    VTASK_REQ_BLOCK
} vtask_class_t;

typedef struct {
    int vtype;
    union {
        void*          generic;
        io_task_data_t iodata;
    } u;
} vtask_data_t;

typedef struct {
    data_item_t   link;
    void*         owner;
    vtask_data_t* data;
    int           vcpu;
    vtask_func_t  entry;
    vtask_class_t cls;  /* LivenessAdmission class (task 1.6); default LIVENESS */
    /* raft-drive-async task 3.1: when non-zero AND data->vtype == VTASK_IO, the
     * carried io_packet_t (data->u.iodata.pkt) is OWNED by the task and must be
     * freed by taskq_item_killer if the task is destroyed while still queued
     * (never executed) at vtaskmgr_fini.  Default 0 preserves the legacy
     * endpoint-owned-packet model used by the libxce client path (xcectx.c's
     * vtask_handle_msg does NOT clear ep->fsm_in.pkt, so the packet lifetime is
     * tied to the endpoint, not the task — freeing it here would double-free).
     * The controller's peer-control dispatch sets this to 1 because it clears
     * ep->fsm_in.pkt and the task becomes the sole owner. */
    int           owns_io_pkt;
} vtask_t;

#define ITPTR_TO_VTASK(p) ((vtask_t*)(((char*)p) - offsetof(vtask_t, link)))

typedef enum {
    VCPU_IDLE = 0,
    VCPU_BUSY,
    VCPU_ERR
} vcpu_state_t;

typedef struct {
    int          cpuid;
    vcpu_state_t state;
    void*        owner;
    cbuf_t       wbuf;
    vtask_t*     task;
    pthread_t    vcpu;
} vcpu_t;

typedef struct {
    int       n_vcpus;
    pthread_t dispatcher;
    vcpu_t*   vcpus;
    cbuf_t    wbuf;
    cbuf_t    wtbuf; /* used to wake up vtaskmgr to check tasks */
    dataq_t   taskq;
    /* LivenessAdmission (raft-drive-async task 1.6): pool-level reservation.
     * liveness_reserved == 0 disables the gate (legacy: nothing gated).
     * blocking_inflight counts in-flight *blocking-admission* holders
     * (VTASK_REQ_BLOCK tasks + commit_submit waits folded onto this seam);
     * capped at max(1, n_vcpus - liveness_reserved). */
    int             liveness_reserved;
    int             blocking_inflight;
    pthread_mutex_t blk_lock; /* guards blocking_inflight check-and-increment */
} vtaskmgr_t;

extern vtask_t* vtask_alloc();
extern int      vtask_init(vtask_t*, vtask_func_t, void*);
extern void     vtask_fini(vtask_t*);

extern int vtask_enq(vtaskmgr_t*, vtask_t*);

/* bug-1864: dequeue the first dispatchable task from a vtaskmgr task queue.
 * is_blocked reports whether an endpoint already has a task running
 * (production: endpoint_is_task_blocked); it is a parameter so a test can
 * make an endpoint unblock part-way through one scan. */
extern vtask_t* vtask_deq_eligible(dataq_t* taskq,
                                   int (*is_blocked)(endpoint_t*));
extern int vtask_handle_msg(vtaskmgr_t*, vtask_func_t, void* arg);

/* LivenessAdmission (raft-drive-async task 1.6) — pool-level reservation.
 *
 * vtaskmgr_set_liveness_reserved: set R, the count of vcpus reserved for
 * never-gated VTASK_LIVENESS work. The controller resolves R via
 * drive_async_liveness_reserved() (the single source of truth) and wires it
 * here at init. R == 0 disables the gate (legacy behavior). */
extern void vtaskmgr_set_liveness_reserved(vtaskmgr_t*, int reserved);

/* Effective blocking-admission cap = max(1, n_vcpus - liveness_reserved)
 * (the cap is floored at 1 so a commit can always run); returns n_vcpus when
 * the gate is disabled (reserved == 0). */
extern int vtaskmgr_blocking_cap(const vtaskmgr_t*);

/* Acquire/release a blocking-admission slot. acquire returns 0 when admitted
 * or -E_XCE_NOT_READY when admitting would breach the reservation (fast
 * reject — never blocks). When the gate is disabled (reserved == 0) acquire
 * always succeeds and is uncounted. This is the seam commit.c folds onto:
 * commit_submit acquires before its blocking wait and releases on wake. */
extern int  vtaskmgr_blocking_acquire(vtaskmgr_t*);
extern void vtaskmgr_blocking_release(vtaskmgr_t*);

/* Classified submit. VTASK_REQ_BLOCK is admission-gated: returns
 * -E_XCE_NOT_READY (task NOT enqueued — caller still owns it) when admitting
 * would breach the reservation. VTASK_LIVENESS / VTASK_LIVE_BLOCK are never
 * gated (always enqueued). On success the task is enqueued and owned by the
 * pool. */
extern int vtask_enq_classified(vtaskmgr_t*, vtask_t*, vtask_class_t);

extern int  vtaskmgr_init(vtaskmgr_t*, int);
extern int  vtaskmgr_start(vtaskmgr_t*);
extern void vtaskmgr_fini(vtaskmgr_t*);
extern void vtaskmgr_wait(vtaskmgr_t*);
extern void vtaskmgr_wakeup(vtaskmgr_t*);

#endif