NAME

Acme::Parataxis::Channel - Bounded Message Queues and CSP Multiplexing for Fibers

SYNOPSIS

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

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

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

    # Consumer: parks while the channel is empty; stops on EOF (undef)
    while ( defined( my $msg = $log->get ) ) {
        say "Received: $msg";
    }
};

DESCRIPTION

A channel is a buffer with two ends and a capacity. put appends to the tail, get removes from the head, and both ends can be shared across any number of concurrent fibers.

Capacity drives backpressure: when the buffer is full, producers park; when it is empty, consumers park. A slow reader naturally pushes back on a fast writer, preventing unbounded queue growth in memory. Setting capacity => 1 creates a rendezvous channel: a put will only complete once a get is actively waiting to take the value.

Every park here is a standard fiber wait. with_timeout and Acme::Parataxis::CancellationToken interrupt a get or put cleanly: an interrupted put leaves nothing in the channel, and an interrupted get leaves the next queued item for subsequent readers.

Calling shutdown closes the write side. Any items already buffered remain available to get. Once drained, subsequent get calls return undef immediately without parking. This is the canonical way to signal EOF down a pipeline and allows Acme::Parataxis::Stream stages to unwind cleanly.

Channels serve as the internal substrate across Parataxis: an Actor mailbox is a channel, a Ticker publishes ticks through a capacity-1 channel, and a Supervisor aggregates worker crashes through one.

METHODS

`new( [%opts] )`

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

Named arguments only:

capacity => $int

Buffer slot limit before put parks (default: 2_000_000_000, effectively unbounded). Must be at least 1. If you want backpressure, always set this explicitly.
timeout => $ms

Default wait bound in milliseconds for any get or put that parks. If the wait exceeds this duration, it throws Acme::Parataxis::Error::Timeout. Default 0 (no timeout).

`get( )`

my $val = $q->get;

Removes and returns the oldest value in the channel, parking the calling fiber if the channel is empty. Once a shut-down channel has been fully drained, get returns undef forever. Honors the channel's configured timeout.

`put( $value )`

$q->put($msg);

Appends $value to the tail, parking the calling fiber if the channel is full. If a deadline or token cuts the wait short, the item is not queued. Honors the channel's configured timeout.

`try_get( )`

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

Non-blocking read. If an item is present, removes it and returns (1, $val). If empty, returns (0, undef) immediately without parking or arming timers.

`try_put( $value )`

$q->try_put($msg) or warn "Buffer full, dropped";

Non-blocking write. If room is available, enqueues $value and returns true. If full, returns false immediately without parking.

`put_priority( $value )`

$q->put_priority($urgent_msg);

Enqueues $value immediately without ever parking, even if the channel is already at or above capacity. Used for out-of-band control signals (such as an Actor's shutdown sentinel) and safe to call from destructors or contexts where yielding is prohibited.

`size( )`

Returns the number of unread items currently buffered.

`capacity( )`

Returns the initial capacity configured during new.

`timeout( )`

Returns the configured default timeout in milliseconds (0 if none).

`shutdown( )`

$q->shutdown;

Closes the channel to future writes. Wakes all fibers currently waiting in get() or in a select read case. Buffered items remain accessible; once empty, all subsequent reads yield undef. Idempotent.

`adjust( $diff )`

$q->adjust(100);    # expand capacity by 100
$q->adjust(-50);    # contract capacity by 50

Adjusts available write capacity by $diff and returns true. If shrinking causes capacity to fall below the current size(), existing buffered items remain intact, but put will park until the queue drains below the new threshold.

`select_waiters( )`

Returns the number of fibers currently parked in a select call involving this channel.

`remove_waiter( $fid )`

$q->remove_waiter($fid);

Unregisters a fiber ID from the channel's internal get, put, and select wait queues, returning the count of entries removed. Used during park interruption cleanup.

CSP MULTIPLEXING: `select`

`Acme::Parataxis::Channel->select( @cases, [%opts] )`

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

A class method that blocks until one of several channel operations can proceed. Probes cases in randomized order to prevent starvation. If an operation is ready immediately, it commits without yielding.

Cases

Each case is an arrayref:

[ $channel, 'get' ] - ready when $channel has a readable value.
[ $channel, 'put', $value ] - ready when $channel has capacity to accept $value.

Options

timeout => $ms

Bounds the wait. If the timeout expires before any case is ready, select returns ( undef, undef ) rather than throwing. If omitted, the shortest positive timeout among the involved channels is used.
default => sub { ... }

If supplied, select will never park. If no channel is immediately ready, it runs the callback and returns ( undef, $default_return ).

Recipes

Draining a queue to EOF

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

fiber {
    while ( defined( my $item = $queue->get ) ) {
        process_item($item);
    }
    say 'All items drained';
};

produce_into($queue);
$queue->shutdown;

Because EOF is signaled as undef, avoid sending raw undef values through the channel; wrap them in arrayrefs or objects if undefined payloads are legitimate data.

Capacity-1 rendezvous

my $req_ch  = Acme::Parataxis::Channel->new( capacity => 1 );
my $resp_ch = Acme::Parataxis::Channel->new( capacity => 1 );

fiber {
    my $req = $req_ch->get;
    $resp_ch->put( handle_request($req) );
};

$req_ch->put($payload);
my $reply = $resp_ch->get;

Multiplexing work and heartbeats

while (1) {
    my ( $ch, $val ) = Acme::Parataxis::Channel->select(
        [ $jobs,   'get' ],
        [ $ticker, 'get' ],
        timeout => 5000,
    );

    if ( !defined $ch ) {
        warn "No activity for 5 seconds";
    }
    elsif ( $ch == $jobs ) {
        last if !defined $val;    # jobs channel was shut down
        handle_job($val);
    }
    elsif ( $ch == $ticker ) {
        send_heartbeat();
    }
}

Behavior Notes

EOF semantics across methods

When a channel is shut down and drained, get returns undef, try_get returns (1, undef), and a select get case returns ($channel, undef). Check for defined values when reading from channels that may close.
Backpressure is a park

When a bounded channel fills, producers park inside the scheduler rather than consuming system memory. This makes cancellation safe: an interrupted put simply aborts its registration, leaving the buffer untouched.
Timeouts apply only to parking operations

A get or put that succeeds immediately never arms a timer. Neither try_get nor try_put ever consults the configured timeout.

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( [%opts] )

get( )

put( $value )

try_get( )

try_put( $value )

put_priority( $value )

size( )

capacity( )

timeout( )

shutdown( )

adjust( $diff )

select_waiters( )

remove_waiter( $fid )

CSP MULTIPLEXING: select

Acme::Parataxis::Channel->select( @cases, [%opts] )