utun2/lib/u_async.h

/* Цикл событий одного потока: сокеты (epoll/poll/select), таймеры и FIFO call_soon.
 * create → регистрация callbacks → poll/mainloop → destroy вне callbacks после остановки владельцев.
 * Таймауты задаются в timebase 0.1мс; arg принадлежит вызывающему, освобождает его callback/владелец.
 * API сокетов/таймеров используется в потоке loop. Из другого потока — post/post_reserved и wakeup;
 * публикация не продлевает время жизни ua/arg, их освобождение нужно согласовать с остановкой producers. */

#ifndef UASYNC_H
#define UASYNC_H

#ifdef __cplusplus
extern "C" {
#endif


#include "platform_compat.h"
#include <stddef.h>
#include <signal.h>
#include "timeout_heap.h"
#include "socket_compat.h"

typedef void (*timeout_callback_t)(void* user_arg);// передаёт user_arg из uasync_set_timeout
typedef void (*socket_callback_t)(int fd, void* user_arg);// передаёт user_arg из uasync_add_socket (for pipe/file)
typedef void (*socket_t_callback_t)(socket_t sock, void* user_arg);// передаёт user_arg из uasync_add_socket_t (for sockets)
// user_arg полезен если нужно передать управляющую структуру. Ее можно выделить в памяти и в ней хранить всё что надо. т.е. при set_timeout передаём и получаем ее в callback-е


// Error type
typedef int err_t;
#define ERR_OK 0
#define ERR_FAIL -1

// Socket node types
#define SOCKET_NODE_TYPE_FD     0  // Regular file descriptor (pipe, file)
#define SOCKET_NODE_TYPE_SOCK   1  // Socket (socket_t)

typedef void (*uasync_post_callback_t)(void* user_arg);

#include "memory_pool.h"

struct timeout_node;  // Forward declaration

/* CPU одного потока: 100% = одно ядро. При работе цикла/запросе снимок обновляется раз в 100 мс, без пробуждений.
 * До первого замера valid=0; window_us растёт до секунды. При задержке замера
 * вклад граничного интервала в среднее оценивается пропорционально времени. */
struct uasync_load {
    double current_percent;
    double average_percent;
    uint64_t sample_time_us;
    uint64_t interval_us;
    uint64_t window_us;
    int valid;
};

struct uasync_load_interval {
    uint64_t start_us, end_us, cpu_ns;
};

struct posted_task {
        uasync_post_callback_t callback;
        void* arg;
        struct posted_task* next;
    };

// Uasync instance structure
struct UASYNC {
    struct memory_pool* timeout_pool;  // Pool for timeout_node allocation
    struct timeout_node* immediate_queue_head;  // FIFO queue for immediate execution
    struct timeout_node* immediate_queue_tail;
    TimeoutHeap* timeout_heap;  // Heap for timeout management
    struct socket_array* sockets; // Array-based socket management
    // Debug counters for memory allocation tracking
    size_t timer_alloc_count;
    size_t timer_free_count;
    size_t socket_alloc_count;
    size_t socket_free_count;
    // Wakeup pipe for interrupting poll
    int wakeup_pipe[2]; // [0] read, [1] write
    int wakeup_initialized;
    // Cached pollfd array for optimized poll operations (fallback for non-Linux)
    struct pollfd* poll_fds;      // Pre-allocated pollfd array
    int poll_fds_capacity;        // Current capacity of poll_fds
    int poll_fds_count;           // Number of valid entries in poll_fds
    int poll_fds_dirty;           // 1 if poll_fds needs rebuild
    // Epoll support (Linux only)
    int epoll_fd;                 // epoll file descriptor (-1 if not using epoll)
    int use_epoll;                // 1 if using epoll, 0 if using poll
    struct posted_task* posted_tasks_head;
    struct posted_task* posted_tasks_tail;

#ifdef _WIN32
    CRITICAL_SECTION posted_lock;
#else
    pthread_mutex_t posted_lock;
#endif
    volatile sig_atomic_t stop;
    volatile sig_atomic_t running;   // 0=shutdown mode (API rejected), 1=full init done
    uint64_t last_poll_exit_us;
    struct uasync_load load;
    struct uasync_load_interval load_history[12]; // 10 периодов + граничные интервалы при задержках
    uint64_t load_previous_us, load_previous_cpu_ns, load_next_us, load_log_us;
    unsigned int load_head, load_count;
    int load_state;              // 0=не начат, 1=работает, -1=ошибка часов
};

// Type definitions
typedef struct UASYNC uasync_t;
typedef struct UASYNC UASYNC_t;

static inline int uasync_is_running(struct UASYNC* ua) {
    if (!ua) return 0;
    return __atomic_load_n(&ua->running, __ATOMIC_ACQUIRE);
}

// Instance API - основной API для работы с uasync
struct UASYNC* uasync_create(void);
// Уничтожение только вне callbacks, после возврата из poll/mainloop.
void uasync_destroy(struct UASYNC* ua, int close_fds);
void uasync_mark_running(struct UASYNC* ua);
void uasync_mark_stopped(struct UASYNC* ua);

// текущее время (timebase 0.1ms)
uint64_t get_time_tb(void);
// текущее время в микросекундах (для burst-измерений)
uint64_t get_time_us(void);

// Timeouts, timebase = 0.1 mS
// Как работать с таймаутами если таймаут может быть использован несколько раз:
// 1. заводим дескриптор таймаута и обнуляем
// 2. перед активацией таймаута проверяем дескриптор (t_id) на null
// 3. при cancel или срабатывании дескриптор обнуляем
// это обеспечит отсутствие утечек и задвоений таймаутов
void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* user_arg, timeout_callback_t callback, const char* name);
err_t uasync_cancel_timeout(struct UASYNC* ua, void* t_id);

// Immediate execution in next mainloop (FIFO order)
void* uasync_call_soon(struct UASYNC* ua, void* user_arg, timeout_callback_t callback);
err_t uasync_call_soon_cancel(struct UASYNC* ua, void* t_id);

// Sockets - for regular file descriptors (pipe, file)
void* uasync_add_socket(struct UASYNC* ua, int fd, socket_callback_t read_cbk, socket_callback_t write_cbk, socket_callback_t except_cbk, const char* name, void* user_arg);
// Sockets - for socket_t (cross-platform sockets)
void* uasync_add_socket_t(struct UASYNC* ua, socket_t sock, socket_t_callback_t read_cbk, socket_t_callback_t write_cbk, socket_t_callback_t except_cbk, const char* name, void* user_arg);
err_t uasync_remove_socket(struct UASYNC* ua, void* s_id);
err_t uasync_remove_socket_t(struct UASYNC* ua, socket_t sock);

// Динамическое управление мониторингом чтения/записи для сокета
err_t uasync_set_socket_read(struct UASYNC* ua, void* s_id, int enable);
err_t uasync_set_socket_write(struct UASYNC* ua, void* s_id, int enable);

// Single iteration of event loop with timeout (timebase units)
void uasync_poll(struct UASYNC* ua, int timeout_tb);
void uasync_drain_immediate(struct UASYNC* ua);

// Цикл до uasync_stop (возвращается после остановки).
void uasync_mainloop(struct UASYNC* ua);
void uasync_stop(struct UASYNC* ua);

// Debug statistics
void uasync_get_stats(struct UASYNC* ua, size_t* timer_alloc, size_t* timer_free, size_t* socket_alloc, size_t* socket_free);

// Актуализирует снимок по сроку 100 мс. Только в потоке uasync; наружу передавать копию через события.
err_t uasync_get_load(struct UASYNC* ua, struct uasync_load* load);

// Handle — кодированный индекс; сохраняется при расширении массива сокетов.
int uasync_lookup_socket(struct UASYNC* ua, int fd, void** socket_id);

// Print all resources (timers, sockets) for debugging
void uasync_print_resources(struct UASYNC* ua, const char* prefix);

// Wakeup mechanism for interrupting poll
int uasync_wakeup(struct UASYNC* ua);
int uasync_get_wakeup_fd(struct UASYNC* ua); // returns write fd for wakeup pipe (for signal handlers)

// сообщить async (из другого thread) чтобы он вызвал callback с аргументом
void uasync_post(struct UASYNC* ua, uasync_post_callback_t callback, void* user_arg);

/* Передать заранее выделенный u_malloc task без аллокаций. uasync освобождает
 * task после callback. Производитель больше не обращается к task после передачи.
 * Отмена — в потоке владельца вне callbacks, после завершения производителя;
 * ERR_OK означает, что task удалён из очереди и освобождён. */
void uasync_post_reserved(struct UASYNC* ua, struct posted_task* task);
err_t uasync_cancel_post(struct UASYNC* ua, struct posted_task* task);

// синхронизация памяти (для доступности из других потоков)
void uasync_memsync(struct UASYNC* ua);


#ifdef __cplusplus
}
#endif
#endif // UASYNC_H