WIP

Author: ioquatix

examples/behaviour/a.out

@@ -0,0 +1,68 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <sys/select.h>
+#include <errno.h>
+#include <string.h>
+
+static int pipefd[2];
+
+void* select_thread(void* arg) {
+    fd_set readfds;
+    struct timeval tv;
+    int ret;
+
+    FD_ZERO(&readfds);
+    FD_SET(pipefd[0], &readfds);
+
+    /* Set a timeout, so select won't block forever if something unexpected happens. */
+    tv.tv_sec = 5;
+    tv.tv_usec = 0;
+
+    printf("Thread: calling select()...\n");
+
+    ret = select(pipefd[0] + 1, &readfds, NULL, NULL, &tv);
+    if (ret == -1) {
+        /* You will often see errno = EBADF after the close() in the other thread. */
+        printf("Thread: select() returned -1, errno=%d (%s)\n", errno, strerror(errno));
+    } else if (ret == 0) {
+        printf("Thread: select() timed out\n");
+    } else {
+        printf("Thread: select() returned %d (pipefd[0] is readable)\n", ret);
+    }
+
+    return NULL;
+}
+
+int main(void) {
+    pthread_t tid;
+    int ret;
+
+    if (pipe(pipefd) == -1) {
+        perror("pipe");
+        exit(EXIT_FAILURE);
+    }
+
+    /* Create a thread that calls select() on the read end of the pipe. */
+    ret = pthread_create(&tid, NULL, select_thread, NULL);
+    if (ret != 0) {
+        fprintf(stderr, "pthread_create failed: %s\n", strerror(ret));
+        exit(EXIT_FAILURE);
+    }
+
+    /* Give the select() thread time to block. */
+    sleep(1);
+
+    /* Close the read end of the pipe from the main thread. */
+    printf("Main: closing pipefd[0]...\n");
+    close(pipefd[0]);
+
+    /* Wait for the select thread to finish. */
+    pthread_join(tid, NULL);
+
+    /* Clean up the write end (though it's not strictly necessary here). */
+    close(pipefd[1]);
+
+    return 0;
+}

examples/behaviour/select_read_close.c

@@ -220,8 +220,10 @@ def io_wait(io, events, timeout = nil)
 				end
 			end
 
+			Fiber.blocking{$stderr.puts "-> (#{fiber}) io_wait(#{io}, #{events}, #{timeout})"}
 			return @selector.io_wait(fiber, io, events)
 		ensure
+			Fiber.blocking{$stderr.puts "<- io_wait(#{io}, #{events}, #{timeout})"}
 			timer&.cancel!
 		end
 
@@ -304,6 +306,7 @@ def transfer
 		# @parameter fiber [Fiber] The fiber to raise the exception on.
 		# @parameter exception [Exception] The exception to raise.
 		def fiber_interrupt(fiber, exception)
+			Fiber.blocking{$stderr.puts "fiber_interrupt(#{fiber}, #{exception})"}
 			unblock(nil, RaiseException.new(fiber, exception))
 		end
 

fixtures/io/event/scheduler.rb

@@ -426,6 +426,7 @@ def process_wait(fiber, pid, flags)
 			end
 
 			def select(duration = nil)
+				Fiber.blocking{$stderr.puts "-> Selecting for #{duration.inspect}..."}
 				if pop_ready
 					# If we have popped items from the ready list, they may influence the duration calculation, so we don't delay the event loop:
 					duration = 0
@@ -436,6 +437,10 @@ def select(duration = nil)
 				priority = Array.new
 
 				@waiting.each do |io, waiter|
+					# If an IO is closed, we can no longer wait on it:
+					next if io.closed?
+					# This is consistent with epoll/kqueue behavior, which will remove the IO from the set of watched IOs if it is closed. This can be a problem if the IO is closed during `pop_ready` above, while another Fiber is waiting on it. The interruption to `io_wait` will be delivered on the next iteration of the event loop, but we can't use it with `IO.select` as it will raise an exception and break the event loop.
+					
 					waiter.each do |fiber, events|
 						if (events & IO::READABLE) > 0
 							readable << io
@@ -463,6 +468,7 @@ def select(duration = nil)
 				# We need to handle interrupts on blocking IO. Every other implementation uses EINTR, but that doesn't work with `::IO.select` as it will retry the call on EINTR.
 				Thread.handle_interrupt(::Exception => :on_blocking) do
 					@blocked = true
+					Fiber.blocking{$stderr.puts "-> IO.select(#{readable.inspect}, #{writable.inspect}, #{priority.inspect}, #{duration.inspect})..."}
 					readable, writable, priority = ::IO.select(readable, writable, priority, duration)
 				rescue ::Exception => error
 					# Requeue below...

lib/io/event/selector/select.rb

@@ -0,0 +1,16 @@
+#!/usr/bin/env ruby
+
+r, w = IO.pipe
+
+thread = Thread.new do
+	# pp IO.select([r])
+	pp r.wait_readable
+rescue => error
+	pp error
+end
+
+Thread.pass until thread.status == "sleep"
+
+r.close
+
+binding.irb

test.rb

@@ -70,28 +70,50 @@ def with_scheduler
 				Fiber.set_scheduler(nil)
 			end
 
-			it "can interrupt reads" do
+			def skip_unless_interruptable
 				skip("IO#close interruption unsupported") unless IO::Event::INTERRUPTABLE
+			end
+			
+			it "can interrupt read" do
+				skip_unless_interruptable
 
 				error = nil
 
 				with_scheduler do
 					buffer = IO::Buffer.new(64)
 
 					Fiber.schedule do
-						$stderr.puts "Reading..."
 						begin
 							buffer.read(input, 1)
 						rescue => error
-						ensure
-							$stderr.puts "Read: #{error}"
+							# Ignore.
+						end
+					end
+					
+					Fiber.schedule do
+						input.close
+					end
+				end
+				
+				expect(error).to be_a(IOError)
+			end
+			
+			it "can interrupt wait_readable" do
+				skip_unless_interruptable
+				
+				error = nil
+				
+				with_scheduler do
+					Fiber.schedule do
+						begin
+							input.wait_readable
+						rescue => error
+							# Ignore.
 						end
 					end
 
 					Fiber.schedule do
-						$stderr.puts "Closing input..."
 						input.close
-						$stderr.puts "Closed input."
 					end
 				end