From e4dd689003b33b98d7db929b9a3b4c6c58f8fad1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Jan 2026 21:42:57 +0000 Subject: [PATCH 1/2] Implement thread pool with work-stealing scheduler This replaces the previous approach of creating a new pthread for every spawn() call with a fixed-size thread pool that reuses worker threads. Workers can steal tasks from each other to balance load automatically. Key changes: - Add work-stealing thread pool with Chase-Lev deques - Workers default to CPU count threads (configurable) - Per-worker local deques with lock-free push/pop - Random work stealing when local deque is empty - Global submission queue for external spawn() calls - Shutdown drains all pending work gracefully Both interpreter and compiler runtime use the same design, maintaining parity between the two backends. This prevents runaway thread creation when spawning many async tasks while still maintaining good parallelism through work stealing. --- include/hemlock_limits.h | 29 + include/thread_pool.h | 170 +++++ runtime/include/thread_pool.h | 115 +++ runtime/src/builtins_async.c | 76 +- runtime/src/thread_pool.c | 598 +++++++++++++++ .../interpreter/builtins/concurrency.c | 115 ++- src/backends/interpreter/internal.h | 1 + .../interpreter/runtime/thread_pool.c | 692 ++++++++++++++++++ src/backends/interpreter/values.c | 3 + .../builtins/thread_pool_worksteal.expected | 3 + .../parity/builtins/thread_pool_worksteal.hml | 36 + 11 files changed, 1760 insertions(+), 78 deletions(-) create mode 100644 include/thread_pool.h create mode 100644 runtime/include/thread_pool.h create mode 100644 runtime/src/thread_pool.c create mode 100644 src/backends/interpreter/runtime/thread_pool.c create mode 100644 tests/parity/builtins/thread_pool_worksteal.expected create mode 100644 tests/parity/builtins/thread_pool_worksteal.hml 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..4c7240b8 --- /dev/null +++ b/runtime/src/thread_pool.c @@ -0,0 +1,598 @@ +/* + * 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; + } + + if (num_workers <= 0) { + num_workers = get_cpu_count(); + } + 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..95df4a92 --- /dev/null +++ b/src/backends/interpreter/runtime/thread_pool.c @@ -0,0 +1,692 @@ +/* + * 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 + if (num_workers <= 0) { + num_workers = get_cpu_count(); + } + 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"); From 456c205a03be63e755d773b40b94c099eb0c3ef2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Jan 2026 01:44:32 +0000 Subject: [PATCH 2/2] Use 2x CPU count for thread pool workers When tasks block on channels or I/O, having only CPU count workers can cause deadlock-like situations where all workers are waiting. Using 2x CPU count provides headroom for blocking tasks. --- runtime/src/thread_pool.c | 3 ++- src/backends/interpreter/runtime/thread_pool.c | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/runtime/src/thread_pool.c b/runtime/src/thread_pool.c index 4c7240b8..b4dd8144 100644 --- a/runtime/src/thread_pool.c +++ b/runtime/src/thread_pool.c @@ -443,8 +443,9 @@ int hml_thread_pool_init(int num_workers) { return 0; } + // Use 2x CPU count to handle blocking tasks (channel waits, I/O) if (num_workers <= 0) { - num_workers = get_cpu_count(); + num_workers = get_cpu_count() * 2; } if (num_workers < HML_THREADPOOL_MIN_WORKERS) { num_workers = HML_THREADPOOL_MIN_WORKERS; diff --git a/src/backends/interpreter/runtime/thread_pool.c b/src/backends/interpreter/runtime/thread_pool.c index 95df4a92..798a090e 100644 --- a/src/backends/interpreter/runtime/thread_pool.c +++ b/src/backends/interpreter/runtime/thread_pool.c @@ -489,8 +489,9 @@ int thread_pool_init(int num_workers) { } // 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(); + num_workers = get_cpu_count() * 2; } if (num_workers < HML_THREADPOOL_MIN_WORKERS) { num_workers = HML_THREADPOOL_MIN_WORKERS;