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

NAME

Acme::Parataxis::Sync::WaitGroup - Coordinate Completion of Multiple Concurrent Fibers

SYNOPSIS

use v5.40;
use Acme::Parataxis qw[async fiber await_sleep];
use Acme::Parataxis::Sync::WaitGroup;

my $wg = Acme::Parataxis::Sync::WaitGroup->new;

async {
    # Register 3 tasks before spawning:
    $wg->add(3);

    for my $id ( 1 .. 3 ) {
        fiber {
            say "Task $id running...";
            await_sleep(50);
            $wg->done;    # decrement counter
        };
    }

    say "Waiting for all tasks to finish...";
    $wg->wait;            # parks until counter hits 0
    say "All tasks complete!";
};

DESCRIPTION

An Acme::Parataxis::Sync::WaitGroup coordinates the completion of multiple asynchronous tasks, providing the fan-in half of a fan-out/fan-in concurrency pattern.

The group maintains an integer counter:

  • add($n) increments the counter by $n.

  • done() decrements the counter by 1.

  • wait() parks the calling fiber until the counter drops to 0.

WaitGroups do not track fibers or job handles; they track a numeric tally. This keeps the primitive lightweight and fast, but requires callers to ensure that every add is matched by exactly one done.

METHODS

new( [ count => $n = 0 ] )

my $wg = Acme::Parataxis::Sync::WaitGroup->new;
my $primed = Acme::Parataxis::Sync::WaitGroup->new( count => 5 );

Constructs a new WaitGroup with an initial counter of $n (default 0).

add( $n )

$wg->add(4);

Adds $n to the counter and returns true. $n must be an integer string or number (matches /\A-?\d+\z/). Negative numbers are supported.

  • Croaks with WaitGroup add() requires an integer if $n is invalid or non-numeric.

  • Croaks with WaitGroup counter would become negative if adding $n drives the counter below zero.

  • If the counter transitions to 0, all fibers currently parked in wait() are unblocked immediately.

Safe to call outside a scheduled fiber.

done( )

$wg->done;

Decrements the counter by 1. Equivalent to $wg->add(-1). Safe to call outside a scheduled fiber.

wait( )

$wg->wait;

Parks the calling fiber until the counter reaches 0. If the counter is already 0, returns true immediately without parking.

Must be called from within a scheduled fiber (croaks otherwise). Bounded waits are supported via with_timeout.

count( ) / remaining( )

Returns the current value of the counter. remaining is an alias. Both work outside of scheduled fibers.

waiters( )

Returns the number of fibers currently parked in wait().

Recipes

Safe fan-out with guaranteed completion

To prevent jobs from stranding the group if an exception occurs, wrap the task in an eval or defer:

use v5.40;
use experimental 'defer';
use Acme::Parataxis qw[async fiber];
use Acme::Parataxis::Sync::WaitGroup;

async {
    my $wg = Acme::Parataxis::Sync::WaitGroup->new;
    $wg->add( scalar @items );

    for my $item (@items) {
        fiber {
            defer { $wg->done };    # always decrements, even if process_item dies!
            process_item($item);
        };
    }

    $wg->wait;
};

Dynamic task registration

A running task can increment the WaitGroup to enroll sub-tasks dynamically:

sub worker ($wg, $task) {
    if ( my @subtasks = split_work($task) ) {
        $wg->add( scalar @subtasks );    # add subtasks before spawning
        for my $st (@subtasks) {
            fiber { worker($wg, $st) };
        }
    }
    $wg->done;
}

Behavior Notes

  • Add before spawn

    Always call $wg->add() before spawning the child fiber. If you call add inside the spawned fiber, the calling fiber might reach wait() while the counter is still 0, causing wait() to return immediately before the children have even begun.

  • Counter underflows

    Calling done() more times than add() throws an exception. Crucially, add() decrements the counter before checking the bounds, so a croaked done() leaves the counter negative. A negative counter causes future wait() calls to return immediately without waiting. Avoid this by matching counts carefully.

  • Re-checking on wake

    wait() re-checks count in a loop upon waking. If a waking fiber is unblocked when count == 0, but a concurrent fiber immediately calls $wg->add(2) before the waiter resumes, the waiter detects the positive count and re-parks automatically until zero is reached again.

SEE ALSO

Acme::Parataxis::Sync, Acme::Parataxis::Sync::Barrier, Acme::Parataxis::Sync::Mutex, Acme::Parataxis::Nursery

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.