view release on metacpan or  search on metacpan

queue.h  view on Meta::CPAN

                                        struct timespec *remaining) {
    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    remaining->tv_sec = deadline->tv_sec - now.tv_sec;
    remaining->tv_nsec = deadline->tv_nsec - now.tv_nsec;
    if (remaining->tv_nsec < 0) {
        remaining->tv_sec--;
        remaining->tv_nsec += 1000000000L;
    }
    return remaining->tv_sec >= 0;
}

/* Convert timeout in seconds (double) to absolute deadline */
static inline void queue_make_deadline(double timeout, struct timespec *deadline) {
    clock_gettime(CLOCK_MONOTONIC, deadline);
    deadline->tv_sec += (time_t)timeout;
    deadline->tv_nsec += (long)((timeout - (double)(time_t)timeout) * 1e9);
    if (deadline->tv_nsec >= 1000000000L) {
        deadline->tv_sec++;
        deadline->tv_nsec -= 1000000000L;
    }
}

/* ================================================================
 * Create / Open / Close
 * ================================================================ */

#define QUEUE_ERR(fmt, ...) do { if (errbuf) snprintf(errbuf, QUEUE_ERR_BUFLEN, fmt, ##__VA_ARGS__); } while(0)

static inline void queue_init_new_header(void *base, uint32_t cap, uint64_t arena_cap,
                                          uint32_t slots_off, uint32_t arena_off,
                                          uint32_t mode, uint64_t total_size) {
    QueueHeader *hdr = (QueueHeader *)base;
    memset(hdr, 0, sizeof(QueueHeader));
    hdr->magic      = QUEUE_MAGIC;
    hdr->version    = QUEUE_VERSION;
    hdr->mode       = mode;
    hdr->capacity   = cap;
    hdr->total_size = total_size;
    hdr->slots_off  = slots_off;
    hdr->arena_off  = arena_off;
    hdr->arena_cap  = arena_cap;
    #define INIT_SEQ(STYPE, C) do { \
        STYPE *s = (STYPE *)((char *)base + slots_off); \
        for (uint32_t _i = 0; _i < (C); _i++) s[_i].sequence = _i; \
    } while(0)
    if      (mode == QUEUE_MODE_INT)   INIT_SEQ(QueueIntSlot,   cap);
    else if (mode == QUEUE_MODE_INT32) INIT_SEQ(QueueInt32Slot, cap);
    else if (mode == QUEUE_MODE_INT16) INIT_SEQ(QueueInt16Slot, cap);
    #undef INIT_SEQ
    __atomic_thread_fence(__ATOMIC_SEQ_CST);
}

/* Securely obtain a fd: create exclusively (O_CREAT|O_EXCL|O_NOFOLLOW at
 * file_mode, default 0600 = owner-only), or attach an existing file
 * (O_RDWR|O_NOFOLLOW, no O_CREAT). Blocks a symlink swap or a pre-seeded/
 * hard-linked backing file; cross-user sharing is opt-in via a wider file_mode. */
static int queue_secure_open(const char *path, mode_t file_mode, char *errbuf) {
    for (int attempt = 0; attempt < 100; attempt++) {
        int fd = open(path, O_RDWR|O_CREAT|O_EXCL|O_NOFOLLOW|O_CLOEXEC, file_mode);
        if (fd >= 0) { (void)fchmod(fd, file_mode); return fd; }   /* exact mode: umask narrowed the O_EXCL create */
        if (errno != EEXIST) { QUEUE_ERR("create(%s): %s", path, strerror(errno)); return -1; }
        fd = open(path, O_RDWR|O_NOFOLLOW|O_CLOEXEC);
        if (fd >= 0) return fd;
        if (errno == ENOENT) continue;
        QUEUE_ERR("open(%s): %s", path, strerror(errno));
        return -1;
    }
    QUEUE_ERR("open(%s): create/attach kept racing", path);
    return -1;
}

static QueueHandle *queue_create(const char *path, uint32_t capacity,
                                  uint32_t mode, uint64_t arena_cap_hint,
                                  mode_t file_mode, char *errbuf) {
    if (errbuf) errbuf[0] = '\0';

    uint32_t cap = queue_next_pow2(capacity);
    if (cap == 0) { QUEUE_ERR("invalid capacity"); return NULL; }
    if (mode > QUEUE_MODE_INT16) { QUEUE_ERR("unknown mode %u", mode); return NULL; }

    uint64_t slots_off = sizeof(QueueHeader);
    uint64_t slot_size = (mode == QUEUE_MODE_INT)   ? sizeof(QueueIntSlot)
                       : (mode == QUEUE_MODE_INT32) ? sizeof(QueueInt32Slot)
                       : (mode == QUEUE_MODE_INT16) ? sizeof(QueueInt16Slot)
                       :                              sizeof(QueueStrSlot);
    uint64_t arena_off = 0, arena_cap = 0;
    uint64_t total_size;

    if (mode == QUEUE_MODE_STR) {
        uint64_t slots_end = slots_off + (uint64_t)cap * slot_size;
        arena_off = (slots_end + 7) & ~(uint64_t)7;
        arena_cap = arena_cap_hint;
        if (arena_cap < 4096) arena_cap = 4096;
        if (arena_cap > UINT32_MAX) arena_cap = UINT32_MAX;
        total_size = arena_off + arena_cap;
    } else {
        total_size = slots_off + (uint64_t)cap * slot_size;
    }

    int anonymous = (path == NULL);
    size_t map_size;
    void *base;

    if (anonymous) {
        /* Anonymous shared mmap — fork-inherited, no filesystem */
        map_size = (size_t)total_size;
        base = mmap(NULL, map_size, PROT_READ | PROT_WRITE,
                     MAP_SHARED | MAP_ANONYMOUS, -1, 0);
        if (base == MAP_FAILED) {
            QUEUE_ERR("mmap(anonymous): %s", strerror(errno));
            return NULL;
        }
        queue_init_new_header(base, cap, arena_cap, slots_off, arena_off, mode, total_size);
    } else {
        /* File-backed shared mmap */
        int fd = queue_secure_open(path, file_mode, errbuf);
        if (fd < 0) return NULL;

        if (flock(fd, LOCK_EX) < 0) {
            QUEUE_ERR("flock(%s): %s", path, strerror(errno));