Wrap loop_yield() in rb_protect()

ext/io/event/selector/uring.c

@@ -563,21 +563,45 @@ void IO_Event_Selector_URing_Completion_cancel_async(struct IO_Event_Selector_UR
 	io_uring_submit_pending(selector);
 }
 
+static
+VALUE IO_Event_Selector_URing_Waiting_wait_yield(VALUE _selector)
+{
+	struct IO_Event_Selector_URing *selector = (struct IO_Event_Selector_URing *)_selector;
+
+	return IO_Event_Selector_loop_yield(&selector->backend);
+}
+
 static
 void IO_Event_Selector_URing_Waiting_cancel_and_wait(struct IO_Event_Selector_URing *selector, struct IO_Event_Selector_URing_Waiting *waiting)
 {
+	int state = 0;
+	VALUE error = Qnil;
+
 	if (waiting->completion) {
 		IO_Event_Selector_URing_Completion_cancel_async(selector, waiting->completion);
 
 		// The kernel may still be reading from or writing to the supplied buffer.
 		// Keep the C frame, buffer lock and completion record alive until the
 		// original operation CQE confirms that it can no longer access memory.
 		while (waiting->completion) {
-			IO_Event_Selector_loop_yield(&selector->backend);
+			int current_state = 0;
+			rb_protect(IO_Event_Selector_URing_Waiting_wait_yield, (VALUE)selector, &current_state);
+
+			if (current_state && !state) {
+				state = current_state;
+				error = rb_errinfo();
+			}
+
+			rb_set_errinfo(Qnil);
 		}
 	}
 
 	IO_Event_Selector_URing_Waiting_cancel(waiting);
+
+	if (state) {
+		rb_set_errinfo(error);
+		rb_jump_tag(state);
+	}
 }
 
 #pragma mark - Process.wait

test/io/event/selector/cancellable.rb

@@ -40,6 +40,45 @@
 			end
 		end
 
+		it "continues cancellation when interrupted again" do
+			skip "Requires the URing selector" unless defined?(IO::Event::Selector::URing) && selector.is_a?(IO::Event::Selector::URing)
+
+			buffer = IO::Buffer.new(64)
+			error = nil
+
+			reader = Fiber.new do
+				begin
+					if selector.method(:io_read).arity == 5
+						selector.io_read(Fiber.current, input, buffer, 0, 1)
+					else
+						selector.io_read(Fiber.current, input, buffer, 1)
+					end
+				rescue Interrupt => exception
+					error = exception
+				end
+			end
+
+			# Submit the read and leave it pending on the empty pipe:
+			reader.transfer
+
+			# The first interruption enters cancel_and_wait and yields while the
+			# original operation is still outstanding:
+			reader.raise(Interrupt)
+			expect(reader).to be(:alive?)
+
+			# A second interruption must be deferred until cancellation cleanup
+			# has consumed the original operation's completion:
+			reader.raise(Interrupt)
+			continued_cancellation = reader.alive?
+
+			if continued_cancellation
+				selector.select(0.1) while reader.alive?
+			end
+
+			expect(continued_cancellation).to be == true
+			expect(error).to be_a(Interrupt)
+		end
+
 		it "can cancel waits" do
 			skip "Single-transfer io_read does not wait for readiness" if defined?(IO::Buffer::VERSION) && IO::Buffer::VERSION >= 3