view release on metacpan or  search on metacpan

Shared.xs  view on Meta::CPAN

                    SvGETMAGIC(wsv);
                    { double w = SvOK(wsv) ? SvNV(wsv) : 0.0;  /* undef -> 0 -> clean croak below, no warning */
                      if (!(w > 0.0) || !isfinite(w))
                          croak("Data::Reservoir::Shared->add_many: weight must be a finite number > 0");
                      ws[i] = w; }
                } else if (el && *el) {
                    STRLEN len;
                    const char *src = SvPVbyte(*el, len); /* may run overload/tie/get-magic = arbitrary Perl */
                    /* Copy bytes into a private mortal SV NOW: a LATER element SvPVbyte can
                     * grow/free THIS element PV, dangling src before the locked loop uses it. */
                    SV *copy = sv_2mortal(newSVpvn(src, len));
                    ps[i] = SvPVX_const(copy);
                    ls[i] = len;
                } else { ps[i] = ""; ls[i] = 0; }
            }
        }
        /* Element get-magic in the resolve loop above (SvPVbyte / SvGETMAGIC on
         * the weight) can also have destroyed self. Outside the `if (cnt)`
         * block: an empty or tied size-0 array skips the loop but still
         * reaches this lock. */
        REEXTRACT(self);
        rsv_rwlock_wrlock(h);
        if (weighted)
            for (i = 0; i < cnt; i++) stored += (UV)rsv_add_weighted_locked(h, ps[i], (uint64_t)ls[i], ws[i]);
        else
            for (i = 0; i < cnt; i++) stored += (UV)rsv_add_locked(h, ps[i], (uint64_t)ls[i]);
        __atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
        rsv_rwlock_wrunlock(h);
    }
    RETVAL = stored;
  OUTPUT:
    RETVAL

# ---- reading the sample ----

SV *
get(self, i)
    SV *self
    UV i
  PREINIT:
    EXTRACT(self);
    const uint8_t *ptr; uint64_t len; int ok;
    uint8_t *tmp = NULL;
  CODE:
    /* copy the item out under the read lock, then build the SV (no croak-capable
     * Perl allocation while the lock is held) */
    Newx(tmp, (size_t)h->item_size, uint8_t); SAVEFREEPV(tmp);   /* cached: matches rsv_get_locked's clamp */
    rsv_rwlock_rdlock(h);
    ok = rsv_get_locked(h, (uint64_t)i, &ptr, &len);
    if (ok) memcpy(tmp, ptr, (size_t)len);
    rsv_rwlock_rdunlock(h);
    RETVAL = ok ? newSVpvn((char *)tmp, (STRLEN)len) : &PL_sv_undef;
  OUTPUT:
    RETVAL

void
sample(self)
    SV *self
  PREINIT:
    EXTRACT(self);
  PPCODE:
    {
        uint64_t k = h->k, isz = h->stride, cnt, i;   /* cached geometry: alloc + copy agree */
        uint8_t *buf;
        /* worst-case snapshot buffer (k slots) allocated BEFORE the lock */
        Newx(buf, (size_t)(k * isz), uint8_t); SAVEFREEPV(buf);
        rsv_rwlock_rdlock(h);
        cnt = rsv_count_locked(h);
        {
            uint64_t smax = rsv_slots_max(h);
            if (cnt > smax) cnt = smax;
            memcpy(buf, rsv_slots(h), (size_t)(cnt * isz));   /* snapshot filled slots */
        }
        rsv_rwlock_rdunlock(h);
        EXTEND(SP, (SSize_t)cnt);
        for (i = 0; i < cnt; i++) {
            uint8_t *slot = buf + i * isz;
            uint64_t len; memcpy(&len, slot, sizeof(uint64_t));
            if (len > h->item_size) len = h->item_size;   /* cached clamp */
            PUSHs(sv_2mortal(newSVpvn((char *)(slot + 8), (STRLEN)len)));
        }
    }

UV
count(self)
    SV *self
  PREINIT:
    EXTRACT(self);
    UV c;
  CODE:
    rsv_rwlock_rdlock(h);
    c = (UV)rsv_count_locked(h);
    rsv_rwlock_rdunlock(h);
    RETVAL = c;
  OUTPUT:
    RETVAL

UV
seen(self)
    SV *self
  PREINIT:
    EXTRACT(self);
    UV s;
  CODE:
    rsv_rwlock_rdlock(h);
    s = (UV)h->hdr->seen;
    rsv_rwlock_rdunlock(h);
    RETVAL = s;
  OUTPUT:
    RETVAL

# ---- mutation / introspection ----

void
clear(self)
    SV *self
  PREINIT:
    EXTRACT(self);
  CODE:
    rsv_rwlock_wrlock(h);
    rsv_clear_locked(h);