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>
#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) */

lib/Acme/Parataxis.c  view on Meta::CPAN

 */
#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();
#elif defined(__linux__)
    return sched_getcpu();
#else
    return -1;
#endif
}

/**
 * @brief Detects the number of logical cores available on the system.
 *
 * @return int CPU count (minimum 1).
 */