/ 
Acme-Parataxis-v0.1.1
/ Acme::Parataxis::Sync::WaitGroup

NAME

Acme::Parataxis::Sync::WaitGroup - Count outstanding jobs and wait for the count to reach zero

SYNOPSIS

use v5.40;
use Acme::Parataxis qw[async fiber];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
async {
    $wg->add(3);                            # three jobs about to start
    my @jobs = map {
        fiber {
            my $id = $_;
            say "job $id running";
            $wg->done;                      # one done per add, exactly once
        }
    } 1 .. 3;
    $wg->wait;                              # parks until the count is back to zero
    $_->await for @jobs;
    say 'every job has reported, count is ', $wg->count;
};
# job 1 running
# job 2 running
# job 3 running
# every job has reported, count is 0

DESCRIPTION

A counter shared between fibers. add( $n ) puts n units on it, done takes one off, and wait parks the calling fiber until nothing is left outstanding. It is the fan-in half of fan-out: start N fibers, have each report when it finishes, and wait for the reports rather than for the fibers.

The group holds no job state and knows no jobs at all. It is a number, and done is a decrement - which is worth keeping in mind, because it means the group cannot tell whether the reports it received are the ones it was waiting for. A job that dies without reporting leaves its unit on the counter forever, and no later wait returns. "A job that dies without done" is the failure mode to design around.

The counter is not a queue and the group does not order anything. Two fibers that call done at the same moment are both accepted; what matters is only that the total reaches zero, not who got there first.

METHODS

new( [ count = $n ] )>

use v5.40;
use Acme::Parataxis::Sync::WaitGroup;
my $fresh = Acme::Parataxis::Sync::WaitGroup->new;
say 'a new group: count ', $fresh->count, ', waiters ', $fresh->waiters;
my $primed = Acme::Parataxis::Sync::WaitGroup->new( count => 3 );
say 'a primed one: count ', $primed->count;
# a new group: count 0, waiters 0
# a primed one: count 3

Creates a group. The only parameter is count, which primes the counter and defaults to 0. Any other key croaks with Unrecognized parameters for "Acme::Parataxis::Sync::WaitGroup" constructor - including remaining, which is a reader but not a parameter.

add( $n )

use v5.40;
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
say 'add(3): ', $wg->add(3), ', count now ', $wg->count;
say 'add(0): ', $wg->add(0), ', count still ', $wg->count;
for my $n ( 2.5, 'x' ) {
    my $trial = Acme::Parataxis::Sync::WaitGroup->new( count => 4 );
    say "add('$n'): ", ( eval { $trial->add($n); 'ok' } || 'croaks' ), ', count still ', $trial->count;
}
# add(3): 1, count now 3
# add(0): 1, count still 3
# add('2.5'): croaks, count still 4
# add('x'): croaks, count still 4

Adds $n to the counter and returns true. The argument is checked by string, against /\A-?\d+\z/, so an integer written as a number, as a string, or as a float that happens to have no fractional part is all accepted, while a real fraction is not. A reference, undef and anything non-numeric croak with WaitGroup add() requires an integer, and the counter is left alone.

$n may be negative, which is how done is implemented. A $n that would take the counter below zero croaks with WaitGroup counter would become negative - but see "A croaking done still moves the counter" for what the counter reads afterwards.

done( )

use v5.40;
use Acme::Parataxis qw[async fiber];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
async {
    $wg->add( 1 );
    say 'done: ', $wg->done, ', count now ', $wg->count;
    say 'one too many: ', ( eval { $wg->done; 'ok' } || 'croaks' );
};
# done: 1, count now 0
# one too many: croaks

Removes one unit and returns true. It is exactly add(-1), with the same validation and the same return value, and it needs a fiber no more than add does - both work off a scheduled fiber.

wait( )

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new( count => 1 );
async {
    my $slow = fiber { await_sleep( 50 ); $wg->done };
    say 'parking on a count of ', $wg->count;
    $wg->wait;
    say 'woken, and the count reads ', $wg->count;
    $slow->await;
};
# parking on a count of 1
# woken, and the count reads 0

Parks the calling fiber until the counter reaches zero, then returns true. Called on a group whose count is already zero it returns immediately without parking, so a wait is never a way to find out whether work is outstanding - check remaining for that. Called off a scheduled fiber it croaks with WaitGroup waits must occur inside a scheduled fiber.

A wait is interruptible, and an interrupted one gives up its place in the queue, leaving the group itself untouched and usable. See "Waiting with a deadline".

count and remaining

use v5.40;
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new( count => 2 );
say 'count: ', $wg->count, ', remaining: ', $wg->remaining;
$wg->done;
say 'after one done: count ', $wg->count, ', remaining ', $wg->remaining;
# count: 2, remaining: 2
# after one done: count 1, remaining 1

The same number, twice over. count is the counter field, and remaining is an ordinary method that returns it, so the two can never disagree. Pick whichever reads better at the call site: count when you are thinking about the arithmetic, remaining when you are thinking about how much is left.

Both are readers, so they work off a scheduled fiber as well as on one, and neither croaks.

waiters

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new( count => 1 );
async {
    my @ws = map { fiber { $wg->wait } } 1 .. 3;
    await_sleep( 10 );
    say 'parked in wait: ', $wg->waiters;
    $wg->done;
    $_->await for @ws;
    say 'once the count reached zero: ', $wg->waiters;
};
# parked in wait: 3
# once the count reached zero: 0

How many fibers are currently parked in wait. The count drops to zero the moment the last unit is taken off, before any of those fibers has run again, so this reads as a queue depth rather than as a list of fibers still to finish. It is a plain method and works off a fiber.

How It Works

The state is one integer and one array of fiber ids. The integer is the counter; the array is the fibers parked in wait, in the order they arrived.

Three methods do all the work. add moves the counter and, if that landed it on zero, calls the private _release, which empties the queue and wakes every fiber in it at once. wait pushes the calling fiber's id and parks it. done is add(-1), so it goes through the same release check and needs no code of its own.

wait is a loop rather than a single park, and that is the one detail worth knowing. A fiber released from wait does not simply return: it re-tests the counter, and if anything has been added since the release it pushes its id back on the queue and parks again. wait therefore means "wait until the count is zero", not "wait until the next time the count is zero" - which is the guarantee you want when a job can add more jobs as it runs.

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new( count => 1 );
async {
    my $w = fiber { $wg->wait };
    await_sleep( 10 );
    say 'parked, and the count is ', $wg->count;
    $wg->add( 2 );
    $wg->done for 1 .. 2;
    await_sleep( 10 );
    say 'back to ', $wg->count . ', but the waiter is still parked: ', $wg->waiters;
    $wg->done;
    $w->await;
    say 'and now it is through, at ', $wg->waiters;
};
# parked, and the count is 1
# back to 1, but the waiter is still parked: 1
# and now it is through, at 0

The park is registered with a cleanup closure that pulls the fiber's id back out of the queue, so a wait that is cut short by a deadline or a cancellation does not leave a stale id behind for the next release to wake.

Recipes

Fan out, then fan in

The shape to reach for. Count the jobs before starting them, let each report when it finishes, and wait for the count.

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
my @squares;
async {
    $wg->add(4);                             # before the jobs, always
    my @jobs = map {
        fiber {
            my $id = $_;
            await_sleep( 10 );
            push @squares, $id * $id;
            $wg->done;
        }
    } 1 .. 4;
    $wg->wait;
    say 'the squares came back ', join ', ', ( sort { $a <=> $b } @squares );
};
# the squares came back 1, 4, 9, 16

The add comes first deliberately. See "Add before you spawn" for what happens when it does not.

A job that cannot strand the group

done in the same place on every path out of the job, so a failure still reports. The eval is what makes that possible: it turns the job's own failure into a value, so the reporting line is reached either way.

use v5.40;
use Acme::Parataxis qw[async fiber];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
my @outcomes;
async {
    my @jobs = map {
        fiber {
            my $id = $_;
            $wg->add( 1 );
            my $ok = eval { die "job $id failed\n" if $id == 3; 1 };
            $wg->done;                       # always, failure or not
            push @outcomes, $id . ( $ok ? ' ok' : ' failed' );
        }
    } 1 .. 4;
    $wg->wait;
    say join( ', ', sort @outcomes ), ' and the count is back to ', $wg->count;
};
# 1 ok, 2 ok, 3 failed, 4 ok and the count is back to 0

Note the add inside the job rather than before it. That is what makes the group safe to add to while it is already being waited on: a job that registers itself is counted before it can finish, so there is no window in which a fast job reports against a counter that does not know about it yet.

Waiting with a deadline

A wait is an ordinary interruptible wait, so with_timeout( $ms, ... ) bounds it. The waiter gives up its place in the queue on the way out, and the group stays usable - which matters, because the jobs it was waiting for are still running and will still report.

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
async {
    $wg->add( 1 );
    fiber { await_sleep( 300 ); $wg->done };
    eval { Acme::Parataxis::with_timeout( 30, sub { $wg->wait } ) };
    my $why = $@ ? ( $@->message =~ s/ at .*//sr ) : 'it returned';
    say "the fan-in gave up: $why";
    say 'the group is still usable, and the count reads ', $wg->count;
};
# the fan-in gave up: The operation timed out
# the group is still usable, and the count reads 1

Behavior Notes

Add before you spawn

add and the jobs it counts have to be ordered, and only one order is safe. A job that calls done before the matching add lands drives the counter below zero and croaks, and the add that follows then lands on a counter that has already been thrown off by the amount.

use v5.40;
use Acme::Parataxis qw[async fiber];
use Acme::Parataxis::Sync::WaitGroup;
my $late = Acme::Parataxis::Sync::WaitGroup->new;
my @after;
async {
    my @jobs = map { fiber { push @after, ( eval { $late->done; 'ok' } || 'croaked' ) } } 1 .. 2;
    $late->add( scalar @jobs );
    $_->await for @jobs;
    say 'counted after spawning: ', join( ', ', @after ), ', ending at ', $late->count;
};
# counted after spawning: croaked, croaked, ending at 0

The ending count of 0 looks like success, and it is the arithmetic of two croaked calls rather than two reports. Count first - either $wg->add( scalar @jobs ) on a list of handles you have already built, or $wg->add(1) as the first line of each job, which is the form "A job that cannot strand the group" uses.

A croaking done still moves the counter

add adjusts the counter and then checks it, so a done that croaks for want of a matching add has already taken its unit off. The croak tells you the call was wrong; it does not undo it.

use v5.40;
use Acme::Parataxis qw[async];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new( count => 1 );
async {
    eval { $wg->done; $wg->done };
    say 'the second done croaked, and the count reads ', $wg->count;
};
# the second done croaked, and the count reads -1

The consequence is worse than the wrong number. wait parks only while the count is above zero, so a counter that has been driven negative makes every later wait return at once - the group reports success while jobs are still running, and no amount of waiting will catch up.

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new( count => 2 );
async {
    eval { $wg->done; $wg->done; $wg->done };     # one more than there was
    say 'the counter now reads ', $wg->count;
    my $stray = fiber { await_sleep( 20 ); eval { $wg->done; 1 } };
    $wg->wait;
    say 'and wait returned at once, with the job still running';
    $stray->await;
};
# the counter now reads -1
# and wait returned at once, with the job still running

So a negative count is not something to test for at the call site; it is something to avoid by counting before spawning. If you are picking a group up from code you do not control, remaining reading below zero is the signal that its accounting has already gone wrong.

A job that dies without done

done is a line of code, and nothing calls it for you. A job that returns early, takes an exception that escapes, or is torn down mid-flight leaves its unit on the counter, and because wait is a loop rather than a single park, nothing will ever satisfy it.

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;
my $wg = Acme::Parataxis::Sync::WaitGroup->new;
async {
    $wg->add( 2 );
    fiber { $wg->done };
    fiber { eval { die "the job blew up\n" }; 1 };
    await_sleep( 20 );
    say 'both jobs have finished, yet the count is ', $wg->count;
    eval { Acme::Parataxis::with_timeout( 20, sub { $wg->wait } ) };
    say 'so waiting gives: ', ( $@ ? ( $@->message =~ s/ at .*//sr ) : 'it returned' );
};
# both jobs have finished, yet the count is 1
# so waiting gives: The operation timed out

Left alone this does not merely return the wrong answer, it hangs: the scheduler eventually reports a deadlock, because the only fiber left is parked on a count nothing is going to move.

The fix is the one in "A job that cannot strand the group" - put the done where every path reaches it. Where that is awkward, a small wrapper is worth having, so the reporting lives in one place rather than in every job.

SEE ALSO

Acme::Parataxis::Sync, Acme::Parataxis::Sync::Barrier, Acme::Parataxis::Sync::Once, Acme::Parataxis::Sync::Mutex, Acme::Parataxis::Sync::RwLock, Acme::Parataxis::Nursery, Acme::Parataxis::Error::Timeout, Acme::Parataxis::Error::Cancelled

AUTHOR

Sanko Robinson https://github.com/sanko

LICENSE

Copyright (C) Sanko Robinson.

This library is free software; you can redistribute it and/or modify it under the terms found in the Artistic License 2.