Class: Takagi::EventBus::AsyncExecutor::ProcessExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/takagi/event_bus/async_executor.rb,
sig/takagi/event_bus/async_executor.rbs

Overview

Process-based executor for multi-reactor workloads

Constant Summary collapse

Job =

Returns:

  • (Object) 

      
    

  







  
    

      Instance Method Summary
      collapse
    


    
    
  



  
Constructor Details

  
    

  

  
    #initialize(processes:, threads:)  ⇒ ProcessExecutor 
  

  

  




  

    
Returns a new instance of ProcessExecutor.


  





  
Parameters:


    
  
    
  • 
      
        processes:
      
      
        (Object)
      
      
      
    
    • 
  
    
  • 
      
        threads:
      
      
        (Object)
      
      
      
    
    • 
  






  

    
      


73
74
75
76
77
78
79
80

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 73

def initialize(processes:, threads:)
  @processes = processes.positive? ? processes : 0
  @threads = threads
  @mutex = Mutex.new
  @jobs = []
  @next_index = 0
  @needs_restart = false
end

    		
  





  




  

    
Instance Method Details


    
      

  

  
    #dispatch(handler, message)  ⇒ nil, untyped 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        handler
      
      
        (Object)
      
      
      
    
    • 
  
    
  • 
      
        message
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (nil, untyped)
      
      
      
    
    • 
  





  

    
      


191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 191

def dispatch(handler, message)
  pool_id = handler.respond_to?(:pool_id) ? handler.pool_id : nil
  if pool_id.nil? || @jobs.empty?
    handler.call(message)
    return
  end

  job = select_job
  payload = [:job, pool_id, message]
  begin
    Marshal.dump(payload, job.io)
  rescue Errno::EPIPE, IOError
    @mutex.synchronize do
      reopen_worker(job.index)
    end
    handler.call(message)
  end
end

    		
  





    
      

  

  
    #ensure_running  ⇒ nil, untyped 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (nil, untyped)
      
      
      
    
    • 
  





  

    
      


106
107
108
109
110
111
112
113

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 106

def ensure_running
  return if @processes.zero?

  @mutex.synchronize do
    restart_workers_locked if @needs_restart && @jobs.any?
    spawn_workers_locked if @jobs.empty?
  end
end

    		
  





    
      

  

  
    #fork_worker(index)  ⇒ Object 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        index
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


155
156
157
158
159
160
161
162
163
164
165
166
167

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 155

def fork_worker(index)
  reader, writer = IO.pipe

  pid = fork do
    writer.close
    run_worker(reader, index)
    exit! 0
  end

  reader.close
  writer.binmode
  Job.new(pid, writer, index)
end

    		
  





    
      

  

  
    #mark_restart_needed  ⇒ Object 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


115
116
117
118
119

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 115

def mark_restart_needed
  @mutex.synchronize do
    @needs_restart = true if @jobs.any?
  end
end

    		
  





    
      

  

  
    #post(handler, message)  ⇒ Object 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        handler
      
      
        (Object)
      
      
      
    
    • 
  
    
  • 
      
        message
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


82
83
84
85

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 82

def post(handler, message)
  ensure_running
  dispatch(handler, message)
end

    		
  





    
      

  

  
    #register_handler(_handler)  ⇒ Object 
  

  

  




  

    
Mark for restart so new handlers are visible in workers


  





  
Parameters:


    
  
    
  • 
      
        _handler
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


88
89
90

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 88

def register_handler(_handler)
  mark_restart_needed
end

    		
  





    
      

  

  
    #reopen_worker(index)  ⇒ Object 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        index
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


218
219
220
221
222
223
224
225
226
227
228
229
230
231

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 218

def reopen_worker(index)
  old = @jobs[index]
  begin
    old.io.close unless old.io.closed?
  rescue StandardError
    nil
  end
  begin
    Process.waitpid(old.pid)
  rescue StandardError
    nil
  end
  @jobs[index] = fork_worker(index)
end

    		
  





    
      

  

  
    #restart_workers_locked  ⇒ Object 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


121
122
123
124

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 121

def restart_workers_locked
  shutdown_workers
  spawn_workers_locked
end

    		
  





    
      

  

  
    #run_worker(reader, index)  ⇒ Object 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        reader
      
      
        (Object)
      
      
      
    
    • 
  
    
  • 
      
        index
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 169

def run_worker(reader, index)
  Signal.trap('TERM') { exit! 0 }
  reader.binmode
  loop do
    payload = Marshal.load(reader) # rubocop:disable Security/MarshalLoad
    type = payload[0]
    case type
    when :shutdown
      break
    when :job
      pool_id = payload[1]
      message = payload[2]
      handler = Takagi::EventBus.handler_for_pool_id(pool_id)
      handler&.call(message)
    end
  rescue EOFError, Errno::EPIPE
    break
  rescue StandardError => e
    warn "EventBus ProcessExecutor[#{index}] error: #{e.class} - #{e.message}"
  end
end

    		
  





    
      

  

  
    #select_job  ⇒ Object 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


210
211
212
213
214
215
216

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 210

def select_job
  @mutex.synchronize do
    job = @jobs[@next_index % @jobs.size]
    @next_index = (@next_index + 1) % @jobs.size
    job
  end
end

    		
  





    
      

  

  
    #shutdown  ⇒ Object 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


96
97
98

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 96

def shutdown
  @mutex.synchronize { shutdown_workers }
end

    		
  





    
      

  

  
    #shutdown_workers  ⇒ Object 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 126

def shutdown_workers
  @jobs.each do |job|
    Marshal.dump([:shutdown], job.io)
  rescue StandardError
    # ignore failures while shutting down
  ensure
    job.io.close unless job.io.closed?
    begin
      Process.kill('TERM', job.pid)
    rescue StandardError
      nil
    end
    begin
      Process.waitpid(job.pid)
    rescue StandardError
      nil
    end
  end
  @jobs.clear
end

    		
  





    
      

  

  
    #spawn_workers_locked  ⇒ nil, untyped 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        (nil, untyped)
      
      
      
    
    • 
  





  

    
      


147
148
149
150
151
152
153

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 147

def spawn_workers_locked
  return if @processes.zero?

  @jobs = Array.new(@processes) { |index| fork_worker(index) }
  @next_index = 0
  @needs_restart = false
end

    		
  





    
      

  

  
    #stats  ⇒ { mode: :processes, size: untyped } 
  

  

  




  

    

  





  

Returns:


    
  
    
  • 
      
      
        ({ mode: :processes, size: untyped })
      
      
      
    
    • 
  





  

    
      


100
101
102

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 100

def stats
  { mode: :processes, size: @jobs.size }
end

    		
  





    
      

  

  
    #unregister_handler(_handler)  ⇒ Object 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        _handler
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


92
93
94

    		
    
      
# File 'lib/takagi/event_bus/async_executor.rb', line 92

def unregister_handler(_handler)
  mark_restart_needed
end