view release on metacpan or  search on metacpan

Shared.xs  view on Meta::CPAN

                    ps[i] = SvPVX_const(copy);
                    ls[i] = len;
                } else { ps[i] = ""; ls[i] = 0; }
            }
        }
        REEXTRACT(self);
        tk_rwlock_wrlock(h);                             /* locked region: NO croak-capable calls */
        for (i = 0; i < cnt; i++) { tk_observe_locked(h, ps[i], ls[i], 0, 0.0); processed++; }   /* decayed: per-add tick */
        __atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);  /* a call always counts, even an empty batch */
        tk_rwlock_wrunlock(h);
    }
    RETVAL = processed;
  OUTPUT:
    RETVAL

SV *
estimate(self, item)
    SV *self
    SV *item
  PREINIT:
    EXTRACT(self);
    STRLEN n;
    const char *s;
    uint64_t raw; double g = 1.0;
  CODE:
    s = SvPVbyte(item, n);                 /* may croak (wide char) -- BEFORE the lock */
    REEXTRACT(self);
    tk_rwlock_rdlock(h);
    raw = tk_estimate_locked(h, s, n, NULL);
    if (h->mode == TK_MODE_DECAYED) g = tk_g(h);
    tk_rwlock_rdunlock(h);
    RETVAL = (h->mode == TK_MODE_DECAYED) ? newSVnv(tk_w_get(raw) / g) : newSVuv((UV)raw);
  OUTPUT:
    RETVAL

SV *
error(self, item)
    SV *self
    SV *item
  PREINIT:
    EXTRACT(self);
    STRLEN n;
    const char *s;
    uint64_t err = 0; double g = 1.0;
  CODE:
    s = SvPVbyte(item, n);                 /* may croak (wide char) -- BEFORE the lock */
    REEXTRACT(self);
    tk_rwlock_rdlock(h);
    (void)tk_estimate_locked(h, s, n, &err);
    if (h->mode == TK_MODE_DECAYED) g = tk_g(h);
    tk_rwlock_rdunlock(h);
    RETVAL = (h->mode == TK_MODE_DECAYED) ? newSVnv(tk_w_get(err) / g) : newSVuv((UV)err);
  OUTPUT:
    RETVAL

void
top(self, ...)
    SV *self
  PREINIT:
    EXTRACT(self);
  PPCODE:
    {
        UV k = (items > 1 && (SvGETMAGIC(ST(1)), SvOK(ST(1)))) ? SvUV(ST(1)) : 0;   /* optional; undef/omitted = 0 = all */
        TkEnt *ents = NULL;
        char  *keys = NULL;
        REEXTRACT(self);
        uint64_t used, ks = h->key_size, i, cap = h->capacity;
        /* Allocate worst-case snapshot buffers (all `capacity` slots) BEFORE the
           lock: Newx can croak on OOM, and under the read lock that longjmp would
           strand the lock.  `used` (below) is clamped to capacity, so these fit. */
        if (cap) {
            Newx(ents, (size_t)cap, TkEnt);          SAVEFREEPV(ents);
            Newx(keys, (size_t)(cap * ks), char);    SAVEFREEPV(keys);
        }
        /* Snapshot every monitored (count, error, key) under the read lock; the
           copied key bytes cannot then be evicted out from under us, and all
           Perl-value building happens after the unlock. */
        tk_rwlock_rdlock(h);
        double g = (h->mode == TK_MODE_DECAYED) ? tk_g(h) : 1.0;   /* decay factor snapshot */
        used = tk_heap_size(h);            /* == used, clamped to capacity (Layer B) */
        for (i = 0; i < used; i++) {
            TkSlot *sl = tk_slot(h, i);
            uint32_t kl = sl->key_len; if (kl > h->key_size) kl = h->key_size;  /* Layer B */
            ents[i].count = sl->count;
            ents[i].error = sl->error;
            ents[i].off   = i * ks;        /* 64-bit byte offset into the key blob */
            ents[i].klen  = kl;
            memcpy(keys + ents[i].off, tk_slot_key(sl), kl);
        }
        tk_rwlock_rdunlock(h);

        if (used) {
            qsort(ents, (size_t)used, sizeof(TkEnt), tk_ent_cmp);
            uint64_t want = (k == 0 || k > used) ? used : (uint64_t)k;
            EXTEND(SP, (SSize_t)want);
            for (i = 0; i < want; i++) {
                HV *hv = newHV();
                hv_stores(hv, "key",   newSVpvn(keys + ents[i].off, ents[i].klen));
                if (h->mode == TK_MODE_DECAYED) {
                    hv_stores(hv, "count", newSVnv(tk_w_get(ents[i].count) / g));
                    hv_stores(hv, "error", newSVnv(tk_w_get(ents[i].error) / g));
                } else {
                    hv_stores(hv, "count", newSVuv((UV)ents[i].count));
                    hv_stores(hv, "error", newSVuv((UV)ents[i].error));
                }
                PUSHs(sv_2mortal(newRV_noinc((SV *)hv)));
            }
        }
    }

void
clear(self)
    SV *self
  PREINIT:
    EXTRACT(self);
  CODE:
    tk_rwlock_wrlock(h);
    tk_clear_locked(h);
    __atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
    tk_rwlock_wrunlock(h);