IPC/COW fork fixes

TODO-CLAUDE.md

@@ -8,30 +8,39 @@ Minigun is a high-performance data processing pipeline framework for Ruby with s
 
 ### Phase 1.0: Cross-Boundary Routing
 
-- [ ] **Cross-Boundary Routing**
-  - Remove "skip" from hanging example tests.
-  - IPC fork getting input via `to` from various sources (IPC, COW, threads, master)
-  - IPC fork doing output routing
-  - IPC to COW, COW to IPC, IPC to master routing
-  - IPC/COW fan-out/fan-in patterns
-  - Ingress delegator for routing to inner stages
+- [x] **Cross-Boundary Routing** ✓ (Completed 2025-01-04)
+  - [x] Remove "skip" from hanging example tests
+  - [x] IPC fork getting input via `to` from various sources (IPC, COW, threads, master)
+  - [x] IPC fork doing output routing
+  - [x] IPC to COW, COW to IPC, IPC to master routing
+  - [x] IPC/COW fan-out/fan-in patterns
   - Routing patterns
-    - [ ] output.to of IpcQueues
-    - [ ] Fork/Thread etc should create an implicit pipeline
-    - [ ] cow_fork getting IPC input via to from IPC
-    - [ ] cow_fork getting IPC input via to from COW
-    - [ ] cow_fork getting IPC input via to from threads
-    - [ ] cow_fork getting IPC input via to from master(?)
-    - [ ] cow_fork doing IPC output
-    - [ ] ipc 2 cow, cow to ipc, ipc to master
-    - [ ] ipc/cow fan-out/fan-in
+    - [x] output.to of IpcQueues - implemented via IpcRoutedOutputQueue
+    - [x] cow_fork getting IPC input via to from IPC
+    - [x] cow_fork getting IPC input via to from COW
+    - [x] cow_fork getting IPC input via to from threads
+    - [x] cow_fork getting IPC input via to from master
+    - [x] cow_fork doing IPC output - COW now uses IpcOutputQueue
+    - [x] ipc 2 cow, cow to ipc, ipc to master - all working
+    - [x] ipc/cow fan-out/fan-in - examples 80, 81, 82, 84 working
     - [ ] routing to inner stages of pipelines
     - [ ] routing to inner stages of cow and ipc fork via an ingress delegator
+  - Additional scenarios
+    - [ ] test reroute with IPC/COW complex scenarios, inner routing, etc. - all tests passing
+    - [ ] producers inside IPC/COW forks
+    - [ ] routing with multiple forked processes - round-robin via IPC workers
+    - [ ] start of IPC/COW stage should not require await - added await: true option
 
 ### Phase 1.1: QoL Improvements
 
+- [ ] **Graceful shutdown**
+  - [ ] signal trapping, child state management/killing
+  - [ ] Kill child threads/forks/ractors
+  - [ ] Ctrl+C once to start graceful shutdown (send end signals from all producers)
+  - [ ] Press Ctrl+C again to force quit.
+
+
 - [ ] to_mermaid
-- [ ] signal trapping, child state management/killing
 - [ ] child culling (look at puma)
 - [ ] supervision tree of processes
 - [ ] htop-like monitoring dashboard (CLI)

examples/78_master_to_ipc_via_to.rb

@@ -21,7 +21,13 @@ class MasterToIpcViaToExample
   def initialize
     @results_a = []
     @results_b = []
-    @mutex = Mutex.new
+    @results_a_file = "/tmp/minigun_78_ipc_a_#{Process.pid}.txt"
+    @results_b_file = "/tmp/minigun_78_ipc_b_#{Process.pid}.txt"
+  end
+
+  def cleanup
+    File.unlink(@results_a_file) if File.exist?(@results_a_file)
+    File.unlink(@results_b_file) if File.exist?(@results_b_file)
   end
 
   pipeline do
@@ -50,8 +56,10 @@ def initialize
         puts "[ProcessA:ipc_fork] Processing #{item[:id]} in PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_a << item.merge(worker_pid: pid)
+        File.open(@results_a_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:value]}:#{pid}"
+          f.flock(File::LOCK_UN)
         end
       end
     end
@@ -63,8 +71,27 @@ def initialize
         puts "[ProcessB:ipc_fork] Processing #{item[:id]} in PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_b << item.merge(worker_pid: pid)
+        File.open(@results_b_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:value]}:#{pid}"
+          f.flock(File::LOCK_UN)
+        end
+      end
+    end
+
+    after_run do
+      # Read results from temp files
+      if File.exist?(@results_a_file)
+        @results_a = File.readlines(@results_a_file).map do |line|
+          id, value, pid = line.strip.split(':')
+          { id: id.to_i, value: value.to_i, worker_pid: pid.to_i }
+        end
+      end
+
+      if File.exist?(@results_b_file)
+        @results_b = File.readlines(@results_b_file).map do |line|
+          id, value, pid = line.strip.split(':')
+          { id: id.to_i, value: value.to_i, worker_pid: pid.to_i }
         end
       end
     end
@@ -111,5 +138,7 @@ def initialize
   rescue NotImplementedError => e
     puts "\nForking not available on this platform: #{e.message}"
     puts "(This is expected on Windows)"
+  ensure
+    example.cleanup
   end
 end

examples/80_ipc_fan_out.rb

@@ -23,7 +23,15 @@ def initialize
     @results_a = []
     @results_b = []
     @results_c = []
-    @mutex = Mutex.new
+    @results_a_file = "/tmp/minigun_80_ipc_a_#{Process.pid}.txt"
+    @results_b_file = "/tmp/minigun_80_ipc_b_#{Process.pid}.txt"
+    @results_c_file = "/tmp/minigun_80_ipc_c_#{Process.pid}.txt"
+  end
+
+  def cleanup
+    File.unlink(@results_a_file) if File.exist?(@results_a_file)
+    File.unlink(@results_b_file) if File.exist?(@results_b_file)
+    File.unlink(@results_c_file) if File.exist?(@results_c_file)
   end
 
   pipeline do
@@ -60,8 +68,10 @@ def initialize
         puts "[ProcessA:ipc_fork] Processing #{item[:id]} in PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_a << item.merge(worker_pid: pid)
+        File.open(@results_a_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:routed_to]}:#{item[:splitter_pid]}:#{pid}"
+          f.flock(File::LOCK_UN)
         end
       end
     end
@@ -72,8 +82,10 @@ def initialize
         puts "[ProcessB:ipc_fork] Processing #{item[:id]} in PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_b << item.merge(worker_pid: pid)
+        File.open(@results_b_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:routed_to]}:#{item[:splitter_pid]}:#{pid}"
+          f.flock(File::LOCK_UN)
         end
       end
     end
@@ -84,8 +96,34 @@ def initialize
         puts "[ProcessC:ipc_fork] Processing #{item[:id]} in PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_c << item.merge(worker_pid: pid)
+        File.open(@results_c_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:routed_to]}:#{item[:splitter_pid]}:#{pid}"
+          f.flock(File::LOCK_UN)
+        end
+      end
+    end
+
+    after_run do
+      # Read results from temp files
+      if File.exist?(@results_a_file)
+        @results_a = File.readlines(@results_a_file).map do |line|
+          id, routed_to, splitter_pid, worker_pid = line.strip.split(':')
+          { id: id.to_i, routed_to: routed_to, splitter_pid: splitter_pid.to_i, worker_pid: worker_pid.to_i }
+        end
+      end
+
+      if File.exist?(@results_b_file)
+        @results_b = File.readlines(@results_b_file).map do |line|
+          id, routed_to, splitter_pid, worker_pid = line.strip.split(':')
+          { id: id.to_i, routed_to: routed_to, splitter_pid: splitter_pid.to_i, worker_pid: worker_pid.to_i }
+        end
+      end
+
+      if File.exist?(@results_c_file)
+        @results_c = File.readlines(@results_c_file).map do |line|
+          id, routed_to, splitter_pid, worker_pid = line.strip.split(':')
+          { id: id.to_i, routed_to: routed_to, splitter_pid: splitter_pid.to_i, worker_pid: worker_pid.to_i }
         end
       end
     end
@@ -140,5 +178,7 @@ def initialize
   rescue NotImplementedError => e
     puts "\nForking not available on this platform: #{e.message}"
     puts "(This is expected on Windows)"
+  ensure
+    example.cleanup
   end
 end

examples/81_ipc_fan_in.rb

@@ -21,7 +21,11 @@ class IpcFanInExample
 
   def initialize
     @results = []
-    @mutex = Mutex.new
+    @results_file = "/tmp/minigun_81_ipc_#{Process.pid}.txt"
+  end
+
+  def cleanup
+    File.unlink(@results_file) if File.exist?(@results_file)
   end
 
   pipeline do
@@ -55,8 +59,20 @@ def initialize
         puts "[Aggregator:ipc_fork] Processing #{item[:id]} from #{item[:source]} in PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results << item.merge(worker_pid: pid)
+        File.open(@results_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:value]}:#{item[:source]}:#{pid}"
+          f.flock(File::LOCK_UN)
+        end
+      end
+    end
+
+    after_run do
+      # Read results from temp file
+      if File.exist?(@results_file)
+        @results = File.readlines(@results_file).map do |line|
+          id, value, source, worker_pid = line.strip.split(':')
+          { id: id, value: value.to_i, source: source, worker_pid: worker_pid.to_i }
         end
       end
     end
@@ -105,5 +121,7 @@ def initialize
   rescue NotImplementedError => e
     puts "\nForking not available on this platform: #{e.message}"
     puts "(This is expected on Windows)"
+  ensure
+    example.cleanup
   end
 end

examples/84_mixed_ipc_cow_fan_out.rb

@@ -23,11 +23,14 @@ def initialize
     @results_ipc_a = []
     @results_ipc_c = []
     @results_cow_b = []
-    @mutex = Mutex.new
-    @results_cow_file = "/tmp/minigun_mixed_fan_out_cow_#{Process.pid}.txt"
+    @results_ipc_a_file = "/tmp/minigun_84_ipc_a_#{Process.pid}.txt"
+    @results_ipc_c_file = "/tmp/minigun_84_ipc_c_#{Process.pid}.txt"
+    @results_cow_file = "/tmp/minigun_84_cow_b_#{Process.pid}.txt"
   end
 
   def cleanup
+    File.unlink(@results_ipc_a_file) if File.exist?(@results_ipc_a_file)
+    File.unlink(@results_ipc_c_file) if File.exist?(@results_ipc_c_file)
     File.unlink(@results_cow_file) if File.exist?(@results_cow_file)
   end
 
@@ -65,8 +68,10 @@ def cleanup
         puts "[ProcessIpcA:ipc_fork] Processing #{item[:id]} in persistent worker PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_ipc_a << item.merge(worker_pid: pid, fork_type: 'IPC')
+        File.open(@results_ipc_a_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:routed_to]}:#{pid}:IPC"
+          f.flock(File::LOCK_UN)
         end
       end
     end
@@ -81,7 +86,7 @@ def cleanup
         # COW-shared input, write to file
         File.open(@results_cow_file, 'a') do |f|
           f.flock(File::LOCK_EX)
-          f.puts "#{item[:id]}:#{pid}:COW"
+          f.puts "#{item[:id]}:#{item[:routed_to]}:#{pid}:COW"
           f.flock(File::LOCK_UN)
         end
       end
@@ -94,18 +99,34 @@ def cleanup
         puts "[ProcessIpcC:ipc_fork] Processing #{item[:id]} in persistent worker PID #{pid}"
         sleep 0.03
 
-        @mutex.synchronize do
-          @results_ipc_c << item.merge(worker_pid: pid, fork_type: 'IPC')
+        File.open(@results_ipc_c_file, 'a') do |f|
+          f.flock(File::LOCK_EX)
+          f.puts "#{item[:id]}:#{item[:routed_to]}:#{pid}:IPC"
+          f.flock(File::LOCK_UN)
         end
       end
     end
 
     after_run do
-      # Read COW results from temp file
+      # Read results from temp files
+      if File.exist?(@results_ipc_a_file)
+        @results_ipc_a = File.readlines(@results_ipc_a_file).map do |line|
+          id, routed_to, worker_pid, fork_type = line.strip.split(':')
+          { id: id.to_i, routed_to: routed_to, worker_pid: worker_pid.to_i, fork_type: fork_type }
+        end
+      end
+
       if File.exist?(@results_cow_file)
         @results_cow_b = File.readlines(@results_cow_file).map do |line|
-          id, pid, fork_type = line.strip.split(':')
-          { id: id.to_i, worker_pid: pid.to_i, fork_type: fork_type }
+          id, routed_to, worker_pid, fork_type = line.strip.split(':')
+          { id: id.to_i, routed_to: routed_to, worker_pid: worker_pid.to_i, fork_type: fork_type }
+        end
+      end
+
+      if File.exist?(@results_ipc_c_file)
+        @results_ipc_c = File.readlines(@results_ipc_c_file).map do |line|
+          id, routed_to, worker_pid, fork_type = line.strip.split(':')
+          { id: id.to_i, routed_to: routed_to, worker_pid: worker_pid.to_i, fork_type: fork_type }
         end
       end
     end