view release on metacpan or  search on metacpan

Shared.xs  view on Meta::CPAN

    const char *class
    int fd
  PREINIT:
    char errbuf[TW_ERR_BUFLEN];
  CODE:
    TwHandle *h = tw_open_fd(fd, errbuf);
    if (!h) croak("Data::TimingWheel::Shared->new_from_fd: %s", errbuf);
    MAKE_OBJ(class, h);
  OUTPUT:
    RETVAL

void
DESTROY(self)
    SV *self
  CODE:
    if (sv_isobject(self) && sv_derived_from(self, "Data::TimingWheel::Shared")) {
        TwHandle *h = INT2PTR(TwHandle*, SvIV(SvRV(self)));
        if (h) { sv_setiv(SvRV(self), 0); tw_destroy(h); }   /* null first: activates EXTRACT's use-after-destroy croak + makes a double DESTROY a no-op */
    }

UV
add(self, delay, payload)
    SV *self
    IV delay
    UV payload
  PREINIT:
    EXTRACT(self);
    int64_t id;
  CODE:
    /* signed delay: a negative value (e.g. from clock skew) clamps to 1 tick per
     * the POD, instead of UV-wrapping to a ~2^64 delay that never fires (slot leak) */
    tw_rwlock_wrlock(h);
    id = tw_add_locked(h, delay < 1 ? 1 : (uint64_t)delay, (uint64_t)payload);
    __atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
    tw_rwlock_wrunlock(h);
    if (id < 0) croak("Data::TimingWheel::Shared->add: timer pool is full (capacity %u)", (unsigned)h->capacity);
    RETVAL = (UV)id;
  OUTPUT:
    RETVAL

int
cancel(self, timer_id)
    SV *self
    UV timer_id
  PREINIT:
    EXTRACT(self);
  CODE:
    tw_rwlock_wrlock(h);
    RETVAL = tw_cancel_locked(h, (uint64_t)timer_id);
    __atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
    tw_rwlock_wrunlock(h);
  OUTPUT:
    RETVAL

void
advance(self, ticks = 1)
    SV *self
    IV ticks
  PREINIT:
    EXTRACT(self);
  PPCODE:
    {
        REEXTRACT(self);   /* `ticks = 1` is converted after PREINIT, so its magic already ran */
        uint64_t *out = NULL, fired = 0, i, cap = h->capacity;
        /* signed: reject a negative tick count (would UV-wrap to ~2^64 and spin the
         * per-tick loop under the write lock, wedging every sharer) */
        if (ticks < 0) croak("Data::TimingWheel::Shared->advance: ticks must be >= 0");
        if (cap) { Newx(out, (size_t)cap, uint64_t); SAVEFREEPV(out); }   /* alloc BEFORE the lock */
        tw_rwlock_wrlock(h);
        fired = cap ? tw_advance_locked(h, (uint64_t)ticks, out, cap) : 0;
        __atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
        tw_rwlock_wrunlock(h);
        EXTEND(SP, (SSize_t)fired);
        for (i = 0; i < fired; i++) PUSHs(sv_2mortal(newSVuv((UV)out[i])));   /* Perl alloc AFTER unlock */
    }

UV
now(self)
    SV *self
  PREINIT:
    EXTRACT(self);
    UV v;
  CODE:
    tw_rwlock_rdlock(h);
    v = (UV)h->hdr->now;
    tw_rwlock_rdunlock(h);
    RETVAL = v;
  OUTPUT:
    RETVAL

UV
count(self)
    SV *self
  PREINIT:
    EXTRACT(self);
    UV v;
  CODE:
    tw_rwlock_rdlock(h);
    v = (UV)h->hdr->count;
    tw_rwlock_rdunlock(h);
    RETVAL = v;
  OUTPUT:
    RETVAL

UV
capacity(self)
    SV *self
  PREINIT:
    EXTRACT(self);
  CODE:
    RETVAL = (UV)h->hdr->capacity;
  OUTPUT:
    RETVAL

UV
num_slots(self)
    SV *self
  PREINIT:
    EXTRACT(self);
  CODE:
    RETVAL = (UV)h->hdr->num_slots;