diff --git a/ext/io/event/selector/uring.c b/ext/io/event/selector/uring.c index 9823f5f1..7bbaea41 100644 --- a/ext/io/event/selector/uring.c +++ b/ext/io/event/selector/uring.c @@ -563,9 +563,20 @@ 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); @@ -573,11 +584,24 @@ void IO_Event_Selector_URing_Waiting_cancel_and_wait(struct IO_Event_Selector_UR // 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, ¤t_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 diff --git a/test/io/event/selector/cancellable.rb b/test/io/event/selector/cancellable.rb index e5ea7fd8..575fe36a 100644 --- a/test/io/event/selector/cancellable.rb +++ b/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