Prevent worker-pool re-entry with Fiber.blocking

ext/io/event/worker_pool.c

@@ -1,5 +1,5 @@
 // Released under the MIT License.
-// Copyright, 2025, by Samuel Williams.
+// Copyright, 2025-2026, by Samuel Williams.
 
 #include "worker_pool.h"
 #include "worker_pool_test.h"
@@ -19,7 +19,9 @@ enum {
 };
 
 static VALUE IO_Event_WorkerPool;
+static VALUE rb_Fiber;
 static ID id_maximum_worker_count;
+static ID id_blocking;
 
 // Thread pool structure
 struct IO_Event_WorkerPool_Worker {
@@ -318,17 +320,11 @@ static VALUE worker_pool_work_block(VALUE _work) {
 static VALUE worker_pool_work_wait(VALUE _work) {
 	struct IO_Event_WorkerPool_Work *work = (void*)_work;
 
-	while (true) {
+	// A stale unblock is not cancellation. Only a non-local exit cancels work.
+	do {
 		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);
-	}
+	} while (!work->completed);
 
 	return Qtrue;
 }
@@ -354,6 +350,21 @@ static VALUE worker_pool_work_ensure(VALUE _work) {
 	return Qnil;
 }
 
+static VALUE worker_pool_work_submit(RB_BLOCK_CALL_FUNC_ARGLIST(fiber, _work)) {
+	struct IO_Event_WorkerPool_Work *work = (void*)_work;
+	struct IO_Event_WorkerPool *pool;
+	TypedData_Get_Struct(work->blocker, struct IO_Event_WorkerPool, &IO_Event_WorkerPool_type, pool);
+
+	pthread_mutex_lock(&pool->mutex);
+	enqueue_work(pool, work);
+	pthread_cond_signal(&pool->work_available);
+	pthread_mutex_unlock(&pool->mutex);
+
+	// Keep cancellation and draining inside the blocking scope too. rb_ensure
+	// preserves the original exception or other non-local exit.
+	return rb_ensure(worker_pool_work_wait, _work, worker_pool_work_ensure, _work);
+}
+
 // Ruby method to submit work and wait for completion
 static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 	struct IO_Event_WorkerPool *pool;
@@ -389,17 +400,13 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 		.fiber = fiber,
 		.next = NULL
 	};
-
-	// Enqueue work:
-	pthread_mutex_lock(&pool->mutex);
-	enqueue_work(pool, &work);
-	pthread_cond_signal(&pool->work_available);
-	pthread_mutex_unlock(&pool->mutex);
 
-	// Block the current fiber until work is completed. If blocking exits via an
-	// exception or another non-local jump, ensure the work is cancelled and fully
-	// drained before Ruby restores and rethrows the original control-flow state.
-	return rb_ensure(worker_pool_work_wait, (VALUE)&work, worker_pool_work_ensure, (VALUE)&work);
+	// A finalizer inside the scheduler's block hook must not offload another
+	// operation on this fiber and consume this work's wakeup. Fiber.blocking
+	// makes nested operations synchronous, while explicit fiber transfers still
+	// work and other fibers retain their own blocking state. Capture the scheduler
+	// above, before Fiber.blocking hides it from rb_fiber_scheduler_current().
+	return rb_block_call(rb_Fiber, id_blocking, 0, NULL, worker_pool_work_submit, (VALUE)&work);
 }
 
 static VALUE worker_pool_allocate(VALUE klass) {
@@ -495,6 +502,9 @@ static VALUE worker_pool_statistics(VALUE self) {
 void Init_IO_Event_WorkerPool(VALUE IO_Event) {
 	// Initialize symbols
 	id_maximum_worker_count = rb_intern("maximum_worker_count");
+	id_blocking = rb_intern("blocking");
+	rb_Fiber = rb_const_get(rb_cObject, rb_intern("Fiber"));
+	rb_global_variable(&rb_Fiber);
 
 	IO_Event_WorkerPool = rb_define_class_under(IO_Event, "WorkerPool", rb_cObject);
 	rb_define_alloc_func(IO_Event_WorkerPool, worker_pool_allocate);

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

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+  - Prevent nested worker-pool waits by running `WorkerPool#call` in a `Fiber.blocking` scope. Blocking operations triggered inside scheduler callbacks, including finalizers, execute synchronously. Cancellation still drains submitted work, and spurious wakeups no longer cancel operations.
   - 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.
 
 ## v1.22.1

test/io/event/worker_pool.rb

@@ -9,6 +9,56 @@
 
 return unless defined?(IO::Event::WorkerPool)
 
+# Simulate a finalizer performing a blocking operation before the scheduler suspends the fiber. The operation must run synchronously to avoid consuming the outer operation's wakeup.
+class ReentrantBlockScheduler < IO::Event::TestScheduler
+	attr_reader :nested_result, :blocking_states
+
+	def initialize(...)
+		super
+		@blocking_states = []
+	end
+
+	def block(blocker, timeout = nil)
+		@blocking_states << Fiber.current.blocking? if blocker.is_a?(IO::Event::WorkerPool)
+
+		# Simulate a finalizer that performs an offloaded blocking operation, exactly once:
+		if blocker.is_a?(IO::Event::WorkerPool) and !@reentered
+			@reentered = true
+			@nested_result = IO::Event::WorkerPool.busy(duration: 0.01)
+		end
+
+		super
+	end
+end
+
+class WaitTimeout < StandardError
+end
+
+# Exercise both exceptional exits from block and synchronous operations during cancellation cleanup.
+class InterruptedBlockScheduler < IO::Event::TestScheduler
+	attr_reader :nested_result, :blocking_states
+
+	def initialize(...)
+		super
+		@blocking_states = []
+	end
+
+	def block(blocker, timeout = nil)
+		return super unless blocker.is_a?(IO::Event::WorkerPool)
+
+		@blocking_states << Fiber.current.blocking?
+		case @blocking_states.size
+		when 1
+			raise WaitTimeout, "original interruption"
+		when 2
+			@nested_result = IO::Event::WorkerPool.busy(duration: 0.01)
+			raise WaitTimeout, "cleanup interruption"
+		end
+
+		super
+	end
+end
+
 describe IO::Event::WorkerPool do
 	with "an instance" do
 		let(:worker_pool) {subject.new}
@@ -303,6 +353,7 @@ def block(...)
 		it "preserves non-exception control flow" do
 			result = nil
 			rescued = false
+			blocking = nil
 
 			Thread.new do
 				scheduler = ThrowingBlockScheduler.new
@@ -313,6 +364,7 @@ def block(...)
 						IO::Event::WorkerPool.busy(duration: 2.0)
 						:missing
 					end
+					blocking = Fiber.current.blocking?
 				end
 				rescued = scheduler.rescued
 			ensure
@@ -322,6 +374,148 @@ def block(...)
 
 			expect(result).to be == :expected
 			expect(rescued).to be == true
+			expect(blocking).to be == false
+		end
+	end
+
+	with "blocking operations inside the scheduler block hook" do
+		it "wakes the outer operation and runs the nested operation" do
+			result = nil
+			nested_result = nil
+			statistics = nil
+			blocking_states = nil
+			blocking = nil
+
+			Thread.new do
+				# The nested operation must finish synchronously without using the single worker:
+				scheduler = ReentrantBlockScheduler.new(maximum_worker_count: 1)
+				Fiber.set_scheduler(scheduler)
+
+				fiber = Fiber.schedule do
+					result = IO::Event::WorkerPool.busy(duration: 0.1)
+					blocking = Fiber.current.blocking?
+				end
+
+				# Bound the outer wait so a lost wakeup fails the test instead of hanging it:
+				Fiber.schedule do
+					deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 2.0
+
+					while result.nil? and Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline
+						sleep(0.01)
+					end
+
+					Fiber.scheduler.fiber_interrupt(fiber, WaitTimeout) if result.nil?
+				end
+
+				scheduler.run
+				nested_result = scheduler.nested_result
+				statistics = scheduler.worker_pool.statistics
+				blocking_states = scheduler.blocking_states
+			ensure
+				Fiber.set_scheduler(nil)
+				scheduler&.close
+			end.join
+
+			# The nested operation should run to completion, rather than being cancelled when the outer operation's wakeup resumes the nested wait (depending on timing, the cancelled operation is either interrupted or skipped entirely):
+			expect(nested_result).to have_keys(
+				executed: be == true,
+				result: be == :completed
+			)
+
+			# The outer operation should complete without being interrupted by the timeout:
+			expect(result[:exception]).to be_nil
+			expect(result).to have_keys(
+				executed: be == true,
+				result: be == :completed
+			)
+			expect(statistics).to have_keys(call_count: be == 1, completed_count: be == 1, current_queue_size: be == 0)
+			expect(blocking_states).to be == [true]
+			expect(blocking).to be == false
+		end
+	end
+
+	with "a blocking wait" do
+		it "allows other fibers to offload operations while waiting" do
+			results = []
+			blocking_states = []
+			statistics = nil
+
+			Thread.new do
+				scheduler = IO::Event::TestScheduler.new(maximum_worker_count: 1)
+				Fiber.set_scheduler(scheduler)
+
+				2.times do
+					Fiber.schedule do
+						blocking_states << Fiber.current.blocking?
+						results << IO::Event::WorkerPool.busy(duration: 0.01)
+						blocking_states << Fiber.current.blocking?
+					end
+				end
+
+				# Both fibers have submitted their operations and suspended:
+				statistics = scheduler.worker_pool.statistics
+				scheduler.run
+			ensure
+				Fiber.set_scheduler(nil)
+			end.value
+
+			expect(statistics[:call_count]).to be == 2
+			expect(results.size).to be == 2
+			results.each do |result|
+				expect(result).to have_keys(executed: be == true, result: be == :completed)
+			end
+			expect(blocking_states).to be == [false, false, false, false]
+		end
+
+		it "keeps cancellation cleanup blocking and restores the fiber after an exception" do
+			result = nil
+			next_result = nil
+			blocking = nil
+			statistics = nil
+			scheduler = nil
+
+			Thread.new do
+				scheduler = InterruptedBlockScheduler.new
+				Fiber.set_scheduler(scheduler)
+
+				Fiber.schedule do
+					result = IO::Event::WorkerPool.busy(duration: 2.0)
+					blocking = Fiber.current.blocking?
+					next_result = IO::Event::WorkerPool.busy(duration: 0.01)
+				end
+
+				scheduler.run
+				statistics = scheduler.worker_pool.statistics
+			ensure
+				Fiber.set_scheduler(nil)
+			end.value
+
+			expect(result).to have_keys(result: be == :exception, exception: be_a(WaitTimeout).and(have_attributes(message: be == "original interruption")))
+			expect(scheduler.nested_result).to have_keys(executed: be == true, result: be == :completed)
+			expect(next_result).to have_keys(executed: be == true, result: be == :completed)
+			expect(statistics).to have_keys(call_count: be == 2, completed_count: be == 2, current_queue_size: be == 0)
+			expect(scheduler.blocking_states).not.to be(:include?, false)
+			expect(blocking).to be == false
+		end
+
+		it "does not cancel work on a spurious wakeup" do
+			result = nil
+
+			Thread.new do
+				scheduler = IO::Event::TestScheduler.new
+				Fiber.set_scheduler(scheduler)
+
+				fiber = Fiber.schedule do
+					result = IO::Event::WorkerPool.busy(duration: 0.1)
+				end
+
+				scheduler.unblock(nil, fiber)
+				scheduler.run
+			ensure
+				Fiber.set_scheduler(nil)
+			end.value
+
+			expect(result).to have_keys(executed: be == true, result: be == :completed)
 		end
 	end
 end