NAME

Acme::Parataxis::Channel - A Simple Message Queue for Inter-Fiber Communication

SYNOPSIS

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

my $log = Acme::Parataxis::Channel->new( capacity => 1024 );

async {
    fiber { $log->put($_) for 1 .. 8 };    # a producer: parks while the channel is full

    say $log->get for 1 .. 8;              # a consumer: parks while it is empty
};

DESCRIPTION

A channel is a buffer with two ends and a capacity. put appends to the tail, get takes from the head, and both ends are open to as many fibers as you like. The capacity is the point: when the buffer is full a producer parks, and when it is empty a consumer parks, so a slow reader pushes back on a fast writer instead of letting a queue grow without bound. capacity => 1 is a rendezvous - a put only completes when a get is waiting for it - and every extra slot is one buffered value.

Every park here is an ordinary fiber wait, so with_timeout and Acme::Parataxis::CancellationToken bound a get or a put like any other wait, and the interruption leaves the channel as it found it: a value that was on its way to a get that got cut short still reaches the next reader, and a put that got cut short never happened at all.

shutdown is the other half of the story. It opens the read side for good, so the values already buffered still come out and get then returns undef - EOF - instead of parking. That is how a producer tells its consumers that no more work is coming, and how a Stream chain unwinds one stage at a time.

A channel is also the substrate under much of the rest of the distribution: an Actor mailbox is one, a Ticker publishes its ticks through a capacity-1 channel, a Supervisor collects child deaths through one, and every Stream stage is a bounded channel plus a fiber.

METHODS

`new( [ ... ] )`

my $q = Acme::Parataxis::Channel->new;                           # effectively unbounded
my $q = Acme::Parataxis::Channel->new( capacity => 1024 );       # then put parks
my $q = Acme::Parataxis::Channel->new( capacity => 8, timeout => 200 );

Named arguments only. capacity is how many values the buffer holds before put parks, and must be at least 1; it defaults to 2_000_000_000, which in practice means unbounded, so ask for a capacity when you want backpressure rather than assuming you got one. timeout is a default wait bound in milliseconds for a get or put that would park: that wait gives up with Acme::Parataxis::Error::Timeout instead, and 0 (the default) or undef means no bound. Dies on a capacity below 1 and on a negative timeout. Both read back with "capacity( )" and "timeout( )".

`get( )`

my $line = $q->get;

Removes and returns the oldest value, parking the calling fiber while the channel is empty. Returns undef forever once a shut-down channel has been drained (see "shutdown( )"). Honours the channel's timeout if it has one.

`put( $value )`

$q->put($msg);

Appends $value to the tail, parking the calling fiber while the channel is full, and honouring the channel's timeout. A put that a deadline or a token cuts short does not happen: the value is not queued and no room is consumed.

`try_get( )`

my ( $ok, $value ) = $q->try_get;

Takes a value if one is there and returns (1, $value), or returns (0, undef) if not. Never parks, so it never waits on the channel's timeout either - this is the polling form of get.

`try_put( $value )`

$q->try_put($msg) or next;    # true when it fitted

Appends $value and returns true if there was room, or returns false without parking if the channel was full. The non-blocking put.

`put_priority( $value )`

$q->put_priority($msg);    # returns immediately, full channel or not

Enqueues $value without ever parking, which is what makes it usable from a destructor or anywhere else that may not yield - an Actor's stop marker is queued this way. It ignores the channel's timeout and does not wait for room: on a full channel it pushes past capacity, and the channel stays over capacity from then on (adjust is the way back). For "put if there is room", use try_put.

`size( )`

my $n = $q->size;

How many values are buffered right now.

`shutdown( )`

$q->shutdown;    # returns true

Opens the read side for good: the values already buffered still come out of get, and once they are gone get returns undef instead of parking, and keeps doing so. Idempotent, and the only way to end a consumer loop. It does not stop producers: a put is still accepted, and on a full channel it can still park.

`adjust( $diff )`

$q->adjust( 1_024 );    # more room
$q->adjust( -512 );       # less

Moves the room available to put by $diff and returns true. Shrinking below what is already buffered is allowed - it just means put waits for the buffer to drain first - so capacity is a budget you can move at runtime rather than a promise.

`capacity( )`

my $slots = $q->capacity;

The capacity new was given, read back unchanged - adjust moves the room without touching it, so the room a put actually sees is this minus size plus everything you have adjusted.

`timeout( )`

my $ms = $q->timeout;    # 0 when there is no bound

The channel's default wait bound in milliseconds, 0 when it has none. new is the only thing that sets it, and select reads it to pick its own deadline.

`select_waiters( )`

my $n = $q->select_waiters;

How many fibers are parked in select on this channel. Introspection; mostly for tests.

`remove_waiter( $fid )`

$q->remove_waiter($fid);

Unregisters a fiber id from this channel's get, put and select queues, returning how many entries that removed (0, 1 or 2). A removed waiter is never woken by a later put, get or shutdown, so the caller owns what happens to that fiber next. The same contract as remove_waiter on a Semaphore or a Signal, and it is what an interrupted park uses to clean up after itself.

`select( %args )`

my ( $ch, $value ) = Acme::Parataxis::Channel->select(
    [ $inbox,  'get'      ],
    [ $outbox, 'put', $msg ],
    timeout    => 1_000,
    default    => sub { undef }
);

A class method that waits until one of several channels can move, and then performs that move. Each case is an array reference: [ $channel, 'get' ] to take a value out, or [ $channel, 'put', $value ] to put one in, where the value is required. Give it at least one case and call it from inside a scheduled fiber.

It returns ( $channel, $value ) for whichever case committed: the channel that fired, and the value read from it or written to it. Cases are probed in random order, so a case that is always ready cannot starve the others, and a ready case commits without yielding at all - a select on ready channels is a non-blocking poll.

timeout => $ms

Bounds the wait. If it fires before any case is ready, select returns ( undef, undef ) rather than throwing. An explicit timeout always wins over the channels' own bounds; without one, the smallest positive bound among the cases applies, and with no bound anywhere select parks until a case is ready (or until the run's deadlock detector has something to say about it).
default => sub { ... }

Run only when no case is ready, and it never parks, so its return value is what select hands back as ( undef, $value ). That is the "nothing was waiting, so ..." branch.

The park is an ordinary fiber wait, so an enclosing with_timeout or cancellation token still throws straight through it - only select's own deadline gives you ( undef, undef ). Only the channels in the cases are registered, so a put or get anywhere else is unaffected by a fiber waiting here.

Croaks on a case that is not a channel, an op that is not 'get' or 'put', a 'put' case with no value, a negative timeout, a default that is not a code ref, no cases at all, and a call outside a scheduled fiber.

Recipes

These are very brief demos that should inspire your work.

Draining a channel to the end

my $q = Acme::Parataxis::Channel->new( capacity => 64 );

fiber {                       # the consumer
    while ( defined( my $line = $q->get ) ) {
        handle($line);       # parks on an empty channel, stops at EOF
    }
    say 'no more work';
};

produce_into($q);            # the producer: puts everything it has
$q->shutdown;                # ... then says so

The loop tests defined because that is how EOF arrives, which means a channel that legitimately carries undef as a value ends the loop early - send such a message as an array ref or an exception object instead.

A request and its one reply

my ( $jobs, $results ) = (
    Acme::Parataxis::Channel->new( capacity => 1 ),
    Acme::Parataxis::Channel->new( capacity => 1 )
);

fiber {                            # the worker
    my $job = $jobs->get;
    $results->put( do_work($job) );
    $results->shutdown;            # one reply, then EOF
};

$jobs->put($work);
my ( $ch, $result ) = Acme::Parataxis::Channel->select( [ $results => 'get' ] );

With no timeout, that select parks until the reply lands - and a cancellation token or an enclosing with_timeout still interrupts it.

Two queues, whichever moves first

my ( $ch, $what ) = Acme::Parataxis::Channel->select(
    [ $jobs,  'get' ],    # work arriving
    [ $ticks, 'get' ],    # ...or a clock tick
    timeout => 2_000
);

return 'nothing happened' if !$ch;

That is select's reason to exist: one wait, several channels, one bound, and ( undef, undef ) as the answer when nothing won.

Behavior Notes

EOF is a value

A drained, shut-down channel makes get return undef forever, but try_get returns (1, undef) and a get case in select returns ( $channel, undef ). All three are the same end-of-stream, so every drain loop has to test the value it got.
Backpressure is a park

A full channel parks producers rather than growing the buffer, and that park is interruptible: a put cut short by a deadline or a token leaves the channel exactly as it was. A slow consumer therefore throttles its producers all the way upstream, which is what makes a Stream pipeline bound in memory for free.
The bound is per wait

timeout applies to a get or put that would actually park, and to select when you do not pass it a timeout of its own. A get that finds a value waiting never arms a timer, and neither try_get nor try_put ever waits on the bound.
No introspection beyond the buffer

capacity and timeout read back what you asked for, size is the buffered count, and select_waiters counts selectors only - there is no waiters to ask how many fibers are parked in a get or a put.

AUTHOR

Sanko Robinson https://github.com/sanko

LICENSE

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

To install Acme::Parataxis, copy and paste the appropriate command in to your terminal.

cpanm

cpanm Acme::Parataxis

CPAN shell

perl -MCPAN -e shell
install Acme::Parataxis

For more information on module installation, please visit the detailed CPAN module installation guide.

	Global
`s`	Focus search bar
`?`	Bring up this help dialog

	GitHub
`g` `p`	Go to pull requests
`g` `i`	Go to GitHub issues (only if GitHub is preferred repository)

	POD
`g` `a`	Go to author
`g` `c`	Go to changes
`g` `i`	Go to issues
`g` `d`	Go to dist
`g` `r`	Go to repository/SCM
`g` `s`	Go to source
`g` `b`	Go to file browse

Search terms
module: (e.g. module:Plugin)
distribution: (e.g. distribution:Dancer auth)
author: (e.g. author:SONGMU Redis)
version: (e.g. version:1.00)

NAME

SYNOPSIS

DESCRIPTION

METHODS

new( [ ... ] )

get( )

put( $value )

try_get( )

try_put( $value )

put_priority( $value )

size( )

shutdown( )

adjust( $diff )

capacity( )

timeout( )

select_waiters( )

remove_waiter( $fid )

select( %args )