Improved implementation of blocking_operation_wait.

Author: ioquatix

fixtures/io/event/test_scheduler.rb

@@ -32,7 +32,11 @@ module IO::Event
 	class TestScheduler
 		def initialize(selector: nil, worker_pool: nil, maximum_worker_count: nil)
 			@selector = selector || ::IO::Event::Selector.new(Fiber.current)
-			@worker_pool = worker_pool || WorkerPool.new(maximum_worker_count: maximum_worker_count)
+			
+			if ::IO::Event.const_defined?(:WorkerPool)
+				@worker_pool = worker_pool || ::IO::Event::WorkerPool.new(maximum_worker_count: maximum_worker_count)
+			end
+			
 			@timers = ::IO::Event::Timers.new
 
 			# Track the number of fibers that are blocked.
@@ -45,10 +49,15 @@ def initialize(selector: nil, worker_pool: nil, maximum_worker_count: nil)
 		# @attribute [IO::Event::Selector] The I/O event selector used for managing fiber scheduling.
 		attr_reader :selector
 
-		# Required fiber scheduler hook - delegates to WorkerPool
-		def blocking_operation_wait(operation)
-			# Submit the operation to the worker pool and wait for completion
-			@worker_pool.call(operation)
+		if ::IO::Event.const_defined?(:WorkerPool)
+			# Optional fiber scheduler hook - delegates to WorkerPool:
+			def blocking_operation_wait(operation)
+				@blocked += 1
+				# Submit the operation to the worker pool and wait for completion
+				@worker_pool&.call(operation)
+			ensure
+				@blocked -= 1
+			end
 		end
 
 		# Required fiber scheduler hooks
@@ -114,7 +123,12 @@ def io_wait(io, events, timeout = nil)
 				end
 			end
 
-			return @selector.io_wait(fiber, io, events)
+			begin
+				@blocked += 1
+				return @selector.io_wait(fiber, io, events)
+			ensure
+				@blocked -= 1
+			end
 		ensure
 			timer&.cancel!
 		end

lib/io/event/debug/selector.rb

@@ -128,6 +128,12 @@ def ready?
 				@selector.ready?
 			end
 
+			# Run the given blocking operation and wait for its completion.
+			def blocking_operation_wait(operation)
+				log("Waiting for blocking operation #{operation.inspect}")
+				@selector.blocking_operation_wait(operation)
+			end
+			
 			# Wait for the given process, forwarded to the underlying selector.
 			def process_wait(*arguments)
 				log("Waiting for process with #{arguments.inspect}")

test/tcp_socket.rb

@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "io/event"
+require "io/event/test_scheduler"
+
+require "socket"
+
+describe TCPSocket do
+	let(:scheduler) {IO::Event::TestScheduler.new}
+	
+	it "can read and write data" do
+		chunk_size = 1024*6
+		buffer_size = 1024*64
+		
+		server_socket = TCPServer.new("localhost", 0)
+		port = server_socket.addr[1]
+		
+		client = TCPSocket.new("localhost", port)
+		server = server_socket.accept
+		
+		Fiber.set_scheduler(scheduler)
+		
+		writers = Thread::Queue.new
+		2.times do |i|
+			Fiber.schedule do
+				buffer = i.to_s * chunk_size
+				
+				128.times do
+					server.write(buffer)
+					server.flush
+				end
+				
+				writers << :done
+			end
+		end
+		
+		Fiber.schedule do
+			2.times do
+				writers.pop
+			end
+			
+			server.close
+		end
+		
+		Fiber.schedule do
+			while result = client.read_nonblock(buffer_size, exception: false)
+				case result
+				when :wait_readable
+					client.wait_readable
+				when :wait_writable
+					client.wait_writable
+				else
+					# Done.
+				end
+			end
+		end
+		
+		scheduler.run
+	ensure
+		Fiber.set_scheduler(nil)
+	end
+end