Acme-Parataxis
view release on metacpan or search on metacpan
builder/Acme/Parataxis/Builder.pm view on Meta::CPAN
my $build = Affix::Build->new(
name => 'parataxis',
flags => {
cflags => join( ' ',
( $Config::Config{ccflags}, '-std=c11', map { '-I' . $_ } ( File::Spec->catdir( $Config::Config{archlibexp}, 'CORE' ), 'src' ) )
),
ldflags => ( $^O eq 'MSWin32' ) ?
( '-L' .
File::Spec->catdir( $Config::Config{archlibexp}, 'CORE' ) . ' -l' .
( $Config::Config{libperl} =~ s/^lib//r =~ s/\.a$//r =~ s/\.lib$//r ) .
' -lws2_32' ) : ( $^O eq 'darwin' ? '-undefined dynamic_lookup' : '' )
},
build_dir => $arch_dir,
clean => 0
);
# Add the C source file to be compiled. It's expected to be in 'lib/Acme/'
$build->add('lib/Acme/Parataxis.c');
say "Compiling and linking...";
$build->compile_and_link();
say "Build complete.";
builder/Acme/Parataxis/Builder.pm view on Meta::CPAN
}
method step_clean() { rmtree( $_, $verbose ) for qw[blib temp]; 0 }
method step_install() {
$self->step_build() unless -d 'blib';
install(
[ from_to => $install_paths->install_map,
verbose => $verbose,
dry_run => $dry_run,
uninstall_shadows => $uninst,
skip => undef,
always_copy => 1
]
);
0;
}
method step_realclean () { rmtree( $_, $verbose ) for qw[blib temp Build _build_params MYMETA.yml MYMETA.json]; 0 }
method step_test() {
$self->step_build() unless -d 'blib';
require TAP::Harness::Env;
eg/symmetric.pl view on Meta::CPAN
# Symmetric coroutines (Producer/Consumer)
my ( $producer, $consumer );
$producer = Acme::Parataxis->new(
code => sub {
for my $item (qw[Apple Banana Cherry]) {
say "Producer: Created $item. Transferring to Consumer...";
$consumer->transfer($item);
say 'Producer: Consumer gave control back. Moving to next item.';
}
say 'Producer: Out of items. Telling consumer to finish.';
$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...';
lib/Acme/Parataxis.c view on Meta::CPAN
/**
* @brief Retrieves the result of a completed job as a Perl SV.
*
* @param idx The job index in the queue.
* @return SV* A mortalized Perl SV containing the result (IV).
*/
DLLEXPORT SV * get_job_result(int idx) {
dTHX;
if (idx < 0 || idx >= MAX_JOBS)
return &PL_sv_undef;
SV * res = &PL_sv_undef;
LOCK(queue_lock);
if (job_slots[idx].status == JOB_DONE || job_slots[idx].status == JOB_BUSY) {
if (job_slots[idx].type == TASK_SLEEP || job_slots[idx].type == TASK_GET_CPU ||
job_slots[idx].type == TASK_READ || job_slots[idx].type == TASK_WRITE) {
res = newSViv(job_slots[idx].output.i);
sv_2mortal(res);
}
}
UNLOCK(queue_lock);
return res;
lib/Acme/Parataxis.c view on Meta::CPAN
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
* threshold is reached.
*
* @return SV* Result of the yield, or undef if no yield occurred.
*/
DLLEXPORT SV * maybe_yield() {
dTHX;
preempt_count++;
if (preempt_threshold > 0 && preempt_count >= preempt_threshold) {
preempt_count = 0;
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.
lib/Acme/Parataxis.c view on Meta::CPAN
// Use Newxz to ensure the context stack is zeroed.
Newxz(c->si->si_cxstack, c->si->si_cxmax, PERL_CONTEXT);
c->si->si_cxix = -1;
c->si->si_type = PERLSI_MAIN;
// 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);
// Initialize stack with a dummy undef at index 0, matching Perl's main stack
AvARRAY(c->curstack)[0] = &PL_sv_undef;
AvFILLp(c->curstack) = 0;
c->stack_sp_offset = 0;
// Link the SI to the AV. Perl uses this linkage during stack unwinding.
c->si->si_stack = c->curstack;
// Allocate Control Stacks
I32 sz = 2048; /* Recursion depth support */
Newx(c->markstack, sz, I32);
lib/Acme/Parataxis.c view on Meta::CPAN
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;
lib/Acme/Parataxis.c view on Meta::CPAN
*
* 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)
return &PL_sv_undef;
para_fiber_t * self = fibers[current_fiber_id];
int parent = self->parent_id;
if (parent != -1 && (!fibers[parent] || fibers[parent]->finished))
parent = self->last_sender;
else if (parent == -1)
parent = self->last_sender;
if (parent >= 0 && (!fibers[parent] || fibers[parent]->finished))
parent = -1;
para_fiber_t * caller = (parent == -1) ? &main_context : fibers[parent];
/* Pass return value to caller */
if (caller->transfer_data != ret_val) {
if (caller->transfer_data && caller->transfer_data != &PL_sv_undef)
SvREFCNT_dec(caller->transfer_data);
caller->transfer_data = ret_val;
if (ret_val && ret_val != &PL_sv_undef)
SvREFCNT_inc(ret_val);
}
perform_switch(parent);
/* Retrieve value passed back during resume */
SV * res = self->transfer_data;
self->transfer_data = &PL_sv_undef;
if (res && res != &PL_sv_undef)
sv_2mortal(res);
return res;
}
/**
* @brief Entry point function for all new fibers.
*
* Sets up the Perl environment (ENTER/SAVETMPS), unpacks arguments,
* calls the user coderef, handles results/errors, and manages the
* fiber's completion lifecycle.
lib/Acme/Parataxis.c view on Meta::CPAN
if (svp)
XPUSHs(*svp);
}
}
PUTBACK;
/* Execute the Perl sub */
int count = call_sv(c->user_cv, G_SCALAR | G_EVAL);
SPAGAIN;
SV * ret_val = &PL_sv_undef;
if (count == 1)
ret_val = POPs;
PUTBACK;
c->finished = true;
/* Cleanup transfer data and store result */
if (c->transfer_data && c->transfer_data != &PL_sv_undef) {
SvREFCNT_dec(c->transfer_data);
c->transfer_data = &PL_sv_undef;
}
if (ret_val && ret_val != &PL_sv_undef) {
SvREFCNT_inc(ret_val);
c->transfer_data = ret_val;
}
/* Update the Perl-level Acme::Parataxis object */
if (c->self_ref && SvROK(c->self_ref)) {
dSP;
ENTER;
SAVETMPS;
PUSHMARK(SP);
lib/Acme/Parataxis.c view on Meta::CPAN
PUTBACK;
call_method("set_result", G_DISCARD);
}
FREETMPS;
LEAVE;
}
FREETMPS;
LEAVE;
/* Final yield back to caller */
coro_yield(c->transfer_data ? c->transfer_data : &PL_sv_undef);
/* Loop indefinitely if resumed after finish */
while (1)
coro_yield(&PL_sv_undef);
}
#ifdef _WIN32
/** @brief Windows fiber callback wrapper. */
static void WINAPI fiber_entry(void * param) { entry_point((para_fiber_t *)param); }
#else
/** @brief POSIX makecontext callback wrapper. */
static void posix_entry(int fiber_id) { entry_point(fibers[fiber_id]); }
#endif
lib/Acme/Parataxis.c view on Meta::CPAN
break;
}
}
if (idx == -1)
return -2;
para_fiber_t * c = (para_fiber_t *)malloc(sizeof(para_fiber_t));
if (!c)
return -3;
memset(c, 0, sizeof(para_fiber_t));
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->transfer_data = &PL_sv_undef;
fibers[idx] = c;
/* Initialize Perl stacks */
init_perl_stacks(c);
#ifdef _WIN32
c->context = CreateFiber(0, fiber_entry, c);
#else
c->stack_sz = 512 * 1024; // 512KB is plenty for Perl fibers
if (posix_memalign(&c->stack_p, 16, c->stack_sz) != 0) {
lib/Acme/Parataxis.c view on Meta::CPAN
* 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.
*/
DLLEXPORT SV * coro_call(int fiber_id, SV * args) {
dTHX;
if (fiber_id < 0 || fiber_id >= MAX_FIBERS || !fibers[fiber_id] || fibers[fiber_id]->finished)
return &PL_sv_undef;
if (fibers[fiber_id]->transfer_data != args) {
if (fibers[fiber_id]->transfer_data && fibers[fiber_id]->transfer_data != &PL_sv_undef)
SvREFCNT_dec(fibers[fiber_id]->transfer_data);
fibers[fiber_id]->transfer_data = args;
if (args && args != &PL_sv_undef)
SvREFCNT_inc(args);
}
fibers[fiber_id]->parent_id = current_fiber_id;
perform_switch(fiber_id);
if (fibers[fiber_id] && fibers[fiber_id]->finished) {
if (fibers[fiber_id]->transfer_data && fibers[fiber_id]->transfer_data != &PL_sv_undef) {
SvREFCNT_dec(fibers[fiber_id]->transfer_data);
fibers[fiber_id]->transfer_data = &PL_sv_undef;
}
}
para_fiber_t * me = (current_fiber_id == -1) ? &main_context : fibers[current_fiber_id];
SV * res = me->transfer_data;
me->transfer_data = &PL_sv_undef;
if (res && res != &PL_sv_undef)
sv_2mortal(res);
return res;
}
/**
* @brief Transfers control directly to another fiber (symmetric).
*
* Suspends the current fiber and switches directly to the target. No
* parent/child relationship is established.
*
* @param target_id Fiber ID to transfer to.
* @param args Arguments to pass to the target.
* @return SV* Data eventually transferred back to this fiber.
*/
DLLEXPORT SV * coro_transfer(int target_id, SV * args) {
dTHX;
if (target_id < -1 || (target_id >= 0 && (target_id >= MAX_FIBERS || !fibers[target_id])))
return &PL_sv_undef;
if (target_id >= 0 && fibers[target_id]->finished)
return &PL_sv_undef;
para_fiber_t * target = (target_id == -1) ? &main_context : fibers[target_id];
if (target->transfer_data != args) {
if (target->transfer_data && target->transfer_data != &PL_sv_undef)
SvREFCNT_dec(target->transfer_data);
target->transfer_data = args;
if (args && args != &PL_sv_undef)
SvREFCNT_inc(args);
}
perform_switch(target_id);
if (target_id >= 0 && fibers[target_id] && fibers[target_id]->finished) {
if (fibers[target_id]->transfer_data && fibers[target_id]->transfer_data != &PL_sv_undef) {
SvREFCNT_dec(fibers[target_id]->transfer_data);
fibers[target_id]->transfer_data = &PL_sv_undef;
}
}
para_fiber_t * me = (current_fiber_id == -1) ? &main_context : fibers[current_fiber_id];
SV * res = me->transfer_data;
me->transfer_data = &PL_sv_undef;
if (res && res != &PL_sv_undef)
sv_2mortal(res);
return res;
}
/** @brief Returns 1 if the fiber has finished execution. */
DLLEXPORT int is_finished(int fiber_id) {
if (fiber_id < 0)
return 0;
return (fibers[fiber_id] && fibers[fiber_id]->finished) ? 1 : 0;
}
lib/Acme/Parataxis.c view on Meta::CPAN
para_fiber_t * c = fibers[fiber_id];
if (!c)
return;
fibers[fiber_id] = NULL;
/* 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(c->stack_p);
#endif
lib/Acme/Parataxis.c view on Meta::CPAN
}
if (c->markstack)
Safefree(c->markstack);
if (c->scopestack)
Safefree(c->scopestack);
if (c->savestack)
Safefree(c->savestack);
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);
}
Safefree(c->tmps_stack);
}
free(c);
}
/**
* @brief Global cleanup function for the fiber and thread pool system.
*
lib/Acme/Parataxis.c view on Meta::CPAN
#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);
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;
}
}
lib/Acme/Parataxis.pm view on Meta::CPAN
}
sub stop () { $IS_RUNNING = 0 }
class #
Acme::Parataxis {
use Carp qw[croak];
field $code : reader : param;
field $is_done = 0;
field $error : reader;
field $result : reader;
field $fid : reader;
field $future : param = undef;
method set_result ($val) {
$result = $val;
$future->set_result($val) if $future;
}
method set_error ($err) {
$error = $err;
$future->set_error($err) if $future;
}
method _clear_result () {
$result = undef;
$error = undef;
}
our %REGISTRY;
ADJUST {
Acme::Parataxis::force_depth_zero($code);
$fid = Acme::Parataxis::create_fiber( $code, $self );
$REGISTRY{$fid} = $self;
builtin::weaken $REGISTRY{$fid};
}
method call (@args) {
lib/Acme/Parataxis.pm view on Meta::CPAN
}
method set_error ($err) {
die 'Future already ready' if $is_ready;
$error = $err;
$is_ready = 1;
$_->($self) for @callbacks;
}
method clear_result () {
$result = undef;
$error = undef;
}
method on_ready ($cb) {
if ($is_ready) { $cb->($self) }
else { push @callbacks, $cb }
}
method await () {
return $self->result if $is_ready;
my $fid = Acme::Parataxis->current_fid;
t/007_data_types.t view on Meta::CPAN
subtest 'Passing object into fiber' => sub {
{
my $obj = Local::Destructor->new('A');
my $fiber = Acme::Parataxis->new(
code => sub ($o) {
isa_ok $o, ['Local::Destructor'], 'Fiber returned object';
return 'OK';
}
);
$fiber->call($obj);
$obj = undef; # Local ref gone
# Fiber should have finished and reported its results, releasing the arg
is $DESTROYED, 1, 'Object destroyed (fiber finished and released args)';
}
};
$DESTROYED = 0;
subtest 'Returning object from fiber' => sub {
my $res;
{
my $fiber = Acme::Parataxis->new( code => sub { Local::Destructor->new('B') } );
$res = $fiber->call();
isa_ok $res, ['Local::Destructor'], 'Fiber returned object';
# Fiber is technically done, but we manually flag it to ensure
# the Perl-side wrapper drops its internal references.
$fiber->is_done();
is $DESTROYED, 0, 'Object still alive in parent var';
}
$res = undef;
# Force a stack cycle to clear the mortal reference returned by the XS call
flush_stack();
is $DESTROYED, 1, 'Object destroyed in parent after release';
};
};
#
done_testing();
( run in 1.203 second using v1.01-cache-2.11-cpan-d80b1682f3f )