Module: ConcurrentWorker::ConcurrentProcess

Defined in:
lib/concurrent_worker/worker.rb

Instance Method Summary collapse

Instance Method Details

#channel_close ⇒ Object 



280
281
282
 
# File 'lib/concurrent_worker/worker.rb', line 280

def channel_close
  @ipc_channel.close
end

#cncr_block ⇒ Object 



243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/concurrent_worker/worker.rb', line 243

def cncr_block
  @ipc_channel = IPCDuplexChannel.new
  @c_pid = fork do
    @ipc_channel.choose_io
    Thread.handle_interrupt(Object => :never) do
      begin
        Thread.handle_interrupt(Object => :immediate) do
          yield_base_block
        end
      rescue
        @ipc_channel.send($!)
      ensure
        @ipc_channel.send(nil)
      end
    end
  end
  @ipc_channel.choose_io
  @recv_thread = result_handle_thread do
    ipc_recv_loop
  end
end

#ipc_recv_loop ⇒ Object 



235
236
237
238
239
240
241
 
# File 'lib/concurrent_worker/worker.rb', line 235

def ipc_recv_loop
  while result = @ipc_channel.recv
    raise result if result.kind_of?(Exception)

    call_result_callbacks(result)
  end
end

#receive_req ⇒ Object 



270
271
272
273
 
# File 'lib/concurrent_worker/worker.rb', line 270

def receive_req
  #called from worker process only
  @ipc_channel.recv
end

#send_req(args) ⇒ Object 



265
266
267
268
 
# File 'lib/concurrent_worker/worker.rb', line 265

def send_req(args)
  #called from main process only
  @ipc_channel.send(args)
end

#send_res(args) ⇒ Object 



275
276
277
278
 
# File 'lib/concurrent_worker/worker.rb', line 275

def send_res(args)
  #called from worker process only
  @ipc_channel.send(args)
end

#wait_cncr_proc ⇒ Object 



284
285
286
287
288
289
290
 
# File 'lib/concurrent_worker/worker.rb', line 284

def wait_cncr_proc
  begin
    Process.waitpid(@c_pid)
  rescue Errno::ECHILD
  end
  @recv_thread && @recv_thread.join
end