view release on metacpan or  search on metacpan

pubsub.h  view on Meta::CPAN

                                uint64_t total_size, uint64_t slots_off,
                                uint64_t data_off, uint32_t msg_size,
                                uint64_t arena_cap) {
    PubSubHeader *hdr = (PubSubHeader *)base;
    memset(hdr, 0, sizeof(PubSubHeader));
    hdr->magic     = PUBSUB_MAGIC;
    hdr->version   = PUBSUB_VERSION;
    hdr->mode      = mode;
    hdr->capacity  = cap;
    hdr->total_size = total_size;
    hdr->slots_off = slots_off;
    hdr->data_off  = data_off;
    hdr->msg_size  = msg_size;
    hdr->arena_cap = arena_cap;
}

/* Returns 1 on success, 0 if the requested capacity * msg_size would
 * exceed the 32-bit arena addressing limit (out_arena_cap is set to 0). */
static int pubsub_calc_layout(uint32_t cap, uint32_t mode, uint32_t msg_size,
                                uint64_t *out_slots_off, uint64_t *out_data_off,
                                uint64_t *out_arena_cap, uint64_t *out_total_size) {
    uint64_t slots_off = sizeof(PubSubHeader);
    uint64_t slot_size = (mode == PUBSUB_MODE_INT)   ? sizeof(PubSubIntSlot)
                       : (mode == PUBSUB_MODE_INT32) ? sizeof(PubSubInt32Slot)
                       : (mode == PUBSUB_MODE_INT16) ? sizeof(PubSubInt16Slot)
                       :                               sizeof(PubSubStrSlot);
    uint64_t data_off = 0, arena_cap = 0, total_size;

    if (mode == PUBSUB_MODE_STR) {
        uint64_t slots_end = slots_off + (uint64_t)cap * slot_size;
        data_off = (slots_end + 63) & ~(uint64_t)63;
        arena_cap = (uint64_t)cap * ((uint64_t)msg_size + 8);
        /* Safety invariant: in any window of `capacity` consecutive publishes
         * (one full slot-ring wrap), the arena must not wrap. Otherwise a
         * publish to slot K could overwrite slot N's still-current arena
         * region without bumping slot N's sequence -- the seqlock would
         * not catch the corruption. arena_cap = capacity*(msg_size+8)
         * guarantees this; silently capping to UINT32_MAX would break the
         * invariant for extreme (capacity, msg_size) combinations. */
        if (arena_cap > UINT32_MAX) {
            *out_arena_cap = 0;
            return 0;
        }
        total_size = data_off + arena_cap;
    } else {
        total_size = slots_off + (uint64_t)cap * slot_size;
    }

    *out_slots_off  = slots_off;
    *out_data_off   = data_off;
    *out_arena_cap  = arena_cap;
    *out_total_size = total_size;
    return 1;
}

/* Securely obtain a fd: create exclusively (O_CREAT|O_EXCL|O_NOFOLLOW at mode,
 * default 0600), or attach an existing file (O_RDWR|O_NOFOLLOW, no O_CREAT). */
static int pubsub_secure_open(const char *path, mode_t mode, char *errbuf) {
    for (int attempt = 0; attempt < 100; attempt++) {
        int fd = open(path, O_RDWR|O_CREAT|O_EXCL|O_NOFOLLOW|O_CLOEXEC, mode);
        if (fd >= 0) { (void)fchmod(fd, mode); return fd; }   /* exact mode: umask narrowed the O_EXCL create */
        if (errno != EEXIST) { PUBSUB_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;   /* creator unlinked between our two opens; retry */
        PUBSUB_ERR("open %s: %s", path, strerror(errno));  /* ELOOP => symlink rejected */
        return -1;
    }
    PUBSUB_ERR("open %s: create/attach kept racing", path);
    return -1;
}

static PubSubHandle *pubsub_create(const char *path, uint32_t capacity,
                                    uint32_t mode, uint32_t msg_size,
                                    mode_t fmode, char *errbuf) {
    if (errbuf) errbuf[0] = '\0';

    uint32_t cap = pubsub_next_pow2(capacity);
    if (cap == 0) { PUBSUB_ERR("invalid capacity"); return NULL; }
    if (mode > PUBSUB_MODE_INT16) { PUBSUB_ERR("unknown mode %u", mode); return NULL; }

    if (mode == PUBSUB_MODE_STR && msg_size == 0)
        msg_size = PUBSUB_DEFAULT_MSG_SIZE;

    uint64_t slots_off, data_off, arena_cap, total_size;
    if (!pubsub_calc_layout(cap, mode, msg_size, &slots_off, &data_off, &arena_cap, &total_size)) {
        PUBSUB_ERR("capacity * (msg_size+8) exceeds 4GB arena limit");
        return NULL;
    }

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

    if (anonymous) {
        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) {
            PUBSUB_ERR("mmap(anonymous): %s", strerror(errno));
            return NULL;
        }
        pubsub_init_header(base, mode, cap, total_size, slots_off, data_off,
                            msg_size, arena_cap);
    } else {
        int fd = pubsub_secure_open(path, fmode, errbuf);
        if (fd < 0) return NULL;

        if (flock(fd, LOCK_EX) < 0) {
            PUBSUB_ERR("flock(%s): %s", path, strerror(errno));
            close(fd); return NULL;
        }

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

        int is_new = (st.st_size == 0);