Ensure work can be cancelled while still in the queue.

Author: ioquatix

ext/io/event/worker_pool.c

@@ -336,19 +336,32 @@ static VALUE worker_pool_call(VALUE self, VALUE _blocking_operation) {
 	pthread_mutex_unlock(&pool->mutex);
 
 	// Block the current fiber until work is completed:
-	int state;
+	int state = 0;
 	while (true) {
-		rb_protect(worker_pool_work_begin, (VALUE)&work, &state);
-
+		int current_state = 0;
+		rb_protect(worker_pool_work_begin, (VALUE)&work, &current_state);
+		if (DEBUG) fprintf(stderr, "-- worker_pool_call:work completed=%d, current_state=%d, state=%d\n", work.completed, current_state, state);
+		
+		// Store the first exception state:
+		if (!state) {
+			state = current_state;
+		}
+		
+		// If the work is still in the queue, we must wait for a worker to complete it (even if cancelled):
 		if (work.completed) {
+			// The work was completed, we can exit the loop:
 			break;
 		} else {
 			if (DEBUG) fprintf(stderr, "worker_pool_call:rb_fiber_scheduler_blocking_operation_cancel\n");
+			// Ensure the blocking operation is cancelled:
 			rb_fiber_scheduler_blocking_operation_cancel(blocking_operation);
-			// The work was not completed, we need to wait for it to be completed.
+			
+			// The work was not completed, we need to wait for it to be completed, so we go around the loop again.
 		}
 	}
 
+	if (DEBUG) fprintf(stderr, "<- worker_pool_call:work completed=%d, state=%d\n", work.completed, state);
+	
 	if (state) {
 		rb_jump_tag(state);
 	} else {

test/io/event/worker_pool.rb

@@ -8,7 +8,7 @@
 
 return unless defined?(IO::Event::WorkerPool)
 
-describe IO::Event::WorkerPool do		
+describe IO::Event::WorkerPool do
 	with "an instance" do
 		let(:worker_pool) {subject.new}
 
@@ -61,7 +61,7 @@
 		end
 	end
 
-	with "TestScheduler integration" do
+	with IO::Event::TestScheduler do
 		let(:scheduler) {IO::Event::TestScheduler.new}
 
 		it "can create a test scheduler" do
@@ -102,7 +102,7 @@
 
 	with "cancellable busy operation" do
 		let(:scheduler) {IO::Event::TestScheduler.new}
-
+		
 		it "can perform a busy operation that completes normally" do
 			start_time = Time.now
 			result = IO::Event::WorkerPool.busy(duration: 0.1)
@@ -144,7 +144,7 @@
 			# The operation should have been interrupted before completion
 			expect(completed).to be == false
 		end
-				
+		
 		it "can be cancelled when executed in a worker pool" do
 			result = nil
 			elapsed = nil
@@ -173,5 +173,27 @@
 			expect(elapsed).to be < 1.0
 			expect(error).to be_nil
 		end
+		
+		it "can be cancelled before even starting" do
+			result = nil
+			
+			Thread.new do
+				Fiber.set_scheduler(scheduler)
+				
+				busy_fiber = Fiber.schedule do
+					result = IO::Event::WorkerPool.busy(duration: 2.0)
+				end
+				
+				Fiber.schedule do
+					Fiber.scheduler.fiber_interrupt(busy_fiber, StandardError)
+				end
+			end.join
+			
+			expect(result).to have_keys(
+				cancelled: be == true,
+				result: be == :exception,
+				exception: be_a(StandardError)
+			)
+		end
 	end
 end