Acme-Parataxis
view release on metacpan or search on metacpan
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [v0.0.10] - 2026-02-22
This version comes with a dynamic thread pool and an improved API.
### Added
- New ergonomic API using exported functions like `async { ... }`, `fiber { ... }`, and `await( $target )`.
### Changed
- Refactored native thread pool to use cond vars (`PARA_COND_*`) instead of busy polling, reducing idle CPU usage to near zero.
- Switched to a global job queue for the thread pool for better load balancing across worker threads.
- Reduced default fiber stack size from 4MB to 512K.
- Worker threads are now only spawned when the first asynchronous job is submitted.
- Increased `MAX_FIBERS` limit to 1024.
- Expose thread pool config with `set_max_threads` and `max_threads`.
## [v0.0.9] - 2026-02-21
Asynchronous HTTP::Tiny is basically a semi-automatic footgun.
### Fixed
- Resolved `AvFILLp(av) == -1` and `!AvREAL(av)` assertion failures in `Perl_pp_entersub` on `DEBUGGING` builds of Perl. This was fixed by ensuring Slot 0 (the argument array) of the next pad depth is correctly initialized during fiber context switch...
### Changed
- Increased fiber stack size to 4MB to provide better support for deep Perl calls and regex operations. This is a temp solution.
## [v0.0.5] - 2026-02-18
### Changed
- I'm honeslty just throwing stuff at the wall. Between my local machines and GH CI workflows, I cannot replicate some of the failures I'm seeing from smokers which makes them virtually impossible to resolve.
## [v0.0.4] - 2026-02-17
### Changed
- Attempt to only spawn max X threads in `t/006_parallel.t` where X is 3 or the `get_thread_pool_size()`? See https://www.cpantesters.org/cpan/report/ecf1410e-0c46-11f1-8628-aee76d8775ea
- Recalculate `PL_curpad = AvARRAY(PL_comppad)` in `swap_perl_state`? See https://www.cpantesters.org/cpan/report/e7244bd8-0c44-11f1-b3ab-94362698fc84
## [v0.0.3] - 2026-02-17
### Changed
- Adding an optional timeout to `await_read` and `await_write`.
- Allow fibers to return complex data (AV*, HV*).
## [v0.0.2] - 2026-02-17
eg/http_tiny.pl
eg/modern.pl
eg/port_scanner.pl
eg/preempt.pl
eg/socket.pl
eg/stress_fibers.pl
eg/stress_io.pl
eg/stress_preempt.pl
eg/symmetric.pl
eg/synopsis.pl
eg/thread_pool.pl
eg/worker_pool.pl
lib/Acme/Parataxis.c
lib/Acme/Parataxis.pm
lib/Acme/Parataxis.pod
t/001_basic.t
t/002_exceptions.t
t/003_signals.t
t/004_synopsis.t
t/005_symmetric.t
t/006_parallel.t
{
"abstract" : "A dangerous hybrid of cooperative fibers and preemptive native threads",
"author" : [
"Sanko Robinson <sanko@cpan.org>"
],
"description" : "Acme::Parataxis is an experimental concurrency platform that merges cooperative user mode fibers with a preemptive native thread pool. It should allow thousands of concurrent 'green threads' to run alongside blocking C-level tasks...
"dynamic_config" : 1,
"generated_by" : "App::mii v1.0.0",
"keywords" : [],
"license" : [
"artistic_2"
],
"meta-spec" : {
"url" : "https://metacpan.org/pod/CPAN::Meta::Spec",
"version" : 2
},
---
abstract: 'A dangerous hybrid of cooperative fibers and preemptive native threads'
author:
- 'Sanko Robinson <sanko@cpan.org>'
build_requires:
Test2::V1: '0'
configure_requires:
Affix: v1.0.7
Affix::Build: '0'
CPAN::Meta: '2.150012'
Exporter: '5.57'
ExtUtils::Helpers: '0.028'
use Acme::Parataxis;
$|++;
Acme::Parataxis::run(
sub {
say 'Main task started';
# Spawn background workers
my $f1 = Acme::Parataxis->spawn(
sub {
say ' Task 1: Sleeping in a native thread pool...';
Acme::Parataxis->await_sleep(1000);
say ' Task 1: Ah! What a nice nap...';
return 42;
}
);
my $f2 = Acme::Parataxis->spawn(
sub {
say ' Task 2: Performing I/O...';
# await_read/write for non-blocking socket handling
return 'I/O Done';
}
);
# Block current fiber until results are ready (without blocking the thread)
say 'Result 1: ' . $f1->await( );
say 'Result 2: ' . $f2->await( );
}
);
```
Or do things the more modern way:
```perl
use v5.40;
# 'await' works on fibers and futures
say 'Result 1: ' . await($f1);
say 'Result 2: ' . await($f2);
};
```
# DESCRIPTION
`Acme::Parataxis` implements a hybrid concurrency model for Perl, greatly inspired by the concurrency system for the
[Wren](https://wren.io/concurrency.html) programming language. It combines cooperative multitasking (fibers) with a
preemptive native thread pool.
Fibers are a mechanism for lightweight concurrency. They are similar to threads but are cooperatively scheduled. While
the OS may switch between threads at any time, a fiber only passes control when explicitly told to do so. This makes
concurrency deterministic and easier to reason about. You (probably) don't have to worry about random context switches
clobbering your data.
Fibers are incredibly lightweight. Each one has its own stack and context, but they don't use OS thread resources. You
can easily create thousands of them without stalling your system.
## A Warning
I had this idea while writing cookbook examples for Affix. I wondered if I could implement a hybrid concurrency model
for Perl from within FFI. This is that unpublished article made into a module. It's fragile. It's dangerous. It's my
attempt at combining cooperative multitasking (green threads or fibers or whatever they're called in the latest edit of
Wikipedia) with a preemptive native thread pool. It's Acme::Parataxis.
This module is experimental and resides in the `Acme::` namespace for a reason. It manually manipulates Perl's
internal stacks and C context. It is very dangerous. It's irresponsible, honestly, that I'm even putting this terrible
idea into the world. Don't use this. Forget you even saw it. Just **reading** this has probably made your projects more
prone to breaking. Reading the package name out loud might cause brain damage to yourself and those within earshot.
Close the browser and clear your history before this does further harm!
# MODERN API
```
async {
await_write($socket);
syswrite($socket, $message);
};
```
## `await_core_id( )`
Returns the ID of the CPU core currently executing the background task. This is a non-blocking operation that offloads
the request to the thread pool and suspends the fiber until the result is ready.
```perl
async {
my $core = await_core_id( );
say "Background task handled by CPU core: $core";
};
```
# CORE CONCEPTS
switch between fibers. When you transfer to a fiber, the current one is suspended, and the target fiber resumes. Unlike
`call( )`, transferring does not establish a parent/child relationship. It's more like a `goto` for execution
contexts.
```
$other_fiber->transfer( );
```
## Fibers vs. Threads
In Parataxis, your **Perl code** always runs on a single OS thread. However, when you call an `await_*` function, the
current fiber is suspended, and the actual blocking work is performed on a **different** OS thread in a native pool.
Once the task completes, your fiber is automatically queued for resumption on the main thread.
# SCHEDULER FUNCTIONS
The following functions are the primary interface for the integrated cooperative scheduler.
## `run( $code )`
Starts the event loop and executes `$code` as the initial fiber. The loop continues to run as long as there are active
fibers or pending background tasks.
Pauses the current fiber and returns control to the scheduler. If `@args` are provided, they are passed to the context
that next resumes this fiber. Arguments can be of any Perl data type.
## `stop( )`
Tells the scheduler to exit the loop after the current iteration. Note that this does not immediately terminate other
fibers; it simply prevents the scheduler from starting new ones.
# THREAD POOL CONFIGURATION
`Acme::Parataxis` uses a native thread pool to handle blocking tasks. While it manages itself automatically, you can
tune its behavior using these functions.
## `set_max_threads( $count )`
Sets the maximum number of worker threads the pool is allowed to spawn. By default, this is set to the number of
logical CPU cores detected on your system (up to a hard limit of 64).
```
# Limit the pool to 4 threads
set_max_threads(4);
```
## `max_threads( )`
Returns the currently configured maximum thread pool size.
# BLOCKING & I/O FUNCTIONS
These functions **suspend** the current fiber and offload the actual blocking work to the native thread pool.
## `await_sleep( $ms )`
Suspends the fiber for `$ms` milliseconds. While the background thread sleeps, other fibers can continue to execute.
## `await_read( $fh, $timeout = 5000 )`
Suspends the fiber until the filehandle `$fh` is ready for reading, or the `$timeout` (in milliseconds) is reached.
```perl
my $status = Acme::Parataxis->await_read($socket);
if ($status > 0) {
my $data = <$socket>;
}
```
## `await_write( $fh, $timeout = 5000 )`
Suspends the fiber until the filehandle `$fh` is ready for writing.
## `await_core_id( )`
Offloads a request to the thread pool and returns the ID of the CPU core that handled the job.
# MANUAL FIBER MANAGEMENT
Advanced users can manage context switching themselves without using the integrated scheduler.
## `new( code => $sub )`
Instantiates a new fiber. The `code` argument must be a subroutine reference.
```perl
```
## `set_preempt_threshold( $val )`
Sets the number of `maybe_yield` increments before a forced yield occurs. Default is 0 (preemption disabled).
# Class Methods
## `tid( )`
Returns the unique OS Thread ID of the main interpreter thread.
## `current_fid( )`
Returns the unique numeric ID of the currently executing fiber, or -1 if called from the "root" (main) context.
## `root( )`
Returns a proxy object representing the initial execution context. This is useful for `transfer( )`ing control back to
the main thread from a symmetric coroutine.
# Acme::Parataxis OBJECT METHODS
## `fid( )`
Returns the unique numeric ID of the fiber object.
## `is_done( )`
Returns true if the fiber has finished execution (either by returning or dying). Once a fiber is done, its internal ID
}
}
}
}
```
# EXAMPLES
## Cooperative Parallelism
This example demonstrates how to perform multiple HTTP requests concurrently on a single interpretation thread.
```perl
use Acme::Parataxis;
# ... (See My::HTTP implementation in INTEGRATING SYNCHRONOUS MODULES) ...
Acme::Parataxis::run(sub {
my $http = My::HTTP->new(verify_SSL => 0);
my @urls = qw[http://example.com http://perl.org];
# Spawn tasks for each URL
$item = $p->transfer( );
}
});
$c->call( ); # Prime consumer
$p->call( ); # Start producer
```
# BEST PRACTICES & GOTCHAS
- **Avoid blocking syscalls:** Never call blocking `sleep( )` or `sysread( )` on the main interpretation thread.
Always use the `await_*` equivalents to offload work to the pool.
- **Thread Safety:** While Perl code remains single-threaded, background tasks run on separate OS threads. Shared
C-level data (if accessed via FFI) must be mutex-protected.
- **Stack Limits:** Each fiber is allocated a 512KB stack by default. This is more than sufficient for most
Perl code and allows for high concurrency with a small memory footprint. Extremely deep recursion or massive regex
backtracking might still hit limits.
- **Efficiency:** The native thread pool is initialized dynamically upon the first asynchronous request. It
starts with a small "seed" pool and grows on demand up to the configured limit. Worker threads use condition
variables to sleep efficiently when idle, ensuring near-zero CPU usage when no background tasks are pending.
- **Reference Cycles:** Be careful when passing fiber objects into their own closures, as this can create
memory leaks.
# GORY TECHNICAL DETAILS
## Architectural Inspiration
The concurrency model in Parataxis is heavily inspired by the **Wren** programming language, specifically its treatment
of fibers as the primary unit of execution and its deterministic cooperative scheduling.
## Stack Virtualization
On Unix-like systems, we use `ucontext.h` to manage stack and register state. On Windows, we leverage the native
`Fiber API`. In both cases, we perform heart surgery on the Perl interpreter by manually teleporting its internal
global pointers (the `PL_*` variables) between contexts.
## Shared CVs and Pad Virtualization
A significant challenge in Perl green threads is the shared nature of PadLists and the global `CvDEPTH` counter. In
debug builds of Perl, calling a shared subroutine from multiple fibers can trigger internal assertions (like
`AvFILLp(av) == -1`). Parataxis includes a specialized workaround that surgically cleans the next landing pad before
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
A fiber being `is_done( )` simply means its Perl code has finished executing. The underlying C-level memory (stacks,
context, etc.) is not immediately freed until the `Acme::Parataxis` object is destroyed or the runtime performs its
final `cleanup( )`. This is why you might see memory usage stay flat even after a fiber finishes, until the garbage
collector finally catches up with the object.
eg/affinity.pl view on Meta::CPAN
use v5.40;
use blib;
use Acme::Parataxis qw[:all];
$|++;
# Demonstrate Thread Affinity / Core ID using modern API
async {
say 'Main thread (TID: ' . tid() . ')';
my @futures;
for ( 1 .. 20 ) {
push @futures, fiber {
# await_core_id() offloads to the pool and returns the core ID
return await_core_id();
};
}
say 'Main: Tasks spawned, waiting for results...';
my %distribution;
eg/basic.pl view on Meta::CPAN
use v5.40;
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";
}
eg/stress_io.pl view on Meta::CPAN
my $start_time = time();
for my $iter ( 1 .. $iterations ) {
my $iter_start = time();
my @futures;
# Spawn concurrent sleep fibers
for my $i ( 1 .. $concurrency ) {
push @futures, fiber {
no warnings 'recursion';
# This offloads to the C-level background thread pool
await_sleep($sleep_ms);
return "Worker $i done";
};
}
# Wait for all to finish
for my $f (@futures) {
await $f;
}
my $elapsed = time() - $iter_start;
eg/synopsis.pl view on Meta::CPAN
say ' Task 2: Calculating... (simulated CPU work)';
my $sum = 0;
for ( 1 .. 100 ) {
$sum += $_;
Acme::Parataxis->maybe_yield(); # Be a good neighbor
}
return $sum;
}
);
# Await results without blocking the main OS thread
say 'Result 1: ' . $f1->await();
say 'Result 2: ' . $f2->await();
}
);
}
eg/thread_pool.pl view on Meta::CPAN
use v5.40;
use blib;
use Acme::Parataxis qw[:all];
use Time::HiRes qw[time];
$|++;
# Demonstrate dynamic thread pool growth and modern API
async {
say 'Main: Initial thread pool size is ' . Acme::Parataxis::get_thread_pool_size();
my @futures;
for my $id ( 1 .. 5 ) {
push @futures, fiber {
my $fid = current_fid();
my $sleep_time = int( rand(1000) ) + 500;
say " [Worker $id] Fiber #$fid will sleep for ${sleep_time}ms in background pool";
my $start = time();
# This yields to main, background thread sleeps, scheduler resumes us
await_sleep($sleep_time);
my $elapsed = int( ( time() - $start ) * 1000 );
say " [Worker $id] Fiber #$fid woke up after ${elapsed}ms!";
return "Done $id";
};
}
say 'Main: All workers spawned. Waiting for completion...';
say 'Main: Thread pool size during load: ' . Acme::Parataxis::get_thread_pool_size();
my $start_time = time();
# Wait for all workers to finish via futures using modern 'await'
await $_ for @futures;
say 'Main: All workers finished!';
printf "Total wallclock time: %.2fs\n", time() - $start_time;
say 'Main: Final thread pool size: ' . Acme::Parataxis::get_thread_pool_size();
};
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 },
{ id => 5, task => 'Generate Report', delay => 900 },
{ id => 6, task => 'Sync Analytics', delay => 1100 }
eg/worker_pool.pl view on Meta::CPAN
# Spawn 3 worker fibers
my @workers;
for my $w_id ( 1 .. 3 ) {
push @workers, Acme::Parataxis->spawn(
sub {
say " [Worker $w_id] Fiber started.";
while ( my $job = shift @queue ) {
say " [Worker $w_id] Processing Job $job->{id}: $job->{task}...";
# Simulate a blocking I/O call (DB query, API request, idk...)
# This yields to the scheduler while the native thread pool sleeps.
Acme::Parataxis->await_sleep( $job->{delay} );
say " [Worker $w_id] Finished Job $job->{id}.";
push @results, "Job $job->{id} complete";
}
return "Worker $w_id finished.";
}
);
}
# Wait for all workers to complete
lib/Acme/Parataxis.c view on Meta::CPAN
/**
* @file Parataxis.c
* @brief Low-level Green Threads (Fibers) and Hybrid Thread Pool for Perl.
*
* @section Overview
* This file implements a cooperative multitasking system (Fibers) integrated
* with a preemptive native thread pool. It allows Perl to run thousands of
* user-mode fibers that can offload blocking C-level tasks to background
* OS threads without stalling the main interpreter.
*
* @section Architecture
* - **Fibers**: The primitive unit of execution. Each fiber has its own OS context
* and a complete set of Perl interpreter stacks (Argument, Mark, Scope, Save, Mortal).
* - **Coroutines**: The execution pattern (yield/call/transfer) used by fibers to
* pass control.
* - **Thread Pool**: A fixed pool of worker threads that poll a job queue for
* blocking operations like sleep, I/O, or heavy computation.
* - **Context Switching**: The `swap_perl_state` function manually saves and restores
* the global state of the Perl interpreter (`PL_*` variables) to allow disjoint
* execution flows.
*
* @section Caveats
* Shared subroutines (CVs) with re-entrant yielding calls are handled by a
* specialized pad-clearing mechanism in `_activate_current_depths` to satisfy
* Perl's internal `AvFILLp` assertions in debug builds.
*/
lib/Acme/Parataxis.c view on Meta::CPAN
#define NO_XSLOCKS
#include "EXTERN.h"
#include "XSUB.h"
#include "perl.h"
#ifdef _WIN32
/** @brief Export macro for Windows DLLs */
#define DLLEXPORT __declspec(dllexport)
/** @brief Handle for the underlying OS fiber context */
typedef LPVOID coro_handle_t;
/** @brief Handle for a native OS thread */
typedef HANDLE para_thread_t;
/** @brief Mutex type for queue synchronization */
typedef CRITICAL_SECTION para_mutex_t;
#define LOCK(m) EnterCriticalSection(&m)
#define UNLOCK(m) LeaveCriticalSection(&m)
#define LOCK_INIT(m) InitializeCriticalSection(&m)
#else
#include <pthread.h>
#include <sched.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <ucontext.h>
#include <unistd.h>
#if defined(__APPLE__) || defined(__FreeBSD__)
#include <sys/sysctl.h>
#include <sys/types.h>
#endif
/** @brief Export macro for Unix systems */
#define DLLEXPORT __attribute__((visibility("default")))
/** @brief Handle for the underlying OS fiber context (ucontext_t) */
typedef ucontext_t coro_handle_t;
/** @brief Handle for a native OS thread (pthread_t) */
typedef pthread_t para_thread_t;
/** @brief Mutex type for queue synchronization (pthread_mutex_t) */
typedef pthread_mutex_t para_mutex_t;
#define LOCK(m) pthread_mutex_lock(&m)
#define UNLOCK(m) pthread_mutex_unlock(&m)
#define LOCK_INIT(m) pthread_mutex_init(&m, NULL)
#endif
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
// Forward declarations
DLLEXPORT SV * coro_yield(SV * ret_val);
DLLEXPORT SV * coro_transfer(int fiber_id, SV * args);
DLLEXPORT void destroy_coro(int fiber_id);
/**
* @brief Get the Operating System's unique Thread ID.
*
* Useful for debugging to prove that background tasks are running on
* different OS threads than the main Perl interpreter.
*
* @return int The TID (Windows) or LWP ID (Linux/BSD/macOS).
*/
int get_os_thread_id() {
#ifdef _WIN32
return (int)GetCurrentThreadId();
#elif defined(__APPLE__)
uint64_t tid;
pthread_threadid_np(NULL, &tid);
return (int)tid;
#elif defined(SYS_gettid)
return (int)syscall(SYS_gettid);
#else
return (int)(intptr_t)pthread_self();
#endif
}
/**
* @brief Pin the current thread to a specific CPU core.
*
* Used by the Thread Pool to ensure worker threads are distributed
* across available hardware cores for maximum parallelism.
*
* @param core_id The zero-based index of the CPU core.
*/
void pin_to_core(int core_id) {
#ifdef _WIN32
DWORD_PTR mask = (1ULL << core_id);
SetThreadAffinityMask(GetCurrentThread(), mask);
#elif defined(__linux__)
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(core_id, &cpuset);
pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
#else
(void)core_id; /* Not supported on macOS/BSD standard APIs */
#endif
}
/**
* @brief Get the index of the CPU core currently executing this thread.
*
* @return int Core ID (0..N) or -1 if unsupported.
*/
int get_current_cpu() {
#ifdef _WIN32
return GetCurrentProcessorNumber();
#elif defined(__linux__)
return sched_getcpu();
#else
return -1;
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
* @brief Generic container for task input/output data.
*/
typedef union {
int64_t i; /**< Integer/Pointer storage */
double d; /**< Floating point storage */
char * s; /**< String storage */
} value_t;
/**
* @struct job_t
* @brief Represents a task in the background thread pool queue.
*/
typedef struct {
int fiber_id; /**< ID of the Fiber that submitted this task */
int target_thread; /**< Index of the assigned worker thread */
int type; /**< Type of task to perform (TASK_*) */
value_t input; /**< Input data for the task */
value_t output; /**< Result data populated by the worker */
int timeout_ms; /**< Timeout duration for I/O tasks */
int status; /**< Current lifecycle state (JOB_*) */
} job_t;
// Global Registry and State
/** @brief Maximum number of concurrent fibers allowed */
#define MAX_FIBERS 1024
/** @brief Array of active fiber structures */
static para_fiber_t * fibers[MAX_FIBERS];
/** @brief The context representing the main Perl thread */
static para_fiber_t main_context;
/** @brief ID of the currently executing fiber (-1 for Main) */
static int current_fiber_id = -1;
/** @brief Size of the background job queue */
#define MAX_JOBS 1024
/** @brief Fixed-size array for background tasks */
static job_t job_slots[MAX_JOBS];
/** @brief Mutex protecting access to the job queue */
static para_mutex_t queue_lock;
#ifdef _WIN32
static CONDITION_VARIABLE queue_cond;
#else
static pthread_cond_t queue_cond;
#endif
static int threads_initialized = 0;
static int system_initialized = 0;
// Forward declarations for thread safety wrappers
#ifdef _WIN32
#define PARA_COND_WAIT(c, m) SleepConditionVariableCS(&c, &m, INFINITE)
#define PARA_COND_SIGNAL(c) WakeConditionVariable(&c)
#define PARA_COND_BROADCAST(c) WakeAllConditionVariable(&c)
#define PARA_COND_INIT(c) InitializeConditionVariable(&c)
#else
#define PARA_COND_WAIT(c, m) pthread_cond_wait(&c, &m)
#define PARA_COND_SIGNAL(c) pthread_cond_signal(&c)
#define PARA_COND_BROADCAST(c) pthread_cond_broadcast(&c)
#define PARA_COND_INIT(c) pthread_cond_init(&c, NULL)
#endif
/** @brief Threshold for automatic preemption (0 to disable) */
static long long preempt_threshold = 0;
/** @brief Count of operations since last preemption yield */
static long long preempt_count = 0;
/** @brief Maximum worker threads allowed in the pool */
#define MAX_THREADS 64
/** @brief Native OS handles for pool threads */
static para_thread_t thread_handles[MAX_THREADS];
/** @brief Maximum allowed threads in the pool */
static int max_thread_pool_size = 0;
/** @brief Number of currently running worker threads */
static int current_thread_count = 0;
/** @brief Flag to signal worker threads to terminate */
static volatile int threads_keep_running = 1;
#ifdef _WIN32
/** @brief Windows-only handle for the main thread converted to fiber */
static void * main_fiber_handle = NULL;
#endif
/** @brief Sets the maximum number of worker threads allowed in the pool. */
DLLEXPORT void set_max_threads(int max) {
if (max > 0 && max <= MAX_THREADS)
max_thread_pool_size = max;
}
/** @brief Forward declaration of worker_thread */
#ifdef _WIN32
DWORD WINAPI worker_thread(LPVOID arg);
#else
void * worker_thread(void * arg);
#endif
/** @brief Internal helper to spawn N threads into the pool */
static void _spawn_workers(int count) {
for (int i = 0; i < count; i++) {
if (current_thread_count >= max_thread_pool_size || current_thread_count >= MAX_THREADS)
break;
int tid = current_thread_count;
#ifdef _WIN32
thread_handles[tid] = CreateThread(NULL, 0, worker_thread, (LPVOID)(intptr_t)tid, 0, NULL);
#else
pthread_create(&thread_handles[tid], NULL, worker_thread, (void *)(intptr_t)tid);
pthread_detach(thread_handles[tid]);
#endif
current_thread_count++;
}
}
/**
* @brief Background Worker Thread Loop.
*
* Each thread pins itself to a core and continuously waits for jobs.
*
* @param arg Integer thread ID passed as a pointer.
*/
#ifdef _WIN32
DWORD WINAPI worker_thread(LPVOID arg) {
#else
void * worker_thread(void * arg) {
#endif
int thread_id = (int)(intptr_t)arg;
int cpu_count = get_cpu_count();
pin_to_core(thread_id % cpu_count);
while (threads_keep_running) {
int found_idx = -1;
LOCK(queue_lock);
while (threads_keep_running) {
for (int i = 0; i < MAX_JOBS; i++) {
if (job_slots[i].status == JOB_NEW) {
job_slots[i].status = JOB_BUSY;
found_idx = i;
break;
}
}
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
lib/Acme/Parataxis.c view on Meta::CPAN
FD_SET(s, &fds);
#else
int fd = (int)job->input.i;
FD_SET(fd, &fds);
#endif
struct timeval tv;
int res;
int elapsed_ms = 0;
int timeout = job->timeout_ms > 0 ? job->timeout_ms : 5000;
while (threads_keep_running) {
tv.tv_sec = 0;
tv.tv_usec = 10000;
fd_set work_fds = fds;
if (job->type == TASK_READ)
#ifdef _WIN32
res = select(0, &work_fds, NULL, NULL, &tv);
#else
res = select(fd + 1, &work_fds, NULL, NULL, &tv);
#endif
lib/Acme/Parataxis.c view on Meta::CPAN
Sleep(1);
#else
usleep(1000);
#endif
}
}
return 0;
}
/**
* @brief Initializes the background thread pool.
*
* Automatically detects the CPU count and spawns worker threads. This function
* is called automatically by `init_system` and `submit_c_job`.
*/
DLLEXPORT void init_threads() {
dTHX;
if (threads_initialized)
return;
LOCK_INIT(queue_lock);
PARA_COND_INIT(queue_cond);
for (int i = 0; i < MAX_JOBS; i++)
job_slots[i].status = JOB_FREE;
if (max_thread_pool_size == 0) {
max_thread_pool_size = get_cpu_count();
if (max_thread_pool_size > MAX_THREADS)
max_thread_pool_size = MAX_THREADS;
}
/* Start with a small "seed" pool of 2 threads */
_spawn_workers(2);
threads_initialized = 1;
}
/**
* @brief Submits a C-level task to the background pool.
*
* @param type The task type constant (TASK_*).
* @param arg Input integer or pointer data.
* @param timeout_ms Timeout for I/O operations.
* @return int The index of the submitted job, or -1 if the queue is full.
*/
DLLEXPORT int submit_c_job(int type, int64_t arg, int timeout_ms) {
if (!threads_initialized)
init_threads();
int idx = -1;
LOCK(queue_lock);
/* Dynamic Scaling: If we have pending jobs and space in the pool, grow! */
int pending_count = 0;
for (int i = 0; i < MAX_JOBS; i++)
if (job_slots[i].status == JOB_NEW)
pending_count++;
if (pending_count > 0 && current_thread_count < max_thread_pool_size)
_spawn_workers(1); /* Grow by 1 on demand */
for (int i = 0; i < MAX_JOBS; i++) {
if (job_slots[i].status == JOB_FREE) {
idx = i;
break;
}
}
if (idx != -1) {
job_slots[idx].fiber_id = current_fiber_id;
lib/Acme/Parataxis.c view on Meta::CPAN
UNLOCK(queue_lock);
return idx;
}
/**
* @brief Polls the queue for any completed background jobs.
*
* @return int Index of a finished job, or -1 if none are ready.
*/
DLLEXPORT int check_for_completion() {
if (!threads_initialized)
init_threads();
int job_idx = -1;
LOCK(queue_lock);
for (int i = 0; i < MAX_JOBS; i++) {
if (job_slots[i].status == JOB_DONE) {
job_idx = i;
break;
}
}
UNLOCK(queue_lock);
return job_idx;
lib/Acme/Parataxis.c view on Meta::CPAN
if (SvROK(cv_ref))
cv = (CV *)SvRV(cv_ref);
else if (SvTYPE(cv_ref) == SVt_PVCV)
cv = (CV *)cv_ref;
if (cv && SvTYPE((SV *)cv) == SVt_PVCV)
((XPVCV *)MUTABLE_PTR(SvANY(cv)))->xcv_depth = 0;
}
/** @brief Returns the ID of the currently executing fiber. */
DLLEXPORT int get_current_parataxis_id() { return current_fiber_id; }
/** @brief Returns the OS-level thread ID of the main interpreter thread. */
DLLEXPORT int get_os_thread_id_export() { return get_os_thread_id(); }
/** @brief Returns the number of worker threads currently running in the pool. */
DLLEXPORT int get_thread_pool_size() { return current_thread_count; }
/** @brief Returns the maximum number of worker threads allowed in the pool. */
DLLEXPORT int get_max_thread_pool_size() { return max_thread_pool_size; }
/** @brief Sets the threshold for automatic yield-based preemption. */
DLLEXPORT void set_preempt_threshold(int64_t threshold) { preempt_threshold = threshold; }
/** @brief Returns the current count towards the preemption threshold. */
DLLEXPORT int64_t get_preempt_count() { return preempt_count; }
/**
* @brief Checks if automatic preemption should occur.
*
* Increments the internal counter and triggers a `coro_yield` if the
lib/Acme/Parataxis.c view on Meta::CPAN
PL_curpad = to->curpad;
// Restore CvDEPTH and clean landing pads
_activate_current_depths(aTHX_ to);
}
/**
* @brief Allocates and initializes new Perl stacks for a fiber.
*
* Each fiber needs a complete set of independent stacks (Argument, Mark,
* Scope, Save, Mortal) to function as a separate execution thread.
*
* @param c The fiber context to initialize.
*/
void init_perl_stacks(para_fiber_t * c) {
dTHX;
// Allocate Stack Info (SI)
Newxz(c->si, 1, PERL_SI);
c->si->si_cxmax = 64;
lib/Acme/Parataxis.c view on Meta::CPAN
c->curstash = PL_curstash;
c->defstash = PL_defstash;
c->errors = PL_errors;
// Start with fresh pads to avoid interfering with caller.
c->comppad = NULL;
c->curpad = NULL;
}
/**
* @brief Initializes the fiber system and converts the main thread.
*
* This function must be called once before any other fiber operations.
* It captures the state of the main Perl interpreter thread.
*
* @return int 0 on success.
*/
DLLEXPORT int init_system() {
dTHX;
if (system_initialized)
return 0;
if (max_thread_pool_size == 0) {
max_thread_pool_size = get_cpu_count();
if (max_thread_pool_size > MAX_THREADS)
max_thread_pool_size = MAX_THREADS;
}
main_context.si = PL_curstackinfo;
main_context.transfer_data = &PL_sv_undef;
main_context.id = -1;
main_context.finished = 0;
main_context.last_sender = -1;
main_context.curpm = PL_curpm;
main_context.curpm_under = PL_curpm_under;
main_context.reg_curpm = PL_reg_curpm;
main_context.defgv = PL_defgv;
main_context.last_in_gv = PL_last_in_gv;
main_context.rs = PL_rs;
main_context.ofsgv = PL_ofsgv;
main_context.ors_sv = PL_ors_sv;
main_context.defoutgv = PL_defoutgv;
main_context.curstash = PL_curstash;
main_context.defstash = PL_defstash;
main_context.errors = PL_errors;
system_initialized = 1;
#ifdef _WIN32
/* Convert the main thread into a fiber so it can be switched out */
if (!main_fiber_handle) {
main_fiber_handle = ConvertThreadToFiber(NULL);
if (!main_fiber_handle) {
if (GetLastError() == ERROR_ALREADY_FIBER)
main_fiber_handle = GetCurrentFiber();
}
}
#endif
init_threads();
return 0;
}
/**
* @brief Performs the low-level OS context switch.
*
* Saves the Perl state and then uses OS primitives (SwitchToFiber or
* swapcontext) to change execution flow.
*
* @param target_id ID of the target fiber (-1 for Main).
lib/Acme/Parataxis.c view on Meta::CPAN
if (target_id == -1)
SwitchToFiber(main_fiber_handle);
else
SwitchToFiber(to->context);
#else
swapcontext(&from->context, &to->context);
#endif
}
/**
* @brief Yields execution back to the caller or the main thread.
*
* Suspends the current fiber and returns a value to the context that
* last resumed or called this fiber.
*
* @param ret_val The Perl SV to "return" to the caller.
* @return SV* The value passed in when this fiber is eventually resumed.
*/
DLLEXPORT SV * coro_yield(SV * ret_val) {
dTHX;
if (current_fiber_id == -1)
lib/Acme/Parataxis.c view on Meta::CPAN
SV * sv = c->tmps_stack[i];
if (sv && sv != &PL_sv_undef)
SvREFCNT_dec(sv);
}
Safefree(c->tmps_stack);
}
free(c);
}
/**
* @brief Global cleanup function for the fiber and thread pool system.
*
* Signals all worker threads to terminate and destroys all remaining
* fibers. Should be called during global destruction or system shutdown.
*/
DLLEXPORT void cleanup() {
dTHX;
if (threads_initialized) {
LOCK(queue_lock);
threads_keep_running = 0;
PARA_COND_BROADCAST(queue_cond);
UNLOCK(queue_lock);
#ifdef _WIN32
/* Wait for threads to finish and close handles */
for (int i = 0; i < current_thread_count; i++) {
if (thread_handles[i]) {
WaitForSingleObject(thread_handles[i], 100);
CloseHandle(thread_handles[i]);
thread_handles[i] = NULL;
}
}
#else
/* Give threads a moment to notice threads_keep_running = 0 */
usleep(10000);
#endif
}
if (current_fiber_id != -1) {
swap_perl_state(fibers[current_fiber_id], &main_context);
current_fiber_id = -1;
}
for (int i = 0; i < MAX_FIBERS; i++)
if (fibers[i])
lib/Acme/Parataxis.pm view on Meta::CPAN
use Time::HiRes qw[usleep];
use Exporter qw[import];
use Carp qw[croak];
our %EXPORT_TAGS = (
all => [
our @EXPORT_OK
= qw[
run spawn yield await stop async fiber
await_sleep await_read await_write await_core_id
current_fid tid root maybe_yield
set_max_threads max_threads
]
]
);
#
our @IPC_BUFFER;
my $lib;
my @SCHEDULER_QUEUE;
my $IS_RUNNING = 0;
sub _bind_functions ($l) {
affix $l, 'init_system', [], Int;
affix $l, 'create_fiber', [ Pointer [SV], Pointer [SV] ], Int;
affix $l, 'coro_call', [ Int, Pointer [SV] ], Pointer [SV];
affix $l, 'coro_transfer', [ Int, Pointer [SV] ], Pointer [SV];
affix $l, 'coro_yield', [ Pointer [SV] ], Pointer [SV];
affix $l, 'is_finished', [Int], Int;
affix $l, 'destroy_coro', [Int], Void;
affix $l, 'force_depth_zero', [ Pointer [SV] ], Void;
affix $l, 'cleanup', [], Void;
affix $l, 'get_os_thread_id_export', [], Int;
affix $l, 'get_current_parataxis_id', [], Int;
affix $l, 'submit_c_job', [ Int, LongLong, Int ], Int;
affix $l, 'check_for_completion', [], Int;
affix $l, 'get_job_result', [Int], Pointer [SV];
affix $l, 'get_job_coro_id', [Int], Int;
affix $l, 'free_job_slot', [Int], Void;
affix $l, 'get_thread_pool_size', [], Int;
affix $l, 'get_max_thread_pool_size', [], Int;
affix $l, 'set_max_threads', [Int], Void;
affix $l, 'set_preempt_threshold', [LongLong], Void;
affix $l, [ 'maybe_yield' => '_maybe_yield' ], [], Pointer [SV];
affix $l, 'get_preempt_count', [], LongLong;
# Capture the main interpreter context eagerly
init_system();
if ( $^O eq 'MSWin32' ) {
my $perl_dll = $Config{libperl};
$perl_dll =~ s/^lib//;
$perl_dll =~ s/\.a$//;
lib/Acme/Parataxis.pm view on Meta::CPAN
sub maybe_yield {
my $invocant = shift;
if ( !defined $invocant || ( ( ref $invocant || $invocant ) ne 'Acme::Parataxis' && !eval { $invocant->isa('Acme::Parataxis') } ) ) {
unshift @_, $invocant if defined $invocant;
}
my $result = Acme::Parataxis::_maybe_yield();
return unless defined $result;
return wantarray ? @$result : $result->[-1];
}
sub tid { get_os_thread_id_export() }
sub current_fid { get_current_parataxis_id() }
sub root { state $root //= Acme::Parataxis::Root->new() }
sub max_threads () { Acme::Parataxis::get_max_thread_pool_size() }
# Scheduler internals
sub _scheduler_enqueue_by_id ($fid) {
if ( my $fiber = Acme::Parataxis->by_id($fid) ) {
push @SCHEDULER_QUEUE, $fiber;
}
}
sub poll_io {
my @ready;
lib/Acme/Parataxis.pod view on Meta::CPAN
use Acme::Parataxis;
$|++;
Acme::Parataxis::run(
sub {
say 'Main task started';
# Spawn background workers
my $f1 = Acme::Parataxis->spawn(
sub {
say ' Task 1: Sleeping in a native thread pool...';
Acme::Parataxis->await_sleep(1000);
say ' Task 1: Ah! What a nice nap...';
return 42;
}
);
my $f2 = Acme::Parataxis->spawn(
sub {
say ' Task 2: Performing I/O...';
# await_read/write for non-blocking socket handling
return 'I/O Done';
}
);
# Block current fiber until results are ready (without blocking the thread)
say 'Result 1: ' . $f1->await( );
say 'Result 2: ' . $f2->await( );
}
);
Or do things the more modern way:
use v5.40;
use Acme::Parataxis qw[:all];
$|++;
lib/Acme/Parataxis.pod view on Meta::CPAN
# 'await' works on fibers and futures
say 'Result 1: ' . await($f1);
say 'Result 2: ' . await($f2);
};
=head1 DESCRIPTION
C<Acme::Parataxis> implements a hybrid concurrency model for Perl, greatly inspired by the concurrency system for the
L<Wren|https://wren.io/concurrency.html> programming language. It combines cooperative multitasking (fibers) with a
preemptive native thread pool.
Fibers are a mechanism for lightweight concurrency. They are similar to threads but are cooperatively scheduled. While
the OS may switch between threads at any time, a fiber only passes control when explicitly told to do so. This makes
concurrency deterministic and easier to reason about. You (probably) don't have to worry about random context switches
clobbering your data.
Fibers are incredibly lightweight. Each one has its own stack and context, but they don't use OS thread resources. You
can easily create thousands of them without stalling your system.
=head2 A Warning
I had this idea while writing cookbook examples for Affix. I wondered if I could implement a hybrid concurrency model
for Perl from within FFI. This is that unpublished article made into a module. It's fragile. It's dangerous. It's my
attempt at combining cooperative multitasking (green threads or fibers or whatever they're called in the latest edit of
Wikipedia) with a preemptive native thread pool. It's Acme::Parataxis.
This module is experimental and resides in the C<Acme::> namespace for a reason. It manually manipulates Perl's
internal stacks and C context. It is very dangerous. It's irresponsible, honestly, that I'm even putting this terrible
idea into the world. Don't use this. Forget you even saw it. Just B<reading> this has probably made your projects more
prone to breaking. Reading the package name out loud might cause brain damage to yourself and those within earshot.
Close the browser and clear your history before this does further harm!
=head1 MODERN API
lib/Acme/Parataxis.pod view on Meta::CPAN
Suspends the current fiber until the provided filehandle is ready for writing, or the timeout is reached.
async {
await_write($socket);
syswrite($socket, $message);
};
=head2 C<await_core_id( )>
Returns the ID of the CPU core currently executing the background task. This is a non-blocking operation that offloads
the request to the thread pool and suspends the fiber until the result is ready.
async {
my $core = await_core_id( );
say "Background task handled by CPU core: $core";
};
=head1 CORE CONCEPTS
=head2 Creating Fibers
lib/Acme/Parataxis.pod view on Meta::CPAN
While C<call( )> and C<yield( )> manage a stack-like chain of execution, C<transfer( )> provides an unstructured way to
switch between fibers. When you transfer to a fiber, the current one is suspended, and the target fiber resumes. Unlike
C<call( )>, transferring does not establish a parent/child relationship. It's more like a C<goto> for execution
contexts.
$other_fiber->transfer( );
=head2 Fibers vs. Threads
In Parataxis, your B<Perl code> always runs on a single OS thread. However, when you call an C<await_*> function, the
current fiber is suspended, and the actual blocking work is performed on a B<different> OS thread in a native pool.
Once the task completes, your fiber is automatically queued for resumption on the main thread.
=head1 SCHEDULER FUNCTIONS
The following functions are the primary interface for the integrated cooperative scheduler.
=head2 C<run( $code )>
Starts the event loop and executes C<$code> as the initial fiber. The loop continues to run as long as there are active
fibers or pending background tasks.
lib/Acme/Parataxis.pod view on Meta::CPAN
Pauses the current fiber and returns control to the scheduler. If C<@args> are provided, they are passed to the context
that next resumes this fiber. Arguments can be of any Perl data type.
=head2 C<stop( )>
Tells the scheduler to exit the loop after the current iteration. Note that this does not immediately terminate other
fibers; it simply prevents the scheduler from starting new ones.
=head1 THREAD POOL CONFIGURATION
C<Acme::Parataxis> uses a native thread pool to handle blocking tasks. While it manages itself automatically, you can
tune its behavior using these functions.
=head2 C<set_max_threads( $count )>
Sets the maximum number of worker threads the pool is allowed to spawn. By default, this is set to the number of
logical CPU cores detected on your system (up to a hard limit of 64).
# Limit the pool to 4 threads
set_max_threads(4);
=head2 C<max_threads( )>
Returns the currently configured maximum thread pool size.
=head1 BLOCKING & I/O FUNCTIONS
These functions B<suspend> the current fiber and offload the actual blocking work to the native thread pool.
=head2 C<await_sleep( $ms )>
Suspends the fiber for C<$ms> milliseconds. While the background thread sleeps, other fibers can continue to execute.
=head2 C<await_read( $fh, $timeout = 5000 )>
Suspends the fiber until the filehandle C<$fh> is ready for reading, or the C<$timeout> (in milliseconds) is reached.
my $status = Acme::Parataxis->await_read($socket);
if ($status > 0) {
my $data = <$socket>;
}
=head2 C<await_write( $fh, $timeout = 5000 )>
Suspends the fiber until the filehandle C<$fh> is ready for writing.
=head2 C<await_core_id( )>
Offloads a request to the thread pool and returns the ID of the CPU core that handled the job.
=head1 MANUAL FIBER MANAGEMENT
Advanced users can manage context switching themselves without using the integrated scheduler.
=head2 C<new( code =E<gt> $sub )>
Instantiates a new fiber. The C<code> argument must be a subroutine reference.
my $fiber = Acme::Parataxis->new(code => sub {
lib/Acme/Parataxis.pod view on Meta::CPAN
}
=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( )>
Returns the unique OS Thread ID of the main interpreter thread.
=head2 C<current_fid( )>
Returns the unique numeric ID of the currently executing fiber, or -1 if called from the "root" (main) context.
=head2 C<root( )>
Returns a proxy object representing the initial execution context. This is useful for C<transfer( )>ing control back to
the main thread from a symmetric coroutine.
=head1 Acme::Parataxis OBJECT METHODS
=head2 C<fid( )>
Returns the unique numeric ID of the fiber object.
=head2 C<is_done( )>
Returns true if the fiber has finished execution (either by returning or dying). Once a fiber is done, its internal ID
lib/Acme/Parataxis.pod view on Meta::CPAN
Acme::Parataxis->await_write($self->{fh}, int($wait * 1000));
}
}
}
}
=head1 EXAMPLES
=head2 Cooperative Parallelism
This example demonstrates how to perform multiple HTTP requests concurrently on a single interpretation thread.
use Acme::Parataxis;
# ... (See My::HTTP implementation in INTEGRATING SYNCHRONOUS MODULES) ...
Acme::Parataxis::run(sub {
my $http = My::HTTP->new(verify_SSL => 0);
my @urls = qw[http://example.com http://perl.org];
# Spawn tasks for each URL
my @futures = map {
lib/Acme/Parataxis.pod view on Meta::CPAN
}
});
$c->call( ); # Prime consumer
$p->call( ); # Start producer
=head1 BEST PRACTICES & GOTCHAS
=over
=item * B<Avoid blocking syscalls:> Never call blocking C<sleep( )> or C<sysread( )> on the main interpretation thread.
Always use the C<await_*> equivalents to offload work to the pool.
=item * B<Thread Safety:> While Perl code remains single-threaded, background tasks run on separate OS threads. Shared
C-level data (if accessed via FFI) must be mutex-protected.
=item * B<Stack Limits:> Each fiber is allocated a 512KB stack by default. This is more than sufficient for most
Perl code and allows for high concurrency with a small memory footprint. Extremely deep recursion or massive regex
backtracking might still hit limits.
=item * B<Efficiency:> The native thread pool is initialized dynamically upon the first asynchronous request. It
starts with a small "seed" pool and grows on demand up to the configured limit. Worker threads use condition
variables to sleep efficiently when idle, ensuring near-zero CPU usage when no background tasks are pending.
=item * B<Reference Cycles:> Be careful when passing fiber objects into their own closures, as this can create
memory leaks.
=back
=head1 GORY TECHNICAL DETAILS
=head2 Architectural Inspiration
lib/Acme/Parataxis.pod view on Meta::CPAN
of fibers as the primary unit of execution and its deterministic cooperative scheduling.
=head2 Stack Virtualization
On Unix-like systems, we use C<ucontext.h> to manage stack and register state. On Windows, we leverage the native
C<Fiber API>. In both cases, we perform heart surgery on the Perl interpreter by manually teleporting its internal
global pointers (the C<PL_*> variables) between contexts.
=head2 Shared CVs and Pad Virtualization
A significant challenge in Perl green threads is the shared nature of PadLists and the global C<CvDEPTH> counter. In
debug builds of Perl, calling a shared subroutine from multiple fibers can trigger internal assertions (like
C<AvFILLp(av) == -1>). Parataxis includes a specialized workaround that surgically cleans the next landing pad before
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
A fiber being C<is_done( )> simply means its Perl code has finished executing. The underlying C-level memory (stacks,
context, etc.) is not immediately freed until the C<Acme::Parataxis> object is destroyed or the runtime performs its
final C<cleanup( )>. This is why you might see memory usage stay flat even after a fiber finishes, until the garbage
collector finally catches up with the object.
t/004_synopsis.t view on Meta::CPAN
# This is the synopsis for Acme::Parataxis but verbose
Acme::Parataxis::run(
sub {
diag 'Main fiber started (FID: ' . Acme::Parataxis->current_fid . ')';
# Spawn background workers
diag 'Spawning Task 1 (Sleep)...';
my $f1 = Acme::Parataxis->spawn(
sub {
diag 'Task 1 started (FID: ' . Acme::Parataxis->current_fid . ')';
diag 'Task 1: Sleeping in a native thread pool (1000ms)...';
Acme::Parataxis->await_sleep(1000);
diag 'Task 1: Ah! What a nice nap...';
return 42;
}
);
diag 'Spawning Task 2 (I/O dummy)...';
my $f2 = Acme::Parataxis->spawn(
sub {
diag 'Task 2 started (FID: ' . Acme::Parataxis->current_fid . ')';
diag 'Task 2: Performing I/O simulation...';
# await_read/write for non-blocking socket handling
return 'I/O Done';
}
);
# Block current fiber until results are ready (without blocking the thread)
diag 'Main: Waiting for Task 1 result...';
my $res1 = $f1->await();
is $res1, 42, 'Task 1 returned expected value';
diag "Main: Task 1 result: $res1";
diag 'Main: Waiting for Task 2 result...';
my $res2 = $f2->await();
is $res2, 'I/O Done', 'Task 2 returned expected value';
diag "Main: Task 2 result: $res2";
}
);
t/006_parallel.t view on Meta::CPAN
use Test2::V1 -ipP;
use blib;
use Acme::Parataxis;
use Time::HiRes qw[time];
$|++;
Acme::Parataxis::run(
sub {
my $start_time = time();
diag "Starting parallel sleeps at $start_time...";
# Check thread pool size
my $pool_size = Acme::Parataxis::get_thread_pool_size();
diag "Detected thread pool size: $pool_size";
# Spawn only as many tasks as we have threads (up to 3) to ensure they run in parallel.
# If pool size is 2, running 3 tasks takes 2 seconds, failing the 1.8s test.
my $num_tasks = $pool_size;
$num_tasks = 3 if $num_tasks > 3;
$num_tasks = 1 if $num_tasks < 1;
diag "Spawning $num_tasks parallel tasks...";
my @futures;
for my $i ( 1 .. $num_tasks ) {
push @futures, Acme::Parataxis->spawn(
sub {
my $id = $i; # Closure capture
( run in 2.381 seconds using v1.01-cache-2.11-cpan-c966e8aa7e8 )