Add better cancellation tests.

Author: samuel-williams-shopify

ext/extconf.rb

@@ -64,6 +64,7 @@
 	if have_header("pthread.h")
 		append_cflags(["-DHAVE_IO_EVENT_WORKER_POOL"])
 		$srcs << "io/event/worker_pool.c"
+		$srcs << "io/event/worker_pool_test.c"
 	end
 end
 

ext/io/event/worker_pool.c

@@ -2,9 +2,11 @@
 // Copyright, 2025, by Samuel Williams.
 
 #include "worker_pool.h"
+#include "worker_pool_test.h"
 #include "fiber.h"
 
 #include <ruby/thread.h>
+#include <ruby/fiber/scheduler.h>
 
 #include <pthread.h>
 #include <stdbool.h>
@@ -66,33 +68,17 @@ static void worker_pool_free(void *ptr) {
 	struct IO_Event_WorkerPool *pool = (struct IO_Event_WorkerPool *)ptr;
 
 	if (pool) {
-		// Signal shutdown
-		pthread_mutex_lock(&pool->mutex);
-		pool->shutdown = true;
-		pthread_cond_broadcast(&pool->work_available);
-		pthread_mutex_unlock(&pool->mutex);
-		
-		// Wait for all workers to finish
-		struct IO_Event_WorkerPool_Worker *thread = pool->workers;
-		while (thread) {
-			rb_funcall(thread->thread, rb_intern("join"), 0);
-			struct IO_Event_WorkerPool_Worker *next = thread->next;
-			free(thread);
-			thread = next;
-		}
-		
-		// Clean up work queue
-		struct IO_Event_WorkerPool_Work *work = pool->work_queue;
-		while (work) {
-			struct IO_Event_WorkerPool_Work *next = work->next;
-			free(work);
-			work = next;
+		// Signal shutdown to all workers
+		if (!pool->shutdown) {
+			pthread_mutex_lock(&pool->mutex);
+			pool->shutdown = true;
+			pthread_cond_broadcast(&pool->work_available);
+			pthread_mutex_unlock(&pool->mutex);
 		}
 
-		pthread_mutex_destroy(&pool->mutex);
-		pthread_cond_destroy(&pool->work_available);
-		
-		free(pool);
+		// Note: We don't free worker structures or wait for threads during GC
+		// as this can cause deadlocks. The Ruby GC will handle the thread objects.
+		// Workers will see the shutdown flag and exit cleanly.
 	}
 }
 
@@ -405,4 +391,7 @@ void Init_IO_Event_WorkerPool(VALUE IO_Event) {
 	rb_define_method(IO_Event_WorkerPool, "call", worker_pool_call, 1);
 
 	rb_define_method(IO_Event_WorkerPool, "statistics", worker_pool_statistics, 0);
+	
+	// Initialize test functions
+	Init_IO_Event_WorkerPool_Test(IO_Event_WorkerPool);
 }

ext/io/event/worker_pool.h

@@ -5,8 +5,4 @@
 
 #include <ruby.h>
 
-#ifdef HAVE_RB_FIBER_SCHEDULER_BLOCKING_OPERATION_EXTRACT
-#include <ruby/fiber/scheduler.h>
-#endif
-
 void Init_IO_Event_WorkerPool(VALUE IO_Event); 

ext/io/event/worker_pool_test.c

@@ -0,0 +1,189 @@
+// worker_pool_test.c - Test functions for WorkerPool cancellation
+// Released under the MIT License.
+// Copyright, 2025, by Samuel Williams.
+
+#include "worker_pool_test.h"
+
+#include <ruby/thread.h>
+
+#include <unistd.h>
+#include <sys/select.h>
+#include <errno.h>
+#include <time.h>
+
+struct BusyOperationData {
+  int read_fd;
+  int write_fd;
+  volatile int cancelled;
+  double duration;  // How long to wait (for testing)
+  clock_t start_time;
+  clock_t end_time;
+  int operation_result;
+  VALUE exception;
+};
+
+// The actual blocking operation that can be cancelled
+static void* busy_blocking_operation(void *data) {
+	struct BusyOperationData *busy_data = (struct BusyOperationData*)data;
+	
+	// Use select() to wait for the pipe to become readable
+	fd_set read_fds;
+	struct timeval timeout;
+	
+	FD_ZERO(&read_fds);
+	FD_SET(busy_data->read_fd, &read_fds);
+	
+	// Set timeout based on duration
+	timeout.tv_sec = (long)busy_data->duration;
+	timeout.tv_usec = ((busy_data->duration - timeout.tv_sec) * 1000000);
+	
+	// This will block until:
+	// 1. The pipe becomes readable (cancellation)
+	// 2. The timeout expires
+	// 3. An error occurs
+	int result = select(busy_data->read_fd + 1, &read_fds, NULL, NULL, &timeout);
+	
+	if (result > 0 && FD_ISSET(busy_data->read_fd, &read_fds)) {
+		// Pipe became readable - we were cancelled
+		char buffer;
+		read(busy_data->read_fd, &buffer, 1);  // Consume the byte
+		busy_data->cancelled = 1;
+		return (void*)-1;  // Indicate cancellation
+	} else if (result == 0) {
+		// Timeout - operation completed normally
+		return (void*)0;  // Indicate success
+	} else {
+		// Error occurred
+		return (void*)-2;  // Indicate error
+	}
+}
+
+// Unblock function that writes to the pipe to cancel the operation
+static void busy_unblock_function(void *data) {
+	struct BusyOperationData *busy_data = (struct BusyOperationData*)data;
+	
+	// Write a byte to the pipe to wake up the select()
+	char wake_byte = 1;
+	write(busy_data->write_fd, &wake_byte, 1);
+	
+	busy_data->cancelled = 1;
+}
+
+// Function for the main operation execution (for rb_rescue)
+static VALUE busy_operation_execute(VALUE data_value) {
+	struct BusyOperationData *busy_data = (struct BusyOperationData*)data_value;
+	
+	// Record start time
+	busy_data->start_time = clock();
+	
+	// Execute the blocking operation
+	void *block_result = rb_nogvl(
+		busy_blocking_operation,
+		busy_data,
+		busy_unblock_function,
+		busy_data,
+		RB_NOGVL_UBF_ASYNC_SAFE | RB_NOGVL_OFFLOAD_SAFE
+	);
+	
+	// Record end time
+	busy_data->end_time = clock();
+	
+	// Store the operation result
+	busy_data->operation_result = (int)(intptr_t)block_result;
+	
+	return Qnil;
+}
+
+// Function for exception handling (for rb_rescue)
+static VALUE busy_operation_rescue(VALUE data_value, VALUE exception) {
+	struct BusyOperationData *busy_data = (struct BusyOperationData*)data_value;
+	
+	// Record end time even in case of exception
+	busy_data->end_time = clock();
+	
+	// Mark that an exception was caught
+	busy_data->exception = exception;
+	
+	return exception;
+}
+
+// Ruby method: IO::Event::WorkerPool.busy(duration: 1.0)
+// This creates a cancellable blocking operation for testing
+static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) {
+	VALUE options = Qnil;
+	double duration = 1.0;  // Default 1 second
+	
+	// Parse arguments
+	if (argc == 1) {
+		options = argv[0];
+		if (RB_TYPE_P(options, T_HASH)) {
+			VALUE duration_value = rb_hash_aref(options, ID2SYM(rb_intern("duration")));
+			if (!NIL_P(duration_value)) {
+				duration = NUM2DBL(duration_value);
+			}
+		} else {
+			duration = NUM2DBL(options);
+		}
+	} else if (argc > 1) {
+		rb_raise(rb_eArgError, "wrong number of arguments (given %d, expected 0..1)", argc);
+	}
+	
+	// Create pipe for cancellation
+	int pipe_fds[2];
+	if (pipe(pipe_fds) != 0) {
+		rb_sys_fail("pipe creation failed");
+	}
+	
+	// Stack allocate and initialize operation data with brace initialization
+	struct BusyOperationData busy_data = {
+		.read_fd = pipe_fds[0],
+		.write_fd = pipe_fds[1],
+		.duration = duration,
+		.exception = Qnil,
+		// All other fields are zero-initialized by default
+	};
+	
+	// Execute the blocking operation with exception handling using function pointers
+	rb_rescue(
+		busy_operation_execute,
+		(VALUE)&busy_data,
+		busy_operation_rescue,
+		(VALUE)&busy_data
+	);
+	
+	// Calculate elapsed time from the state stored in busy_data
+	double elapsed = ((double)(busy_data.end_time - busy_data.start_time)) / CLOCKS_PER_SEC;
+	
+	// Cleanup pipes
+	close(busy_data.read_fd);
+	close(busy_data.write_fd);
+	
+	// Create result hash using the state from busy_data
+	VALUE result = rb_hash_new();
+	rb_hash_aset(result, ID2SYM(rb_intern("duration")), DBL2NUM(duration));
+	rb_hash_aset(result, ID2SYM(rb_intern("elapsed")), DBL2NUM(elapsed));
+	
+	// Determine result based on operation outcome
+	if (busy_data.exception != Qnil) {
+		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("exception")));
+		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qtrue);
+		rb_hash_aset(result, ID2SYM(rb_intern("exception")), busy_data.exception);
+	} else if (busy_data.operation_result == -1) {
+		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("cancelled")));
+		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qtrue);
+	} else if (busy_data.operation_result == 0) {
+		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("completed")));
+		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qfalse);
+	} else {
+		rb_hash_aset(result, ID2SYM(rb_intern("result")), ID2SYM(rb_intern("error")));
+		rb_hash_aset(result, ID2SYM(rb_intern("cancelled")), Qfalse);
+	}
+	
+	return result;
+}
+
+// Initialize the test functions
+void Init_IO_Event_WorkerPool_Test(VALUE IO_Event_WorkerPool) {
+	// Add test methods to IO::Event::WorkerPool class
+	rb_define_singleton_method(IO_Event_WorkerPool, "busy", worker_pool_test_busy, -1);
+} 

ext/io/event/worker_pool_test.h

@@ -0,0 +1,9 @@
+// worker_pool_test.h - Header for WorkerPool test functions
+// Released under the MIT License.
+// Copyright, 2025, by Samuel Williams.
+
+#pragma once
+
+#include <ruby.h>
+
+void Init_IO_Event_WorkerPool_Test(VALUE IO_Event_WorkerPool);

fixtures/io/event/test_scheduler.rb

@@ -128,7 +128,7 @@ def kernel_sleep(duration = nil)
 		end
 
 		def fiber(&block)
-			Fiber.new(&block).tap(&:resume)
+			Fiber.new(&block).tap(&:transfer)
 		end
 
 		# Run the scheduler event loop