kqueue.c

Go to the documentation of this file.

#define _POSIX_C_SOURCE 200809L
#include <cwist/sys/io/cwist_io.h>
#include <cwist/core/mem/alloc.h>
#include <cwist/core/mem/gc.h>
#include <sys/types.h>
#include <sys/event.h>
#include <sys/time.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>
 
/* ident 0 is never a job: jobs use their wrapper's address as ident. */
#define KQ_WAKE_IDENT 0
 
struct cwist_io_queue {
    int kq_fd;
    atomic_bool running;
    /* Jobs accepted by submit and not yet taken by a runner. */
    atomic_size_t pending_jobs;
};
 
cwist_io_queue *cwist_io_queue_create(size_t capacity) {
    (void)capacity;
    cwist_io_queue *q = cwist_alloc(sizeof(cwist_io_queue));
    if (!q) return NULL;
    if (cwist_full_gc_enabled()) cwist_gc_scope_disown(q);
 
    q->kq_fd = kqueue();
    if (q->kq_fd < 0) {
        perror("kqueue");
        cwist_free(q);
        return NULL;
    }
 
    /* The stop wake. No EV_CLEAR: once stop triggers it, it stays active and
     * every runner's kevent() returns, not just the first one to look. */
    struct kevent kev;
    EV_SET(&kev, KQ_WAKE_IDENT, EVFILT_USER, EV_ADD | EV_ENABLE, 0, 0, NULL);
    if (kevent(q->kq_fd, &kev, 1, NULL, 0, NULL) < 0) {
        perror("kevent wake");
        close(q->kq_fd);
        cwist_free(q);
        return NULL;
    }
    atomic_init(&q->running, true);
    atomic_init(&q->pending_jobs, 0);
    return q;
}
 
struct job_wrapper {
    cwist_job_func func;
    void *arg;
};
 
bool cwist_io_queue_submit(cwist_io_queue *q, cwist_job_func func, void *arg) {
    if (!q || !func) return false;
 
    /* Count the job before checking running, so a runner that sees
     * running == false and pending_jobs == 0 knows no submit is in flight. */
    atomic_fetch_add_explicit(&q->pending_jobs, 1, memory_order_acq_rel);
    if (!atomic_load_explicit(&q->running, memory_order_acquire)) {
        atomic_fetch_sub_explicit(&q->pending_jobs, 1, memory_order_acq_rel);
        return false;
    }
 
    struct job_wrapper *job = cwist_alloc(sizeof(struct job_wrapper));
    if (!job) {
        atomic_fetch_sub_explicit(&q->pending_jobs, 1, memory_order_acq_rel);
        return false;
    }
    if (cwist_full_gc_enabled()) cwist_gc_scope_disown(job);
    job->func = func;
    job->arg = arg;
 
    struct kevent kev;
    // We use EVFILT_USER to trigger a custom event.
    // Ident = pointer to job (unique ID)
    EV_SET(&kev, (uintptr_t)job, EVFILT_USER, EV_ADD | EV_ENABLE | EV_ONESHOT, NOTE_TRIGGER, 0,
           job);
 
    if (kevent(q->kq_fd, &kev, 1, NULL, 0, NULL) < 0) {
        perror("kevent submit");
        cwist_free(job);
        atomic_fetch_sub_explicit(&q->pending_jobs, 1, memory_order_acq_rel);
        return false;
    }
    return true;
}
 
void cwist_io_queue_run(cwist_io_queue *q) {
    if (!q) return;
    struct kevent events[32];
    while (atomic_load_explicit(&q->running, memory_order_acquire) ||
           atomic_load_explicit(&q->pending_jobs, memory_order_acquire) > 0) {
        int n = kevent(q->kq_fd, NULL, 0, events, 32, NULL);
        if (n < 0) {
            if (errno == EINTR) continue;
            perror("kevent wait");
            break;
        }
 
        int ran = 0;
        for (int i = 0; i < n; i++) {
            if (events[i].filter != EVFILT_USER || events[i].ident == KQ_WAKE_IDENT) continue;
            struct job_wrapper *job = (struct job_wrapper *)events[i].udata;
            if (!job) continue;
            atomic_fetch_sub_explicit(&q->pending_jobs, 1, memory_order_acq_rel);
            job->func(job->arg);
            cwist_free(job);
            ran++;
 
            // Full-GC mode: catch whatever this job forgot to cwist_free()
            // right away instead of letting it pile up for the lifetime of
            // this (long-lived, pooled) worker thread, as io_queue.c does.
            if (cwist_full_gc_enabled()) {
                cwist_gc_scope_flush();
            }
        }
 
        /* Stopping, but another runner still holds the last jobs: the wake
         * event keeps kevent() returning at once, so back off instead of
         * spinning until that runner takes them. */
        if (ran == 0 && n > 0 && !atomic_load_explicit(&q->running, memory_order_acquire)) {
            struct timespec ts = {0, 1000000}; /* 1ms */
            nanosleep(&ts, NULL);
        }
    }
}
 
void cwist_io_queue_stop(cwist_io_queue *q) {
    if (!q) return;
    atomic_store_explicit(&q->running, false, memory_order_release);
    struct kevent kev;
    EV_SET(&kev, KQ_WAKE_IDENT, EVFILT_USER, 0, NOTE_TRIGGER, 0, NULL);
    if (kevent(q->kq_fd, &kev, 1, NULL, 0, NULL) < 0) {
        perror("kevent stop");
    }
}
 
void cwist_io_queue_destroy(cwist_io_queue *q) {
    if (!q) return;
    atomic_store_explicit(&q->running, false, memory_order_release);
 
    /* Drop the level-triggered wake first, or the drain below never ends. */
    struct kevent kev;
    EV_SET(&kev, KQ_WAKE_IDENT, EVFILT_USER, EV_DELETE, 0, 0, NULL);
    (void)kevent(q->kq_fd, &kev, 1, NULL, 0, NULL);
 
    struct kevent events[32];
    const struct timespec zero = {0, 0};
    int n;
    while ((n = kevent(q->kq_fd, NULL, 0, events, 32, &zero)) > 0) {
        for (int i = 0; i < n; i++) {
            if (events[i].filter == EVFILT_USER && events[i].udata) {
                cwist_free((void *)events[i].udata);
            }
        }
    }
    close(q->kq_fd);
    cwist_free(q);
}