Parataxis in the Near Future

[sanko]

I've been looking at other concurrency systems as I tinker and I am determined to bring what I've found to this project. No more introduction, let me toss my notes here.

Nursery pattern

I'm imagining a system like Python's TaskGroup or StructuredTaskScope from Java 21. Right now, fibers can be spawned detached. If one fails, throws an exception, or leaks, it's orphaned. Structured concurrency would enforce lifetimes onto fibers, binding them to the lexical block that created them. Perl's ref counter should make this an elegant solution.

Here's the API in my head:

Acme::Parataxis->nursery(sub ($n) {
    $n->spawn(sub { fetch_user_profile() });
    $n->spawn(sub { fetch_user_orders() });
    # Block suspends until both tasks finish or either fails
});

The nursery spawns child fibers and will not exit until all fibers are complete. If any child fails with an unhandled exception, all sibling fibers in this particular nursery are cancelled and the nursery throws an error.

Cooperative Cancellation and Deadlines

This is straight from tokio.¹

Right now, there's no way to interrupt a parataxis fiber that's blocked by await_sleep, await, or a ::Channel->get call. What I need looks like this:

• Cancellation tokens which we'd pass down call trees. Fibers will check $tok->is_cancelled or raise an interrupt at suspension points.
• Timeout wrappers that enforce deadlines on any yielding block:

my $val = Acme::Parataxis->with_timeout(2500, sub {
    $channel->get;
}); # Dies with 'TimeoutException' and marks the pending wait cancelled

• Cancellation points that can immediately throw a catchable exception to unwind its savestack and trigger object destructors.

CSP Channel Multiplexing

Communicating Sequential Processes (CSP) is just a fancy way to say select( ... ) over channels. I'm borrowing this directly from Rust's crossbeam_channel.

Currently, ::Channel supports get and put (which are blocking) but CSP requires non-deterministic multiplexing, the ability to wait on multiple channels at the same time without busy-waiting.

This could eventually look like this:

my $q = Acme::Parataxis::Channel->new( capacity => 1024, timeout => 500, ... );
# Wait on whichever channel is ready first
my ($event, $value) = $q->select(
    [ $ch_events => 'get' ], # rece
    [ $ch_metrics => 'put', $metric_data ], # sends data into a channel
    timeout => 500,
    default => sub { 'fallback' } # when no channels become ready in the timeout period
);

Implementing it might be tricky. A fiber will register a synthetic waiter across multiple channels and whichever channel satisfies the condition first unblocks the fiber and unregisters it from the other channels.

Hypothetical snippet once I get this working:

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

async {
    my $jobs_ch     = Acme::Parataxis::Channel->new( capacity => 10 );
    my $shutdown_ch = Acme::Parataxis::Channel->new( capacity => 1 );

    # Producer Fiber: Generates work
    fiber {
        for my $i ( 1 .. 5 ) {
            await_sleep(50);
            $jobs_ch->put("job-$i");
        }
        await_sleep(100);
        $shutdown_ch->put("STOP");
    };

    # Consumer / Worker Fiber using CSP select
    fiber {
        my $running = 1;
        while ($running) {
            my ( $chosen, $msg ) = Acme::Parataxis::Channel->select(
                [ $shutdown_ch => 'get' ],
                [ $jobs_ch     => 'get' ],
                default        => sub { 'idle' }
            );

            if ( defined $chosen && $chosen == $shutdown_ch ) {
                say "Worker received shutdown signal: $msg. Exiting loop.";
                $running = 0;
            }
            elsif ( defined $chosen && $chosen == $jobs_ch ) {
                say "Worker processed: $msg";
            }
            else {
                # Ran the 'default' branch because nothing was waiting
                say "Worker: No work ready, doing housekeeping...";
                await_sleep(20);
            }
        }
    };
};

Stackful iterators

This is common across a lot of languages... Python, Ruby, Javascript, etc...

It might look like this:

class Acme::Parataxis::Generator {
    field $fiber;
    
    method next () { ... }
}

# Example usage:
my $gen = Acme::Parataxis::Generator->new(sub ($yield) {
    for my $item (@large_tree) {
        $yield->($item); # Suspends fiber and yields value
    }
});

while (defined(my $val = $gen->next)) {
    say 'Got ' . $val;
}

This would be amazing for batched network access.

Actor pattern

I kept this away for a long time because it's a mess when trying to insert into a language that's already unfriendly to concurrency but fibers lend themselves to it.

New ask and send (or tell) methods that opens up the status so a supervisor can automatically manage children. If an actor terminates on an error, its supervisor can automatically restart it according to OTP restart strategy (one-for-one, one-for-all, etc.).

my $actor = Acme::Parataxis::Actor->spawn(sub ($self, $msg) {
    if ($msg->{cmd} eq 'ping') {
        return 'pong';
    }
});
my $reply = $actor->ask({ cmd => 'ping' })->await;

More Primitives

Mutextes, WaitGroups (like Go's sync.WaitGroup), RwLocks, etc.

While a binary semaphore functions as a mutex, a true Mutex tracks fiber ownership, prevents release from non-owners, and enables read-heavy concurrency with a shared or exclusive read/write lock.

Barriers that holds a group of fibers to synchronize at a phase boundary before any are allowed to proceed.

Deadlock tracing

The most important missing feature. It's almost impossible to debug Parataxis because I never got around to this. I should be able to do Acme::Parataxis->dump_fibers(); and it prints all living fibers, their state (WAITING, RUNNING, SUSPENDED), what resources are they parked on (Semaphore, Cannel, etc), and the perl-level stack trace where they called yield.

My only idea to make this work is with weak references to the bottom layer.

Output might look like this on error:

FATAL: deadlock detected:
  - Fiber #3 waiting on Semaphore 0x55d... (at worker.pl line 42)
  - Fiber #4 waiting on Channel 0x55e... (at worker.pl line 87)

But it would return structured data by default.

Footnotes

1. Go and C# also have similar systems ↩