Make channel cleanup idempotent

Assisted-By: devx/3236e566-7538-432e-a30a-2bdf37265ed4

Author: samuel-williams-shopify

lib/async/container/channel.rb

@@ -25,12 +25,14 @@ def initialize(timeout: 1.0)
 
 			# Close the input end of the pipe.
 			def close_read
-				@in.close
+				@in.close unless @in.closed?
+			rescue IOError, Errno::EBADF
 			end
 
 			# Close the output end of the pipe.
 			def close_write
-				@out.close
+				@out.close unless @out.closed?
+			rescue IOError, Errno::EBADF
 			end
 
 			# Close both ends of the pipe.

lib/async/container/forked.rb

@@ -101,14 +101,15 @@ def exec(*arguments, ready: true, **options)
 				def self.fork(**options)
 					# $stderr.puts fork: caller
 					self.new(**options) do |process|
+						# Fork from `Thread.new` so the child does not inherit the parent fiber scheduler or the current caller's fiber stack. Only this short-lived thread is copied into the child process:
 						::Thread.new do
 							::Process.fork do
 								# We use `Thread.current.raise(...)` so that exceptions are filtered through `Thread.handle_interrupt` correctly.
 								Signal.trap(:INT){::Thread.current.raise(Interrupt)}
 								Signal.trap(:TERM){::Thread.current.raise(Interrupt)}  # Same as SIGINT.
 								Signal.trap(:HUP){::Thread.current.raise(Restart)}
 
-								# This could be a configuration option:
+								# Reset `SignalException` delivery because CRuby inherits the `Thread.handle_interrupt` mask stack across `Thread.new`. Async deliberately masks signal exceptions, and the signal traps above should be delivered promptly:
 								::Thread.handle_interrupt(SignalException => :immediate) do
 									yield Instance.for(process)
 								rescue Interrupt

lib/async/container/threaded.rb

@@ -114,7 +114,7 @@ def exec(*arguments, ready: true, **options)
 				def self.fork(**options)
 					self.new(**options) do |thread|
 						::Thread.new do
-							# This could be a configuration option (see forked implementation too):
+							# Reset `SignalException` delivery because CRuby inherits the `Thread.handle_interrupt` mask stack across `Thread.new`. Async deliberately masks signal exceptions, and signal-driven thread control should be delivered promptly:
 							::Thread.handle_interrupt(SignalException => :immediate) do
 								yield Instance.for(thread)
 							end

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Make channel close operations idempotent, so cleanup paths can safely close already-closed pipe ends.
+
 ## v0.37.0
 
   - Rename `ASYNC_CONTAINER_GRACEFUL_TIMEOUT` to `ASYNC_CONTAINER_GRACEFUL_STOP` and apply it at the controller level as `GRACEFUL_STOP`. `Group#stop` now only applies the shutdown policy it is given.

test/async/container/channel.rb

@@ -30,6 +30,32 @@
 		expect(channel.receive).to be_nil
 	end
 
+	with "#close" do
+		it "can close more than once" do
+			channel.close
+			
+			expect do
+				channel.close
+			end.not.to raise_exception
+		end
+		
+		it "can close the input end after it was already closed" do
+			channel.in.close
+			
+			expect do
+				channel.close_read
+			end.not.to raise_exception
+		end
+		
+		it "can close the output end after it was already closed" do
+			channel.out.close
+			
+			expect do
+				channel.close_write
+			end.not.to raise_exception
+		end
+	end
+	
 	with "timeout" do
 		let(:channel) {subject.new(timeout: 0.001)}