view release on metacpan or  search on metacpan

buf_generic.h  view on Meta::CPAN

static inline void buf_rwlock_wrunlock(BufHandle *h) {
    __atomic_store_n(&h->hdr->rwlock, 0, __ATOMIC_RELEASE);
    if (__atomic_load_n(&h->hdr->rwlock_waiters, __ATOMIC_RELAXED) > 0)
        syscall(SYS_futex, &h->hdr->rwlock, FUTEX_WAKE, INT_MAX, NULL, NULL, 0);
}

/* ---- Seqlock ---- */

static inline uint32_t buf_seqlock_read_begin(BufHeader *hdr) {
    int spin = 0;
    for (;;) {
        uint32_t s = __atomic_load_n(&hdr->seq, __ATOMIC_ACQUIRE);
        if (__builtin_expect((s & 1) == 0, 1)) return s;
        if (__builtin_expect(spin < 100000, 1)) {
            buf_spin_pause();
            spin++;
            continue;
        }
        uint32_t val = __atomic_load_n(&hdr->rwlock, __ATOMIC_RELAXED);
        if (val >= BUF_RWLOCK_WRITER_BIT) {
            uint32_t pid = val & BUF_RWLOCK_PID_MASK;
            if (!buf_pid_alive(pid)) {
                buf_recover_stale_lock(hdr, val);
                spin = 0;
                continue;
            }
        }
        struct timespec ts = {0, 1000000};
        nanosleep(&ts, NULL);
        spin = 0;
    }
}

static inline int buf_seqlock_read_retry(uint32_t *seq, uint32_t start) {
    /* Acquire FENCE (LoadLoad): the section's data loads must retire before we
     * re-read seq; a plain acquire load is the wrong direction (ARM64 torn read). */
    __atomic_thread_fence(__ATOMIC_ACQUIRE);
    return __atomic_load_n(seq, __ATOMIC_RELAXED) != start;
}

static inline void buf_seqlock_write_begin(uint32_t *seq) {
    __atomic_add_fetch(seq, 1, __ATOMIC_RELEASE);  /* seq becomes odd */
    /* StoreStore: publish the odd seq before the section's data writes, else an
     * ARM64 reader could see an even seq with half-written data (Linux smp_wmb). */
    __atomic_thread_fence(__ATOMIC_RELEASE);
}

static inline void buf_seqlock_write_end(uint32_t *seq) {
    __atomic_add_fetch(seq, 1, __ATOMIC_RELEASE);
}

/* ---- mmap create/open ---- */

/* Race-safe create-or-attach with restrictive perms + symlink refusal.
 * The backing file is created 0600 by default (see file_mode); a local peer
 * cannot pre-create it as a symlink (O_NOFOLLOW) nor open a wider-permissioned
 * copy.  Sets *created=1 iff this call won the O_EXCL race and made the file. */
static int buf_secure_open(const char *path, mode_t file_mode, int *created, 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); *created = 1; return fd; }  /* exact mode: umask narrowed the create */
        if (errno != EEXIST) {
            snprintf(errbuf, BUF_ERR_BUFLEN, "create(%s): %s", path, strerror(errno));
            return -1;
        }
        fd = open(path, O_RDWR | O_NOFOLLOW | O_CLOEXEC);
        if (fd >= 0) { *created = 0; return fd; }
        if (errno == ENOENT) continue;   /* creator unlinked between our two opens; retry */
        snprintf(errbuf, BUF_ERR_BUFLEN, "open(%s): %s", path, strerror(errno));
        return -1;
    }
    snprintf(errbuf, BUF_ERR_BUFLEN, "open(%s): create/attach kept racing", path);
    return -1;
}

static BufHandle *buf_create_map(const char *path, uint64_t capacity,
                                  uint32_t elem_size, uint32_t variant_id,
                                  mode_t file_mode, char *errbuf) {
    errbuf[0] = '\0';
    int created = 0;
    int fd = buf_secure_open(path, file_mode, &created, errbuf);
    if (fd < 0) {
        return NULL;
    }

    /* Lock file for init race prevention */
    if (flock(fd, LOCK_EX) < 0) {
        snprintf(errbuf, BUF_ERR_BUFLEN, "flock(%s): %s", path, strerror(errno));
        close(fd);
        return NULL;
    }

    uint64_t reader_slots_off = sizeof(BufHeader); /* 128 */
    uint64_t reader_slots_size = (uint64_t)BUF_READER_SLOTS * sizeof(BufReaderSlot);
    uint64_t data_off = reader_slots_off + reader_slots_size; /* cache-aligned */
    if (elem_size > 0 && capacity > (UINT64_MAX - data_off) / elem_size) {
        snprintf(errbuf, BUF_ERR_BUFLEN, "buffer size overflow");
        flock(fd, LOCK_UN);
        close(fd);
        return NULL;
    }
    uint64_t total_size = data_off + capacity * elem_size;

    struct stat st;
    if (fstat(fd, &st) < 0) {
        snprintf(errbuf, BUF_ERR_BUFLEN, "fstat(%s): %s", path, strerror(errno));
        flock(fd, LOCK_UN);
        close(fd);
        return NULL;
    }

    if (!created && st.st_size > 0 && (uint64_t)st.st_size < sizeof(BufHeader)) {
        snprintf(errbuf, BUF_ERR_BUFLEN, "%s: file too small (%lld)", path, (long long)st.st_size);
        flock(fd, LOCK_UN); close(fd); return NULL;
    }
    if (created || st.st_size == 0) {
        if (ftruncate(fd, (off_t)total_size) < 0) {
            snprintf(errbuf, BUF_ERR_BUFLEN, "ftruncate(%s): %s", path, strerror(errno));
            flock(fd, LOCK_UN);
            close(fd);
            return NULL;