Data-HierTimingWheel-Shared
view release on metacpan or search on metacpan
billions, in a fixed amount of memory. It is the multi-level
generalisation of Data::TimingWheel::Shared: where the single-level wheel
revisits a far-future timer once per rotation, this one parks it in a
coarse level and only touches it as its time approaches.
Time advances in integer ticks. There are "num_levels" cascading wheels of
"num_slots" (= S) buckets each; a level-"k" slot spans "S**k" ticks, so
the whole structure schedules any delay in "[1, S**num_levels)". A timer
is placed in the lowest level whose range covers its delay. On each tick,
level 0 fires the timers in its current slot; when level 0 completes a
rotation, the next level's current slot cascades down -- its timers are
redistributed into finer levels by their remaining delay -- recursively up
the levels. A timer with delay "D" fires at exactly tick "D", just like
the single-level wheel, but a far-future timer costs O(1) instead of one
visit per rotation.
Each timer carries an arbitrary 64-bit payload (e.g. a job id) returned
when it fires. Timers live in a fixed pool of "capacity" slots; scheduling
beyond it, or with a delay at or beyond "S**num_levels", croaks.
Because the wheels live in a shared mapping, several processes schedule
returning 1 if it was cancelled or 0 if it had already fired or the id is
not active.
Advancing the clock
my @due = $tw->advance($ticks); # advance by $ticks (default 1)
my @due = $tw->advance; # advance by one tick
"advance" moves the wheel forward by $ticks ticks (default 1) and returns
the list of payloads of every timer that came due during those ticks, in
fire order. Timers that fire are removed automatically. Cost is O(ticks +
fired) amortised; cascades happen only when a level rolls over.
Introspection and lifecycle
$tw->now; # absolute tick count since creation (or last clear)
$tw->count; # number of pending timers
$tw->num_slots; # slots per level (S)
$tw->num_levels; # number of levels (L)
$tw->max_delay; # largest schedulable delay (S**L - 1)
$tw->capacity; # maximum concurrent timers
$tw->clear; # cancel all timers and reset the clock to 0
$tw->stats; # { now, count, num_slots, num_levels, max_delay, capacity, ops, mmap_size }
eg/cross_process.pl view on Meta::CPAN
push @pids, $pid;
}
waitpid $_, 0 for @pids;
printf "parent: %d timers scheduled by %d children\n\n", $tw->count, $kids;
# advance the clock past every delay and count how many fire
my $total_fired = 0;
$total_fired += scalar $tw->advance(1) for 1 .. 5000;
printf "after 5000 ticks: %d timers fired, %d still pending\n", $total_fired, $tw->count;
printf "(every delay was in 1..5000, so all should have fired -- through 2 levels of cascades)\n";
hiertimingwheel.h view on Meta::CPAN
/*
* hiertimingwheel.h -- Shared-memory hierarchical timing wheel for Linux
*
* O(1) timer scheduling at any delay: a hierarchical (Varghese-Lauck) timing
* wheel of num_levels cascading wheels, each with num_slots (=S) buckets. A
* level-k slot spans S^k ticks, so the structure schedules any delay in
* [1, S^num_levels). Scheduling and cancelling a timer are O(1); a far-future
* timer waits in a coarse level and cascades down to finer levels as its time
* approaches, so it is touched only once per coarse tick (versus a single-level
* wheel that revisits it every rotation). The wheels live in a shared mapping so
* several processes schedule into and advance one clock; a write-preferring futex
* rwlock with reader-slot dead-process recovery guards mutation. Each timer
* carries an arbitrary 64-bit payload returned when it fires.
*
* Layout: Header -> reader_slots[1024] -> occ bitmap -> buckets[num_levels*num_slots] -> timers[capacity]
*/
#ifndef HW_H
hiertimingwheel.h view on Meta::CPAN
}
/* ================================================================
* Hierarchical timing-wheel operations (callers hold the lock)
*
* num_levels cascading wheels of num_slots (=S) buckets each; a level-k slot
* spans tick[k]=S^k ticks, so the whole structure schedules any delay in
* [1, S^num_levels). A timer stores its absolute expiry; it is placed in the
* lowest level whose range covers its remaining delay, at slot (expiry/tick[k])%S.
* On each tick level 0 fires its current slot; when level 0 wraps, the next
* level's now-current slot cascades down -- its timers are re-binned into finer
* levels by remaining delay -- recursively up the levels. Timers live in a fixed
* pool with a free list; each bucket is a doubly-linked list (prev/next) so
* cancellation is O(1). Every timer index and bucket index read from shared
* memory is bounds checked, so a corrupt link can never drive an OOB access.
* ================================================================ */
/* unlink timer `t` from its current bucket's doubly-linked list */
static void hw_unlink(HwHandle *h, uint32_t t) {
HwTimer *tm = hw_timer(h, t);
uint32_t pv = tm->prev, nx = tm->next;
hiertimingwheel.h view on Meta::CPAN
if (!(tm->state & 1u)) return 0; /* free / already fired */
if ((tm->state >> 1) != gen) return 0; /* stale id: this slot was reused by a later timer */
hw_unlink(h, idx);
hw_free(h, idx);
if (h->hdr->count) h->hdr->count--;
return 1;
}
/* re-bin every timer in level-k's now-current bucket into a finer level (called
* when `now` has just reached a multiple of tick[k]). If level k also wrapped,
* cascade level k+1 first (top-down) so higher levels feed into this one. */
static void hw_cascade(HwHandle *h, uint32_t k) {
uint32_t S = h->num_slots;
uint64_t slot = (h->hdr->now / h->tick[k]) % S;
if (slot == 0 && k + 1 < h->num_levels) hw_cascade(h, k + 1); /* level k wrapped too */
uint64_t b = (uint64_t)k * S + slot;
uint32_t t = hw_slots(h)[b];
hw_slots(h)[b] = HW_NIL; /* detach the whole list */
uint64_t guard = 0;
while (HW_TIMER_OK(h, t) && guard++ <= (uint64_t)h->capacity) {
HwTimer *tm = hw_timer(h, t);
uint32_t nx = tm->next;
if (tm->state & 1u) /* re-insert into a lower level (or level-0 current slot) */
hw_link(h, t, hw_bucket_for(h, tm->expiry, h->hdr->now));
t = nx;
hiertimingwheel.h view on Meta::CPAN
/* advance the wheel by `ticks`, collecting fired payloads into out[] (capped at
* out_cap); returns the number fired. (caller holds the write lock) */
static uint64_t hw_advance_locked(HwHandle *h, uint64_t ticks, uint64_t *out, uint64_t out_cap) {
uint64_t fired = 0;
uint32_t S = h->num_slots;
if (S == 0 || h->num_levels == 0) return 0;
for (uint64_t j = 0; j < ticks; j++) {
h->hdr->now++;
uint64_t now = h->hdr->now;
if (now % S == 0 && h->num_levels > 1) /* level 0 wrapped -> cascade higher levels down */
hw_cascade(h, 1);
uint64_t b = now % S; /* level-0 current slot (tick[0]==1) */
uint32_t t = hw_slots(h)[b];
uint64_t guard = 0;
while (HW_TIMER_OK(h, t) && guard++ <= (uint64_t)h->capacity) {
HwTimer *tm = hw_timer(h, t);
if (!(tm->state & 1u)) break; /* Layer B: corrupt link into a freed node */
uint32_t nx = tm->next;
hw_unlink(h, t); /* every timer in level-0's current slot is due now */
hw_free(h, t);
if (h->hdr->count) h->hdr->count--;
lib/Data/HierTimingWheel/Shared.pm view on Meta::CPAN
cancelling are O(1) at any delay>, from one tick to billions, in a fixed amount
of memory. It is the multi-level generalisation of L<Data::TimingWheel::Shared>:
where the single-level wheel revisits a far-future timer once per rotation, this
one parks it in a coarse level and only touches it as its time approaches.
Time advances in integer B<ticks>. There are C<num_levels> cascading wheels of
C<num_slots> (= S) buckets each; a level-C<k> slot spans C<S**k> ticks, so the
whole structure schedules any delay in C<[1, S**num_levels)>. A timer is placed
in the lowest level whose range covers its delay. On each tick, level 0 fires the
timers in its current slot; when level 0 completes a rotation, the next level's
current slot B<cascades down> -- its timers are redistributed into finer levels
by their remaining delay -- recursively up the levels. A timer with delay C<D>
fires at exactly tick C<D>, just like the single-level wheel, but a far-future
timer costs O(1) instead of one visit per rotation.
Each timer carries an arbitrary 64-bit B<payload> (e.g. a job id) returned when
it fires. Timers live in a fixed pool of C<capacity> slots; scheduling beyond it,
or with a delay at or beyond C<S**num_levels>, croaks.
Because the wheels live in a shared mapping, B<several processes schedule into
and advance one clock>: any process that opens the same backing file, inherits
lib/Data/HierTimingWheel/Shared.pm view on Meta::CPAN
if it was cancelled or 0 if it had already fired or the id is not active.
=head2 Advancing the clock
my @due = $tw->advance($ticks); # advance by $ticks (default 1)
my @due = $tw->advance; # advance by one tick
C<advance> moves the wheel forward by C<$ticks> ticks (default 1) and returns the
list of payloads of every timer that came due during those ticks, in fire order.
Timers that fire are removed automatically. Cost is O(ticks + fired) amortised;
cascades happen only when a level rolls over.
=head2 Introspection and lifecycle
$tw->now; # absolute tick count since creation (or last clear)
$tw->count; # number of pending timers
$tw->num_slots; # slots per level (S)
$tw->num_levels; # number of levels (L)
$tw->max_delay; # largest schedulable delay (S**L - 1)
$tw->capacity; # maximum concurrent timers
$tw->clear; # cancel all timers and reset the clock to 0
t/01-basic.t view on Meta::CPAN
is $tw->num_levels, 3, 'num_levels';
is $tw->max_delay, 16 ** 3 - 1, 'max_delay == S**L - 1';
is $tw->capacity, 100, 'capacity';
is $tw->now, 0, 'fresh: now 0';
is $tw->count, 0, 'fresh: no timers';
is_deeply [$tw->advance(1)], [], 'advancing an empty wheel fires nothing';
is $tw->now, 1, 'advance moves the clock';
}
# THE ORACLE: a timer fires exactly `delay` ticks after scheduling, across delays
# that span every level and force cascades. Tiny wheel (S=4, L=3, max 63) so the
# 63-delay timer drops through all three levels.
{
my $S = 4;
my $tw = Data::HierTimingWheel::Shared->new(undef, $S, 3, 1000);
my %expect; # payload -> the tick it must fire on (== its delay)
my $p = 1;
for my $delay (1, 2, 3, 4, 5, 7, 8, 15, 16, 17, 31, 32, 48, 63) {
$tw->add($delay, $p);
$expect{$p} = $delay;
$p++;
}
is $tw->count, scalar(keys %expect), 'all timers pending';
my %got;
for my $t (1 .. 70) { $got{$_} = $t for $tw->advance(1) }
my $bad = 0;
for my $pl (keys %expect) { $bad++ if ($got{$pl} // -1) != $expect{$pl} }
is $bad, 0, 'every timer fires on exactly its delay tick (S=4, L=3, all levels + cascades)';
is $tw->count, 0, 'all timers fired';
is $tw->now, 70, 'clock advanced 70 ticks';
}
# a second oracle at a wider geometry (S=8, L=3, max 511), delays landing in
# level 2 and cascading down through levels 1 and 0
{
my $tw = Data::HierTimingWheel::Shared->new(undef, 8, 3, 1000);
my %expect;
my $p = 1;
xt/fork_hw.t view on Meta::CPAN
use Test::More;
use POSIX qw(_exit);
plan skip_all => 'author test' unless $ENV{AUTHOR_TESTING};
use Data::HierTimingWheel::Shared;
# An anonymous MAP_SHARED hierarchical wheel inherited across fork: children each
# schedule a disjoint block of timers at delays spanning several levels,
# concurrently (contending on the free-list and bucket lists under the rwlock).
# The parent then advances past every delay and must collect exactly one fire per
# scheduled timer -- no lost schedules, no double fires, no corrupted lists or
# cascades under contention.
my $kids = 4;
my $per = 5_000;
my $maxd = 4000; # spans levels 0..1 for S=64
my $cap = $kids * $per + 16;
my $tw = Data::HierTimingWheel::Shared->new(undef, 64, 3, $cap); # max delay 64**3 - 1
my @pids;
for my $c (0 .. $kids - 1) {
my $pid = fork // die "fork: $!";
if (!$pid) {
xt/fork_hw.t view on Meta::CPAN
}
waitpid $_, 0 for @pids;
is $tw->count, $kids * $per, 'every child scheduled its timers (no lost schedules)';
my %fired;
my $dupes = 0;
for my $t (1 .. $maxd) {
for my $p ($tw->advance(1)) { $dupes++ if $fired{$p}++; }
}
is scalar(keys %fired), $kids * $per, 'every scheduled timer fired exactly once (cascades intact under contention)';
is $dupes, 0, 'no timer fired twice';
is $tw->count, 0, 'no timers left pending after advancing past every delay';
done_testing;
( run in 0.641 second using v1.01-cache-2.11-cpan-e7c6538aa59 )