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

NAME

Acme::Parataxis::Semaphore - Counting semaphore for fiber synchronization

SYNOPSIS

use Acme::Parataxis;
use Acme::Parataxis::Semaphore;

my $sem = Acme::Parataxis::Semaphore->new( count => 3 );

async {
    fiber { $sem->down; ... };    # waits for a permit
    $sem->up;                     # and hands one back
};

DESCRIPTION

A counting semaphore: an integer that hands out permits, and parks any fiber that asks for one when there are none left. There is no owner, so one fiber can down while another ups, up may be called before any down, and the two halves need not live in the same place.

Waiters park rather than spin, and are resumed in the order they parked. A parked fiber is an ordinary Acme::Parataxis wait, so with_timeout and cancellation tokens interrupt it like anything else, and an interrupted waiter removes itself from the queue rather than being woken later by a permit it no longer wants.

This is the primitive the rest of the library is built from. A Acme::Parataxis::Channel keeps one behind each of its ends, Acme::Parataxis::RateLimiter keeps one as its token bucket, and Acme::Parataxis::Sync::Mutex is a semaphore with an owner. Reach for it directly when you want to cap concurrency or make one fiber wait for another's permission.

METHODS

new( [ count => $n ] )

my $sem = Acme::Parataxis::Semaphore->new;             # one permit
my $sem = Acme::Parataxis::Semaphore->new( count => 3 );

Creates a semaphore holding count permits, defaulting to 1. Keyword-only, and count is not validated: see "Nothing checks the count".

down( )

$sem->down;        # take a permit, waiting if there is none

Takes a permit and decrements the count, parking the calling fiber first if the count is zero. Returns true.

Waiters are released in the order they parked, one per up. A fiber that is woken but finds the count at zero again re-parks rather than stealing the permit from someone else.

Must run inside a scheduled fiber only if it would actually have to wait. See "Usable as a plain counter off-fiber".

up( )

$sem->up;          # hand back a permit, releasing one waiter

Returns a permit and wakes exactly one waiter, if any are queued. Returns true.

try( )

if ( $sem->try ) {
    ...            # took a permit
}

Takes a permit and returns true if one was available, or returns false immediately if not. Never parks, so it is safe off-fiber and safe to call when you would rather do something else than wait.

adjust( $diff )

$sem->adjust(5);       # add five permits at once
$sem->adjust(-2);      # take two away

Adds $diff to the count, which may be negative, and then releases up to the new count worth of waiters. Returns true.

This is the bulk-release you want at shutdown, or when a producer knows exactly how much it is about to hand over. Note that each waiter it releases goes on to take a permit of its own, so adjust(5) on an empty semaphore with four waiters leaves the count at 1, not 5. A negative $diff releases nobody; see "What adjust really releases".

wait( )

$sem->wait;        # block until a permit exists -- does not take one

Blocks until the count is greater than zero and returns true without decrementing. It is a wait for the semaphore to become non-empty, not an acquire, and it is the one method here whose name will mislead you; see "wait does not take a permit".

guard( )

{
    my $permit = $sem->guard;
    my $held   = $sem->count;    # the permit is held for exactly this scope
}
# released here, including if the scope is left by an exception

Takes a permit the way down does, blocking if it has to, and returns an Acme::Parataxis::Semaphore::Guard object that hands the permit back when it is destroyed. Use it when a permit should be held for the length of a scope, so an early return or a die cannot leak it.

The guard is a plain destructor-based object with no other methods, and it is not idempotent: calling ->DESTROY on it yourself and then letting it go out of scope releases the permit twice.

count( )

$sem->count;       # permits currently available

waiters( )

my $n   = $sem->waiters;       # how many fibers are parked
my @ids = $sem->waiters;       # their fiber ids

In scalar context, the number of fibers currently parked. In list context, their fiber ids, in the order they will be released. Those are the scheduler's own ids, which get recycled once a fiber is reaped, so treat them as a debugging aid rather than as handles to keep.

remove_waiter( $fid )

my $removed = $sem->remove_waiter($fid);

Drops a fiber id from the wait queue and returns how many entries went away, so 0 or 1. A removed waiter is never woken by a later up or adjust.

You rarely call this yourself. It is the deregistration hook that the park installs on the fiber's behalf, and every Parataxis primitive that can park implements it - a Acme::Parataxis::Channel sums it across both of its internal semaphores, and Acme::Parataxis::Future, Acme::Parataxis::Signal and the Acme::Parataxis::Sync classes each have their own.

Note that it edits the queue and nothing else. Removing a waiter does not wake that fiber; it only guarantees that this semaphore never will. You still have to release the fiber some other way, or it stays parked.

Recipes

Capping concurrency

The ordinary use: a pool of count permits, and every unit of work takes one before it starts. Nothing here counts anything, because the semaphore does.

async {
    my $pool = Acme::Parataxis::Semaphore->new( count => 4 );   # four at a time

    my @work = map {
        my $item = $_;
        fiber {
            $pool->down;                    # park here while all four are busy
            my $result = process($item);
            $pool->up;                      # ... and hand the permit straight back
            $result;
        };
    } @items;

    my @results = map { $_->await } @work;
};

The permits are returned explicitly rather than through a guard, because the fiber spans more than one lexical scope and the up sits next to the work it releases.

A gate everyone waits on

A semaphore with no permits is a one-way door: every fiber that arrives parks, and each up admits exactly one. Set it to 0 and up it N times to let an N-fiber group through together.

async {
    my $gate = Acme::Parataxis::Semaphore->new( count => 0 );
    my @ready = map { fiber { $gate->down; go() } } 1 .. 3;
    await_sleep(50);

    is scalar $gate->waiters, 3, 'all three are parked';
    $gate->adjust(3);                        # open it for all three at once

    $_->await for @ready;
};

Doing the work if there is room, and skipping it if not

try never parks, so it turns a semaphore into a "is there capacity right now" test. That is what you want for optional work that is not worth queueing for.

async {
    my $pool = Acme::Parataxis::Semaphore->new( count => 2 );

    for my $task (@nice_to_have) {
        next unless $pool->try;      # skip it if both permits are taken
        fiber {
            my $r = process($task);
            $pool->up;
            $r;
        };
    }
};

Letting everyone go at shutdown

adjust releases a whole group in one call. Release exactly as many as are waiting, so you neither strand a fiber nor hand out permits nobody will take.

async {
    my $gate = Acme::Parataxis::Semaphore->new( count => 0 );
    my @waiters = map { my $id = $_; fiber { $gate->down; $id } } @ids;

    # ... shutdown is signalled ...

    $gate->adjust( scalar $gate->waiters );   # let them all through
    my @results = map { $_->await } @waiters;
};

Holding a permit for a scope

When the permit belongs to a block rather than to one operation, guard releases it on every way out of that block, including an exception.

my $exclusive = Acme::Parataxis::Semaphore->new( count => 1 );

sub with_exclusive {
    my ( $sem, $body ) = @_;
    my $permit = $sem->guard;           # blocks until we have it
    return $body->();
}                                       # released here, die or not

with_exclusive( $exclusive, sub { do_the_exclusive_thing() } );

Behavior Notes

wait does not take a permit

wait is not an acquire. It blocks until the count is above zero and then returns having left the count alone, so two fibers can both wait on a one-permit semaphore and both return.

async {
    my $s = Acme::Parataxis::Semaphore->new( count => 1 );
    $s->wait;
    $s->wait;
    is $s->count, 1, 'still one: wait never consumed anything';
};

The trap is a loop. Because wait does not decrement, while ( $s->wait ) { ... } against a semaphore that already has a permit never parks and never blocks; it is a busy loop that the scheduler will happily run forever. If you want a permit, you want down.

Nothing in the library calls wait itself. It is there for "tell me when this becomes non-empty", and even that is racy on its own, since nothing is claimed on your behalf.

Usable as a plain counter off-fiber

down only insists on being in a fiber when it would actually have to wait, so a semaphore with permits to spare works fine from the mainline, and up, try and adjust never need a fiber at all.

my $s = Acme::Parataxis::Semaphore->new( count => 1 );
$s->down;                              # fine: a permit was available
$s->down;                              # Semaphore waits must occur inside a scheduled fiber

That makes a semaphore usable as a shared counter between a fiber and the outside world, but it also means the croak is conditional: the same call site works until the moment it would block, and then starts dying. Do not rely on down being safe off-fiber.

An interrupted waiter deregisters

A down cut short by with_timeout or a cancellation token takes itself out of the queue, so a later up does not wake a fiber that has already moved on. The permit that up adds is not lost, but nothing takes it either.

async {
    my $s = Acme::Parataxis::Semaphore->new( count => 0 );
    fiber { eval { with_timeout( 60, sub { $s->down } ); 1 } };
    await_sleep(120);                   # past the deadline

    is scalar $s->waiters, 0, 'the cancelled waiter removed itself';
    $s->up;
    is $s->count, 1, 'so the permit just sits there';
};

What adjust really releases

adjust releases up to the new count, not up to $diff, and a negative $diff releases nobody at all.

async {
    my $s = Acme::Parataxis::Semaphore->new( count => 0 );
    my @w = map { fiber { $s->down; 1 } } 1 .. 4;
    await_sleep(30);

    $s->adjust(2);                       # releases exactly 2
    await_sleep(20);
    is scalar $s->waiters, 2, 'two still waiting';
    is $s->count, 0, 'and the count is back to zero';

    $s->adjust(-1);                      # releases nobody, and the count goes negative
    await_sleep(20);
    is scalar $s->waiters, 2, 'still two waiting';
    is $s->count, -1;

    $s->adjust(9);                       # releases the rest; each takes a permit of its own
    await_sleep(20);
    is scalar $s->waiters, 0, 'and the last two are through';
    is $s->count, 6, 'which is -1 + 9 - 2';
};

Nothing checks the count

There is no validation on the count argument. Negative and fractional values are both accepted, and count => undef quietly becomes 1.

my $neg  = Acme::Parataxis::Semaphore->new( count => -5 );
my $frac = Acme::Parataxis::Semaphore->new( count => 2.5 );

A negative count is not an error, it is a semaphore that can never be taken from without an adjust or up first, so every down parks forever. A fractional one works, since down only cares whether the count is above zero, but there is no reason to want it.

SEE ALSO

Acme::Parataxis, Acme::Parataxis::Channel, Acme::Parataxis::Sync::Mutex, Acme::Parataxis::RateLimiter

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.