Data-SortedSet-Shared
view release on metacpan or search on metacpan
The total order is (score, member): members with equal scores are
ordered by member id, which gives a well-defined rank and a
deterministic "pop_min"/"pop_max".
Multiple processes can map the same set and read and write it
concurrently; access is serialized by a write-preferring futex rwlock
that recovers automatically if a lock holder dies (see "CRASH SAFETY").
Members are 64-bit integers. For string-keyed sets, see "String-keyed
sets" and Data::SortedSet::Shared::Strings (bundled). Scores must not be
NaN. Linux-only. Requires 64-bit Perl.
METHODS
Constructors
my $z = Data::SortedSet::Shared->new($path, $max [, $mode]);
my $z = Data::SortedSet::Shared->new(undef, $max); # anonymous
my $z = Data::SortedSet::Shared->new_memfd($name, $max);
my $z = Data::SortedSet::Shared->new_from_fd($fd);
$path is the backing file ("undef" for an anonymous mapping); $max is
the maximum number of members. When reopening an existing file or memfd,
Mutators
$z->add($member, $score); # 1 new, 0 existing (score updated), undef if full
$z->incr($member, $delta); # add to the score (creating at $delta); returns new score
$z->remove($member); # true if removed, false if absent
my $n = $z->add_many([ [$m1,$s1], [$m2,$s2], ... ]); # bulk; returns count of new
$z->clear;
"add" inserts a new member or updates an existing member's score,
returning 1 or 0 respectively, or "undef" if the pool is full and the
member is new. $score may be any finite or infinite value but not NaN
(croaks). "incr" creates an absent member at $delta (like Redis ZINCRBY)
and croaks if the result would be NaN, or if the pool is full and the
member is new.
"add_many" applies a whole batch under a single lock; each row is an
"[member, score]" arrayref, malformed or NaN-scored rows are skipped,
and it stops at $max. It returns the number of members newly inserted,
which can be fewer than the number of new rows if the pool fills
mid-batch.
Lookup and count
$z->score($member); # the score, or undef if absent
$z->exists($member);
$z->count; # number of members
$z->rank($member); # 0-based rank (lowest score = 0), or undef
$z->rev_rank($member); # rank from the top, or undef
SV *
add(self, member, score)
SV *self
IV member
NV score
PREINIT:
EXTRACT(self);
int rc;
CODE:
if (score != score) croak("add: score must not be NaN");
ss_rwlock_wrlock(h);
rc = ss_add_locked(h, (int64_t)member, (double)score);
__atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
ss_rwlock_wrunlock(h);
RETVAL = (rc < 0) ? &PL_sv_undef : newSViv(rc);
OUTPUT:
RETVAL
SV *
score(self, member)
NV
incr(self, member, delta)
SV *self
IV member
NV delta
PREINIT:
EXTRACT(self);
double out;
int rc;
CODE:
if (delta != delta) croak("incr: delta must not be NaN");
ss_rwlock_wrlock(h);
rc = ss_incr_locked(h, (int64_t)member, (double)delta, &out);
__atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
ss_rwlock_wrunlock(h);
if (rc == -1) croak("incr: max_entries exhausted");
if (rc == -2) croak("incr: result is NaN");
RETVAL = out;
OUTPUT:
RETVAL
void
pop_min(self)
SV *self
PREINIT:
EXTRACT(self);
PPCODE:
SSize_t nr = av_len(av) + 1;
ss_rwlock_wrlock(h);
for (SSize_t i = 0; i < nr; i++) {
SV **rv = av_fetch(av, i, 0);
if (!rv || !SvROK(*rv) || SvTYPE(SvRV(*rv)) != SVt_PVAV) continue; /* skip malformed */
AV *row = (AV *)SvRV(*rv);
if (av_len(row) + 1 < 2) continue;
SV **ms = av_fetch(row, 0, 0), **sv = av_fetch(row, 1, 0);
if (!ms || !sv) continue;
double score = SvNV(*sv);
if (score != score) continue; /* skip NaN */
int rc = ss_add_locked(h, (int64_t)SvIV(*ms), score);
if (rc == 1) added++;
else if (rc == -1) break; /* pool full */
}
__atomic_fetch_add(&h->hdr->stat_ops, 1, __ATOMIC_RELAXED);
ss_rwlock_wrunlock(h);
}
RETVAL = added;
OUTPUT:
RETVAL
lib/Data/SortedSet/Shared.pm view on Meta::CPAN
The total order is B<(score, member)>: members with equal scores are ordered by
member id, which gives a well-defined rank and a deterministic
C<pop_min>/C<pop_max>.
Multiple processes can map the same set and read and write it concurrently;
access is serialized by a write-preferring futex rwlock that recovers
automatically if a lock holder dies (see L</CRASH SAFETY>).
Members are 64-bit integers. For B<string-keyed> sets, see L</String-keyed sets>
and L<Data::SortedSet::Shared::Strings> (bundled). Scores must not be NaN.
B<Linux-only>. Requires 64-bit Perl.
=head1 METHODS
=head2 Constructors
my $z = Data::SortedSet::Shared->new($path, $max [, $mode]);
my $z = Data::SortedSet::Shared->new(undef, $max); # anonymous
my $z = Data::SortedSet::Shared->new_memfd($name, $max);
my $z = Data::SortedSet::Shared->new_from_fd($fd);
lib/Data/SortedSet/Shared.pm view on Meta::CPAN
=head2 Mutators
$z->add($member, $score); # 1 new, 0 existing (score updated), undef if full
$z->incr($member, $delta); # add to the score (creating at $delta); returns new score
$z->remove($member); # true if removed, false if absent
my $n = $z->add_many([ [$m1,$s1], [$m2,$s2], ... ]); # bulk; returns count of new
$z->clear;
C<add> inserts a new member or updates an existing member's score, returning 1 or
0 respectively, or C<undef> if the pool is full and the member is new. C<$score>
may be any finite or infinite value but B<not NaN> (croaks). C<incr> creates an
absent member at C<$delta> (like Redis ZINCRBY) and croaks if the result would be
NaN, or if the pool is full and the member is new.
C<add_many> applies a whole batch under a single lock; each row is an
C<[member, score]> arrayref, malformed or NaN-scored rows are skipped, and it
stops at C<$max>. It returns the number of members B<newly inserted>, which can
be fewer than the number of new rows if the pool fills mid-batch.
=head2 Lookup and count
$z->score($member); # the score, or undef if absent
$z->exists($member);
$z->count; # number of members
$z->rank($member); # 0-based rank (lowest score = 0), or undef
$z->rev_rank($member); # rank from the top, or undef
lib/Data/SortedSet/Shared/Strings.pm view on Meta::CPAN
my $id = $self->{keys}->id_of($str);
return defined $id ? $self->{set}->remove($id) : 0;
}
sub add_many {
my ($self, $rows) = @_;
Carp::croak("add_many: expected an arrayref") unless ref $rows eq 'ARRAY';
my @id_rows;
for my $r (@$rows) {
next unless ref $r eq 'ARRAY' && @$r >= 2;
next if $r->[1] != $r->[1]; # skip a NaN score before interning (no ghost key slot)
my $id = $self->{keys}->intern($r->[0]);
last unless defined $id; # key table full -> stop
push @id_rows, [ $id, $r->[1] ];
}
return $self->{set}->add_many(\@id_rows);
}
sub clear { $_[0]{set}->clear; $_[0]{keys}->clear; return }
# ---- lookup (id_of the key; undef short-circuits) ----
lib/Data/SortedSet/Shared/Strings.pm view on Meta::CPAN
$z->sync; $z->unlink; $z->stats; # stats: { set => {...}, keys => {...} }
C<remove> leaves the key interned (ids are stable); the key universe (C<max_keys>)
must therefore accommodate every distinct key ever added, not just those currently
present. C<clear> is the one exception -- it resets B<both> the set and the key
table, so ids minted before a C<clear> no longer resolve. C<add> returns C<undef>
(and C<incr> croaks) if the key table or the member pool is full. C<add_many>
interns each row's key in order, stopping only if the key table fills, then
bulk-adds; it returns the number of members B<newly inserted> (fewer than the rows
given if either pool fills -- keys interned past the member pool's capacity stay
interned but unadded). Malformed rows and NaN-scored rows are skipped without
interning their key.
=head1 SEE ALSO
L<Data::SortedSet::Shared>, L<Data::Intern::Shared>.
=head1 AUTHOR
vividsnow
sortedset.h view on Meta::CPAN
/* remove: 1 if removed, 0 if absent */
static int ss_remove_locked(SsHandle *h, int64_t member) {
double old;
if (!ss_idx_get(h, member, &old)) return 0;
ss_tree_del(h, old, member);
ss_idx_del(h, member);
return 1;
}
/* incr by delta. *out = new score. returns 1 (created), 0 (updated), -1 (full),
-2 (result is NaN) */
static int ss_incr_locked(SsHandle *h, int64_t member, double delta, double *out) {
double old;
if (ss_idx_get(h, member, &old)) {
double ns = old + delta; *out = ns;
if (ns != ns) return -2;
if (ns != old) { ss_tree_del(h, old, member); ss_tree_add(h, ns, member); ss_idx_set(h, member, ns); }
return 0;
}
if (h->hdr->count >= h->hdr->max_entries) return -1;
*out = delta;
}
ok $sc_ok, 'score() matches oracle for every member';
ok !$z->exists(-12345) && !defined($z->score(-12345)), 'absent member: exists 0, score undef';
# re-add semantics
my ($any) = keys %oracle;
is $z->add($any, $oracle{$any}), 0, 're-add an existing member returns 0';
ok !$z->exists(2_000_000_000), 'fresh member absent before add';
is $z->add(2_000_000_000, 5), 1, 'add of a new member returns 1';
# NaN
eval { $z->add(1, ("NaN" + 0)) }; like $@, qr/NaN/, 'NaN score croaks';
# full pool
my $f = Data::SortedSet::Shared->new(undef, 4);
is $f->add(1, 1), 1, 'add 1/4';
is $f->add(2, 2), 1, 'add 2/4';
is $f->add(3, 3), 1, 'add 3/4';
is $f->add(4, 4), 1, 'add 4/4';
ok !defined($f->add(5, 5)), 'add past max_entries returns undef';
ok $f->_validate, 'small full set is valid';
t/03-query.t view on Meta::CPAN
{
my $iv_min = -(~0 >> 1) - 1; # -2**63
my $zc = Data::SortedSet::Shared->new(undef, 10);
$zc->add($iv_min, 5);
$zc->add(1, 5);
is $zc->count_in_score(5, 5), 2, 'count_in_score includes member=INT64_MIN at the boundary score';
is_deeply [$zc->range_by_score(5, 5)], [$iv_min, 1], 'INT64_MIN member sorts before a positive member at equal score';
is $zc->rank($iv_min), 0, 'INT64_MIN member ranks first at equal score';
}
# +/-Inf scores are allowed (only NaN croaks): -Inf sorts first, +Inf last
{
my $inf = "Inf" + 0;
my $zi = Data::SortedSet::Shared->new(undef, 10);
$zi->add(1, $inf);
$zi->add(2, -$inf);
$zi->add(3, 0);
$zi->add(4, 1e308); # large finite, below +Inf
ok $zi->_validate, 'tree valid with +/-Inf scores';
is_deeply [$zi->range_by_rank(0, -1)], [2, 3, 4, 1], '-Inf first, +Inf last, finite in between';
my ($pmin) = $zi->peek_min;
t/04-mutate.t view on Meta::CPAN
is $e->incr(7, 3), 3, 'incr on absent member creates at delta';
is $e->score(7), 3, 'incr created the member';
is $e->incr(7, 2.5), 5.5, 'incr existing returns new score';
is $e->add(7, 10), 0, 'add existing returns 0 (update)';
is $e->score(7), 10, 'add updated the score';
is $e->rank(7), 0, 'single member rank 0';
is_deeply [$e->pop_min], [7, 10], 'pop_min returns the only element';
is $e->count, 0, 'empty after popping last';
ok $e->_validate, 'empty tree valid';
# NaN-result guards
eval { $e->incr(5, ("NaN" + 0)) }; like $@, qr/NaN/, 'incr with NaN delta croaks';
$e->add(99, ("Inf" + 0));
eval { $e->incr(99, -("Inf" + 0)) }; like $@, qr/NaN/, 'incr to a NaN result croaks';
$e->clear;
# pop ordering with ties
$e->add($_, $_ % 5) for 1 .. 50;
my ($pm, $ps) = $e->pop_min;
is $ps, 0, 'pop_min takes the lowest score';
my ($xm, $xs) = $e->pop_max;
is $xs, 4, 'pop_max takes the highest score';
ok $e->_validate, 'valid after pops';
t/05-bulk.t view on Meta::CPAN
use Data::SortedSet::Shared;
my $z = Data::SortedSet::Shared->new(undef, 10000);
my @rows = map { [ $_, ($_ * 7) % 50 ] } 1 .. 3000;
is $z->add_many(\@rows), 3000, 'add_many returns count of new members';
is $z->count, 3000, 'count after add_many';
ok $z->_validate, 'valid after add_many';
my $ok = 1; for (1 .. 3000) { $ok = 0, last unless $z->score($_) == ($_ * 7) % 50 }
ok $ok, 'scores correct after add_many';
# updates + malformed rows + a NaN-scored row + new
my $a2 = $z->add_many([ [1, 999], [2, 888], "bad", [], [3], [3003, ("NaN" + 0)], [3001, 5], [3002, 6] ]);
is $a2, 2, 'add_many counts only new (updates + malformed + NaN skipped)';
ok !$z->exists(3003), 'NaN-scored row is skipped, not inserted';
is $z->score(1), 999, 'add_many updated an existing score';
is $z->count, 3002, 'count after mixed add_many';
ok $z->_validate, 'valid after mixed add_many';
eval { $z->add_many("nope") }; like $@, qr/arrayref/, 'add_many non-arrayref croaks';
my $f = Data::SortedSet::Shared->new(undef, 3);
is $f->add_many([ [1,1],[2,2],[3,3],[4,4],[5,5] ]), 3, 'add_many stops at max_entries';
is $f->count, 3, 'count capped at max_entries';
my $s = $z->stats;
t/06-strings.t view on Meta::CPAN
# the parent class's new_strings() convenience constructor (delegates to Strings->new)
{
require Data::SortedSet::Shared;
my $z2 = Data::SortedSet::Shared->new_strings(max => 1000);
isa_ok $z2, 'Data::SortedSet::Shared::Strings', 'new_strings returns a Strings object';
is $z2->add("x", 10), 1, 'new_strings: add works';
is $z2->score("x"), 10, 'new_strings: score round-trip';
is $z2->count, 1, 'new_strings: count';
}
# add_many skips a NaN-scored row WITHOUT interning its key (no ghost slot)
{
my $z3 = Data::SortedSet::Shared::Strings->new(max => 100);
my $before = $z3->key_table->count;
my $n = $z3->add_many([ ["good", 5], ["bad", "NaN"+0], ["good2", 7] ]);
is $n, 2, 'add_many: NaN-scored row not added';
ok !$z3->exists("bad"), 'add_many: NaN-row member absent';
is $z3->key_table->count, $before + 2, 'add_many: NaN-row key not interned (no ghost slot)';
}
done_testing;
xt/invalid_args.t view on Meta::CPAN
use strict;
use warnings;
use Test::More;
use Data::SortedSet::Shared;
my $z = Data::SortedSet::Shared->new(undef, 100);
eval { Data::SortedSet::Shared->new(undef, 0) }; ok $@, 'max_entries 0 rejected';
eval { $z->add(1, ("NaN" + 0)) }; like $@, qr/NaN/, 'NaN score croaks';
eval { $z->incr(1, ("NaN" + 0)) }; like $@, qr/NaN/, 'NaN incr delta croaks';
eval { $z->add_many("nope") }; like $@, qr/arrayref/, 'add_many non-arrayref croaks';
eval { $z->each("notcode") }; ok $@, 'each non-coderef croaks';
eval { $z->range_by_score(1, 2, bogus => 1) }; like $@, qr/unknown option/, 'unknown range option croaks';
eval { $z->range_by_score(1, 2, "withscores") }; like $@, qr/key => value/, 'odd range-option count croaks';
my $full = Data::SortedSet::Shared->new(undef, 2);
$full->add(1, 1); $full->add(2, 2);
eval { $full->incr(3, 5) }; like $@, qr/exhausted|max_entries/, 'incr on a full pool croaks';
eval { Data::SortedSet::Shared->new_from_fd(-1) }; ok $@, 'new_from_fd of an invalid fd croaks';
( run in 1.408 second using v1.01-cache-2.11-cpan-9581c071862 )