Fix lost wakeup when the waiting fiber re-enters WorkerPool#call

ext/io/event/worker_pool.c

@@ -7,6 +7,7 @@
 
 #include <ruby/thread.h>
 #include <ruby/fiber/scheduler.h>
+#include <ruby/st.h>
 
 #include <pthread.h>
 #include <stdbool.h>
@@ -37,14 +38,23 @@ struct IO_Event_WorkerPool_Worker {
 
 // Work item structure
 struct IO_Event_WorkerPool_Work {
+	struct IO_Event_WorkerPool *pool;
+
 	rb_fiber_scheduler_blocking_operation_t *blocking_operation;
 
+	// Set by the worker (holding the GVL) before it unblocks the waiting fiber, exactly once:
 	bool completed;
 
+	// Set when a wakeup of the waiting fiber has been attributed to this work's completion. See `worker_pool_work_wakeup`.
+	bool woken;
+
 	VALUE scheduler;
 	VALUE blocker;
 	VALUE fiber;
 
+	// The work the same fiber was already waiting for when this work was submitted, if any. This happens when the scheduler's `block` hook performs a blocking operation before the fiber suspends, e.g. a GC finalizer that unlinks a file.
+	struct IO_Event_WorkerPool_Work *outer;
+
 	struct IO_Event_WorkerPool_Work *next;
 };
 
@@ -64,6 +74,9 @@ struct IO_Event_WorkerPool {
 	size_t completed_count;
 	size_t cancelled_count;
 
+	// Maps each fiber waiting in `call` to the innermost work it is waiting for (protected by the GVL):
+	st_table *waiting;
+
 	bool shutdown;
 };
 
@@ -83,6 +96,11 @@ static void worker_pool_free(void *ptr) {
 		// 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.
+
+		if (pool->waiting) {
+			st_free_table(pool->waiting);
+			pool->waiting = NULL;
+		}
 	}
 }
 
@@ -110,7 +128,9 @@ static void worker_pool_compact(void *ptr)
 
 // Size functions for Ruby GC
 static size_t worker_pool_size(const void *ptr) {
-	return sizeof(struct IO_Event_WorkerPool);
+	const struct IO_Event_WorkerPool *pool = (const struct IO_Event_WorkerPool *)ptr;
+
+	return sizeof(struct IO_Event_WorkerPool) + (pool->waiting ? st_memsize(pool->waiting) : 0);
 }
 
 // Ruby TypedData structures
@@ -315,27 +335,40 @@ static VALUE worker_pool_work_block(VALUE _work) {
 	return Qnil;
 }
 
+// Account for an exit from the scheduler's block hook, including an exceptional exit.
+//
+// Each completed work unblocks its fiber exactly once, but wakeups are addressed to the fiber, not to the work. If the fiber re-enters `call` while waiting for outer work, the outer work's wakeup can resume the nested wait instead, so attribute the wakeup to the innermost work (starting from `work`) which has completed but not yet been woken. If there is no such work, the wakeup is spurious (e.g. a stale `unblock`) and is ignored.
+// An exception may arrive after a completion wakeup has resumed the fiber. The scheduler doesn't tell us whether that wakeup was consumed, so account for it conservatively: replaying an unconsumed wakeup is preferable to leaving completed outer work blocked forever.
+static void worker_pool_work_wakeup(struct IO_Event_WorkerPool_Work *work) {
+	for (; work; work = work->outer) {
+		if (work->completed && !work->woken) {
+			work->woken = true;
+			return;
+		}
+	}
+}
+
 static VALUE worker_pool_work_wait(VALUE _work) {
 	struct IO_Event_WorkerPool_Work *work = (void*)_work;
-
-	while (true) {
+
+	// Only an exception or other non-local exit cancels the work (see `worker_pool_work_ensure`). Any other wakeup was either for this work, for outer work (re-issued when this call returns), or spurious, so keep waiting:
+	while (!work->woken) {
 		worker_pool_work_block(_work);
-		if (DEBUG) fprintf(stderr, "-- worker_pool_work_wait:work completed=%d\n", work->completed);
-
-		if (work->completed) {
-			break;
-		}
-
-		if (DEBUG) fprintf(stderr, "worker_pool_work_wait:rb_fiber_scheduler_blocking_operation_cancel\n");
-		rb_fiber_scheduler_blocking_operation_cancel(work->blocking_operation);
+		worker_pool_work_wakeup(work);
+		if (DEBUG) fprintf(stderr, "-- worker_pool_work_wait:work completed=%d woken=%d\n", work->completed, work->woken);
 	}
-
+	
 	return Qtrue;
 }
 
 static VALUE worker_pool_work_ensure(VALUE _work) {
 	struct IO_Event_WorkerPool_Work *work = (void*)_work;
-
+
+	// A normal return from `worker_pool_work_wait` already accounted for its wakeup. An exceptional exit may have consumed a wakeup before reaching that accounting:
+	if (!work->woken) {
+		worker_pool_work_wakeup(work);
+	}
+
 	while (!work->completed) {
 		if (DEBUG) fprintf(stderr, "worker_pool_work_ensure:rb_fiber_scheduler_blocking_operation_cancel\n");
 		rb_fiber_scheduler_blocking_operation_cancel(work->blocking_operation);
@@ -349,8 +382,26 @@ static VALUE worker_pool_work_ensure(VALUE _work) {
 			// will restore and rethrow the original control-flow state after this returns.
 			rb_set_errinfo(Qnil);
 		}
+		worker_pool_work_wakeup(work);
 	}
-
+
+	// This work is no longer being waited for:
+	st_data_t fiber = (st_data_t)work->fiber;
+	if (work->outer) {
+		st_insert(work->pool->waiting, fiber, (st_data_t)work->outer);
+	} else {
+		st_delete(work->pool->waiting, &fiber, NULL);
+	}
+
+	// If we consumed the wakeup for outer work, the fiber won't be resumed again for it, so re-issue it:
+	for (struct IO_Event_WorkerPool_Work *outer = work->outer; outer; outer = outer->outer) {
+		if (outer->woken) {
+			if (DEBUG) fprintf(stderr, "worker_pool_work_ensure:rb_fiber_scheduler_unblock outer=%p\n", outer);
+			outer->woken = false;
+			rb_fiber_scheduler_unblock(outer->scheduler, outer->blocker, outer->fiber);
+		}
+	}
+
 	return Qnil;
 }
 
@@ -382,14 +433,24 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 
 	// Create work item
 	struct IO_Event_WorkerPool_Work work = {
+		.pool = pool,
 		.blocking_operation = blocking_operation,
 		.completed = false,
+		.woken = false,
 		.scheduler = scheduler,
 		.blocker = self,
 		.fiber = fiber,
+		.outer = NULL,
 		.next = NULL
 	};
-
+
+	// Track the work this fiber is waiting for, so that a nested call can account for its wakeup:
+	st_data_t outer = 0;
+	if (st_lookup(pool->waiting, (st_data_t)fiber, &outer)) {
+		work.outer = (struct IO_Event_WorkerPool_Work *)outer;
+	}
+	st_insert(pool->waiting, (st_data_t)fiber, (st_data_t)&work);
+
 	// Enqueue work:
 	pthread_mutex_lock(&pool->mutex);
 	enqueue_work(pool, &work);
@@ -409,6 +470,8 @@ static VALUE worker_pool_allocate(VALUE klass) {
 	// Initialize to NULL/zero so we can detect uninitialized pools
 	memset(pool, 0, sizeof(struct IO_Event_WorkerPool));
 
+	pool->waiting = st_init_numtable();
+
 	return self;
 }
 

ext/io/event/worker_pool_test.c

@@ -18,6 +18,7 @@ struct BusyOperationData {
   int read_fd;
   int write_fd;
   volatile int cancelled;
+  volatile int executed;
   double duration;  // How long to wait (for testing)
   clock_t start_time;
   clock_t end_time;
@@ -29,6 +30,9 @@ struct BusyOperationData {
 static void* busy_blocking_operation(void *data) {
 	struct BusyOperationData *busy_data = (struct BusyOperationData*)data;
 
+	// Record that the operation actually ran (a cancelled operation may be skipped entirely):
+	busy_data->executed = 1;
+
 	// Use select() to wait for the pipe to become readable
 	fd_set read_fds;
 	struct timeval timeout;
@@ -143,6 +147,7 @@ static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) {
 		.read_fd = pipe_fds[0],
 		.write_fd = pipe_fds[1],
 		.cancelled = 0,
+		.executed = 0,
 		.duration = duration,
 		.start_time = 0,
 		.end_time = 0,
@@ -165,6 +170,7 @@ static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) {
 	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));
+	rb_hash_aset(result, ID2SYM(rb_intern("executed")), busy_data.executed ? Qtrue : Qfalse);
 
 	// Determine result based on operation outcome
 	if (busy_data.exception != Qnil) {

releases.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
   - Add `IO::Event::Futex` on Linux with Ruby 4.1+, including atomic value operations and blocking and scheduler-aware single and vector waits over shared memory. Futexes retain a counted allocation lock until explicitly closed or finalized; pending waits prevent closing until completion or cancellation has finished.
+  - Fix a lost wakeup in `IO::Event::WorkerPool` when the waiting fiber re-enters `WorkerPool#call` from inside the scheduler's `block` hook, e.g. a GC finalizer that closes or unlinks a `Tempfile`. The outer wait no longer hangs, including when the nested wait is interrupted after resuming, and the nested operation is no longer cancelled by an unrelated wakeup. More generally, a wakeup without an exception no longer cancels the operation; only an exception or other non-local exit does.
 
 ## v1.22.1