Acme-Parataxis
view release on metacpan or search on metacpan
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.
*/
#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0601
#endif
#else
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE 600
#endif
#ifndef _BSD_SOURCE
#define _BSD_SOURCE
#endif
#ifndef _DEFAULT_SOURCE
#define _DEFAULT_SOURCE
#endif
#ifdef __APPLE__
#ifndef _DARWIN_C_SOURCE
#define _DARWIN_C_SOURCE
#endif
#endif
#endif
#define PERL_NO_GET_CONTEXT
#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 <signal.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <ucontext.h>
#include <unistd.h>
#include <sys/mman.h>
#if defined(__APPLE__) || defined(__FreeBSD__)
#include <sys/sysctl.h>
lib/Acme/Parataxis.c view on Meta::CPAN
/** @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);
#ifndef _WIN32
static void install_stack_guard(void);
#endif
/*
* Assembly-based coroutine context switching.
*
* glibc's swapcontext() saves/restores the signal mask (rt_sigprocmask) on every context switch which dominates the
* cost of fiber switches. On x86_64 we instead switch with a tiny assembly routine that only saves the callee-saved
* registers and the stack pointer, avoiding the syscall entirely. All other platforms keep the portable ucontext path
* for now...
*/
#if defined(__x86_64__) && !defined(_WIN32) && defined(__ELF__)
#define USE_ASM_CORO 1
#endif
typedef struct para_fiber_t para_fiber_t;
/* C-level entry point invoked when a freshly created fiber starts running. */
void para_entry_point(para_fiber_t * c);
#if defined(USE_ASM_CORO)
/**
* @brief Raw register-only context switch.
*
* Saves the callee-saved registers and the current stack pointer into *from, restores them from *to, then returns
* (popping the return address off the target stack). A freshly created fiber's stack is pre-arranged so that the return
* address lands in para_trampoline.
*
* @param from Pointer to the storage slot holding the current stack pointer.
* @param to Pointer to the storage slot holding the target stack pointer.
*/
extern void para_coro_switch(void ** from, void ** to);
/** @brief Initial jump target for brand-new fiber stacks. */
extern void para_trampoline(void);
__asm__(
".text\n"
".p2align 4\n"
".globl para_coro_switch\n"
".type para_coro_switch, @function\n"
"para_coro_switch:\n"
" pushq %rbx\n"
" pushq %rbp\n"
" pushq %r12\n"
" pushq %r13\n"
" pushq %r14\n"
" pushq %r15\n"
" movq %rsp, (%rdi)\n"
" movq (%rsi), %rsp\n"
" popq %r15\n"
" popq %r14\n"
" popq %r13\n"
" popq %r12\n"
" popq %rbp\n"
" popq %rbx\n"
" ret\n"
".size para_coro_switch, .-para_coro_switch\n"
".p2align 4\n"
".globl para_trampoline\n"
".type para_trampoline, @function\n"
"para_trampoline:\n"
" popq %rdi\n"
" call para_entry_point\n"
" ud2\n"
".size para_trampoline, .-para_trampoline\n");
#endif /* USE_ASM_CORO */
/**
* @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();
lib/Acme/Parataxis.c view on Meta::CPAN
#endif
}
/**
* @struct para_fiber_t
* @brief The complete execution context of a Perl Fiber.
*
* This structure encapsulates both the OS-level register state (via context) and the entire internal state of the Perl
* interpreter required to pause and resume execution of Perl code.
*/
typedef struct para_fiber_t {
coro_handle_t context; /**< OS-specific context handle */
#ifndef _WIN32
void * stack_p; /**< Pointer to dynamically allocated fiber stack (Unix only) */
size_t stack_sz; /**< Size of the allocated stack (Unix only) */
#ifdef USE_ASM_CORO
void * rsp; /**< Saved stack pointer for the assembly switch (Unix x86_64) */
#endif
#endif
/*
* Perl Interpreter State Pointers.
* These must be saved and restored during every context switch.
*/
PERL_SI * si; /**< Current Stack Info (tracks recursion and eval frames) */
AV * curstack; /**< The active Argument Stack (AV*) */
SV * mainstack; /**< Main argument stack target for exit()/POPSTACK_TO (per-fiber) */
SSize_t stack_sp_offset; /**< Stack Pointer offset from stack base */
I32 * markstack; /**< Base of the Mark Stack (tracks list start points) */
I32 * markstack_ptr; /**< Current pointer into the Mark Stack */
I32 * markstack_max; /**< Limit of the Mark Stack */
I32 * scopestack; /**< Base of the Scope Stack (tracks block nesting) */
I32 scopestack_ix; /**< Current index in the Scope Stack */
I32 scopestack_max; /**< Limit of the Scope Stack */
#ifdef DEBUGGING
const char ** scopestack_name; /**< DEBUGGING-only names parallel to scopestack */
#endif
ANY * savestack; /**< Base of the Save Stack (tracks local/my variables for cleanup) */
I32 savestack_ix; /**< Current index in the Save Stack */
I32 savestack_max; /**< Limit of the Save Stack */
SV ** tmps_stack; /**< Base of the Mortal Stack (tracks SVs needing refcnt decrement) */
I32 tmps_ix; /**< Current index in the Mortal Stack */
I32 tmps_floor; /**< Current floor of the Mortal Stack */
I32 tmps_max; /**< Limit of the Mortal Stack */
JMPENV * top_env; /**< Pointer to the top exception environment (eval/die buffers) */
COP * curcop; /**< Current Op Pointer (location in the source/bytecode) */
OP * op; /**< Current Operation being executed */
PAD * comppad; /**< Current lexical Pad (variable storage) */
SV ** curpad; /**< Array pointer to the current lexical Pad */
PMOP * curpm; /**< Current pattern match state */
PMOP * curpm_under; /**< Current pattern match state under */
PMOP * reg_curpm; /**< Current regex match state */
GV * defgv; /**< The $_ global */
GV * last_in_gv; /**< GV used in last <FH> */
SV * rs; /**< The $/ global */
GV * ofsgv; /**< The $, global */
SV * ors_sv; /**< The $\ global */
GV * defoutgv; /**< The default output filehandle */
HV * curstash; /**< Current package stash */
HV * defstash; /**< Default package stash */
SV * errors; /**< Outstanding queued errors */
SV * user_cv; /**< The Perl sub/coderef this fiber is running */
SV * self_ref; /**< The Acme::Parataxis Perl object wrapper */
SV * transfer_data; /**< Arguments or return values passed during yield/transfer */
int id; /**< Numeric ID of this fiber */
int finished; /**< Flag: 1 if the fiber has completed its entry_point */
int started; /**< Flag: 1 once the fiber has actually begun running */
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 */
#ifdef _WIN32
/* exit() interception (x64 Windows only). The CRT longjmp cannot cross stacks, so an exit() thrown on a fiber stack
* must be captured there and re-raised on the caller's stack (see parataxis_pp_exit / para_entry_point /
* coro_call). exit_pending is set when exit() was called by this fiber or a fiber it called; exit_status carries
* the requested exit code. */
int exit_pending;
int exit_status;
#endif
} 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 */
///@}
/**
* @union value_t
* @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_*) */
lib/Acme/Parataxis.c view on Meta::CPAN
return coro_yield(&PL_sv_undef);
}
return &PL_sv_undef;
}
/**
* @brief Restores subroutine call depths and cleans argument pads.
*
* This function iterates the context stack and restores CvDEPTH for active subroutines in two passes to safely handle
* recursive calls.
*
* Pass 1: Restores CvDEPTH for all active frames.
* Pass 2: Surgicaly cleans Slot 0 of the *next* pad depth for each CV.
*
* @param to The fiber being resumed.
*/
static void _activate_current_depths(pTHX_ para_fiber_t * to) {
PERL_SI * si = to->si;
if (!si || !si->si_cxstack)
return;
/* Pass 1: Restore CvDEPTH for all active frames */
for (I32 i = 0; i <= si->si_cxix; i++) {
PERL_CONTEXT * cx = &(si->si_cxstack[i]);
if (CxTYPE(cx) == CXt_SUB || CxTYPE(cx) == CXt_FORMAT) {
CV * cv = cx->blk_sub.cv;
if (cv && SvTYPE((SV *)cv) == SVt_PVCV)
CvDEPTH(cv) = cx->blk_sub.olddepth + 1;
}
}
/* Pass 2: Clean the landing pads for the NEXT call in each CV */
for (I32 i = 0; i <= si->si_cxix; i++) {
PERL_CONTEXT * cx = &(si->si_cxstack[i]);
if (CxTYPE(cx) == CXt_SUB || CxTYPE(cx) == CXt_FORMAT) {
CV * cv = cx->blk_sub.cv;
if (cv && SvTYPE((SV *)cv) == SVt_PVCV) {
PADLIST * pl = CvPADLIST(cv);
I32 next_depth = CvDEPTH(cv) + 1;
if (pl && next_depth <= PadlistMAX(pl)) {
AV * next_pad = (AV *)PadlistARRAY(pl)[next_depth];
if (next_pad && SvTYPE(next_pad) == SVt_PVAV) {
SV ** array = AvARRAY(next_pad);
if (array && AvMAX(next_pad) >= 0) {
SV * args = array[0];
if (args && SvTYPE(args) == SVt_PVAV) {
AvFILLp((AV *)args) = -1;
AvREAL_off((AV *)args);
}
}
}
}
}
}
}
}
/**
* @brief Swaps the internal Perl Interpreter state pointers.
*
* This is the core of the fiber implementation. It manually saves all global pointers that define the "state" of the
* Perl virtual machine for the current context and restores them for the target context.
*
* @param from Context being paused.
* @param to Context being resumed.
*/
void swap_perl_state(para_fiber_t * from, para_fiber_t * to) {
dTHX;
/* Save current state into 'from' context */
from->si = PL_curstackinfo;
// The Argument Stack (Main Perl stack)
from->curstack = PL_curstack;
from->mainstack = PL_mainstack;
from->stack_sp_offset = PL_stack_sp - PL_stack_base;
// The Mark Stack (Tracks where lists begin on the argument stack)
from->markstack = PL_markstack;
from->markstack_ptr = PL_markstack_ptr;
from->markstack_max = PL_markstack_max;
// The Scope Stack (Tracks block entry/exit for cleanup)
from->scopestack = PL_scopestack;
from->scopestack_ix = PL_scopestack_ix;
from->scopestack_max = PL_scopestack_max;
#ifdef DEBUGGING
from->scopestack_name = PL_scopestack_name;
#endif
// The Save Stack (Tracks 'local' variables and destructors)
from->savestack = PL_savestack;
from->savestack_ix = PL_savestack_ix;
from->savestack_max = PL_savestack_max;
// The Mortal Stack (Tracks temporary SVs that need decrementing)
from->tmps_stack = PL_tmps_stack;
from->tmps_ix = PL_tmps_ix;
from->tmps_floor = PL_tmps_floor;
from->tmps_max = PL_tmps_max;
// Exception Environment (setjmp/longjmp buffers for eval/die)
from->top_env = PL_top_env;
// Op and Pad pointers (Where we are in the bytecode)
from->curcop = PL_curcop;
from->op = PL_op;
from->comppad = PL_comppad;
from->curpad = PL_curpad;
from->curpm = PL_curpm;
from->curpm_under = PL_curpm_under;
from->reg_curpm = PL_reg_curpm;
from->defgv = PL_defgv;
from->last_in_gv = PL_last_in_gv;
from->rs = PL_rs;
from->ofsgv = PL_ofsgv;
from->ors_sv = PL_ors_sv;
from->defoutgv = PL_defoutgv;
from->curstash = PL_curstash;
from->defstash = PL_defstash;
from->errors = PL_errors;
/* Load target state from 'to' context */
PL_curstackinfo = to->si;
PL_curstack = to->curstack;
PL_mainstack = to->mainstack;
// Re-calculate stack bounds based on the new array (AV)
PL_stack_base = AvARRAY(PL_curstack);
PL_stack_max = PL_stack_base + AvMAX(PL_curstack);
PL_stack_sp = PL_stack_base + to->stack_sp_offset;
AvFILLp(PL_curstack) = to->stack_sp_offset; // Keep stack AV metadata synced
PL_markstack = to->markstack;
PL_markstack_ptr = to->markstack_ptr;
PL_markstack_max = to->markstack_max;
PL_scopestack = to->scopestack;
PL_scopestack_ix = to->scopestack_ix;
PL_scopestack_max = to->scopestack_max;
#ifdef DEBUGGING
PL_scopestack_name = to->scopestack_name;
#endif
PL_savestack = to->savestack;
PL_savestack_ix = to->savestack_ix;
PL_savestack_max = to->savestack_max;
PL_tmps_stack = to->tmps_stack;
PL_tmps_ix = to->tmps_ix;
PL_tmps_floor = to->tmps_floor;
PL_tmps_max = to->tmps_max;
PL_top_env = to->top_env;
PL_curcop = to->curcop;
PL_op = to->op;
PL_comppad = to->comppad;
PL_curpm = to->curpm;
PL_curpm_under = to->curpm_under;
PL_reg_curpm = to->reg_curpm;
PL_defgv = to->defgv;
PL_last_in_gv = to->last_in_gv;
PL_rs = to->rs;
PL_ofsgv = to->ofsgv;
PL_ors_sv = to->ors_sv;
PL_defoutgv = to->defoutgv;
PL_curstash = to->curstash;
PL_defstash = to->defstash;
PL_errors = to->errors;
if (PL_comppad)
PL_curpad = AvARRAY(PL_comppad);
else
PL_curpad = to->curpad;
// Restore CvDEPTH and clean landing pads
_activate_current_depths(aTHX_ to);
}
/**
* @brief Restores a context's saved Perl interpreter state in-place.
*
* Loads every saved interpreter global from @p to without performing an OS context switch. Used when a fiber longjmps
* out of the system (e.g. via exit()) so that the top-level main context's perl-owned stacks are current again before
* the jump propagates to perl_run/perl_destruct.
*
* Unlike swap_perl_state this deliberately leaves PL_top_env alone: the caller manages the jump environment so the
* rethrow continues past the wrapper that caught the exit.
*
* @param to The context whose saved state should become current.
*/
void restore_perl_state(para_fiber_t * to) {
dTHX;
PL_curstackinfo = to->si;
PL_curstack = to->curstack;
PL_mainstack = to->mainstack;
PL_stack_base = AvARRAY(PL_curstack);
PL_stack_max = PL_stack_base + AvMAX(PL_curstack);
PL_stack_sp = PL_stack_base + to->stack_sp_offset;
AvFILLp(PL_curstack) = to->stack_sp_offset;
PL_markstack = to->markstack;
PL_markstack_ptr = to->markstack_ptr;
PL_markstack_max = to->markstack_max;
PL_scopestack = to->scopestack;
PL_scopestack_ix = to->scopestack_ix;
PL_scopestack_max = to->scopestack_max;
#ifdef DEBUGGING
PL_scopestack_name = to->scopestack_name;
#endif
PL_savestack = to->savestack;
PL_savestack_ix = to->savestack_ix;
PL_savestack_max = to->savestack_max;
PL_tmps_stack = to->tmps_stack;
PL_tmps_ix = to->tmps_ix;
PL_tmps_floor = to->tmps_floor;
PL_tmps_max = to->tmps_max;
PL_curcop = to->curcop;
PL_op = to->op;
PL_comppad = to->comppad;
PL_curpm = to->curpm;
PL_curpm_under = to->curpm_under;
PL_reg_curpm = to->reg_curpm;
PL_defgv = to->defgv;
PL_last_in_gv = to->last_in_gv;
PL_rs = to->rs;
PL_ofsgv = to->ofsgv;
PL_ors_sv = to->ors_sv;
PL_defoutgv = to->defoutgv;
PL_curstash = to->curstash;
PL_defstash = to->defstash;
PL_errors = to->errors;
if (PL_comppad)
PL_curpad = AvARRAY(PL_comppad);
else
PL_curpad = to->curpad;
}
lib/Acme/Parataxis.c view on Meta::CPAN
while (root->si_prev)
root = root->si_prev;
si->si_prev = root;
}
c->si = si;
}
/**
* @brief Resets a fiber's Perl control stacks back to their initial state.
*
* Called when a fiber is destroyed and its memory parked in the reuse cache. The stacks keep whatever size they grew to
* and are simply re-initialized; resetting the bounds conservatively back to the starting size just means perl grows
* them again if a new fiber recurses deep.
*
* @param c The fiber context to reset.
*/
static void reset_perl_stacks(pTHX_ para_fiber_t * c) {
I32 sz = FIBER_STACK_DEPTH;
PERL_SI * si = c->si;
/* Release any mortal SVs still parked on the tmps stack */
if (c->tmps_stack) {
for (I32 i = 0; i <= c->tmps_ix; i++) {
SV * sv = c->tmps_stack[i];
if (sv && sv != &PL_sv_undef)
SvREFCNT_dec(sv);
}
}
if (c->curstack) {
/* Do NOT av_clear the fiber's argument stack here. Slots below the last-saved stack pointer may already have
* been popped and freed during earlier resume/yield cycles, so clearing them would double-decrement live SVs.
* The AV is reused verbatim; new pushes overwrite the stale slots before they are ever read. */
AvARRAY(c->curstack)[0] = &PL_sv_undef;
AvFILLp(c->curstack) = 0;
}
c->stack_sp_offset = 0;
if (si) {
si->si_cxix = -1;
si->si_stack = c->curstack;
}
/* exit()/POPSTACK_TO(PL_mainstack) inside this fiber must stop at the fiber's own argument stack instead of
* unwinding the shared main contexts whose pads are not current while a fiber runs. */
c->mainstack = (SV *)c->curstack;
c->markstack_ptr = c->markstack;
*c->markstack_ptr = 0;
c->markstack_max = c->markstack + sz;
c->scopestack_ix = 0;
c->scopestack_max = sz;
c->savestack_ix = 0;
c->savestack_max = sz;
c->tmps_ix = -1;
c->tmps_floor = -1;
c->tmps_max = sz;
/* Inherit globals from the current interpreter state */
c->curcop = PL_curcop;
c->op = PL_op;
c->top_env = PL_top_env;
c->curpm = PL_curpm;
c->curpm_under = PL_curpm_under;
c->reg_curpm = NULL;
c->defgv = PL_defgv;
c->last_in_gv = PL_last_in_gv;
c->rs = PL_rs;
c->ofsgv = PL_ofsgv;
c->ors_sv = PL_ors_sv;
c->defoutgv = PL_defoutgv;
c->curstash = PL_curstash;
c->defstash = PL_defstash;
c->errors = PL_errors;
c->comppad = NULL;
c->curpad = NULL;
}
/**
* @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. The control stacks share a single allocation block for speed.
*
* @param c The fiber context to initialize.
*/
void init_perl_stacks(para_fiber_t * c) {
dTHX;
alloc_perl_stacks(aTHX_ c);
if (!c->si)
return;
// Allocate Argument Stack (AV)
c->curstack = newAV();
AvREAL_off(c->curstack); // Stacks do not 'own' their elements in the refcnt sense
av_extend(c->curstack, 128);
/* The control stacks are uninitialized beyond the SI header, so mark the tmps stack empty before reset runs its
* release loop. */
c->tmps_ix = -1;
c->tmps_floor = -1;
reset_perl_stacks(aTHX_ c);
}
/**
* @brief Frees the Perl stacks.
*
* @param c The fiber context whose stacks should be released.
*/
static void free_perl_stacks(pTHX_ para_fiber_t * c) {
if (c->curstack) {
/* Skip av_clear: stale slots may already be freed (see reset_perl_stacks). */
SvREFCNT_dec((SV *)c->curstack);
c->curstack = NULL;
}
if (c->si) {
Safefree(c->si->si_cxstack);
lib/Acme/Parataxis.c view on Meta::CPAN
/*
* Lay out the fresh stack for the assembly switch. When the switch routine resumes this context it first pops the
* six dummy saved registers, then "returns" into para_trampoline. The trampoline pops the fiber pointer and
* tail-calls para_entry_point with the correct ABI stack alignment.
*/
void ** slot = (void **)((char *)c->stack_p + c->stack_sz);
slot -= 8; /* 6 saved regs + return address + fiber pointer */
for (int i = 0; i < 6; i++)
slot[i] = NULL;
slot[6] = (void *)¶_trampoline;
slot[7] = c;
c->rsp = slot;
#else
getcontext(&c->context);
c->context.uc_stack.ss_sp = c->stack_p;
c->context.uc_stack.ss_size = c->stack_sz;
c->context.uc_link = &main_context.context;
makecontext(&c->context, (void (*)())posix_entry, 1, idx);
#endif
#endif
}
/**
* @brief Allocates and prepares a new Fiber context.
*
* @param user_code Coderef to execute in the fiber.
* @param self_ref Acme::Parataxis object to notify on completion.
* @return int Unique ID of the new fiber, or negative on error.
*/
DLLEXPORT int create_fiber(SV * user_code, SV * self_ref) {
dTHX;
if (!free_slots_seeded) {
for (int i = 0; i < MAX_FIBERS; i++)
free_slots[i] = MAX_FIBERS - 1 - i;
free_slot_count = MAX_FIBERS;
free_slots_seeded = 1;
}
int idx;
if (free_slot_count > 0) {
idx = free_slots[--free_slot_count];
}
else {
/* Safety net: scan for a slot if the free list is ever exhausted. */
idx = -1;
for (int i = 0; i < MAX_FIBERS; i++) {
if (fibers[i] == NULL && !fiber_destroyed[i]) {
idx = i;
break;
}
}
}
if (idx == -1)
return -2;
para_fiber_t * c = NULL;
#ifndef _WIN32
if (fiber_cache_count > 0)
c = fiber_cache[--fiber_cache_count];
#endif
if (c) {
/* Recycle a parked fiber context: re-inherit globals, clear pads */
reset_perl_stacks(aTHX_ c);
}
else {
c = (para_fiber_t *)malloc(sizeof(para_fiber_t));
if (!c)
return -3;
memset(c, 0, sizeof(para_fiber_t));
/* Initialize Perl stacks */
init_perl_stacks(c);
if (!c->si) {
free(c);
return -3;
}
#ifndef _WIN32
c->stack_sz = FIBER_STACK_SZ;
c->stack_p = alloc_fiber_stack(c->stack_sz);
if (!c->stack_p) {
free_perl_stacks(aTHX_ c);
free(c);
return -3;
}
#else
c->context = NULL;
#endif
}
/* Reset the coderef's call depth so the fiber starts clean */
if (user_code && user_code != &PL_sv_undef)
force_depth_zero(user_code);
c->user_cv = user_code;
if (user_code && user_code != &PL_sv_undef)
SvREFCNT_inc(user_code);
c->self_ref = self_ref;
if (self_ref && self_ref != &PL_sv_undef)
SvREFCNT_inc(self_ref);
c->id = idx;
c->parent_id = -1;
c->last_sender = -1;
c->finished = 0;
c->started = 0;
#ifdef _WIN32
c->exit_pending = 0;
c->exit_status = 0;
#endif
c->transfer_data = &PL_sv_undef;
fibers[idx] = c;
arm_fiber_context(c, idx);
return idx;
}
/**
* @brief Resumes a fiber (asymmetric call).
*
* Suspends the caller and switches execution to the specified fiber. Sets the caller as the 'parent' for future yields.
*
* @param fiber_id Fiber ID to call.
* @param args Perl SV (usually arrayref) to pass as arguments to the fiber.
* @return SV* Result yielded by the fiber.
lib/Acme/Parataxis.c view on Meta::CPAN
* fiber can be woken by a stale completion targeting this id (released via free_job_slot). */
fiber_destroyed[fiber_id] = 1;
}
else {
if (free_slot_count < MAX_FIBERS)
free_slots[free_slot_count++] = fiber_id;
}
/* Unwind pads */
if (c->si)
_clear_pads_in_stack(aTHX_ c->si);
/* Release Perl references */
if (c->user_cv && c->user_cv != &PL_sv_undef) {
SvREFCNT_dec(c->user_cv);
c->user_cv = NULL;
}
if (c->self_ref && c->self_ref != &PL_sv_undef) {
SvREFCNT_dec(c->self_ref);
c->self_ref = NULL;
}
if (c->transfer_data && c->transfer_data != &PL_sv_undef) {
SvREFCNT_dec(c->transfer_data);
c->transfer_data = NULL;
}
/* Early exit if Perl is already shutting down */
if (PL_dirty) {
#ifndef _WIN32
if (c->stack_p)
free_fiber_stack(c->stack_p, c->stack_sz);
#endif
free_perl_stacks(aTHX_ c);
free(c);
return;
}
#ifdef _WIN32
if (c->context)
DeleteFiber(c->context);
#else
/* Park the whole context (Perl stacks + OS stack) for reuse */
if (fiber_cache_count < MAX_FIBER_CACHE) {
reset_perl_stacks(aTHX_ c);
fiber_cache[fiber_cache_count++] = c;
return;
}
if (c->stack_p) {
free_fiber_stack(c->stack_p, c->stack_sz);
c->stack_p = NULL;
}
#endif
free_perl_stacks(aTHX_ c);
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;
#ifdef _WIN32
/* Restore the original exit op so any exit() during global destruction (after this DLL could be unmapped) behaves
* like a plain perl exit. */
if (parataxis_saved_pp_exit && PL_ppaddr[OP_EXIT] == parataxis_pp_exit)
PL_ppaddr[OP_EXIT] = parataxis_saved_pp_exit;
#endif
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
/* Wake any workers blocked in select() so they observe threads_keep_running = 0 */
if (shutdown_pipe[1] >= 0) {
char byte = 1;
ssize_t ignored = write(shutdown_pipe[1], &byte, 1);
(void)ignored;
}
/* 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])
destroy_coro(i);
#ifndef _WIN32
#ifdef __linux__
while (stack_cache_count > 0)
munmap((char *)stack_cache[--stack_cache_count] - fiber_guard_sz, FIBER_STACK_SZ + fiber_guard_sz);
#else
while (stack_cache_count > 0)
munmap((char *)stack_cache[--stack_cache_count] - fiber_guard_sz, FIBER_STACK_SZ + fiber_guard_sz);
#endif
if (shutdown_pipe[1] >= 0) {
close(shutdown_pipe[0]);
close(shutdown_pipe[1]);
shutdown_pipe[0] = shutdown_pipe[1] = -1;
}
#endif
if (main_context.transfer_data && main_context.transfer_data != &PL_sv_undef) {
SvREFCNT_dec(main_context.transfer_data);
main_context.transfer_data = &PL_sv_undef;
}
}
( run in 2.060 seconds using v1.01-cache-2.11-cpan-4716db65a44 )