diff --git a/include/hemlock_limits.h b/include/hemlock_limits.h index dcfa0044..fd25072a 100644 --- a/include/hemlock_limits.h +++ b/include/hemlock_limits.h @@ -124,6 +124,35 @@ #define HML_ASCII_PRINTABLE_START 32 #define HML_ASCII_PRINTABLE_END 127 +// ========== THREAD POOL CONFIGURATION ========== + +// Default number of worker threads (0 = auto-detect based on CPU count) +#define HML_THREADPOOL_DEFAULT_WORKERS 0 + +// Maximum number of worker threads +#define HML_THREADPOOL_MAX_WORKERS 256 + +// Minimum number of worker threads +#define HML_THREADPOOL_MIN_WORKERS 2 + +// Initial capacity of per-worker work-stealing deque +#define HML_THREADPOOL_DEQUE_INITIAL_CAPACITY 64 + +// Maximum capacity of per-worker work-stealing deque +#define HML_THREADPOOL_DEQUE_MAX_CAPACITY 65536 + +// Global submission queue capacity +#define HML_THREADPOOL_SUBMISSION_QUEUE_CAPACITY 4096 + +// Number of steal attempts before sleeping +#define HML_THREADPOOL_STEAL_ATTEMPTS 32 + +// Sleep duration when no work is available (microseconds) +#define HML_THREADPOOL_IDLE_SLEEP_US 100 + +// Work stealing random seed multiplier (for LFSR) +#define HML_THREADPOOL_STEAL_SEED_MULT 1103515245 + // ========== SANDBOX CONFIGURATION ========== // Sandbox restriction flags (bitmask) diff --git a/include/thread_pool.h b/include/thread_pool.h new file mode 100644 index 00000000..e80608d1 --- /dev/null +++ b/include/thread_pool.h @@ -0,0 +1,170 @@ +/* + * Hemlock Thread Pool with Work-Stealing Scheduler + * + * A fixed-size thread pool where workers can steal work from each other + * to balance load. Uses Chase-Lev work-stealing deques for efficient + * lock-free operations on the local end. + * + * Design: + * - Each worker has a local deque (double-ended queue) + * - Workers push/pop from bottom of their own deque (LIFO - cache locality) + * - Workers steal from top of other workers' deques (FIFO - oldest tasks) + * - External submissions go to a global queue with condition variable + * - Workers check local deque, then global queue, then steal from others + */ + +#ifndef HEMLOCK_THREAD_POOL_H +#define HEMLOCK_THREAD_POOL_H + +#include +#include +#include +#include "hemlock_limits.h" + +// Forward declarations +typedef struct ThreadPool ThreadPool; +typedef struct Worker Worker; +typedef struct WorkStealingDeque WorkStealingDeque; +typedef struct WorkItem WorkItem; + +// Work item callback type +// The callback receives the work item data and returns a result +typedef void* (*WorkItemFunc)(void* data, void* ctx); + +// Work item - a unit of work to execute +typedef struct WorkItem { + WorkItemFunc func; // Function to execute + void* data; // Data to pass to function + void* ctx; // Context (ExecutionContext for interpreter) + void* result; // Result storage (set by callback) + atomic_int completed; // 1 when work is done + pthread_mutex_t* mutex; // For waiting on completion + pthread_cond_t* cond; // For signaling completion + int has_waiter; // 1 if someone is waiting for result + struct WorkItem* next; // For linked list (global queue) +} WorkItem; + +// Chase-Lev work-stealing deque +// Lock-free for push/pop (owner), uses CAS for steal +typedef struct WorkStealingDeque { + WorkItem** items; // Circular buffer of work items + atomic_long bottom; // Bottom index (owner pushes/pops here) + atomic_long top; // Top index (thieves steal from here) + int capacity; // Current capacity + int max_capacity; // Maximum allowed capacity + pthread_mutex_t resize_lock; // Lock for resizing only +} WorkStealingDeque; + +// Per-worker state +typedef struct Worker { + int id; // Worker ID (0 to num_workers-1) + pthread_t thread; // Worker thread + ThreadPool* pool; // Back-pointer to pool + WorkStealingDeque* deque; // Local work-stealing deque + unsigned int steal_seed; // Random seed for choosing steal victim + atomic_int active; // 1 if actively working, 0 if idle + atomic_long tasks_executed; // Statistics: number of tasks executed + atomic_long tasks_stolen; // Statistics: number of tasks stolen +} Worker; + +// Global submission queue (MPSC - multiple producers, single consumers) +typedef struct SubmissionQueue { + WorkItem* head; // Head of linked list + WorkItem* tail; // Tail of linked list + int count; // Number of items in queue + int capacity; // Maximum capacity + pthread_mutex_t mutex; // Protects queue access + pthread_cond_t not_empty; // Signal when items are available +} SubmissionQueue; + +// Thread pool statistics +typedef struct ThreadPoolStats { + long total_tasks_submitted; // Total tasks submitted to pool + long total_tasks_completed; // Total tasks completed + long total_steals; // Total successful steals + long total_steal_attempts; // Total steal attempts +} ThreadPoolStats; + +// The thread pool +typedef struct ThreadPool { + int num_workers; // Number of worker threads + Worker* workers; // Array of workers + SubmissionQueue* submission; // Global submission queue + atomic_int shutdown; // 1 when shutting down + atomic_int started; // 1 when workers have started + pthread_mutex_t start_mutex; // For coordinating startup + pthread_cond_t start_cond; // For signaling startup complete + ThreadPoolStats stats; // Pool statistics +} ThreadPool; + +// ========== THREAD POOL API ========== + +// Initialize the global thread pool +// num_workers: number of worker threads (0 = auto-detect based on CPU count) +// Returns: 0 on success, -1 on error +int thread_pool_init(int num_workers); + +// Shutdown the global thread pool +// Waits for all pending work to complete +void thread_pool_shutdown(void); + +// Submit a work item to the thread pool +// If called from a worker thread, pushes to local deque +// Otherwise, pushes to global submission queue +// func: function to execute +// data: data to pass to function +// ctx: execution context +// Returns: WorkItem pointer (can be used to wait for result) +WorkItem* thread_pool_submit(WorkItemFunc func, void* data, void* ctx); + +// Submit work and wait for result (blocking) +// Returns: result from work item function +void* thread_pool_submit_wait(WorkItemFunc func, void* data, void* ctx); + +// Wait for a work item to complete +// Returns: result from work item function +void* work_item_wait(WorkItem* item); + +// Free a work item (must be called after waiting or if not waiting) +void work_item_free(WorkItem* item); + +// Check if current thread is a pool worker +// Returns: worker ID (0 to num_workers-1) or -1 if not a worker +int thread_pool_current_worker_id(void); + +// Get thread pool statistics +ThreadPoolStats thread_pool_get_stats(void); + +// Check if thread pool is initialized +int thread_pool_is_initialized(void); + +// Get the global thread pool (for internal use) +ThreadPool* thread_pool_get(void); + +// ========== WORK-STEALING DEQUE API (internal) ========== + +// Create a new work-stealing deque +WorkStealingDeque* deque_new(int initial_capacity); + +// Free a work-stealing deque +void deque_free(WorkStealingDeque* deque); + +// Push a work item to the bottom of the deque (owner only) +// Returns: 0 on success, -1 if full +int deque_push(WorkStealingDeque* deque, WorkItem* item); + +// Pop a work item from the bottom of the deque (owner only) +// Returns: work item or NULL if empty +WorkItem* deque_pop(WorkStealingDeque* deque); + +// Steal a work item from the top of the deque (any thread) +// Returns: work item or NULL if empty +WorkItem* deque_steal(WorkStealingDeque* deque); + +// Check if deque is empty +int deque_is_empty(WorkStealingDeque* deque); + +// Get number of items in deque (approximate) +long deque_size(WorkStealingDeque* deque); + +#endif // HEMLOCK_THREAD_POOL_H diff --git a/runtime/include/thread_pool.h b/runtime/include/thread_pool.h new file mode 100644 index 00000000..f36cd833 --- /dev/null +++ b/runtime/include/thread_pool.h @@ -0,0 +1,115 @@ +/* + * Hemlock Runtime Thread Pool with Work-Stealing Scheduler + * + * A fixed-size thread pool where workers can steal work from each other + * to balance load. Uses Chase-Lev work-stealing deques for efficient + * lock-free operations on the local end. + */ + +#ifndef HEMLOCK_RUNTIME_THREAD_POOL_H +#define HEMLOCK_RUNTIME_THREAD_POOL_H + +#include +#include +#include + +// Thread pool configuration constants +#define HML_THREADPOOL_DEFAULT_WORKERS 0 // 0 = auto-detect +#define HML_THREADPOOL_MAX_WORKERS 256 +#define HML_THREADPOOL_MIN_WORKERS 2 +#define HML_THREADPOOL_DEQUE_INITIAL_CAPACITY 64 +#define HML_THREADPOOL_DEQUE_MAX_CAPACITY 65536 +#define HML_THREADPOOL_SUBMISSION_QUEUE_CAPACITY 4096 +#define HML_THREADPOOL_STEAL_ATTEMPTS 32 +#define HML_THREADPOOL_IDLE_SLEEP_US 100 +#define HML_THREADPOOL_STEAL_SEED_MULT 1103515245 + +// Forward declarations +typedef struct HmlThreadPool HmlThreadPool; +typedef struct HmlWorker HmlWorker; +typedef struct HmlWorkStealingDeque HmlWorkStealingDeque; +typedef struct HmlWorkItem HmlWorkItem; + +// Work item callback type +typedef void* (*HmlWorkItemFunc)(void* data, void* ctx); + +// Work item - a unit of work to execute +typedef struct HmlWorkItem { + HmlWorkItemFunc func; // Function to execute + void* data; // Data to pass to function + void* ctx; // Context + void* result; // Result storage + atomic_int completed; // 1 when work is done + pthread_mutex_t* mutex; // For waiting on completion + pthread_cond_t* cond; // For signaling completion + int has_waiter; // 1 if someone is waiting for result + struct HmlWorkItem* next; // For linked list +} HmlWorkItem; + +// Chase-Lev work-stealing deque +typedef struct HmlWorkStealingDeque { + HmlWorkItem** items; + atomic_long bottom; + atomic_long top; + int capacity; + int max_capacity; + pthread_mutex_t resize_lock; +} HmlWorkStealingDeque; + +// Per-worker state +typedef struct HmlWorker { + int id; + pthread_t thread; + HmlThreadPool* pool; + HmlWorkStealingDeque* deque; + unsigned int steal_seed; + atomic_int active; + atomic_long tasks_executed; + atomic_long tasks_stolen; +} HmlWorker; + +// Global submission queue +typedef struct HmlSubmissionQueue { + HmlWorkItem* head; + HmlWorkItem* tail; + int count; + int capacity; + pthread_mutex_t mutex; + pthread_cond_t not_empty; +} HmlSubmissionQueue; + +// The thread pool +typedef struct HmlThreadPool { + int num_workers; + HmlWorker* workers; + HmlSubmissionQueue* submission; + atomic_int shutdown; + atomic_int started; + pthread_mutex_t start_mutex; + pthread_cond_t start_cond; +} HmlThreadPool; + +// ========== THREAD POOL API ========== + +// Initialize the global thread pool +int hml_thread_pool_init(int num_workers); + +// Shutdown the global thread pool +void hml_thread_pool_shutdown(void); + +// Submit a work item to the thread pool +HmlWorkItem* hml_thread_pool_submit(HmlWorkItemFunc func, void* data, void* ctx); + +// Wait for a work item to complete +void* hml_work_item_wait(HmlWorkItem* item); + +// Free a work item +void hml_work_item_free(HmlWorkItem* item); + +// Check if current thread is a pool worker +int hml_thread_pool_current_worker_id(void); + +// Check if thread pool is initialized +int hml_thread_pool_is_initialized(void); + +#endif // HEMLOCK_RUNTIME_THREAD_POOL_H diff --git a/runtime/src/builtins_async.c b/runtime/src/builtins_async.c index 2c163901..9e0a810a 100644 --- a/runtime/src/builtins_async.c +++ b/runtime/src/builtins_async.c @@ -5,12 +5,50 @@ */ #include "builtins_internal.h" +#include "thread_pool.h" #include #include #include static atomic_int g_next_task_id = 1; +// Helper function to free a task +static void task_free(HmlTask *task) { + if (!task) return; + + // Free function and args + hml_release(&task->function); + for (int i = 0; i < task->num_args; i++) { + hml_release(&task->args[i]); + } + free(task->args); + + // Free mutex and cond + if (task->mutex) { + pthread_mutex_destroy((pthread_mutex_t*)task->mutex); + free(task->mutex); + } + if (task->cond) { + pthread_cond_destroy((pthread_cond_t*)task->cond); + free(task->cond); + } + if (task->thread) { + free(task->thread); + } + + free(task); +} + +// Decrement task reference count and free if it reaches 0 +static void hml_task_release(HmlTask *task) { + if (!task) return; + + int new_count = __atomic_sub_fetch(&task->ref_count, 1, __ATOMIC_SEQ_CST); + if (new_count <= 0) { + task_free(task); + } +} + // Define ffi_type for HmlValue struct (16 bytes: 4 type + 4 padding + 8 union) static ffi_type *hml_value_elements[] = { &ffi_type_uint32, // HmlValueType (enum) @@ -67,9 +105,10 @@ static HmlValue call_hemlock_function_ffi(void *fn_ptr, void *closure_env, HmlVa return result; } -// Thread wrapper function -static void* task_thread_wrapper(void* arg) { - HmlTask *task = (HmlTask*)arg; +// Thread pool work item callback for task execution +static void* task_pool_execute(void* data, void* ctx_unused) { + (void)ctx_unused; + HmlTask *task = (HmlTask*)data; // Mark as running pthread_mutex_lock((pthread_mutex_t*)task->mutex); @@ -88,10 +127,15 @@ static void* task_thread_wrapper(void* arg) { pthread_mutex_lock((pthread_mutex_t*)task->mutex); task->result = result; task->state = HML_TASK_COMPLETED; - pthread_cond_signal((pthread_cond_t*)task->cond); + pthread_cond_broadcast((pthread_cond_t*)task->cond); pthread_mutex_unlock((pthread_mutex_t*)task->mutex); - return NULL; + // Clean up detached tasks + if (task->detached) { + hml_task_release(task); + } + + return task; } HmlValue hml_spawn(HmlValue fn, HmlValue *args, int num_args) { @@ -134,9 +178,14 @@ HmlValue hml_spawn(HmlValue fn, HmlValue *args, int num_args) { pthread_mutex_init((pthread_mutex_t*)task->mutex, NULL); pthread_cond_init((pthread_cond_t*)task->cond, NULL); - // Create thread - task->thread = malloc(sizeof(pthread_t)); - pthread_create((pthread_t*)task->thread, NULL, task_thread_wrapper, task); + // task->thread not used with thread pool, set to NULL + task->thread = NULL; + + // Submit task to thread pool + HmlWorkItem* work = hml_thread_pool_submit(task_pool_execute, task, NULL); + if (!work) { + hml_runtime_error("Failed to submit task to thread pool"); + } // Return task value HmlValue result; @@ -160,16 +209,14 @@ HmlValue hml_join(HmlValue task_val) { hml_runtime_error("cannot join detached task"); } - // Wait for task to complete + // Wait for task to complete using condition variable + // With thread pool, task->thread is not used (pool manages threads) pthread_mutex_lock((pthread_mutex_t*)task->mutex); while (task->state != HML_TASK_COMPLETED) { pthread_cond_wait((pthread_cond_t*)task->cond, (pthread_mutex_t*)task->mutex); } - pthread_mutex_unlock((pthread_mutex_t*)task->mutex); - - // Join the thread - pthread_join(*(pthread_t*)task->thread, NULL); task->joined = 1; + pthread_mutex_unlock((pthread_mutex_t*)task->mutex); // Return result (retained) HmlValue result = task->result; @@ -192,8 +239,9 @@ void hml_detach(HmlValue task_val) { return; // Already detached } + // With thread pool, just mark as detached + // Pool manages its own threads, so no pthread_detach needed task->detached = 1; - pthread_detach(*(pthread_t*)task->thread); } // task_debug_info(task) - Print debug information about a task diff --git a/runtime/src/thread_pool.c b/runtime/src/thread_pool.c new file mode 100644 index 00000000..b4dd8144 --- /dev/null +++ b/runtime/src/thread_pool.c @@ -0,0 +1,599 @@ +/* + * Hemlock Runtime Thread Pool with Work-Stealing Scheduler + * + * Implementation of a work-stealing thread pool for async task execution. + */ + +#include "thread_pool.h" +#include +#include +#include +#include +#include + +// Thread-local storage for current worker +static __thread HmlWorker* tls_current_worker = NULL; + +// Global thread pool instance +static HmlThreadPool* g_thread_pool = NULL; +static pthread_mutex_t g_pool_mutex = PTHREAD_MUTEX_INITIALIZER; + +// ========== WORK-STEALING DEQUE IMPLEMENTATION ========== + +static HmlWorkStealingDeque* deque_new(int initial_capacity) { + HmlWorkStealingDeque* deque = malloc(sizeof(HmlWorkStealingDeque)); + if (!deque) return NULL; + + deque->items = calloc(initial_capacity, sizeof(HmlWorkItem*)); + if (!deque->items) { + free(deque); + return NULL; + } + + atomic_init(&deque->bottom, 0); + atomic_init(&deque->top, 0); + deque->capacity = initial_capacity; + deque->max_capacity = HML_THREADPOOL_DEQUE_MAX_CAPACITY; + pthread_mutex_init(&deque->resize_lock, NULL); + + return deque; +} + +static void deque_free(HmlWorkStealingDeque* deque) { + if (!deque) return; + pthread_mutex_destroy(&deque->resize_lock); + free(deque->items); + free(deque); +} + +static int deque_resize(HmlWorkStealingDeque* deque) { + pthread_mutex_lock(&deque->resize_lock); + + int old_capacity = deque->capacity; + int new_capacity = old_capacity * 2; + + if (new_capacity > deque->max_capacity) { + pthread_mutex_unlock(&deque->resize_lock); + return -1; + } + + HmlWorkItem** new_items = calloc(new_capacity, sizeof(HmlWorkItem*)); + if (!new_items) { + pthread_mutex_unlock(&deque->resize_lock); + return -1; + } + + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + long bottom = atomic_load_explicit(&deque->bottom, memory_order_acquire); + + for (long i = top; i < bottom; i++) { + new_items[i % new_capacity] = deque->items[i % old_capacity]; + } + + HmlWorkItem** old_items = deque->items; + deque->items = new_items; + deque->capacity = new_capacity; + + pthread_mutex_unlock(&deque->resize_lock); + free(old_items); + + return 0; +} + +static int deque_push(HmlWorkStealingDeque* deque, HmlWorkItem* item) { + long bottom = atomic_load_explicit(&deque->bottom, memory_order_relaxed); + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + + long size = bottom - top; + if (size >= deque->capacity - 1) { + if (deque_resize(deque) != 0) { + return -1; + } + } + + deque->items[bottom % deque->capacity] = item; + atomic_thread_fence(memory_order_release); + atomic_store_explicit(&deque->bottom, bottom + 1, memory_order_relaxed); + + return 0; +} + +static HmlWorkItem* deque_pop(HmlWorkStealingDeque* deque) { + long bottom = atomic_load_explicit(&deque->bottom, memory_order_relaxed) - 1; + atomic_store_explicit(&deque->bottom, bottom, memory_order_relaxed); + atomic_thread_fence(memory_order_seq_cst); + + long top = atomic_load_explicit(&deque->top, memory_order_relaxed); + + if (top <= bottom) { + HmlWorkItem* item = deque->items[bottom % deque->capacity]; + + if (top == bottom) { + if (!atomic_compare_exchange_strong_explicit( + &deque->top, &top, top + 1, + memory_order_seq_cst, memory_order_relaxed)) { + item = NULL; + } + atomic_store_explicit(&deque->bottom, bottom + 1, memory_order_relaxed); + } + return item; + } else { + atomic_store_explicit(&deque->bottom, bottom + 1, memory_order_relaxed); + return NULL; + } +} + +static HmlWorkItem* deque_steal(HmlWorkStealingDeque* deque) { + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + atomic_thread_fence(memory_order_seq_cst); + long bottom = atomic_load_explicit(&deque->bottom, memory_order_acquire); + + if (top < bottom) { + HmlWorkItem* item = deque->items[top % deque->capacity]; + + if (!atomic_compare_exchange_strong_explicit( + &deque->top, &top, top + 1, + memory_order_seq_cst, memory_order_relaxed)) { + return NULL; + } + return item; + } + return NULL; +} + +// ========== SUBMISSION QUEUE IMPLEMENTATION ========== + +static HmlSubmissionQueue* submission_queue_new(int capacity) { + HmlSubmissionQueue* queue = malloc(sizeof(HmlSubmissionQueue)); + if (!queue) return NULL; + + queue->head = NULL; + queue->tail = NULL; + queue->count = 0; + queue->capacity = capacity; + pthread_mutex_init(&queue->mutex, NULL); + pthread_cond_init(&queue->not_empty, NULL); + + return queue; +} + +static void submission_queue_free(HmlSubmissionQueue* queue) { + if (!queue) return; + + pthread_mutex_lock(&queue->mutex); + HmlWorkItem* item = queue->head; + while (item) { + HmlWorkItem* next = item->next; + hml_work_item_free(item); + item = next; + } + pthread_mutex_unlock(&queue->mutex); + + pthread_mutex_destroy(&queue->mutex); + pthread_cond_destroy(&queue->not_empty); + free(queue); +} + +static int submission_queue_push(HmlSubmissionQueue* queue, HmlWorkItem* item) { + pthread_mutex_lock(&queue->mutex); + + if (queue->count >= queue->capacity) { + pthread_mutex_unlock(&queue->mutex); + return -1; + } + + item->next = NULL; + + if (queue->tail) { + queue->tail->next = item; + queue->tail = item; + } else { + queue->head = queue->tail = item; + } + queue->count++; + + pthread_cond_signal(&queue->not_empty); + pthread_mutex_unlock(&queue->mutex); + + return 0; +} + +static HmlWorkItem* submission_queue_pop(HmlSubmissionQueue* queue) { + pthread_mutex_lock(&queue->mutex); + + HmlWorkItem* item = queue->head; + if (item) { + queue->head = item->next; + if (!queue->head) { + queue->tail = NULL; + } + queue->count--; + item->next = NULL; + } + + pthread_mutex_unlock(&queue->mutex); + return item; +} + +static HmlWorkItem* submission_queue_pop_wait(HmlSubmissionQueue* queue, int timeout_us) { + pthread_mutex_lock(&queue->mutex); + + if (!queue->head) { + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + ts.tv_nsec += timeout_us * 1000; + if (ts.tv_nsec >= 1000000000) { + ts.tv_sec++; + ts.tv_nsec -= 1000000000; + } + pthread_cond_timedwait(&queue->not_empty, &queue->mutex, &ts); + } + + HmlWorkItem* item = queue->head; + if (item) { + queue->head = item->next; + if (!queue->head) { + queue->tail = NULL; + } + queue->count--; + item->next = NULL; + } + + pthread_mutex_unlock(&queue->mutex); + return item; +} + +// ========== WORK ITEM IMPLEMENTATION ========== + +static HmlWorkItem* work_item_new(HmlWorkItemFunc func, void* data, void* ctx) { + HmlWorkItem* item = malloc(sizeof(HmlWorkItem)); + if (!item) return NULL; + + item->func = func; + item->data = data; + item->ctx = ctx; + item->result = NULL; + atomic_init(&item->completed, 0); + item->mutex = NULL; + item->cond = NULL; + item->has_waiter = 0; + item->next = NULL; + + return item; +} + +void hml_work_item_free(HmlWorkItem* item) { + if (!item) return; + + if (item->mutex) { + pthread_mutex_destroy(item->mutex); + free(item->mutex); + } + if (item->cond) { + pthread_cond_destroy(item->cond); + free(item->cond); + } + free(item); +} + +static void work_item_setup_wait(HmlWorkItem* item) { + if (!item->mutex) { + item->mutex = malloc(sizeof(pthread_mutex_t)); + item->cond = malloc(sizeof(pthread_cond_t)); + pthread_mutex_init(item->mutex, NULL); + pthread_cond_init(item->cond, NULL); + } + item->has_waiter = 1; +} + +static void work_item_signal_complete(HmlWorkItem* item) { + atomic_store_explicit(&item->completed, 1, memory_order_release); + + if (item->has_waiter && item->mutex && item->cond) { + pthread_mutex_lock(item->mutex); + pthread_cond_signal(item->cond); + pthread_mutex_unlock(item->mutex); + } +} + +void* hml_work_item_wait(HmlWorkItem* item) { + if (!item) return NULL; + + if (atomic_load_explicit(&item->completed, memory_order_acquire)) { + return item->result; + } + + work_item_setup_wait(item); + + pthread_mutex_lock(item->mutex); + while (!atomic_load_explicit(&item->completed, memory_order_acquire)) { + pthread_cond_wait(item->cond, item->mutex); + } + pthread_mutex_unlock(item->mutex); + + return item->result; +} + +// ========== WORKER IMPLEMENTATION ========== + +static int worker_random_victim(HmlWorker* worker) { + int num_workers = worker->pool->num_workers; + if (num_workers <= 1) return -1; + + worker->steal_seed = worker->steal_seed * HML_THREADPOOL_STEAL_SEED_MULT + 1; + int victim = (worker->steal_seed >> 16) % num_workers; + + if (victim == worker->id) { + victim = (victim + 1) % num_workers; + } + + return victim; +} + +static HmlWorkItem* worker_steal(HmlWorker* worker) { + HmlThreadPool* pool = worker->pool; + + for (int attempts = 0; attempts < HML_THREADPOOL_STEAL_ATTEMPTS; attempts++) { + int victim_id = worker_random_victim(worker); + if (victim_id < 0) break; + + HmlWorker* victim = &pool->workers[victim_id]; + HmlWorkItem* item = deque_steal(victim->deque); + + if (item) { + atomic_fetch_add(&worker->tasks_stolen, 1); + return item; + } + } + + return NULL; +} + +static HmlWorkItem* worker_get_work(HmlWorker* worker) { + HmlThreadPool* pool = worker->pool; + + HmlWorkItem* item = deque_pop(worker->deque); + if (item) return item; + + item = submission_queue_pop(pool->submission); + if (item) return item; + + item = worker_steal(worker); + if (item) return item; + + return NULL; +} + +static void worker_execute(HmlWorker* worker, HmlWorkItem* item) { + atomic_store(&worker->active, 1); + + void* result = item->func(item->data, item->ctx); + item->result = result; + + work_item_signal_complete(item); + + atomic_fetch_add(&worker->tasks_executed, 1); + atomic_store(&worker->active, 0); +} + +static void* worker_thread_main(void* arg) { + HmlWorker* worker = (HmlWorker*)arg; + HmlThreadPool* pool = worker->pool; + + tls_current_worker = worker; + + // Block all signals in worker thread + sigset_t set; + sigfillset(&set); + pthread_sigmask(SIG_BLOCK, &set, NULL); + + // Signal that we're ready + pthread_mutex_lock(&pool->start_mutex); + pthread_cond_signal(&pool->start_cond); + pthread_mutex_unlock(&pool->start_mutex); + + // Main work loop + while (!atomic_load(&pool->shutdown)) { + HmlWorkItem* item = worker_get_work(worker); + + if (item) { + worker_execute(worker, item); + + if (!item->has_waiter) { + hml_work_item_free(item); + } + } else { + item = submission_queue_pop_wait(pool->submission, + HML_THREADPOOL_IDLE_SLEEP_US); + if (item) { + worker_execute(worker, item); + if (!item->has_waiter) { + hml_work_item_free(item); + } + } + } + } + + // Drain remaining work + HmlWorkItem* item; + while ((item = worker_get_work(worker)) != NULL) { + worker_execute(worker, item); + if (!item->has_waiter) { + hml_work_item_free(item); + } + } + + tls_current_worker = NULL; + return NULL; +} + +// ========== THREAD POOL IMPLEMENTATION ========== + +static int get_cpu_count(void) { + long count = sysconf(_SC_NPROCESSORS_ONLN); + if (count <= 0) count = 4; + return (int)count; +} + +int hml_thread_pool_init(int num_workers) { + pthread_mutex_lock(&g_pool_mutex); + + if (g_thread_pool) { + pthread_mutex_unlock(&g_pool_mutex); + return 0; + } + + // Use 2x CPU count to handle blocking tasks (channel waits, I/O) + if (num_workers <= 0) { + num_workers = get_cpu_count() * 2; + } + if (num_workers < HML_THREADPOOL_MIN_WORKERS) { + num_workers = HML_THREADPOOL_MIN_WORKERS; + } + if (num_workers > HML_THREADPOOL_MAX_WORKERS) { + num_workers = HML_THREADPOOL_MAX_WORKERS; + } + + HmlThreadPool* pool = malloc(sizeof(HmlThreadPool)); + if (!pool) { + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + + pool->num_workers = num_workers; + atomic_init(&pool->shutdown, 0); + atomic_init(&pool->started, 0); + pthread_mutex_init(&pool->start_mutex, NULL); + pthread_cond_init(&pool->start_cond, NULL); + + pool->submission = submission_queue_new(HML_THREADPOOL_SUBMISSION_QUEUE_CAPACITY); + if (!pool->submission) { + free(pool); + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + + pool->workers = calloc(num_workers, sizeof(HmlWorker)); + if (!pool->workers) { + submission_queue_free(pool->submission); + free(pool); + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + + for (int i = 0; i < num_workers; i++) { + HmlWorker* worker = &pool->workers[i]; + worker->id = i; + worker->pool = pool; + worker->steal_seed = (unsigned int)(i * 1103515245 + 12345); + atomic_init(&worker->active, 0); + atomic_init(&worker->tasks_executed, 0); + atomic_init(&worker->tasks_stolen, 0); + + worker->deque = deque_new(HML_THREADPOOL_DEQUE_INITIAL_CAPACITY); + if (!worker->deque) { + for (int j = 0; j < i; j++) { + deque_free(pool->workers[j].deque); + } + free(pool->workers); + submission_queue_free(pool->submission); + free(pool); + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + } + + g_thread_pool = pool; + + // Start worker threads + pthread_mutex_lock(&pool->start_mutex); + + for (int i = 0; i < num_workers; i++) { + HmlWorker* worker = &pool->workers[i]; + pthread_create(&worker->thread, NULL, worker_thread_main, worker); + } + + for (int i = 0; i < num_workers; i++) { + pthread_cond_wait(&pool->start_cond, &pool->start_mutex); + } + + atomic_store(&pool->started, 1); + pthread_mutex_unlock(&pool->start_mutex); + + pthread_mutex_unlock(&g_pool_mutex); + return 0; +} + +void hml_thread_pool_shutdown(void) { + pthread_mutex_lock(&g_pool_mutex); + + if (!g_thread_pool) { + pthread_mutex_unlock(&g_pool_mutex); + return; + } + + HmlThreadPool* pool = g_thread_pool; + + atomic_store(&pool->shutdown, 1); + + pthread_mutex_lock(&pool->submission->mutex); + pthread_cond_broadcast(&pool->submission->not_empty); + pthread_mutex_unlock(&pool->submission->mutex); + + for (int i = 0; i < pool->num_workers; i++) { + pthread_join(pool->workers[i].thread, NULL); + } + + for (int i = 0; i < pool->num_workers; i++) { + deque_free(pool->workers[i].deque); + } + free(pool->workers); + + submission_queue_free(pool->submission); + + pthread_mutex_destroy(&pool->start_mutex); + pthread_cond_destroy(&pool->start_cond); + free(pool); + + g_thread_pool = NULL; + + pthread_mutex_unlock(&g_pool_mutex); +} + +HmlWorkItem* hml_thread_pool_submit(HmlWorkItemFunc func, void* data, void* ctx) { + HmlThreadPool* pool = g_thread_pool; + + if (!pool) { + if (hml_thread_pool_init(HML_THREADPOOL_DEFAULT_WORKERS) != 0) { + return NULL; + } + pool = g_thread_pool; + } + + HmlWorkItem* item = work_item_new(func, data, ctx); + if (!item) return NULL; + + if (tls_current_worker && tls_current_worker->pool == pool) { + if (deque_push(tls_current_worker->deque, item) == 0) { + return item; + } + } + + if (submission_queue_push(pool->submission, item) != 0) { + hml_work_item_free(item); + return NULL; + } + + return item; +} + +int hml_thread_pool_current_worker_id(void) { + if (tls_current_worker) { + return tls_current_worker->id; + } + return -1; +} + +int hml_thread_pool_is_initialized(void) { + return g_thread_pool != NULL; +} diff --git a/src/backends/interpreter/builtins/concurrency.c b/src/backends/interpreter/builtins/concurrency.c index 68082bff..7580f881 100644 --- a/src/backends/interpreter/builtins/concurrency.c +++ b/src/backends/interpreter/builtins/concurrency.c @@ -4,17 +4,13 @@ // Global task ID counter (atomic for thread-safety in concurrent spawns) static atomic_int next_task_id = 1; -// Thread wrapper function that executes a task -static void* task_thread_wrapper(void* arg) { - Task *task = (Task*)arg; +// Task execution function for thread pool +// This is the work item callback that executes the async task +static void* task_pool_execute(void* data, void* ctx_unused) { + (void)ctx_unused; // ExecutionContext is stored in task->ctx + Task *task = (Task*)data; Function *fn = task->function; - // Block all signals in worker thread - only main thread should handle signals - // This prevents signal handlers from corrupting task state during execution - sigset_t set; - sigfillset(&set); - pthread_sigmask(SIG_BLOCK, &set, NULL); - // Mark as running (thread-safe) pthread_mutex_lock((pthread_mutex_t*)task->task_mutex); task->state = TASK_RUNNING; @@ -50,6 +46,10 @@ static void* task_thread_wrapper(void* arg) { task->result = malloc(sizeof(Value)); *task->result = result; task->state = TASK_COMPLETED; + // Signal anyone waiting on join + if (task->thread) { // thread field reused as condition variable + pthread_cond_broadcast((pthread_cond_t*)task->thread); + } pthread_mutex_unlock((pthread_mutex_t*)task->task_mutex); // Release function environment (reference counted) @@ -61,7 +61,7 @@ static void* task_thread_wrapper(void* arg) { task_release(task); } - return NULL; + return task; // Return task pointer (not used, but required by API) } Value builtin_spawn(Value *args, int num_args, ExecutionContext *ctx) { @@ -107,20 +107,30 @@ Value builtin_spawn(Value *args, int num_args, ExecutionContext *ctx) { int task_id = atomic_fetch_add(&next_task_id, 1); Task *task = task_new(task_id, fn, task_args, task_num_args, fn->closure_env); - // Allocate pthread_t - task->thread = malloc(sizeof(pthread_t)); + // Allocate condition variable for join() to wait on + // We reuse the task->thread field to store the condition variable + task->thread = malloc(sizeof(pthread_cond_t)); if (!task->thread) { fprintf(stderr, "Runtime error: Memory allocation failed\n"); exit(1); } - - // Create thread to execute task - int rc = pthread_create((pthread_t*)task->thread, NULL, task_thread_wrapper, task); - if (rc != 0) { - fprintf(stderr, "Runtime error: Failed to create thread: %d\n", rc); + pthread_cond_init((pthread_cond_t*)task->thread, NULL); + + // Submit task to thread pool + // The pool will execute task_pool_execute with the task as data + WorkItem* work = thread_pool_submit(task_pool_execute, task, NULL); + if (!work) { + fprintf(stderr, "Runtime error: Failed to submit task to thread pool\n"); + pthread_cond_destroy((pthread_cond_t*)task->thread); + free(task->thread); + task->thread = NULL; + task_free(task); exit(1); } + // We don't need to track the WorkItem - the task has its own completion mechanism + // The work item will be freed by the pool after execution + return val_task(task); } @@ -157,20 +167,20 @@ Value builtin_join(Value *args, int num_args, ExecutionContext *ctx) { // Mark as joined task->joined = 1; - pthread_mutex_unlock((pthread_mutex_t*)task->task_mutex); - - // Wait for thread to complete (outside of mutex to avoid deadlock) - if (task->thread) { - int rc = pthread_join(*(pthread_t*)task->thread, NULL); - if (rc != 0) { - runtime_error(ctx, "pthread_join failed: %d", rc); - return val_null(); + // Wait for task to complete using condition variable + // The task->thread field now holds a condition variable instead of pthread_t + while (task->state != TASK_COMPLETED) { + if (task->thread) { + pthread_cond_wait((pthread_cond_t*)task->thread, (pthread_mutex_t*)task->task_mutex); + } else { + // No condition variable - spin briefly using nanosleep + pthread_mutex_unlock((pthread_mutex_t*)task->task_mutex); + struct timespec ts = { .tv_sec = 0, .tv_nsec = 100000 }; // 100 microseconds + nanosleep(&ts, NULL); + pthread_mutex_lock((pthread_mutex_t*)task->task_mutex); } } - // Access exception state and result (thread-safe) - pthread_mutex_lock((pthread_mutex_t*)task->task_mutex); - // Check if task threw an exception if (task->ctx->exception_state.is_throwing) { // Re-throw the exception in the current context @@ -231,20 +241,12 @@ Value builtin_detach(Value *args, int num_args, ExecutionContext *ctx) { return val_null(); } - // Mark as detached + // Mark as detached - with thread pool, no pthread_detach needed + // The pool manages its own threads t->detached = 1; pthread_mutex_unlock((pthread_mutex_t*)t->task_mutex); - // Detach the pthread (fire and forget) - if (t->thread) { - int rc = pthread_detach(*(pthread_t*)t->thread); - if (rc != 0) { - runtime_error(ctx, "pthread_detach failed: %d", rc); - return val_null(); - } - } - return val_null(); } @@ -276,41 +278,26 @@ Value builtin_detach(Value *args, int num_args, ExecutionContext *ctx) { int task_id = atomic_fetch_add(&next_task_id, 1); Task *task = task_new(task_id, fn, task_args, task_num_args, fn->closure_env); - // Mark as detached before starting thread + // Mark as detached before submitting to pool task->detached = 1; - // Allocate pthread_t - task->thread = malloc(sizeof(pthread_t)); - if (!task->thread) { - runtime_error(ctx, "Memory allocation failed"); - return val_null(); - } + // No condition variable needed for detached tasks (no one will join) + task->thread = NULL; - // CRITICAL: Retain task to prevent premature cleanup during pthread_detach - // Without this, the worker thread may complete and free the task before - // we finish calling pthread_detach, leading to use-after-free + // CRITICAL: Retain task to prevent premature cleanup + // The worker will release the task when done (via task->detached check) task_retain(task); // ref_count: 1 -> 2 - // Create thread to execute task - int rc = pthread_create((pthread_t*)task->thread, NULL, task_thread_wrapper, task); - if (rc != 0) { - runtime_error(ctx, "Failed to create thread: %d", rc); - free(task->thread); - task_release(task); // Release our temporary reference - return val_null(); - } - - // Detach the pthread immediately (fire and forget) - // Safe to access task->thread because we're holding a reference - rc = pthread_detach(*(pthread_t*)task->thread); - if (rc != 0) { - runtime_error(ctx, "pthread_detach failed: %d", rc); + // Submit task to thread pool (fire and forget) + WorkItem* work = thread_pool_submit(task_pool_execute, task, NULL); + if (!work) { + runtime_error(ctx, "Failed to submit task to thread pool"); task_release(task); // Release our temporary reference return val_null(); } - // Release our temporary reference - worker thread will clean up when done - // ref_count: 2 -> 1 (worker thread holds the remaining reference) + // Release our reference - worker thread will clean up when done + // ref_count: 2 -> 1 (worker thread holds the remaining reference via task->detached cleanup) task_release(task); return val_null(); diff --git a/src/backends/interpreter/internal.h b/src/backends/interpreter/internal.h index 2008a40e..faa6c1b9 100644 --- a/src/backends/interpreter/internal.h +++ b/src/backends/interpreter/internal.h @@ -4,6 +4,7 @@ #include "interpreter.h" #include "ast.h" #include "hemlock_limits.h" +#include "thread_pool.h" #include // ========== CONTROL FLOW STATE ========== diff --git a/src/backends/interpreter/runtime/thread_pool.c b/src/backends/interpreter/runtime/thread_pool.c new file mode 100644 index 00000000..798a090e --- /dev/null +++ b/src/backends/interpreter/runtime/thread_pool.c @@ -0,0 +1,693 @@ +/* + * Hemlock Thread Pool with Work-Stealing Scheduler + * + * Implementation of a work-stealing thread pool for async task execution. + * Workers steal from each other when idle to balance load automatically. + */ + +#include "thread_pool.h" +#include +#include +#include +#include +#include + +// Thread-local storage for current worker +static __thread Worker* tls_current_worker = NULL; + +// Global thread pool instance +static ThreadPool* g_thread_pool = NULL; +static pthread_mutex_t g_pool_mutex = PTHREAD_MUTEX_INITIALIZER; + +// ========== WORK-STEALING DEQUE IMPLEMENTATION ========== +// Based on the Chase-Lev work-stealing deque algorithm + +WorkStealingDeque* deque_new(int initial_capacity) { + WorkStealingDeque* deque = malloc(sizeof(WorkStealingDeque)); + if (!deque) return NULL; + + deque->items = calloc(initial_capacity, sizeof(WorkItem*)); + if (!deque->items) { + free(deque); + return NULL; + } + + atomic_init(&deque->bottom, 0); + atomic_init(&deque->top, 0); + deque->capacity = initial_capacity; + deque->max_capacity = HML_THREADPOOL_DEQUE_MAX_CAPACITY; + pthread_mutex_init(&deque->resize_lock, NULL); + + return deque; +} + +void deque_free(WorkStealingDeque* deque) { + if (!deque) return; + pthread_mutex_destroy(&deque->resize_lock); + free(deque->items); + free(deque); +} + +// Resize the deque (double capacity) +static int deque_resize(WorkStealingDeque* deque) { + pthread_mutex_lock(&deque->resize_lock); + + int old_capacity = deque->capacity; + int new_capacity = old_capacity * 2; + + if (new_capacity > deque->max_capacity) { + pthread_mutex_unlock(&deque->resize_lock); + return -1; // Cannot grow further + } + + WorkItem** new_items = calloc(new_capacity, sizeof(WorkItem*)); + if (!new_items) { + pthread_mutex_unlock(&deque->resize_lock); + return -1; + } + + // Copy items to new buffer + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + long bottom = atomic_load_explicit(&deque->bottom, memory_order_acquire); + + for (long i = top; i < bottom; i++) { + new_items[i % new_capacity] = deque->items[i % old_capacity]; + } + + WorkItem** old_items = deque->items; + deque->items = new_items; + deque->capacity = new_capacity; + + pthread_mutex_unlock(&deque->resize_lock); + + // Free old buffer (safe because we hold resize_lock during transition) + free(old_items); + + return 0; +} + +int deque_push(WorkStealingDeque* deque, WorkItem* item) { + long bottom = atomic_load_explicit(&deque->bottom, memory_order_relaxed); + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + + long size = bottom - top; + if (size >= deque->capacity - 1) { + // Need to resize + if (deque_resize(deque) != 0) { + return -1; // Deque is full and cannot grow + } + } + + deque->items[bottom % deque->capacity] = item; + atomic_thread_fence(memory_order_release); + atomic_store_explicit(&deque->bottom, bottom + 1, memory_order_relaxed); + + return 0; +} + +WorkItem* deque_pop(WorkStealingDeque* deque) { + long bottom = atomic_load_explicit(&deque->bottom, memory_order_relaxed) - 1; + atomic_store_explicit(&deque->bottom, bottom, memory_order_relaxed); + atomic_thread_fence(memory_order_seq_cst); + + long top = atomic_load_explicit(&deque->top, memory_order_relaxed); + + if (top <= bottom) { + // Non-empty + WorkItem* item = deque->items[bottom % deque->capacity]; + + if (top == bottom) { + // Last item - need CAS to compete with stealers + if (!atomic_compare_exchange_strong_explicit( + &deque->top, &top, top + 1, + memory_order_seq_cst, memory_order_relaxed)) { + // Lost race to stealer + item = NULL; + } + atomic_store_explicit(&deque->bottom, bottom + 1, memory_order_relaxed); + } + return item; + } else { + // Empty - restore bottom + atomic_store_explicit(&deque->bottom, bottom + 1, memory_order_relaxed); + return NULL; + } +} + +WorkItem* deque_steal(WorkStealingDeque* deque) { + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + atomic_thread_fence(memory_order_seq_cst); + long bottom = atomic_load_explicit(&deque->bottom, memory_order_acquire); + + if (top < bottom) { + // Non-empty + WorkItem* item = deque->items[top % deque->capacity]; + + if (!atomic_compare_exchange_strong_explicit( + &deque->top, &top, top + 1, + memory_order_seq_cst, memory_order_relaxed)) { + // Lost race - another thread stole or owner popped + return NULL; + } + return item; + } + return NULL; +} + +int deque_is_empty(WorkStealingDeque* deque) { + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + long bottom = atomic_load_explicit(&deque->bottom, memory_order_acquire); + return top >= bottom; +} + +long deque_size(WorkStealingDeque* deque) { + long top = atomic_load_explicit(&deque->top, memory_order_acquire); + long bottom = atomic_load_explicit(&deque->bottom, memory_order_acquire); + return bottom - top; +} + +// ========== SUBMISSION QUEUE IMPLEMENTATION ========== + +static SubmissionQueue* submission_queue_new(int capacity) { + SubmissionQueue* queue = malloc(sizeof(SubmissionQueue)); + if (!queue) return NULL; + + queue->head = NULL; + queue->tail = NULL; + queue->count = 0; + queue->capacity = capacity; + pthread_mutex_init(&queue->mutex, NULL); + pthread_cond_init(&queue->not_empty, NULL); + + return queue; +} + +static void submission_queue_free(SubmissionQueue* queue) { + if (!queue) return; + + // Free any remaining items + pthread_mutex_lock(&queue->mutex); + WorkItem* item = queue->head; + while (item) { + WorkItem* next = item->next; + work_item_free(item); + item = next; + } + pthread_mutex_unlock(&queue->mutex); + + pthread_mutex_destroy(&queue->mutex); + pthread_cond_destroy(&queue->not_empty); + free(queue); +} + +static int submission_queue_push(SubmissionQueue* queue, WorkItem* item) { + pthread_mutex_lock(&queue->mutex); + + if (queue->count >= queue->capacity) { + pthread_mutex_unlock(&queue->mutex); + return -1; // Queue full + } + + item->next = NULL; + + if (queue->tail) { + queue->tail->next = item; + queue->tail = item; + } else { + queue->head = queue->tail = item; + } + queue->count++; + + pthread_cond_signal(&queue->not_empty); + pthread_mutex_unlock(&queue->mutex); + + return 0; +} + +static WorkItem* submission_queue_pop(SubmissionQueue* queue) { + pthread_mutex_lock(&queue->mutex); + + WorkItem* item = queue->head; + if (item) { + queue->head = item->next; + if (!queue->head) { + queue->tail = NULL; + } + queue->count--; + item->next = NULL; + } + + pthread_mutex_unlock(&queue->mutex); + return item; +} + +// Wait for an item with timeout (returns NULL if timeout or shutdown) +static WorkItem* submission_queue_pop_wait(SubmissionQueue* queue, int timeout_us) { + pthread_mutex_lock(&queue->mutex); + + if (!queue->head) { + // Wait with timeout + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + ts.tv_nsec += timeout_us * 1000; + if (ts.tv_nsec >= 1000000000) { + ts.tv_sec++; + ts.tv_nsec -= 1000000000; + } + pthread_cond_timedwait(&queue->not_empty, &queue->mutex, &ts); + } + + WorkItem* item = queue->head; + if (item) { + queue->head = item->next; + if (!queue->head) { + queue->tail = NULL; + } + queue->count--; + item->next = NULL; + } + + pthread_mutex_unlock(&queue->mutex); + return item; +} + +// ========== WORK ITEM IMPLEMENTATION ========== + +static WorkItem* work_item_new(WorkItemFunc func, void* data, void* ctx) { + WorkItem* item = malloc(sizeof(WorkItem)); + if (!item) return NULL; + + item->func = func; + item->data = data; + item->ctx = ctx; + item->result = NULL; + atomic_init(&item->completed, 0); + item->mutex = NULL; + item->cond = NULL; + item->has_waiter = 0; + item->next = NULL; + + return item; +} + +void work_item_free(WorkItem* item) { + if (!item) return; + + if (item->mutex) { + pthread_mutex_destroy(item->mutex); + free(item->mutex); + } + if (item->cond) { + pthread_cond_destroy(item->cond); + free(item->cond); + } + free(item); +} + +static void work_item_setup_wait(WorkItem* item) { + if (!item->mutex) { + item->mutex = malloc(sizeof(pthread_mutex_t)); + item->cond = malloc(sizeof(pthread_cond_t)); + pthread_mutex_init(item->mutex, NULL); + pthread_cond_init(item->cond, NULL); + } + item->has_waiter = 1; +} + +static void work_item_signal_complete(WorkItem* item) { + atomic_store_explicit(&item->completed, 1, memory_order_release); + + if (item->has_waiter && item->mutex && item->cond) { + pthread_mutex_lock(item->mutex); + pthread_cond_signal(item->cond); + pthread_mutex_unlock(item->mutex); + } +} + +void* work_item_wait(WorkItem* item) { + if (!item) return NULL; + + // Fast path: already completed + if (atomic_load_explicit(&item->completed, memory_order_acquire)) { + return item->result; + } + + // Set up waiting + work_item_setup_wait(item); + + pthread_mutex_lock(item->mutex); + while (!atomic_load_explicit(&item->completed, memory_order_acquire)) { + pthread_cond_wait(item->cond, item->mutex); + } + pthread_mutex_unlock(item->mutex); + + return item->result; +} + +// ========== WORKER IMPLEMENTATION ========== + +// Get a random victim worker for stealing (excluding self) +static int worker_random_victim(Worker* worker) { + int num_workers = worker->pool->num_workers; + if (num_workers <= 1) return -1; + + // Simple LCG random number generator + worker->steal_seed = worker->steal_seed * HML_THREADPOOL_STEAL_SEED_MULT + 1; + int victim = (worker->steal_seed >> 16) % num_workers; + + // Skip self + if (victim == worker->id) { + victim = (victim + 1) % num_workers; + } + + return victim; +} + +// Try to steal work from other workers +static WorkItem* worker_steal(Worker* worker) { + ThreadPool* pool = worker->pool; + + for (int attempts = 0; attempts < HML_THREADPOOL_STEAL_ATTEMPTS; attempts++) { + int victim_id = worker_random_victim(worker); + if (victim_id < 0) break; + + Worker* victim = &pool->workers[victim_id]; + WorkItem* item = deque_steal(victim->deque); + + if (item) { + atomic_fetch_add(&worker->tasks_stolen, 1); + return item; + } + } + + return NULL; +} + +// Get next work item for this worker +static WorkItem* worker_get_work(Worker* worker) { + ThreadPool* pool = worker->pool; + + // 1. Check local deque first (LIFO - good cache locality) + WorkItem* item = deque_pop(worker->deque); + if (item) return item; + + // 2. Check global submission queue + item = submission_queue_pop(pool->submission); + if (item) return item; + + // 3. Try to steal from other workers + item = worker_steal(worker); + if (item) return item; + + return NULL; +} + +// Execute a work item +static void worker_execute(Worker* worker, WorkItem* item) { + atomic_store(&worker->active, 1); + + // Execute the work + void* result = item->func(item->data, item->ctx); + item->result = result; + + // Signal completion + work_item_signal_complete(item); + + atomic_fetch_add(&worker->tasks_executed, 1); + atomic_store(&worker->active, 0); +} + +// Worker thread main loop +static void* worker_thread_main(void* arg) { + Worker* worker = (Worker*)arg; + ThreadPool* pool = worker->pool; + + // Set thread-local worker pointer + tls_current_worker = worker; + + // Block all signals - only main thread should handle signals + sigset_t set; + sigfillset(&set); + pthread_sigmask(SIG_BLOCK, &set, NULL); + + // Signal that we're ready + pthread_mutex_lock(&pool->start_mutex); + pthread_cond_signal(&pool->start_cond); + pthread_mutex_unlock(&pool->start_mutex); + + // Main work loop + while (!atomic_load(&pool->shutdown)) { + WorkItem* item = worker_get_work(worker); + + if (item) { + worker_execute(worker, item); + + // If no waiter, free the item + if (!item->has_waiter) { + work_item_free(item); + } + } else { + // No work available - wait on submission queue with timeout + item = submission_queue_pop_wait(pool->submission, + HML_THREADPOOL_IDLE_SLEEP_US); + if (item) { + worker_execute(worker, item); + if (!item->has_waiter) { + work_item_free(item); + } + } + } + } + + // Drain remaining work before exiting + WorkItem* item; + while ((item = worker_get_work(worker)) != NULL) { + worker_execute(worker, item); + if (!item->has_waiter) { + work_item_free(item); + } + } + + tls_current_worker = NULL; + return NULL; +} + +// ========== THREAD POOL IMPLEMENTATION ========== + +static int get_cpu_count(void) { + long count = sysconf(_SC_NPROCESSORS_ONLN); + if (count <= 0) count = 4; // Default fallback + return (int)count; +} + +int thread_pool_init(int num_workers) { + pthread_mutex_lock(&g_pool_mutex); + + if (g_thread_pool) { + pthread_mutex_unlock(&g_pool_mutex); + return 0; // Already initialized + } + + // Determine number of workers + // Use 2x CPU count to handle blocking tasks (channel waits, I/O) + if (num_workers <= 0) { + num_workers = get_cpu_count() * 2; + } + if (num_workers < HML_THREADPOOL_MIN_WORKERS) { + num_workers = HML_THREADPOOL_MIN_WORKERS; + } + if (num_workers > HML_THREADPOOL_MAX_WORKERS) { + num_workers = HML_THREADPOOL_MAX_WORKERS; + } + + // Allocate pool + ThreadPool* pool = malloc(sizeof(ThreadPool)); + if (!pool) { + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + + pool->num_workers = num_workers; + atomic_init(&pool->shutdown, 0); + atomic_init(&pool->started, 0); + pthread_mutex_init(&pool->start_mutex, NULL); + pthread_cond_init(&pool->start_cond, NULL); + memset(&pool->stats, 0, sizeof(ThreadPoolStats)); + + // Create submission queue + pool->submission = submission_queue_new(HML_THREADPOOL_SUBMISSION_QUEUE_CAPACITY); + if (!pool->submission) { + free(pool); + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + + // Allocate workers + pool->workers = calloc(num_workers, sizeof(Worker)); + if (!pool->workers) { + submission_queue_free(pool->submission); + free(pool); + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + + // Initialize workers + for (int i = 0; i < num_workers; i++) { + Worker* worker = &pool->workers[i]; + worker->id = i; + worker->pool = pool; + worker->steal_seed = (unsigned int)(i * 1103515245 + 12345); + atomic_init(&worker->active, 0); + atomic_init(&worker->tasks_executed, 0); + atomic_init(&worker->tasks_stolen, 0); + + worker->deque = deque_new(HML_THREADPOOL_DEQUE_INITIAL_CAPACITY); + if (!worker->deque) { + // Cleanup on error + for (int j = 0; j < i; j++) { + deque_free(pool->workers[j].deque); + } + free(pool->workers); + submission_queue_free(pool->submission); + free(pool); + pthread_mutex_unlock(&g_pool_mutex); + return -1; + } + } + + g_thread_pool = pool; + + // Start worker threads + pthread_mutex_lock(&pool->start_mutex); + + for (int i = 0; i < num_workers; i++) { + Worker* worker = &pool->workers[i]; + int rc = pthread_create(&worker->thread, NULL, worker_thread_main, worker); + if (rc != 0) { + fprintf(stderr, "Failed to create worker thread %d: %d\n", i, rc); + // Continue with fewer workers + } + } + + // Wait for all workers to signal ready + for (int i = 0; i < num_workers; i++) { + pthread_cond_wait(&pool->start_cond, &pool->start_mutex); + } + + atomic_store(&pool->started, 1); + pthread_mutex_unlock(&pool->start_mutex); + + pthread_mutex_unlock(&g_pool_mutex); + return 0; +} + +void thread_pool_shutdown(void) { + pthread_mutex_lock(&g_pool_mutex); + + if (!g_thread_pool) { + pthread_mutex_unlock(&g_pool_mutex); + return; + } + + ThreadPool* pool = g_thread_pool; + + // Signal shutdown + atomic_store(&pool->shutdown, 1); + + // Wake up all workers waiting on submission queue + pthread_mutex_lock(&pool->submission->mutex); + pthread_cond_broadcast(&pool->submission->not_empty); + pthread_mutex_unlock(&pool->submission->mutex); + + // Wait for workers to exit + for (int i = 0; i < pool->num_workers; i++) { + pthread_join(pool->workers[i].thread, NULL); + } + + // Free workers + for (int i = 0; i < pool->num_workers; i++) { + deque_free(pool->workers[i].deque); + } + free(pool->workers); + + // Free submission queue + submission_queue_free(pool->submission); + + // Free pool + pthread_mutex_destroy(&pool->start_mutex); + pthread_cond_destroy(&pool->start_cond); + free(pool); + + g_thread_pool = NULL; + + pthread_mutex_unlock(&g_pool_mutex); +} + +WorkItem* thread_pool_submit(WorkItemFunc func, void* data, void* ctx) { + ThreadPool* pool = g_thread_pool; + + if (!pool) { + // Auto-initialize with default workers + if (thread_pool_init(HML_THREADPOOL_DEFAULT_WORKERS) != 0) { + return NULL; + } + pool = g_thread_pool; + } + + WorkItem* item = work_item_new(func, data, ctx); + if (!item) return NULL; + + // If called from a worker thread, push to local deque + if (tls_current_worker && tls_current_worker->pool == pool) { + if (deque_push(tls_current_worker->deque, item) == 0) { + return item; + } + // Fall through to submission queue if deque is full + } + + // Push to global submission queue + if (submission_queue_push(pool->submission, item) != 0) { + work_item_free(item); + return NULL; + } + + return item; +} + +void* thread_pool_submit_wait(WorkItemFunc func, void* data, void* ctx) { + WorkItem* item = thread_pool_submit(func, data, ctx); + if (!item) return NULL; + + void* result = work_item_wait(item); + work_item_free(item); + return result; +} + +int thread_pool_current_worker_id(void) { + if (tls_current_worker) { + return tls_current_worker->id; + } + return -1; +} + +ThreadPoolStats thread_pool_get_stats(void) { + ThreadPoolStats stats = {0}; + ThreadPool* pool = g_thread_pool; + + if (!pool) return stats; + + for (int i = 0; i < pool->num_workers; i++) { + stats.total_tasks_completed += atomic_load(&pool->workers[i].tasks_executed); + stats.total_steals += atomic_load(&pool->workers[i].tasks_stolen); + } + + return stats; +} + +int thread_pool_is_initialized(void) { + return g_thread_pool != NULL; +} + +ThreadPool* thread_pool_get(void) { + return g_thread_pool; +} diff --git a/src/backends/interpreter/values.c b/src/backends/interpreter/values.c index be591995..ee7f85bb 100644 --- a/src/backends/interpreter/values.c +++ b/src/backends/interpreter/values.c @@ -704,7 +704,10 @@ void task_free(Task *task) { if (task->ctx) { exec_context_free(task->ctx); } + // task->thread now holds a pthread_cond_t for join() synchronization + // (with thread pool, we use condition variable instead of pthread_t) if (task->thread) { + pthread_cond_destroy((pthread_cond_t*)task->thread); free(task->thread); } if (task->task_mutex) { diff --git a/tests/parity/builtins/thread_pool_worksteal.expected b/tests/parity/builtins/thread_pool_worksteal.expected new file mode 100644 index 00000000..1b9991eb --- /dev/null +++ b/tests/parity/builtins/thread_pool_worksteal.expected @@ -0,0 +1,3 @@ +Spawned 50 tasks +Total: i32 +done diff --git a/tests/parity/builtins/thread_pool_worksteal.hml b/tests/parity/builtins/thread_pool_worksteal.hml new file mode 100644 index 00000000..ab443f21 --- /dev/null +++ b/tests/parity/builtins/thread_pool_worksteal.hml @@ -0,0 +1,36 @@ +// Test thread pool work stealing scheduler +// Spawns many tasks to verify pool doesn't create runaway threads + +async fn heavy_task(id: i32): i32 { + // Do some work + let sum = 0; + let i = 0; + while (i < 1000) { + sum = sum + i * id; + i = i + 1; + } + return sum; +} + +// Spawn many tasks - with old pthread-per-task model this would create 50 threads +// With thread pool, only cpu_count threads are used +let tasks = []; +let i = 0; +while (i < 50) { + tasks.push(spawn(heavy_task, i)); + i = i + 1; +} + +print("Spawned 50 tasks"); + +// Collect all results +let total = 0; +i = 0; +while (i < tasks.length) { + let result = join(tasks[i]); + total = total + result; + i = i + 1; +} + +print("Total: " + typeof(total)); +print("done");