Data-PubSub-Shared

 view release on metacpan or  search on metacpan

Shared.xs  view on Meta::CPAN

    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    RETVAL = (UV)__atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
  OUTPUT:
    RETVAL

bool
has_overflow(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
    RETVAL = (sub->cursor < wp && wp - sub->cursor > sub->capacity);
  OUTPUT:
    RETVAL

UV
cursor(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    if (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) {
        uint64_t new_cursor = (uint64_t)SvUV(ST(1));
        REEXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
        sub->cursor = new_cursor;
    }
    RETVAL = (UV)sub->cursor;
  OUTPUT:
    RETVAL

void
reset(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    sub->cursor = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);

void
reset_oldest(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);
    sub->cursor = (wp > sub->capacity) ? wp - sub->capacity : 0;

UV
poll_cb(self, cb)
    SV *self
    SV *cb
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
    int64_t value;
  CODE:
    RETVAL = 0;
    /* Keep the backing handle alive across the callback: a callback that
     * drops self's last reference would free `sub` mid-loop (UAF).  Guard
     * the referent (the blessed handle), not the RV container -- assigning
     * undef to self overwrites the container and frees the referent. */
    SV *psx_guard = SvRV(self);
    SvREFCNT_inc_simple_void_NN(psx_guard);
    SAVEFREESV(psx_guard);
    while (pubsub_int_poll(sub, &value)) {
        dSP;
        ENTER; SAVETMPS;
        PUSHMARK(SP);
        mXPUSHi((IV)value);
        PUTBACK;
        call_sv(cb, G_DISCARD);
        FREETMPS; LEAVE;
        /* call_sv ran arbitrary Perl: the callback may have DESTROYed or
         * replaced the subscriber.  Re-extract before the next poll. */
        REEXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
        RETVAL++;
    }
  OUTPUT:
    RETVAL

void
drain_notify(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
    int64_t value;
    uint32_t max_count;
  PPCODE:
    pubsub_sub_eventfd_consume(sub);
    max_count = (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) ? (uint32_t)SvUV(ST(1)) : UINT32_MAX;
    REEXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
    while (max_count-- > 0 && pubsub_int_poll(sub, &value))
        mXPUSHi((IV)value);

void
eventfd_set(self, fd)
    SV *self
    int fd
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    sub->notify_fd = fd;

IV
fileno(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int::Sub", self);
  CODE:
    RETVAL = sub->notify_fd;
  OUTPUT:
    RETVAL

MODULE = Data::PubSub::Shared  PACKAGE = Data::PubSub::Shared::Str

SV *
new(class, path, capacity, ...)
    const char *class

Shared.xs  view on Meta::CPAN

    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    RETVAL = (UV)__atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
  OUTPUT:
    RETVAL

bool
has_overflow(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
    RETVAL = (sub->cursor < wp && wp - sub->cursor > sub->capacity);
  OUTPUT:
    RETVAL

UV
cursor(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    if (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) {
        uint64_t new_cursor = (uint64_t)SvUV(ST(1));
        REEXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
        sub->cursor = new_cursor;
    }
    RETVAL = (UV)sub->cursor;
  OUTPUT:
    RETVAL

void
reset(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    sub->cursor = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);

void
reset_oldest(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);
    sub->cursor = (wp > sub->capacity) ? wp - sub->capacity : 0;

UV
poll_cb(self, cb)
    SV *self
    SV *cb
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
    const char *str;
    uint32_t len;
    bool utf8;
  CODE:
    RETVAL = 0;
    /* Keep the backing handle alive across the callback: a callback that
     * drops self's last reference would free `sub` mid-loop (UAF).  Guard
     * the referent (the blessed handle), not the RV container -- assigning
     * undef to self overwrites the container and frees the referent. */
    SV *psx_guard = SvRV(self);
    SvREFCNT_inc_simple_void_NN(psx_guard);
    SAVEFREESV(psx_guard);
    while (pubsub_str_poll(sub, &str, &len, &utf8) == 1) {
        dSP;
        ENTER; SAVETMPS;
        SV *sv = newSVpvn(str, len);
        if (utf8) SvUTF8_on(sv);
        PUSHMARK(SP);
        mXPUSHs(sv);
        PUTBACK;
        call_sv(cb, G_DISCARD);
        FREETMPS; LEAVE;
        /* call_sv ran arbitrary Perl: the callback may have DESTROYed or
         * replaced the subscriber.  Re-extract before the next poll. */
        REEXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
        RETVAL++;
    }
  OUTPUT:
    RETVAL

void
drain_notify(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
    const char *str;
    uint32_t len;
    bool utf8;
    uint32_t max_count;
  PPCODE:
    pubsub_sub_eventfd_consume(sub);
    max_count = (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) ? (uint32_t)SvUV(ST(1)) : UINT32_MAX;
    REEXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
    while (max_count-- > 0 && pubsub_str_poll(sub, &str, &len, &utf8) == 1) {
        SV *sv = newSVpvn(str, len);
        if (utf8) SvUTF8_on(sv);
        mXPUSHs(sv);
    }

void
eventfd_set(self, fd)
    SV *self
    int fd
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    sub->notify_fd = fd;

IV
fileno(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Str::Sub", self);
  CODE:
    RETVAL = sub->notify_fd;
  OUTPUT:

Shared.xs  view on Meta::CPAN

    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    RETVAL = (UV)__atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
  OUTPUT:
    RETVAL

bool
has_overflow(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
    RETVAL = (sub->cursor < wp && wp - sub->cursor > sub->capacity);
  OUTPUT:
    RETVAL

UV
cursor(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    if (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) {
        uint64_t new_cursor = (uint64_t)SvUV(ST(1));
        REEXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
        sub->cursor = new_cursor;
    }
    RETVAL = (UV)sub->cursor;
  OUTPUT:
    RETVAL

void
reset(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    sub->cursor = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);

void
reset_oldest(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);
    sub->cursor = (wp > sub->capacity) ? wp - sub->capacity : 0;

UV
poll_cb(self, cb)
    SV *self
    SV *cb
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
    int32_t value;
  CODE:
    RETVAL = 0;
    /* Keep the backing handle alive across the callback: a callback that
     * drops self's last reference would free `sub` mid-loop (UAF).  Guard
     * the referent (the blessed handle), not the RV container -- assigning
     * undef to self overwrites the container and frees the referent. */
    SV *psx_guard = SvRV(self);
    SvREFCNT_inc_simple_void_NN(psx_guard);
    SAVEFREESV(psx_guard);
    while (pubsub_int32_poll(sub, &value)) {
        dSP;
        ENTER; SAVETMPS;
        PUSHMARK(SP);
        mXPUSHi((IV)value);
        PUTBACK;
        call_sv(cb, G_DISCARD);
        FREETMPS; LEAVE;
        /* call_sv ran arbitrary Perl: the callback may have DESTROYed or
         * replaced the subscriber.  Re-extract before the next poll. */
        REEXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
        RETVAL++;
    }
  OUTPUT:
    RETVAL

void
drain_notify(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
    int32_t value;
    uint32_t max_count;
  PPCODE:
    pubsub_sub_eventfd_consume(sub);
    max_count = (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) ? (uint32_t)SvUV(ST(1)) : UINT32_MAX;
    REEXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
    while (max_count-- > 0 && pubsub_int32_poll(sub, &value))
        mXPUSHi((IV)value);

void
eventfd_set(self, fd)
    SV *self
    int fd
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    sub->notify_fd = fd;

IV
fileno(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int32::Sub", self);
  CODE:
    RETVAL = sub->notify_fd;
  OUTPUT:
    RETVAL

MODULE = Data::PubSub::Shared  PACKAGE = Data::PubSub::Shared::Int16

SV *
new(class, path, capacity, ...)
    const char *class

Shared.xs  view on Meta::CPAN

    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    RETVAL = (UV)__atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
  OUTPUT:
    RETVAL

bool
has_overflow(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_RELAXED);
    RETVAL = (sub->cursor < wp && wp - sub->cursor > sub->capacity);
  OUTPUT:
    RETVAL

UV
cursor(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    if (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) {
        uint64_t new_cursor = (uint64_t)SvUV(ST(1));
        REEXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
        sub->cursor = new_cursor;
    }
    RETVAL = (UV)sub->cursor;
  OUTPUT:
    RETVAL

void
reset(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    sub->cursor = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);

void
reset_oldest(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    uint64_t wp = __atomic_load_n(&sub->hdr->write_pos, __ATOMIC_ACQUIRE);
    sub->cursor = (wp > sub->capacity) ? wp - sub->capacity : 0;

UV
poll_cb(self, cb)
    SV *self
    SV *cb
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
    int16_t value;
  CODE:
    RETVAL = 0;
    /* Keep the backing handle alive across the callback: a callback that
     * drops self's last reference would free `sub` mid-loop (UAF).  Guard
     * the referent (the blessed handle), not the RV container -- assigning
     * undef to self overwrites the container and frees the referent. */
    SV *psx_guard = SvRV(self);
    SvREFCNT_inc_simple_void_NN(psx_guard);
    SAVEFREESV(psx_guard);
    while (pubsub_int16_poll(sub, &value)) {
        dSP;
        ENTER; SAVETMPS;
        PUSHMARK(SP);
        mXPUSHi((IV)value);
        PUTBACK;
        call_sv(cb, G_DISCARD);
        FREETMPS; LEAVE;
        /* call_sv ran arbitrary Perl: the callback may have DESTROYed or
         * replaced the subscriber.  Re-extract before the next poll. */
        REEXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
        RETVAL++;
    }
  OUTPUT:
    RETVAL

void
drain_notify(self, ...)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
    int16_t value;
    uint32_t max_count;
  PPCODE:
    pubsub_sub_eventfd_consume(sub);
    max_count = (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) ? (uint32_t)SvUV(ST(1)) : UINT32_MAX;
    REEXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
    while (max_count-- > 0 && pubsub_int16_poll(sub, &value))
        mXPUSHi((IV)value);

void
eventfd_set(self, fd)
    SV *self
    int fd
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    sub->notify_fd = fd;

IV
fileno(self)
    SV *self
  PREINIT:
    EXTRACT_SUB("Data::PubSub::Shared::Int16::Sub", self);
  CODE:
    RETVAL = sub->notify_fd;
  OUTPUT:
    RETVAL