mizu.h

Go to the documentation of this file.

 
#ifndef MIZU_MIZU_H
#define MIZU_MIZU_H
 
#include <stddef.h>
#include <stdint.h>
 
#ifdef __cplusplus
#  include <atomic>
#  define MIZU_ATOMIC(T) std::atomic<T>
#  define MIZU_STATIC_ASSERT static_assert
extern "C" {
#else
#  include <stdatomic.h>
#  define MIZU_ATOMIC(T) _Atomic T
#  define MIZU_STATIC_ASSERT _Static_assert
#endif
 
#if defined(MIZU_SHARED)
#  if defined(_WIN32)
#    if defined(MIZU_BUILDING)
#      define MIZU_API __declspec(dllexport)
#    else
#      define MIZU_API __declspec(dllimport)
#    endif
#  else
#    define MIZU_API __attribute__((visibility("default")))
#  endif
#else
#  define MIZU_API
#endif
 
#define MIZU_VERSION_MAJOR 0
#define MIZU_VERSION_MINOR 1
#define MIZU_VERSION_PATCH 0
 
MIZU_API const char *mizu_version(void);
 
// Status and errors ------------------------------------------------------------
 
typedef enum mizu_status_e {
  MIZU_OK = 0,
  MIZU_FULL,
  MIZU_TIMEOUT,
  MIZU_CLOSED,
  MIZU_PEER_GONE,
  MIZU_ERR
} mizu_status;
 
typedef enum mizu_errcat_e {
  MIZU_ERRCAT_NONE = 0,
  MIZU_ERRCAT_NOSPACE,      
  MIZU_ERRCAT_NOMEMORY,     
  MIZU_ERRCAT_EXISTS,       
  MIZU_ERRCAT_EXHAUSTED,    
  MIZU_ERRCAT_STOPPED,      
  MIZU_ERRCAT_STAGE,        
  MIZU_ERRCAT_INTERRUPTED,  
  MIZU_ERRCAT_OTHER
} mizu_errcat;
 
MIZU_API mizu_errcat mizu_last_error_category(void);
MIZU_API const char *mizu_last_error_message(void);
 
// Opaque handles ---------------------------------------------------------------
 
typedef struct mizu_channel_s mizu_channel;
typedef struct mizu_pool_s mizu_pool;
typedef struct mizu_shm_s mizu_shm;
typedef struct mizu_binding_s mizu_binding;
 
// Wire format: channel region ----------------------------------------------------
 
#define MIZU_MAGIC        0x4D495A43u   
#define MIZU_ABI_VERSION  1u
 
typedef struct mizu_preamble_s {
  uint32_t magic;
  uint32_t version;          
  uint32_t cap;              
  uint32_t slot;             
  uint64_t host_pid;         
  uint64_t arena_size;       
  uint64_t drop_offset;      
  uint64_t drop_size;        
  uint64_t livedir_offset;   
  uint64_t livedir_size;
} mizu_preamble;
 
MIZU_STATIC_ASSERT(sizeof(mizu_preamble) == 64, "mizu_preamble is the wire format");
 
#define MIZU_DROP_R       0x52u   
#define MIZU_DROP_SOURCE  0x53u   
#define MIZU_ENTITY_HOST  0
#define MIZU_ENTITY_PEER  1
#define MIZU_ENTITY_OFFSET(i)  ((size_t) 128 + 64 * (size_t) (i))
#define MIZU_FIXED_LAYOUT_SIZE ((size_t) 512)
 
#define MIZU_OFF_READY      ((size_t) 64)
#define MIZU_OFF_CLOSED     ((size_t) 68)
#define MIZU_OFF_PEER_PID   ((size_t) 72)
#define MIZU_OFF_FLAGS      ((size_t) 80)
#define MIZU_FLAG_SPIN      1u   
#define MIZU_ENTITY_EPOCH   0
#define MIZU_ENTITY_PARKED  4
#define MIZU_ENTITY_REG     8
#define MIZU_ENTITY_IDENT   16
 
#define MIZU_OFF_HP_TAIL    ((size_t) 256)
#define MIZU_OFF_HP_HEAD    ((size_t) 320)
#define MIZU_OFF_PH_TAIL    ((size_t) 384)
#define MIZU_OFF_PH_HEAD    ((size_t) 448)
 
// Wire format: payload framing -------------------------------------------------
 
typedef enum mizu_kind_e {
  MIZU_KIND_INLINE = 0,
  MIZU_KIND_ARENA,
  MIZU_KIND_SHM_RAW,
  MIZU_KIND_RAWVEC,
  MIZU_KIND_SHM_VEC,
  MIZU_KIND_REF,
  MIZU_KIND_NIL,
  MIZU_KIND_STR1,
  MIZU_KIND_RAWSPILL
} mizu_kind;
 
#define MIZU_AUX_F_KEEPERLESS UINT64_C(1)
 
#define MIZU_STR1_NA UINT64_MAX   
#define MIZU_CE_NATIVE 0u
#define MIZU_CE_UTF8   1u
#define MIZU_CE_LATIN1 2u
#define MIZU_CE_BYTES  3u
 
typedef struct mizu_slot_hdr_s {
  uint32_t kind;             
  uint32_t len;
  uint64_t aux;
} mizu_slot_hdr;
 
MIZU_STATIC_ASSERT(sizeof(mizu_slot_hdr) == 16, "mizu_slot_hdr is the wire format");
 
typedef enum mizu_type_e {
  MIZU_TYPE_LGL = 10,         
  MIZU_TYPE_INT = 13,         
  MIZU_TYPE_REAL = 14,        
  MIZU_TYPE_CPLX = 15,        
  MIZU_TYPE_STR = 16,         
  MIZU_TYPE_VEC = 19,         
  MIZU_TYPE_RAW = 24,         
  MIZU_TYPE_INT64 = 32        
} mizu_type;
 
#define MIZU_NA_INT32     INT32_MIN              
#define MIZU_NA_INT64     INT64_MIN              
#define MIZU_NA_REAL_BITS 0x7FF80000000007A2ULL  
MIZU_API size_t mizu_type_elt_size(int type);
 
#define MIZU_ZC_REFCOUNT_OFF ((size_t) 24)
#define MIZU_ZC_FLAGS_OFF    ((size_t) 28)
#define MIZU_ZC_FLAG_REFHELD 1u
 
// Wire format: MIZH / MIZS / MIZL region layouts ---------------------------------
 
#define MIZU_HDR_FLAGS_OFF ((size_t) 32)
#define MIZU_HDR_FLAG_S4   1u
 
#define MIZU_HDR_VALID_OFF   ((size_t) 40)
#define MIZU_HDR_VALID_COUNT ((size_t) 48)
 
#define MIZU_MIZL_S4 0x40000000
 
#define MIZU_MAGIC_VEC   0x4D495A48u  
#define MIZU_MAGIC_STR   0x4D495A53u  
#define MIZU_MAGIC_LIST  0x4D495A4Cu  
#define MIZU_HEADER_SIZE 64
 
#define MIZU_NAME_MAX 30  
// Wire format: pool region -------------------------------------------------------
 
#define MIZU_POOL_MAGIC  0x4D495A50u  
#define MIZU_MAX_WORKERS 64
 
typedef struct mizu_pool_hdr_s {
  uint32_t magic;
  uint32_t version;
  uint32_t max_workers;      
  uint32_t max_submitters;   
  uint32_t inj_cap;          
  uint32_t deque_cap;        
  uint32_t result_slots;     
  uint32_t slot;             
  uint64_t owner_pid;        
  uint64_t livedir_offset;
  uint64_t livedir_size;
  MIZU_ATOMIC(uint64_t) worker_ident;
} mizu_pool_hdr;
 
MIZU_STATIC_ASSERT(sizeof(mizu_pool_hdr) == 64, "mizu_pool_hdr is the wire format");
 
typedef struct mizu_wk_slot_s {
  MIZU_ATOMIC(int32_t)  status;        
  int32_t          id;
  int64_t          pid;           
  MIZU_ATOMIC(int32_t)  park_state;    
  MIZU_ATOMIC(uint32_t) park_epoch;
  int64_t          deque_buf_off;
  int32_t          deque_cap;
  MIZU_ATOMIC(int32_t)  in_flight_rs;
  MIZU_ATOMIC(uint64_t) in_flight_seq;
  MIZU_ATOMIC(int64_t)  deque_bottom;
  MIZU_ATOMIC(int32_t)  retire;
  uint8_t          pad0[4];
  MIZU_ATOMIC(int64_t)  deque_top;
  uint64_t         live_dev;      
  uint64_t         live_ino;      
  MIZU_ATOMIC(uint64_t) stat_tasks;
  MIZU_ATOMIC(uint64_t) stat_steals;
  MIZU_ATOMIC(uint64_t) stat_inj;
  MIZU_ATOMIC(uint64_t) stat_parks;
  MIZU_ATOMIC(uint64_t) stat_helps;
} mizu_wk_slot;
 
MIZU_STATIC_ASSERT(sizeof(mizu_wk_slot) == 128, "mizu_wk_slot is the wire format");
 
typedef struct mizu_sub_slot_s {
  MIZU_ATOMIC(int32_t)  status;        
  MIZU_ATOMIC(uint32_t) park_epoch;
  int64_t          pid;
  uint32_t         rs_start;      
  uint32_t         rs_count;      
  uint64_t         live_dev;
  uint64_t         live_ino;
  MIZU_ATOMIC(uint64_t) stat_spills;
  MIZU_ATOMIC(uint64_t) stat_spill_reuse;
  uint8_t          pad[8];
} mizu_sub_slot;
 
MIZU_STATIC_ASSERT(sizeof(mizu_sub_slot) == 64, "mizu_sub_slot is the wire format");
 
typedef struct mizu_rs_hdr_s {
  MIZU_ATOMIC(int32_t)  status;        
  MIZU_ATOMIC(int32_t)  waiter_slot;   
  MIZU_ATOMIC(uint64_t) sequence;      
  MIZU_ATOMIC(int32_t)  worker_slot;   
  uint32_t         pad;
  mizu_slot_hdr     ph;
} mizu_rs_hdr;
 
MIZU_STATIC_ASSERT(sizeof(mizu_rs_hdr) == 40, "mizu_rs_hdr is the wire format");
 
typedef struct mizu_entry_hdr_s {
  uint64_t task_id;           
  uint64_t rs_seq;            
  uint32_t rs_index;
  uint16_t submitter_slot;
  uint16_t flags;
  mizu_slot_hdr ph;
} mizu_entry_hdr;
 
MIZU_STATIC_ASSERT(sizeof(mizu_entry_hdr) == 40, "mizu_entry_hdr is the wire format");
 
#define MIZU_ENTRY_RUNNER 1u
 
typedef enum mizu_wk_status_e { MIZU_WK_FREE = 0, MIZU_WK_CLAIMING, MIZU_WK_LIVE,
                               MIZU_WK_LEAVING, MIZU_WK_REAPING } mizu_wk_status;
typedef enum mizu_sub_status_e { MIZU_SUB_FREE = 0, MIZU_SUB_LIVE,
                                MIZU_SUB_REAPING } mizu_sub_status;
typedef enum mizu_rs_status_e { MIZU_RS_FREE = 0, MIZU_RS_PENDING, MIZU_RS_OK,
                               MIZU_RS_ERR, MIZU_RS_CANCEL,
                               MIZU_RS_DIED, MIZU_RS_LOST } mizu_rs_status;
typedef enum mizu_park_state_e { MIZU_WPK_RUNNING = 0, MIZU_WPK_IDLE,
                                MIZU_WPK_PARKED, MIZU_WPK_WAKING } mizu_park_state;
 
#define MIZU_INJ_META_SIZE     ((size_t) 128)
#define MIZU_INJ_TAIL_OFF      ((size_t) 0)
#define MIZU_INJ_HEAD_OFF      ((size_t) 64)
#define MIZU_TIER_READY_OFF    ((size_t) 0)
#define MIZU_TIER_FULL_OFF     ((size_t) 64)
#define MIZU_CTRL_SHUTDOWN_OFF ((size_t) 0)
#define MIZU_CTRL_PARKED_OFF   ((size_t) 64)
#define MIZU_CTRL_HELP_OFF     ((size_t) 128)
 
// Regions (mizu_shm) ----------------------------------------------------------------
 
#define MIZU_OPEN_VIEW_NOCOUNT 1u 
MIZU_API mizu_status mizu_shm_create(mizu_shm **out, size_t size);
MIZU_API mizu_status mizu_shm_open(mizu_shm **out, const char *name);
MIZU_API mizu_status mizu_shm_open_rw(mizu_shm **out, const char *name,
                                   int populate);
MIZU_API mizu_status mizu_shm_open_view(mizu_shm **out, const char *name);
MIZU_API mizu_status mizu_shm_open_view_flags(mizu_shm **out, const char *name,
                                           uint32_t flags);
MIZU_API void mizu_shm_close(mizu_shm *, int unlink);
MIZU_API void *mizu_shm_addr(mizu_shm *);
MIZU_API size_t mizu_shm_size(const mizu_shm *);
MIZU_API const char *mizu_shm_name(const mizu_shm *);
 
MIZU_API char **mizu_shm_reap(int *n);
 
MIZU_API void mizu_zc_ref(mizu_shm *);
MIZU_API void mizu_zc_unref(mizu_shm *);
MIZU_API uint32_t mizu_zc_refcount(const mizu_shm *);
MIZU_API uint32_t mizu_zc_flags(const mizu_shm *);
MIZU_API void mizu_zc_flag_refheld(mizu_shm *);
 
// Channel verbs --------------------------------------------------------------------
 
typedef struct mizu_channel_opts_s {
  uint32_t size;
  uint32_t capacity;      
  uint32_t slot_size;     
  uint64_t arena_size;    
  uint32_t flags;         
  const uint8_t *drop;    
  uint64_t drop_size;
} mizu_channel_opts;
 
#define MIZU_CHANNEL_SPIN MIZU_FLAG_SPIN
 
MIZU_API void mizu_channel_opts_init(mizu_channel_opts *);
 
MIZU_API mizu_status mizu_channel_create(mizu_channel **out,
                                      const mizu_channel_opts *,
                                      const mizu_binding *);
MIZU_API mizu_status mizu_channel_token(const mizu_channel *,
                                     char *buf, size_t cap);
MIZU_API mizu_status mizu_channel_attach(mizu_channel **out, const char *token,
                                      const mizu_binding *);
MIZU_API void mizu_channel_drop(const mizu_channel *, const uint8_t **bytes,
                              uint64_t *n);
MIZU_API mizu_status mizu_channel_ready_set(mizu_channel *);
MIZU_API mizu_status mizu_channel_ready_wait(mizu_channel *, double timeout_ms);
 
MIZU_API mizu_status mizu_channel_send(mizu_channel *, void *obj);
MIZU_API mizu_status mizu_channel_send_batch(mizu_channel *, void **objs,
                                          size_t n, size_t *accepted_out);
MIZU_API mizu_status mizu_channel_recv(mizu_channel *, void **obj_out,
                                    double timeout_ms);
MIZU_API mizu_status mizu_channel_recv_batch(mizu_channel *, void **objs,
                                          size_t cap, size_t *n_out,
                                          double timeout_ms);
 
typedef void (*mizu_obj_sink)(void *ctx, size_t i, void *obj);
MIZU_API mizu_status mizu_channel_recv_batch_fn(mizu_channel *, size_t cap,
                                             size_t *n_out, mizu_obj_sink,
                                             void *ctx, double timeout_ms);
 
MIZU_API mizu_status mizu_channel_close(mizu_channel *, double timeout_ms);
MIZU_API mizu_status mizu_channel_close_signal(mizu_channel *);
MIZU_API int mizu_channel_alive(const mizu_channel *);
MIZU_API void mizu_channel_destroy(mizu_channel *);
MIZU_API void mizu_channel_abandon(mizu_channel *);
 
typedef struct mizu_channel_info_s {
  uint32_t size;
  const char *name;
  int32_t side;             
  uint32_t capacity, slot_size;
  uint64_t arena_size, inline_max;
  int32_t spin, ready, closed;
  int64_t peer_pid;
  uint64_t tx_sent, tx_published, tx_consumed, rx_consumed, rx_published;
  uint64_t fl_entries, fl_hits;                 
  uint64_t open_hits, open_misses;              
  uint64_t ledger_entries;                      
  uint64_t zc_open_hits, zc_open_misses;        
} mizu_channel_info;
 
MIZU_API mizu_status mizu_channel_info_get(const mizu_channel *,
                                        mizu_channel_info *out);
 
MIZU_API mizu_errcat mizu_channel_errcat(const mizu_channel *);
MIZU_API const char *mizu_channel_error(const mizu_channel *);
 
// Pool verbs ---------------------------------------------------------------------
 
typedef struct mizu_pool_opts_s {
  uint32_t size;
  uint32_t max_workers;     
  uint32_t max_submitters;  
  uint32_t injection_cap;   
  uint32_t per_worker_cap;  
  uint32_t result_slots;    
  uint32_t slot_size;       
} mizu_pool_opts;
 
MIZU_API void mizu_pool_opts_init(mizu_pool_opts *);
 
typedef struct mizu_task_s {
  uint64_t word;
} mizu_task;
 
#define MIZU_TASK_SEQ_BITS 40
#define MIZU_TASK_SEQ_MAX ((UINT64_C(1) << MIZU_TASK_SEQ_BITS) - 1)
 
static inline uint64_t mizu_task_seq(const mizu_task *t) {
  return t->word & MIZU_TASK_SEQ_MAX;
}
static inline uint32_t mizu_task_rs_index(const mizu_task *t) {
  return (uint32_t) (t->word >> MIZU_TASK_SEQ_BITS);
}
static inline mizu_task mizu_task_make(uint64_t seq, uint32_t rs_index) {
  mizu_task t;
  t.word = (seq & MIZU_TASK_SEQ_MAX) |
           ((uint64_t) rs_index << MIZU_TASK_SEQ_BITS);
  return t;
}
 
MIZU_API mizu_status mizu_pool_create(mizu_pool **out, const mizu_pool_opts *,
                                   const mizu_binding *);
MIZU_API mizu_status mizu_pool_token(const mizu_pool *, char *buf, size_t cap);
MIZU_API mizu_status mizu_pool_ready_wait(mizu_pool *, const uint32_t *slots,
                                       size_t n, double timeout_ms);
MIZU_API void mizu_pool_destroy(mizu_pool *);
MIZU_API void mizu_pool_abandon(mizu_pool *);
 
MIZU_API mizu_status mizu_pool_attach(mizu_pool **out, const char *token,
                                   const mizu_binding *);
 
typedef enum mizu_worker_exit_e { MIZU_EXIT_SHUTDOWN = 0,
                                 MIZU_EXIT_OWNER_GONE,
                                 MIZU_EXIT_RETIRED,
                                 MIZU_EXIT_ERROR } mizu_worker_exit;
 
MIZU_API mizu_status mizu_pool_worker_join(mizu_pool **out, const char *token,
                                        uint32_t slot, const mizu_binding *);
MIZU_API mizu_worker_exit mizu_pool_worker_run(mizu_pool *);
MIZU_API int mizu_pool_lame_duck(mizu_pool *);
MIZU_API mizu_status mizu_pool_leave(mizu_pool *);
 
MIZU_API mizu_status mizu_pool_retire(mizu_pool *, uint32_t slot);
 
typedef enum mizu_step_result_e { MIZU_STEP_IDLE = 0, MIZU_STEP_TASK = 1,
                                 MIZU_STEP_SHUTDOWN = -1,
                                 MIZU_STEP_RETIRED = -2 } mizu_step_result;
MIZU_API int mizu_pool_step(mizu_pool *, double timeout_ms);
 
MIZU_API mizu_status mizu_pool_submit(mizu_pool *, void *task_obj,
                                   mizu_task *out, double timeout_ms);
MIZU_API mizu_status mizu_pool_submit_batch(mizu_pool *, void **objs, size_t n,
                                         mizu_task *out, size_t *n_out,
                                         double timeout_ms);
 
typedef void *(*mizu_obj_supply)(void *ctx, size_t i);
MIZU_API mizu_status mizu_pool_submit_batch_fn(mizu_pool *, mizu_obj_supply,
                                            void *ctx, size_t n,
                                            mizu_task *out, size_t *n_out,
                                            double timeout_ms);
 
MIZU_API mizu_status mizu_pool_collect(mizu_pool *, const mizu_task *,
                                    void **value_out, double timeout_ms);
MIZU_API mizu_status mizu_pool_collect_any(mizu_pool *, const mizu_task *,
                                        size_t n, size_t *index_out,
                                        void **value_out, double timeout_ms);
MIZU_API mizu_status mizu_pool_collect_all(mizu_pool *, const mizu_task *,
                                        size_t n, void **values_out,
                                        size_t *err_index_out,
                                        size_t *consumed_out,
                                        double timeout_ms);
 
MIZU_API mizu_status mizu_pool_collect_all_fn(mizu_pool *, const mizu_task *,
                                           size_t n, mizu_obj_sink,
                                           void *ctx,
                                           size_t *err_index_out,
                                           size_t *consumed_out,
                                           double timeout_ms);
 
typedef enum mizu_trace_event_e { MIZU_TRACE_SUBMIT = 0, MIZU_TRACE_START,
                                 MIZU_TRACE_DONE, MIZU_TRACE_ERROR,
                                 MIZU_TRACE_DROP,
                                 MIZU_TRACE_REHOME } mizu_trace_event;
typedef void (*mizu_trace_fn)(mizu_trace_event event, uint64_t task_id,
                             void *ctx);
MIZU_API mizu_status mizu_pool_set_trace(mizu_pool *, mizu_trace_fn, void *ctx);
 
MIZU_API int mizu_pool_cancel(mizu_pool *, const mizu_task *);
MIZU_API int mizu_pool_task_release(mizu_pool *, const mizu_task *);
MIZU_API int mizu_pool_task_state(mizu_pool *, const mizu_task *);
MIZU_API mizu_status mizu_pool_stop(mizu_pool *, double timeout_ms);
 
typedef struct mizu_pool_status_s {
  uint32_t size;
  const char *name;
  int32_t role;             
  uint32_t max_workers, max_submitters, injection_cap, result_slots,
           slot_size;
  uint32_t n_workers, n_submitters;
  uint8_t worker_state[MIZU_MAX_WORKERS];   
  uint8_t sub_state[64];                   
  int64_t deque_depth[MIZU_MAX_WORKERS];    
  uint64_t parked_mask;                    
  uint32_t tasks_by_state[7];              
  uint32_t inj_queued[64];                 
  int32_t shutdown;
} mizu_pool_status;
 
typedef enum mizu_role_e { MIZU_ROLE_CONTROLLER = 0, MIZU_ROLE_WORKER,
                          MIZU_ROLE_SUBMITTER } mizu_role;
 
typedef struct mizu_worker_stat_s {
  int32_t status, park_state;
  int64_t pid;
  uint64_t tasks, steals, injections, parks, helps;
  int64_t deque_top, deque_bottom;
  int32_t in_flight_rs;
} mizu_worker_stat;
 
typedef struct mizu_sub_stat_s {
  int32_t status;                   
  int64_t pid;
  uint32_t rs_start, rs_count;      
  uint64_t injected, claimed;       
  uint64_t spills, spill_reuse;     
  int32_t ready;                    
  int32_t full_waiter;              
} mizu_sub_stat;
 
typedef struct mizu_pool_dump_s {
  uint32_t size;
  mizu_pool_status status;
  uint32_t n_workers;                 
  mizu_worker_stat workers[MIZU_MAX_WORKERS];
  uint32_t n_submitters;              
  mizu_sub_stat submitters[64];
  uint64_t fl_entries, fl_bytes, fl_hits;       
  uint64_t open_hits, open_misses;              
  uint64_t collect_parks;                       
  uint64_t ledger_entries;                      
  int32_t help_wanted;
} mizu_pool_dump;
 
typedef struct mizu_rs_row_s {
  uint32_t slot;
  int32_t status;                   
  uint64_t sequence;
  int32_t worker_slot;
  int32_t waiter_slot;
} mizu_rs_row;
 
MIZU_API mizu_status mizu_pool_status_get(const mizu_pool *,
                                       mizu_pool_status *out);
MIZU_API mizu_status mizu_pool_dump_get(const mizu_pool *, mizu_pool_dump *out);
MIZU_API mizu_status mizu_pool_tasks_get(const mizu_pool *, mizu_rs_row *out,
                                      uint32_t cap, uint32_t *n_out);
 
MIZU_API mizu_errcat mizu_pool_errcat(const mizu_pool *);
MIZU_API const char *mizu_pool_error(const mizu_pool *);
 
#ifdef __cplusplus
}
#endif
 
#endif