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

NAME

Acme::Parataxis::Semaphore - Counting Semaphore for Fiber Synchronization and Resource Pools

SYNOPSIS

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

async {
    # Limit concurrency to 3 simultaneous fibers:
    my $pool = Acme::Parataxis::Semaphore->new( count => 3 );

    my @workers = map {
        my $id = $_;
        fiber {
            $pool->down;    # acquire permit (parks if count == 0)
            say "Worker $id acquired permit";
            await_sleep(50);
            $pool->up;      # release permit back to pool
        };
    } 1 .. 6;

    $_->await for @workers;
};

DESCRIPTION

A counting semaphore manages a pool of abstract permits. Fibers request permits via down (parking if none are available) and return permits via up.

Semaphores have no concept of thread or fiber ownership: one fiber can call down while an entirely different fiber calls up, and permits can be added before any fiber requests them.

Waiters park cooperatively in FIFO order without polling. Parked waits integrate with Parataxis structured concurrency: they can be interrupted cleanly via with_timeout or cancellation tokens, and interrupted fibers automatically unregister from the wait queue without consuming leaked permits.

Acme::Parataxis::Semaphore is a foundational building block for the rest of the distribution: channels use two internal semaphores for read and write queues, rate limiters use one as a token bucket, and mutexes wrap a semaphore with strict owner tracking.

METHODS

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

my $sem = Acme::Parataxis::Semaphore->new;             # default 1 permit
my $sem = Acme::Parataxis::Semaphore->new( count => 5 );

Constructs a new semaphore with $n initial permits. Keyword-only.

down( [$reason = 'Semaphore down'] )

$sem->down;

Decrements the permit count by 1. If the permit count is zero or negative, the calling fiber parks until a permit is returned via up() or adjust(). Returns true.

Must be called from within a scheduled fiber if the call needs to park.

up( )

$sem->up;

Increments the permit count by 1 and wakes the longest-waiting parked fiber, if any. Returns true.

try( )

if ( $sem->try ) {
    # permit acquired without parking
} else {
    # no permits available
}

Attempts to acquire a permit without parking. If count is greater than zero, decrements the count and returns true. Otherwise, returns false immediately. Safe to call outside of a scheduled fiber.

adjust( $diff )

$sem->adjust(5);     # add 5 permits and wake up to 5 waiters
$sem->adjust(-2);    # subtract 2 permits (does not wake waiters)

Modifies the permit count by $diff (which may be positive or negative). If the count becomes positive, wakes up to min(count, waiters) parked fibers. Used during bulk releases, capacity resizes, or shutdown teardowns.

guard( )

{
    my $guard = $sem->guard;    # acquires permit via down()
    do_protected_work();
}
# permit returned automatically via up() when $guard goes out of scope

Acquires a permit via down() and returns an Acme::Parataxis::Semaphore::Guard object. When the guard is destroyed (via lexical scope exit or an unhandled exception), it calls up() automatically.

wait( )

$sem->wait;

Blocks until count is greater than zero without decrementing the permit count. This is a non-consuming readiness check, not an acquisition.

count( )

Returns the number of permits currently available.

waiters( )

my $count = $sem->waiters;
my @fids  = $sem->waiters;

In scalar context, returns the number of fibers currently parked on this semaphore. In list context, returns their numeric fiber IDs in FIFO queue order.

remove_waiter( $fid )

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

Removes a specific fiber ID from the wait queue without waking it. Returns the number of entries removed (0 or 1). Used internally for cancellation cleanup.

Recipes

One-way gating (coordinating simultaneous start)

Initialize the semaphore to 0 to block all incoming fibers until a trigger opens the gate:

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

    for my $i ( 1 .. 4 ) {
        fiber {
            $gate->down;
            say "Runner $i started!";
        };
    }

    await_sleep(50);
    say "Releasing all runners at once:";
    $gate->adjust(4);    # wakes all 4 waiting fibers simultaneously
};

Scoped permit management

sub process_transaction ($sem, $data) {
    my $guard = $sem->guard;    # acquires permit
    execute_remote_call($data); # if this dies, $guard->DESTROY safely calls ->up()
}

Behavior Notes

  • wait() does not acquire

    wait() blocks until a permit exists, but does not decrement the counter. Calling while ($sem->wait) { ... } against a non-empty semaphore produces an infinite, non-yielding busy loop. Use down() to consume permits.

  • Automatic deregistration on cancellation

    If a fiber parked in down() is interrupted by a timeout or cancellation token, it unregisters from the semaphore's wait list. When a permit is subsequently returned via up(), it bypasses the cancelled fiber and goes to the next active waiter.

  • Negative counts

    Calling adjust() with a negative number can cause count to drop below zero. A negative count requires an equivalent number of up() calls before any future down() operations will be allowed through.

SEE ALSO

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

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.