/ 
Acme-Parataxis-v0.1.1
/ Acme::Parataxis::Signal

NAME

Acme::Parataxis::Signal - One-shot flag with a FIFO queue of waiters

SYNOPSIS

use Acme::Parataxis;
use Acme::Parataxis::Signal;

my $ready = Acme::Parataxis::Signal->new( count => false );

async {
    fiber { $ready->wait; say 'the work finished' };
    $ready->send;                            # releases the fiber above
};

DESCRIPTION

A signal is a flag and a queue. send sets the flag or hands it straight to whoever is waiting, and wait parks the calling fiber until it arrives. Waiters park rather than spin, and are released in the order they arrived.

The flag is what makes a signal safe to fire early. A send that finds nobody waiting is remembered, so the next wait returns immediately rather than blocking on something that already happened. That is the difference between a signal and a plain join, and it is why you can signal before the fibers that care have even been spawned.

A signal releases one waiter per send, and broadcast releases all of them. There is no reset: once the queue drains, the flag is not set again unless something sets it.

METHODS

new( [ count => $bool ] )

my $ready = Acme::Parataxis::Signal->new( count => false );   # starts unsignalled
my $done  = Acme::Parataxis::Signal->new;                     # starts signalled

Creates a signal. count is really a boolean and it defaults to true, so a plain new is already signalled and its first wait will not block. Pass count => false for the usual case, where the signal is something still to happen.

The reader count returns that boolean, so test it for truth rather than comparing it to a number.

send( )

$ready->send;      # release one waiter, or remember the signal

Wakes the longest-waiting fiber, or invokes the longest-waiting callback, and returns true. If there is nobody waiting, the signal is remembered instead, so the next wait does not block.

broadcast( )

$ready->broadcast;   # release every waiter that is queued right now

Wakes every currently queued waiter and returns true. A waiter registered by a callback during the broadcast is not part of it - the wake budget is fixed when the broadcast starts.

Unlike send, a broadcast with nobody to wake does not remember anything. It also cancels a signal that was already remembered, so do not use it to clear a queue you have not drained; see "broadcast cancels a pending signal".

wait( )

$ready->wait;            # park until signalled
$ready->wait(\&callback);  # or register a callback instead of parking

With no argument, consumes a remembered signal if there is one and returns, otherwise parks the current fiber until the signal arrives. Always returns undef; use awaited to ask how many are waiting.

Must run inside a scheduled fiber, unless a signal is already pending, in which case it returns without parking and needs no fiber at all.

With a coderef, registers that callback and returns immediately without parking, so it is safe to call from anywhere, including off-fiber. If a signal is already pending the callback runs before wait returns. See "Callbacks run where you send from".

Only a real coderef is treated as a callback. undef means the no-argument form, and any other value - including 0 - is registered as if it were a fiber id; see "wait takes one optional argument".

awaited( )

my $n = $ready->awaited;      # fibers plus callbacks

The number of waiters queued on this signal, counting both parked fibers and registered callbacks. Zero when nobody is waiting, which is the cheapest way to ask whether a send is going to be remembered or delivered.

remove_waiter( $waiter )

my $removed = $ready->remove_waiter(\&callback);

Removes a callback or a fiber id from the queue and returns how many entries went away. Consumes nothing and never calls a callback. It does not wake anything either: a fiber you remove this way is simply never signalled again.

You rarely need this. It is the hook the park installs to deregister an interrupted wait, and every Parataxis primitive that can park has one - a Acme::Parataxis::Semaphore, a Acme::Parataxis::Channel, a Acme::Parataxis::Future or any of the Acme::Parataxis::Sync classes.

Recipes

A one-shot "the work finished" event

The ordinary use: something happens once, and a fiber is waiting for it. One send releases one waiter, so this is the shape to reach for when there is a single consumer.

async {
    my $finished = Acme::Parataxis::Signal->new( count => false );

    my $consumer = fiber { $finished->wait; consume() };

    build_everything();
    $finished->send;             # the consumer was parked; this releases it

    $consumer->await;
};

One event, many waiters

A single send releases exactly one waiter, so a group that all need the same event wants broadcast. This is the shape for a fan-out, and for shutdown: everyone queued at the moment of the broadcast goes at once, rather than being drained one permit at a time.

async {
    my $abort = Acme::Parataxis::Signal->new( count => false );
    my @running = map { fiber { $abort->wait; rollback() } } @transactions;

    # ... something has gone wrong ...

    $abort->broadcast;           # every rollback starts now
    $_->await for @running;
};

One send in place of the broadcast would release a single transaction and leave the rest parked, which is the usual way this pattern deadlocks.

Signalling before anyone is listening

Because an unclaimed send is remembered, a signal needs no coordination with its waiters. Fire it early and the fiber that arrives last still sees it.

my $shutdown = Acme::Parataxis::Signal->new( count => false );

async {
    $shutdown->send;         # nobody is waiting yet -- this is remembered
};

async {
    install_handler();        # however long this takes
    $shutdown->wait;          # returns at once: the signal was already sent
    shut_down();
};

Not spawning a fiber just to be woken

A callback is a waiter that runs at send time rather than when the scheduler next runs a fiber. That is cheaper than a fiber, and it is the right shape for bookkeeping.

async {
    my $drained = Acme::Parataxis::Signal->new( count => false );

    $drained->wait( sub { report_drained() } );    # returns immediately
    if ( !@queue ) {
        $drained->send;                            # the callback runs right here
    }
};

Note the ordering: the callback is registered before the queue is measured, so a queue that empties in between is not missed.

Asking whether anyone is listening

awaited turns send from "remember or deliver" into something you can reason about, and it is the cleanest way to spot a leak of parked fibers.

async {
    my $s = Acme::Parataxis::Signal->new( count => false );
    fiber { $s->wait } for 1 .. 3;
    await_sleep(50);

    is $s->awaited, 3, 'three fibers are parked';
    $s->broadcast;
    is $s->awaited, 0, 'and the broadcast drained them';
};

Behavior Notes

A new signal starts signalled

The count parameter defaults to true, so a signal built with a plain new is already fired. Its first wait returns immediately, and only the waiters after that one block.

async {
    my $done = Acme::Parataxis::Signal->new;      # already signalled!
    my $fired = 0;
    fiber { $done->wait; $fired++ };
    await_sleep(30);

    is $fired, 1, 'the fiber never parked';
};

That default is right when the signal means "this already happened" - the first wait is the answer, and there is nothing to wait for. It is wrong for anything else, and passing count => false costs one word.

One send releases one waiter

A send hands the signal to exactly one waiter, the longest-waiting one, and does nothing about the rest. If several fibers are waiting on the same event, one send is not enough to release them, and the leftovers stay parked.

async {
    my $s = Acme::Parataxis::Signal->new( count => false );
    my @done;
    my @w = map { my $i = $_; fiber { $s->wait; push @done, $i } } 1 .. 3;
    await_sleep(30);

    $s->send;                      # releases one
    await_sleep(30);
    is scalar @done, 1, 'one of the three';

    $s->broadcast;                 # releases the rest
    $_->await for @w;
    is scalar @done, 3, 'now all three';
};

Reach for broadcast whenever the event is for a group, and for a Acme::Parataxis::Semaphore or Acme::Parataxis::Channel when you need a count of events rather than a flag.

broadcast cancels a pending signal

broadcast clears the flag whether or not it had anything to wake, so a signal that was remembered is discarded rather than delivered. The next wait parks as though nothing had happened.

async {
    my $s = Acme::Parataxis::Signal->new( count => false );
    $s->send;                      # remembered: nobody is waiting
    $s->broadcast;                 # ... and thrown away
    ok !$s->count, 'no signal left';

    my $fired = 0;
    fiber { $s->wait; $fired++ };
    await_sleep(30);
    is $fired, 0, 'so the next wait parks';
    $s->send;
};

If you want to clear the queue without losing a signal that is already banked, drain it with send calls instead - one per waiter - because send leaves the flag alone when it finds somebody to release.

The flag is a flag, not a counter

Several sends with nobody waiting bank exactly one wake-up. The signal remembers that it fired; it does not remember how many times.

async {
    my $s = Acme::Parataxis::Signal->new( count => false );
    $s->send for 1 .. 5;

    my @out;
    fiber { $s->wait; push @out, 'first' };
    fiber { $s->wait; push @out, 'second' };
    await_sleep(40);

    is_deeply \@out, ['first'], 'five sends bought one wake-up';
    $s->send;
};

Use a Acme::Parataxis::Semaphore or a Acme::Parataxis::Channel when the number of events matters.

Callbacks run where you send from

A callback registered with wait(\&cb) is invoked inside send or broadcast, in the sending fiber, at the moment of the send. It is not deferred to the next scheduler pass. That is what makes it cheap, and it is also what makes it dangerous: it runs code you did not schedule, in a context you did not choose.

The queue is strictly FIFO, but a callback and a fiber in the same queue look different from the outside, because a fiber only resumes when the scheduler gets to it.

async {
    my $s = Acme::Parataxis::Signal->new( count => false );
    my @order;
    $s->wait( sub { push @order, 'callback' } );
    fiber { $s->wait; push @order, 'fiber' };

    $s->broadcast;
    is_deeply \@order, ['callback'], 'the callback ran during the send';
    await_sleep(30);
    is_deeply \@order, [ 'callback', 'fiber' ], 'the fiber ran afterwards';
};

A callback that throws takes the rest of the broadcast with it. The slot is consumed before the callback runs, so the signal is not left for the next waiter, and the exception propagates out of the send or broadcast that triggered it, stranding everyone still queued. Wrap a callback if it can fail.

wait returns undef

Every form of wait returns undef, both when it parks and when it returns immediately. It is a statement, not a predicate. To ask whether anyone is waiting, use awaited; to ask whether a signal is pending, read count.

my $already = Acme::Parataxis::Signal->new;          # signalled
is $already->wait, undef, 'a signalled signal returns at once, with no value';

async {
    my $s = Acme::Parataxis::Signal->new( count => false );
    my $parked = 'unset';
    fiber { $parked = $s->wait };
    await_sleep(30);
    is $parked, 'unset', 'it parked, so it has not returned at all';
    $s->send;
    await_sleep(30);
    is $parked, undef, 'and a released wait returns undef too';
};

wait takes one optional argument

wait dispatches on whether its argument is defined, so a non-coderef argument is quietly treated as a fiber id.

my $s = Acme::Parataxis::Signal->new( count => false );
$s->wait(undef);      # the no-argument form: parks, or croaks off-fiber
$s->wait(0);          # registers fiber id 0 as a waiter, and does nothing else
$s->wait(sub { 1 });  # what you meant

wait(0) is accepted and increments awaited, so a stray value can leave a queue that can never drain correctly. Pass a coderef or nothing.

SEE ALSO

Acme::Parataxis, Acme::Parataxis::Semaphore, Acme::Parataxis::Future

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.