view release on metacpan or  search on metacpan

reqrep.h  view on Meta::CPAN

            return h;
        }
        reqrep_int_init_header(base, req_cap, resp_slots_n, total_size,
                                req_slots_off, resp_off, resp_stride);
        flock(fd, LOCK_UN); close(fd);
        ReqRepHandle *h = reqrep_setup_handle(base, map_size, path, -1);
        if (!h) { munmap(base, map_size); return NULL; }
        return h;
    }

    reqrep_int_init_header(base, req_cap, resp_slots_n, total_size,
                            req_slots_off, resp_off, resp_stride);
    ReqRepHandle *h = reqrep_setup_handle(base, map_size, path, -1);
    if (!h) { munmap(base, map_size); return NULL; }
    return h;
}

static ReqRepHandle *reqrep_create_int_memfd(const char *name, uint32_t req_cap,
                                              uint32_t resp_slots_n, char *errbuf) {
    if (errbuf) errbuf[0] = '\0';
    req_cap = reqrep_next_pow2(req_cap);
    if (req_cap == 0) { REQREP_ERR("invalid req_cap"); return NULL; }
    if (resp_slots_n == 0) { REQREP_ERR("resp_slots must be > 0"); return NULL; }

    uint32_t req_slots_off, resp_off, resp_stride;
    uint64_t total_size;
    if (reqrep_int_compute_layout(req_cap, resp_slots_n, &req_slots_off,
                                   &resp_off, &resp_stride, &total_size) < 0) {
        REQREP_ERR("layout overflow: req_cap too large for uint32 offsets");
        return NULL;
    }

    int fd = memfd_create(name ? name : "reqrep_int", MFD_CLOEXEC | MFD_ALLOW_SEALING);
    if (fd < 0) { REQREP_ERR("memfd_create: %s", strerror(errno)); return NULL; }
    if (ftruncate(fd, (off_t)total_size) < 0) {
        REQREP_ERR("ftruncate: %s", strerror(errno)); close(fd); return NULL;
    }
    (void)fcntl(fd, F_ADD_SEALS, F_SEAL_SHRINK | F_SEAL_GROW);
    void *base = mmap(NULL, (size_t)total_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    if (base == MAP_FAILED) { REQREP_ERR("mmap: %s", strerror(errno)); close(fd); return NULL; }

    reqrep_int_init_header(base, req_cap, resp_slots_n, total_size,
                            req_slots_off, resp_off, resp_stride);
    ReqRepHandle *h = reqrep_setup_handle(base, (size_t)total_size, NULL, fd);
    if (!h) { munmap(base, (size_t)total_size); close(fd); return NULL; }
    return h;
}

/* --- Int request queue: lock-free Vyukov MPMC --- */

static inline int reqrep_int_try_send(ReqRepHandle *h, int64_t value, uint64_t *out_id) {
    int32_t rslot = reqrep_slot_acquire(h);
    if (rslot < 0) return -3;

    RespSlotHeader *rs = reqrep_resp_slot(h, (uint32_t)rslot);
    uint32_t gen = __atomic_load_n(&rs->generation, __ATOMIC_ACQUIRE);

    ReqRepHeader *hdr = h->hdr;
    /* Use the attach-cached, validated slot base -- never re-derive from the
     * peer-writable hdr->req_slots_off, which a lock-violating peer could
     * corrupt to relocate the array under us. */
    ReqIntSlot *slots = (ReqIntSlot *)h->req_slots;
    uint32_t mask = h->req_cap_mask;
    uint64_t pos = __atomic_load_n(&hdr->req_tail, __ATOMIC_RELAXED);

    for (;;) {
        ReqIntSlot *slot = &slots[pos & mask];
        uint64_t seq = __atomic_load_n(&slot->sequence, __ATOMIC_ACQUIRE);
        int64_t diff = (int64_t)seq - (int64_t)pos;
        if (diff == 0) {
            if (__atomic_compare_exchange_n(&hdr->req_tail, &pos, pos + 1,
                    1, __ATOMIC_RELAXED, __ATOMIC_RELAXED)) {
                slot->value = value;
                slot->resp_slot = (uint32_t)rslot;
                slot->resp_gen = gen;
                __atomic_store_n(&slot->sequence, pos + 1, __ATOMIC_RELEASE);
                __atomic_add_fetch(&hdr->stat_requests, 1, __ATOMIC_RELAXED);
                reqrep_wake_consumers(hdr);
                *out_id = REQREP_MAKE_ID((uint32_t)rslot, gen);
                return 1;
            }
        } else if (diff < 0) {
            __atomic_add_fetch(&hdr->stat_send_full, 1, __ATOMIC_RELAXED);
            reqrep_slot_release(h, (uint32_t)rslot);
            return 0;
        } else {
            pos = __atomic_load_n(&hdr->req_tail, __ATOMIC_RELAXED);
        }
    }
}

static int reqrep_int_send_wait(ReqRepHandle *h, int64_t value,
                                 uint64_t *out_id, double timeout) {
    int r = reqrep_int_try_send(h, value, out_id);
    if (r == 1) return 1;
    if (timeout == 0) return r;
    ReqRepHeader *hdr = h->hdr;
    struct timespec deadline, remaining;
    int has_deadline = (timeout > 0);
    if (has_deadline) reqrep_make_deadline(timeout, &deadline);
    for (;;) {
        /* Wait on slot_futex if no slots (-3), send_futex if queue full (0) */
        uint32_t *futex_word = (r == -3) ? &hdr->slot_futex : &hdr->send_futex;
        uint32_t *waiter_cnt = (r == -3) ? &hdr->slot_waiters : &hdr->send_waiters;

        uint32_t fseq = __atomic_load_n(futex_word, __ATOMIC_ACQUIRE);
        r = reqrep_int_try_send(h, value, out_id);
        if (r == 1) return 1;

        __atomic_add_fetch(waiter_cnt, 1, __ATOMIC_RELEASE);
        /* StoreLoad barrier + re-check (see reqrep_send_wait): a consumer that
         * drained the queue / freed a slot in the gap is seen here, else the
         * wakeup is lost.  temp r2: a failed re-check must not re-select the futex. */
        __atomic_thread_fence(__ATOMIC_SEQ_CST);
        { int r2 = reqrep_int_try_send(h, value, out_id);
          if (r2 == 1) { __atomic_sub_fetch(waiter_cnt, 1, __ATOMIC_RELEASE); return r2; } }
        struct timespec *pts = NULL;
        if (has_deadline) {
            if (!reqrep_remaining_time(&deadline, &remaining)) {
                __atomic_sub_fetch(waiter_cnt, 1, __ATOMIC_RELEASE);
                return r;