Acme-Parataxis
view release on metacpan or search on metacpan
# PREEMPTION
## `maybe_yield( )`
Increments an internal operation counter for the current fiber. If the counter reaches the threshold set by
`set_preempt_threshold`, the fiber automatically yields.
```perl
while (my $row = $sth->fetch) {
process($row);
Acme::Parataxis->maybe_yield( ); # Cooperatively prevent starvation
}
```
## `set_preempt_threshold( $val )`
Sets the number of `maybe_yield` increments before a forced yield occurs. Default is 0 (preemption disabled).
# Class Methods
every context switch to satisfy these assertions without clobbering active lexical state.
## `eval` vs. `try/catch`
While `feature 'try'` is available in modern Perl, manually teleporting interpreter state can occasionally confuse the
compiler's expectations for stack unwinding. Standard `eval { ... }` remains the most predictable way to handle
exceptions within fibers.
## Signal Handling
Signals are delivered to the main process thread. Perl handles these at 'safe points,' which in this module typically
occur during a context switch (yield, transfer, or call). If you send a signal while a fiber is suspended, it will
generally be processed when the fiber is resumed and hits the next internal Perl opcode.
## The 'Final Transfer' Requirement
In a symmetric coroutine model (using `transfer( )`), fibers don't have a natural 'parent' to return to. I've added
fallback logic to return to the `last_sender` or the main thread on exit but it's good practice to explicitly
`transfer( )` back to a partner fiber or the `root( )` context to ensure your application logic remains predictable.
Leaving a fiber to just 'fall off the end' is like walking out of a room without closing the door; eventually, the
draft will bother someone.
## `is_done( )` vs. Destruction
eg/basic.pl view on Meta::CPAN
use blib;
use Acme::Parataxis;
#
say 'Main thread (TID: ' . Acme::Parataxis->tid . ')';
# Create a worker parataxis
my $worker = Acme::Parataxis->new(
code => sub ($name) {
say "---> Worker '$name' started (FID: " . Acme::Parataxis->current_fid . ")";
for my $i ( 1 .. 3 ) {
say "---> Worker '$name' processing step $i";
my $input = Acme::Parataxis->yield( 'Result from step ' . $i );
say "---> Worker '$name' received: $input";
}
return 'Final success';
}
);
# Drive the worker
say 'Main: Starting worker...';
my $res = $worker->call('Alice');
eg/symmetric.pl view on Meta::CPAN
$consumer->transfer(undef);
}
);
$consumer = Acme::Parataxis->new(
code => sub {
# Initial yield to wait for the first item from the Producer
my $item = Acme::Parataxis->yield();
while ( defined $item ) {
say " Consumer: Received '$item'. Processing...";
say ' Consumer: Done processing. Transferring back to Producer...';
# Transfer back to producer and wait for the NEXT item
$item = $producer->transfer();
}
say ' Consumer: Shutting down.';
Acme::Parataxis->root->transfer(); # Return to main
}
);
say 'Main: Starting the dance...';
eg/worker_pool.pl view on Meta::CPAN
#!/usr/bin/env perl
use v5.40;
use lib 'lib';
use blib;
use Acme::Parataxis;
use Time::HiRes qw[time];
$|++;
# This simulates a pool of workers processing a queue of jobs.
# Each job 'blocks' by sleeping in the native thread pool (simulating I/O),
# allowing other fibers to continue running concurrently on the main thread.
# I haven't really made good use for any of this because it doesn't work everywhere yet...
Acme::Parataxis::run(
sub {
my @jobs = (
{ id => 1, task => 'Fetch User Data', delay => 800 },
{ id => 2, task => 'Process Payment', delay => 1200 },
{ id => 3, task => 'Send Email', delay => 500 },
{ id => 4, task => 'Update Inventory', delay => 1500 },
lib/Acme/Parataxis.c view on Meta::CPAN
int id; /**< Numeric ID of this fiber */
int finished; /**< Flag: 1 if the fiber has completed its entry_point */
int parent_id; /**< ID of the fiber that 'called' this one (asymmetric) */
int last_sender; /**< ID of the fiber that last switched control to this one */
} para_fiber_t;
/** @name Job Status Constants */
///@{
#define JOB_FREE 0 /**< Slot is available for new tasks */
#define JOB_NEW 1 /**< Task is submitted but not yet picked up by a worker */
#define JOB_BUSY 2 /**< Task is currently being processed by a worker thread */
#define JOB_DONE 3 /**< Task has completed and results are ready */
///@}
/** @name Task Type Constants */
///@{
#define TASK_SLEEP 0 /**< Sleep for N milliseconds */
#define TASK_GET_CPU 1 /**< Retrieve current core ID */
#define TASK_READ 2 /**< Wait for read-readiness on a file descriptor */
#define TASK_WRITE 3 /**< Wait for write-readiness on a file descriptor */
///@}
lib/Acme/Parataxis.c view on Meta::CPAN
}
}
if (found_idx != -1 || !threads_keep_running)
break;
PARA_COND_WAIT(queue_cond, queue_lock);
}
UNLOCK(queue_lock);
if (found_idx != -1 && threads_keep_running) {
job_t * job = &job_slots[found_idx];
// ... processing ...
if (job->type == TASK_SLEEP) {
int ms = (int)job->input.i;
#ifdef _WIN32
Sleep(ms);
#else
usleep(ms * 1000);
#endif
job->output.i = ms;
}
lib/Acme/Parataxis.pod view on Meta::CPAN
is established. Like C<call>, it supports passing arbitrary Perl data via C<@args>.
=head1 PREEMPTION
=head2 C<maybe_yield( )>
Increments an internal operation counter for the current fiber. If the counter reaches the threshold set by
C<set_preempt_threshold>, the fiber automatically yields.
while (my $row = $sth->fetch) {
process($row);
Acme::Parataxis->maybe_yield( ); # Cooperatively prevent starvation
}
=head2 C<set_preempt_threshold( $val )>
Sets the number of C<maybe_yield> increments before a forced yield occurs. Default is 0 (preemption disabled).
=head1 Class Methods
=head2 C<tid( )>
lib/Acme/Parataxis.pod view on Meta::CPAN
every context switch to satisfy these assertions without clobbering active lexical state.
=head2 C<eval> vs. C<try/catch>
While C<feature 'try'> is available in modern Perl, manually teleporting interpreter state can occasionally confuse the
compiler's expectations for stack unwinding. Standard C<eval { ... }> remains the most predictable way to handle
exceptions within fibers.
=head2 Signal Handling
Signals are delivered to the main process thread. Perl handles these at 'safe points,' which in this module typically
occur during a context switch (yield, transfer, or call). If you send a signal while a fiber is suspended, it will
generally be processed when the fiber is resumed and hits the next internal Perl opcode.
=head2 The 'Final Transfer' Requirement
In a symmetric coroutine model (using C<transfer( )>), fibers don't have a natural 'parent' to return to. I've added
fallback logic to return to the C<last_sender> or the main thread on exit but it's good practice to explicitly
C<transfer( )> back to a partner fiber or the C<root( )> context to ensure your application logic remains predictable.
Leaving a fiber to just 'fall off the end' is like walking out of a room without closing the door; eventually, the
draft will bother someone.
=head2 C<is_done( )> vs. Destruction
t/003_signals.t view on Meta::CPAN
code => sub {
diag 'Inside fiber: Yielding READY...';
Acme::Parataxis->yield('READY');
diag 'Inside fiber: Resumed after signal delivery.';
return 'Finished';
}
);
diag 'Calling fiber (First step)...';
my $y = $fiber->call();
is $y, 'READY', 'Fiber suspended at yield';
diag 'Sending signal to parent process...';
kill 'INT', $$;
diag 'Signal count before resume: ' . $signaled;
diag 'Resuming fiber...';
$fiber->call();
is $signaled, 1, 'Signal handled between yield and resume';
diag "Signal count final: $signaled";
};
done_testing();
t/014_http_pool.t view on Meta::CPAN
http://example.com
https://www.google.com/
https://www.perl.org/
https://metacpan.org/
https://www.cpan.org/
https://github.com/
];
my %results;
my $worker_count = 3;
my @workers;
diag "Main: Starting worker pool with $worker_count fibers to process " . scalar(@queue) . " URLs...";
for my $w_id ( 1 .. $worker_count ) {
push @workers, Acme::Parataxis->spawn(
sub {
my $fid = Acme::Parataxis->current_fid;
while (1) {
my $url = shift @queue;
last unless $url;
my $res = $http->get($url);
$results{$url} = $res;
t/015_http_mock_pool.t view on Meta::CPAN
}
);
# Testing reentrancy: Use a SINGLE HTTP::Tiny object across multiple concurrent fibers
my $http = Acme::Parataxis::Test::MockPoolHTTP->new( timeout => 5, verify_SSL => 0 );
my $url_base = "http://127.0.0.1:$server_port/";
my @queue = ($url_base) x 10;
my @results;
my $worker_count = 3;
my @workers;
diag "Main: Starting worker pool with $worker_count fibers to process " . scalar(@queue) . " requests...";
for my $w_id ( 1 .. $worker_count ) {
push @workers, Acme::Parataxis->spawn(
sub {
while (1) {
my $url = shift @queue;
last unless $url;
my $res = $http->get($url);
push @results, $res;
}
( run in 1.945 second using v1.01-cache-2.11-cpan-995e09ba956 )