Acme-Parataxis
view release on metacpan or search on metacpan
CONTRIBUTING.md view on Meta::CPAN
./Build
This builds `Acme::Parataxis` and installs a local xmake into `blib/`.
### Running the tests
./Build test
## Development workflow
This project uses **native Perl `class` syntax** (`use v5.40`). Please keep that style and avoid adding comments unless they clarify non-obvious behavior.
We use [Test2::V0](https://metacpan.org/pod/Test2::V0) for tests. Follow the existing test structure when adding coverage.
Before submitting, please:
1. **Run the full suite** and confirm it passes.
2. Format your code so it matches the surrounding style (the repo has a
`[Code::TidyAll](https://metacpan.org/pod/Code::TidyAll)` configuration in `.tidyallrc`).
3. Add a changelog entry under `## [Unreleased]` in `Changes.md`.
If you change the public API of a class, update the corresponding `.pod` and any examples under `eg/`.
## Submitting changes
1. Fork the repository on GitHub.
2. Create a feature branch: `git checkout -b my-change`
3. Commit your changes with a clear, concise message (see the existing `git log` for style).
4. Push your branch and open a Pull Request against `main` (previously `master`).
Please make sure your PR:
- Describes the problem being solved
- References any related issue
- Includes tests for any new behavior
- Does not push version-number bumps or changelog edits unless they are part of the change (maintainers handle release versions).
## Reporting bugs
## [v0.1.1] - 2026-10-25
Borrowing patterns from Rust, Erlang, Java, etc. to bring more familiar concurrency primitives to Parataxis. No real changes to the core this time.
### Added
- `Acme::Parataxis::Blocking`: CPU offloading onto a bounded pool of cloned interpreters via `spawn_blocking` (lazily loads threads/shared modules on first use).
- `spawn_blocking_fork`: Process-based CPU offloading for non-ithreads Perls with IPC via `Storable`, bounded worker pools, and signal-based cancellation via `kill_blocking_fork`.
- `Acme::Parataxis::Actor`: Mailbox-driven actors with bounded queues, backpressure, `ask` (returning a `Future`), `send`, and graceful teardown.
- `Acme::Parataxis::Supervisor`: OTP-style supervision trees supporting `OneForOne`, `OneForAll`, and `RestForOne` restart strategies with configurable restart budgets.
- Synchronization primitives:
- `Acme::Parataxis::Sync::Mutex`: Non-reentrant lock with strict FIFO hand-off and ownership tracking.
- `Acme::Parataxis::Sync::RwLock`: Writer-preferring, reentrant read/write lock.
- `Acme::Parataxis::Sync::WaitGroup`: Fiber job counter (`add`, `done`, `wait`).
- `Acme::Parataxis::Sync::Barrier`: Reusable phase rendezvous point.
- `Acme::Parataxis::Sync::Once`: Race-safe, single-execution initializer.
- Structured concurrency & flow control:
- `nursery( sub ($n) { ... } )`: Structured fiber scopes where child errors cancel siblings and propagate outward.
- `pmap( { concurrency => $n, }, sub ( $item ) { ... }, @items )`: Parallel list mapping over a bounded fiber pool.
- `with_timeout( $ms, [ $token, ] $code )` and `with_cancel( $code )`: Cooperative deadline and cancellation scopes.
- `Acme::Parataxis::CancellationToken`: Cooperative park-interrupt token.
- `Acme::Parataxis::Local`: Fiber-local ambient storage.
- `Acme::Parataxis::Generator`: Stackful, lazy pull iterator.
- `Acme::Parataxis::Monitor`: Non-owning fiber observer.
- Channels & Streams:
- `Acme::Parataxis::Channel::select`: CSP-style waiting across multiple channels with random case selection to prevent starvation.
- `try_get` / `try_put` non-blocking channel operations and per-channel timeout configurations.
- `Acme::Parataxis::Stream`: Backpressure-driven channel pipelines (`map`, `filter`, `batch`, `batch_time`, `throttle`).
- `Acme::Parataxis::Ticker`: Drift-free interval timer.
- `Acme::Parataxis::RateLimiter`: Token bucket limiter driven by `Ticker`.
- Software Transactional Memory (STM): `Acme::Parataxis::TVar` cells and `atomically` transactions with conflict detection and `retry()`.
- Future combinators: `wait_all` and `wait_any`.
- Event loop & OS integration:
- Event loop drivers: `Acme::Parataxis::Driver::Mojo` and `Acme::Parataxis::Driver::IOAsync`.
- Transparent unblocking (`Acme::Parataxis::Compat`): Opt-in cooperative `CORE::GLOBAL` overrides for `sleep`, `read`, and `sysread`.
- Deterministic virtual clock via `run( virtual => 1 )` and `advance( $ms )`.
The core exports the scheduler interface (`async`, `run`), fiber lifecycle tools (`fiber`, `spawn`, `yield`,
`await`), async I/O primitives (`await_sleep`, `await_read`, `await_write`), structured concurrency (`nursery`),
and cooperative cancellation (`with_timeout`, `with_cancel`).
Higher-level abstractions are separated into dedicated companion classes:
- [Acme::Parataxis::Actor](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3AActor) - Isolated state behind a private fiber with an incoming mailbox.
- [Acme::Parataxis::Channel](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3AChannel) - Bounded CSP message queues with backpressure and multiplexed `select`.
- [Acme::Parataxis::Blocking](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3ABlocking) - Offload heavy CPU work to cloned interpreters or forked processes.
- [Acme::Parataxis::Future](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3AFuture) - Single-assignment asynchronous values and completion hooks.
- [Acme::Parataxis::Stream](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3AStream) - FRP-style streaming pipelines over bounded channels.
- [Acme::Parataxis::Sync](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3ASync) - Mutexes, read/write locks, wait groups, barriers, and initializers.
- [Acme::Parataxis::TVar](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3ATVar) - Software Transactional Memory (STM) with automatic retry on conflict.
- [Acme::Parataxis::Compat](https://metacpan.org/pod/Acme%3A%3AParataxis%3A%3ACompat) - Gevent-style transparent unblocking for legacy `sleep` and `read` calls.
This project was greatly inspired by the concurrency system for the [Wren](https://wren.io/concurrency.html)
programming language and my own curiosity; I wanted to see how hard it would be to fiddle with perl's core from an FFI
and nothing requires more fiddling than picking up the full context and swapping it around at runtime.
# FUNCTIONS
Importable individually or with the `:all` tag.
## `async { ... }`
eg/sha_walk.pl view on Meta::CPAN
use File::Temp ();
use Time::HiRes ();
my $FILE_COUNT = shift // 20;
my $FILE_BYTES = shift // 262_144;
# The install that makes sleep/read/sysread cooperative. It has to sit above the code that reads files, which is
# below, and that is the whole placement rule: an override is resolved when each call site is *compiled*, so it
# covers this file's reads and nothing from a module loaded before it.
BEGIN { Acme::Parataxis->enable_transparent_unblocking() }
# CLEANUP => 0 on purpose, and this is not a style choice - it is the fix for a trap that only bites when the
# threaded branch at the end of this file actually runs.
#
# `threads->create` clones the whole perl stack, so every thread gets its *own* copy of the object graph, with its
# own refcounts. The clone's copy of $ROOT is not kept alive by this interpreter's copy of it, so when a thread
# exits, that refcount goes to zero, DESTROY runs, and File::Temp rmtree's the directory out from under the
# threads still working in it.
#
# The symptom is not an error you would guess at: the first couple of threads succeed, the directory is deleted,
# and everything after that dies with ENOENT. "Open failed: No such file or directory" on files that demonstrably
# existed a second earlier. It looks exactly like a race in the code being benchmarked, and it is not.
eg/tui_bubbletea.pl view on Meta::CPAN
affix 'kernel32', 'GetConsoleMode', [ LongLong, Pointer [Void] ], Int;
affix 'kernel32', 'SetConsoleMode', [ LongLong, ULong ], Int;
affix 'kernel32', 'GetConsoleScreenBufferInfo', [ LongLong, Pointer [Void] ], Int;
affix 'kernel32', 'GetNumberOfConsoleInputEvents', [ LongLong, Pointer [Void] ], Int;
affix 'kernel32', 'ReadConsoleInputA', [ LongLong, Pointer [Void], ULong, Pointer [Void] ], Int;
}
}
#
# A bubbletea-inspired TUI using Acme::Parataxis fibers.
#
# Architecture (Elm-like / bubbletea-style):
# - input fiber: reads keyboard -> MsgKey
# - resize fiber: $SIG{WINCH} or Win32 polling -> MsgResize
# - tick fiber: await_sleep(1000) -> MsgTick
# - main loop: Channel.get() -> Update(model, msg) -> render via View()
#
# Windows notes:
# - Raw mode, VT processing, and input are handled via kernel32 FFI
# - select() only works on sockets on Windows, so await_read/await_write
# (which use select in the thread pool) don't work on console handles.
# The input fiber uses GetNumberOfConsoleInputEvents + ReadConsoleInputA
eg/tui_bubbletea.pl view on Meta::CPAN
return ( $rows, $cols );
}
# ANSI helpers
sub hide_cursor { print "\e[?25l" }
sub show_cursor { print "\e[?25h" }
sub cls() {"\e[H\e[2J"}
sub clear_screen { print cls }
sub move_to { my ( $r, $c ) = @_; print "\e[${r};${c}H" }
# Message types (plain hashrefs bubbletea-style)
sub MsgKey { { type => 'key', key => $_[0] } }
sub MsgTick { { type => 'tick', count => $_[0] } }
sub MsgResize { { type => 'resize', width => $_[0], height => $_[1] } }
sub MsgQuit { { type => 'quit' } }
# Update / View (pure functions over model)
sub Update ( $model, $msg ) {
my %m = %$model;
if ( $msg->{type} eq 'key' ) {
my $key = $msg->{key};
eg/tui_bubbletea.pl view on Meta::CPAN
return \%m, undef;
}
sub View ($model) {
my $w = $model->{width};
my $h = $model->{height};
my $buf = '';
my $list = '';
# title bar
my $title = 'Parataxis TUI Demo (bubbletea-style)';
$buf .= "\e[1;36m$title\e[0m\r\n";
$buf .= "\e[2m${w}x${h}\e[0m\r\n";
$buf .= "\r\n";
# list items
for my $i ( 0 .. $#{ $model->{items} } ) {
my $item = $model->{items}[$i];
my $mark = $model->{toggled}{$i} ? "\e[32m[x]\e[0m" : "[ ]";
my $cursor = $i == $model->{cursor} ? "\e[1;33m>\e[0m" : ' ';
$list .= "$cursor $mark $item\r\n";
lib/Acme/Parataxis.pm view on Meta::CPAN
# blessed object). This mirrors the shift/unshift invocant dance used elsewhere, but reads $_[0] instead of
# shifting, so calling it never reifies the caller's pad @_. A reified @_ left behind in a parked frame trips
# pp_entersub's assert(!AvREAL(av))/assert(AvFILLp(av) == -1) when a different fiber later enters the same sub at
# the same depth.
sub _arg_offset {
my $self = $_[0];
return ( defined $self && ( ( ref $self || $self ) eq __PACKAGE__ || ( builtin::blessed($self) && $self->isa(__PACKAGE__) ) ) ) ? 1 : 0;
}
# Aggregate a list of futures into one that resolves when every input has: the result is an arrayref of each
# input's result, in input order, or the stored error of the first input to fail (reject-fast, Promise.all style).
# Inputs are not cancelled on a rejection; they are independent and keep running, and their later results are
# discarded. Zero futures resolve immediately with [] (matching Promise.all([])). An already-resolved input
# fires its on_ready callback inline, so a wait_all of ready inputs resolves inline (no fiber needed) and any
# mixed ready/pending input resolves the moment the last straggler lands. Inputs must be Acme::Parataxis::Future
# objects else croaks. Also callable as Acme::Parataxis->wait_all(...).
sub wait_all {
my $o = _arg_offset( $_[0] );
my @futures = @_[ $o .. $#_ ];
@_ = ();
croak 'wait_all() requires Acme::Parataxis::Future inputs'
lib/Acme/Parataxis.pod view on Meta::CPAN
=over
=item * L<Acme::Parataxis::Actor> - Isolated state behind a private fiber with an incoming mailbox.
=item * L<Acme::Parataxis::Channel> - Bounded CSP message queues with backpressure and multiplexed C<select>.
=item * L<Acme::Parataxis::Blocking> - Offload heavy CPU work to cloned interpreters or forked processes.
=item * L<Acme::Parataxis::Future> - Single-assignment asynchronous values and completion hooks.
=item * L<Acme::Parataxis::Stream> - FRP-style streaming pipelines over bounded channels.
=item * L<Acme::Parataxis::Sync> - Mutexes, read/write locks, wait groups, barriers, and initializers.
=item * L<Acme::Parataxis::TVar> - Software Transactional Memory (STM) with automatic retry on conflict.
=item * L<Acme::Parataxis::Compat> - Gevent-style transparent unblocking for legacy C<sleep> and C<read> calls.
=back
This project was greatly inspired by the concurrency system for the L<Wren|https://wren.io/concurrency.html>
programming language and my own curiosity; I wanted to see how hard it would be to fiddle with perl's core from an FFI
and nothing requires more fiddling than picking up the full context and swapping it around at runtime.
=head1 FUNCTIONS
Importable individually or with the C<:all> tag.
lib/Acme/Parataxis/Compat.pm view on Meta::CPAN
use Time::HiRes (); # loaded only: the overrides call Time::HiRes::time() fully qualified
#
# Transparent unblocking (CORE::GLOBAL overrides). This module is an opt-in
# convenience: Acme::Parataxis->enable_transparent_unblocking() installs overrides
# that make the blocking builtins `sleep`, `read` and `sysread` cooperative inside
# scheduled fibers, and delegate to the raw CORE:: builtin everywhere else (top
# level, foreign threads), so nothing outside the scheduler changes.
#
# The override subs are installed, not compiled with the module, so only code that
# is *compiled after installation* is affected - exactly the gevent-style contract.
# Every override calls the real builtin through its CORE:: name, so none of them can
# recurse into themselves.
#
my $INSTALLED = 0;
my @SAVED; # [ name, saved-glob ] per installed override, for disable()
# The readiness budget, in ms, that the read/sysread overrides give each await_read probe. A probe that comes
# back short of it means a real wait timed out, so the override keeps waiting rather than block the process.
my $PROBE_MS = 5000;
lib/Acme/Parataxis/Compat.pod view on Meta::CPAN
read( $sock, $buf, 64 ); # frames onto await_read; yields while waiting
say "f2: read ${\length($buf)} bytes";
};
$napper->await;
$reader->await;
};
=head1 DESCRIPTION
C<Acme::Parataxis::Compat> provides an opt-in, C<gevent>-style compatibility layer for legacy synchronous code. By
installing C<CORE::GLOBAL> overrides for C<sleep>, C<read>, and C<sysread>, existing modules and synchronous loops run
cooperatively inside fibers without freezing the entire process:
=over
=item * C<sleep> becomes L<await_sleep|Acme::Parataxis/await_sleep>
The calling fiber yields for the requested duration while other fibers proceed. Fractional seconds are honored with
millisecond precision (C<sleep 0.05> sleeps 50ms, whereas C<CORE::sleep> truncates this to 0). When called outside a
fiber, it falls back immediately to C<CORE::sleep>. The C<$_> default argument is supported.
lib/Acme/Parataxis/Supervisor.pm view on Meta::CPAN
use v5.40;
no warnings qw[experimental::class recursion];
use feature 'class';
#
# OTP-style supervisor trees: children are supervised actors (or nested supervisors) that are
# restarted whenever they die, according to a restart strategy, until a restart budget runs out -
# at which point the tree is torn down and run() fails with an aggregate
# Acme::Parataxis::Error::Supervisor. This is the "heal-fast" half of supervision's "supervision is
# deliberately out of scope" guardrail, and it is pure Perl on top of the existing park/wake
# machinery: one supervision fiber owns the tree and learns a child died through that child's own
# teardown (an actor's on_death hook, or a wrapper fiber around a nested supervisor's run).
class Acme::Parataxis::Supervisor v0.1.3 {
use Acme::Parataxis qw[fiber];
use Acme::Parataxis::Channel;
use Acme::Parataxis::Error;
lib/Acme/Parataxis/Sync/Mutex.pm view on Meta::CPAN
my $waiter = shift @waiters;
next unless defined Acme::Parataxis->by_id($waiter);
$next = $waiter;
last;
}
$owner = $next; # hand-off is atomic: nobody else can steal it before the waiter runs
Acme::Parataxis::_scheduler_enqueue_by_id($next) if defined $next;
return 1;
}
# Return a Scope::Guard-style object that unlocks on destruction.
method guard {
$self->lock;
Acme::Parataxis::Sync::Mutex::Guard->new( mutex => $self );
}
method waiters { scalar @waiters }
method remove_waiter ($fid) { # unregister a parked waiter (used by the lock park's interruption path)
my $before = @waiters;
@waiters = grep { $_ != $fid } @waiters;
return $before - @waiters;
t/160_supervisor.t view on Meta::CPAN
use v5.40;
use blib;
use Acme::Parataxis qw[async fiber yield];
use Acme::Parataxis::Actor;
use Acme::Parataxis::Supervisor;
use Acme::Parataxis::Error;
use Test2::V1 -ipP;
$|++;
# OTP-style supervisor trees. Children are supervised actors (or nested supervisors) that
# are restarted when they die, per restart strategy, until a restart budget runs out - at which
# point the tree stops and run() fails with an aggregate Acme::Parataxis::Error::Supervisor.
#
# Every assertion below is paired with a control: an untouched sibling, a second strategy, a budget
# nobody can exhaust, or the same crash loop under a window that does not expire.
# A factory that starts one supervised actor. The handler answers 'hi', dies on 'boom', and echoes
# anything else, so a restarted child is distinguishable from a dead one by asking it something.
sub make_child ( $name, $capacity = 4 ) {
return sub {
Acme::Parataxis::Actor->spawn(
t/201_actor_registry.t view on Meta::CPAN
async {
my $actor = Acme::Parataxis::Actor->spawn( sub ( $self, $msg ) { return 1 }, 4 );
like dies { $actor->swap(42) }, qr[CODE], 'swap() rejects a non-CODE';
$actor->stop;
yield while $actor->is_alive;
like dies {
$actor->swap( sub {1} )
}, qr[no longer running], 'swap() on a dead actor croaks';
};
};
subtest 'a supervisor-style respawn inherits the registered name' => sub {
async {
my $actor = Acme::Parataxis::Actor->spawn( sub ( $self, $msg ) { die "boom\n" }, 4, supervised => 1, name => 'svc' );
my $reply = $actor->ask('anything');
eval { $reply->await; 1 };
while ( $actor->is_alive ) {yield}
is Acme::Parataxis->actor('svc'), undef, 'the crashed actor released the name';
my $fresh = $actor->respawn;
is $fresh->name, 'svc', 'the restarted actor keeps the name';
is Acme::Parataxis->actor('svc'), $fresh, 'and answers the lookup, so the registry key survives restarts';
$fresh->stop;
t/354_transparent.t view on Meta::CPAN
my $ms = ( time - $t0 ) * 1000;
is $rc, 5, "$builtin asked for 10, got 5, and returned the short count instead of waiting";
is $buf, 'hello', "$builtin filled the caller buffer with the bytes that had arrived";
ok $ms < 2000, "the $builtin returned promptly ($ms ms) instead of parking the OS thread";
ok $ticked, "a sibling fiber ran during the $builtin, so the rest of the run kept going";
}
};
subtest 'the caller handle is handed back in the mode it arrived in' => sub {
# The override makes the handle non-blocking only for the duration of its own read, then puts the mode back.
# Anything else would break the gevent-style contract in reverse: code compiled *before* installation still gets a
# raw blocking read from that same handle, and would start failing with EAGAIN if we left it flipped.
#
# Asked through IO::Handle->blocking, which is the interface the override itself uses and the only portable one -
# Fcntl has no F_GETFL or F_SETFL on Win32 at all. It is not an interface that works *everywhere* either: Winsock's
# FIONBIO can set a socket's mode but has no way to report one, so the getter answers undef there. The capability
# is probed rather than assumed, so a platform that cannot answer is a skip and not a false pass.
my ( $a, $b ) = socket_pair();
plan skip_all => 'this platform cannot report a handle mode through IO::Handle' unless defined IO::Handle::blocking($a);
for my $want ( 1, 0 ) { # 1 = the socket default (blocking), 0 = the caller set non-blocking itself
IO::Handle::blocking( $a, $want );
( run in 2.139 seconds using v1.01-cache-2.11-cpan-b554fb43c41 )