Handle closed IO while invoking #select. (#138)

Author: samuel-williams-shopify

ext/io/event/selector/uring.c

@@ -18,7 +18,7 @@
 enum {
 	DEBUG = 0,
 	DEBUG_COMPLETION = 0,
-	DEBUG_IO_READ = 1,
+	DEBUG_CQE = 0,
 };
 
 enum {URING_ENTRIES = 64};
@@ -552,7 +552,10 @@ VALUE io_wait_transfer(VALUE _arguments) {
 
 	if (DEBUG) fprintf(stderr, "io_wait_transfer:waiting=%p, result=%d\n", (void*)arguments->waiting, arguments->waiting->result);
 
-	if (arguments->waiting->result) {
+	int32_t result = arguments->waiting->result;
+	if (result < 0) {
+		rb_syserr_fail(-result, "io_wait_transfer:io_uring_poll_add");
+	} else if (result > 0) {
 		// We explicitly filter the resulting events based on the requested events.
 		// In some cases, poll will report events we didn't ask for.
 		return RB_INT2NUM(events_from_poll_flags(arguments->waiting->result & arguments->flags));
@@ -1059,7 +1062,7 @@ unsigned select_process_completions(struct IO_Event_Selector_URing *selector) {
 	}
 
 	io_uring_for_each_cqe(ring, head, cqe) {
-		if (DEBUG) fprintf(stderr, "select_process_completions: cqe res=%d user_data=%p\n", cqe->res, (void*)cqe->user_data);
+		if (DEBUG_CQE) fprintf(stderr, "select_process_completions: cqe res=%d user_data=%p\n", cqe->res, (void*)cqe->user_data);
 
 		++completed;
 

lib/io/event/selector/select.rb

@@ -421,19 +421,25 @@ def select(duration = nil)
 				writable = Array.new
 				priority = Array.new
 
-				@waiting.each do |io, waiter|
-					waiter.each do |fiber, events|
-						if (events & IO::READABLE) > 0
-							readable << io
-						end
-						
-						if (events & IO::WRITABLE) > 0
-							writable << io
-						end
-						
-						if (events & IO::PRIORITY) > 0
-							priority << io
+				@waiting.delete_if do |io, waiter|
+					if io.closed?
+						true
+					else
+						waiter.each do |fiber, events|
+							if (events & IO::READABLE) > 0
+								readable << io
+							end
+							
+							if (events & IO::WRITABLE) > 0
+								writable << io
+							end
+							
+							if (events & IO::PRIORITY) > 0
+								priority << io
+							end
 						end
+
+						false
 					end
 				end
 

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Improved consistency of handling closed IO when invoking `#select`.
+
 ## v1.10.0
 
   - `IO::Event::Profiler` is moved to dedicated gem: [fiber-profiler](https://github.com/socketry/fiber-profiler).

test/io/event/selector.rb

@@ -424,6 +424,45 @@ def transfer
 				expect(writable2).to be == true
 			end
 		end
+		
+		it "can handle io being closed by another fiber while waiting" do
+			error = nil
+			
+			wait_fiber = Fiber.new do
+				wait_fiber_started = true
+				events << :wait_readable
+				begin
+					result = selector.io_wait(Fiber.current, local, IO::READABLE)
+					$stderr.puts "result: #{result}"
+					events << :readable
+				rescue => error
+					# This isn't a reliable state transition.
+					# events << :error
+				end
+			end
+			
+			close_fiber = Fiber.new do
+				events << :close_io
+				local.close
+			end
+			
+			events << :transfer
+			wait_fiber.transfer
+			close_fiber.transfer
+			
+			expect do
+				events << :select
+				selector.select(0)
+			end.not.to raise_exception
+			
+			expect(events).to be == [
+				:transfer, :wait_readable, :close_io, :select
+			]
+
+			# io_uring will raise an EBADF error if the IO is closed while waiting.
+			# But other selectors are not capable of detecting this.
+			# expect(error).to be_nil
+		end
 	end
 
 	with "#io_read" do