Module: ConcurrentWorker::ConcurrentProcess

Defined in:

lib/concurrent_worker.rb

Defined Under Namespace

Classes: IPCDuplexChannel

Instance Method Summary

• #cncr_block ⇒ Object
• #receive_req ⇒ Object
• #send_req(args) ⇒ Object
• #send_res(args) ⇒ Object
• #set_rcv_thread ⇒ Object
• #wait_cncr_proc ⇒ Object

Instance Method Details

#cncr_block ⇒ Object

# File 'lib/concurrent_worker.rb', line 327

def cncr_block
  @ipc_channel = IPCDuplexChannel.new
  @c_pid = fork do
    @ipc_channel.choose_io
    begin
      yield_base_block
    rescue
      @ipc_channel.send($!)
    ensure
      @ipc_channel.send(:worker_loop_finished)
    end
  end
  @ipc_channel.choose_io
  @rcv_thread = set_rcv_thread
end

#receive_req ⇒ Object

# File 'lib/concurrent_worker.rb', line 348

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

#send_req(args) ⇒ Object

# File 'lib/concurrent_worker.rb', line 343

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

#send_res(args) ⇒ Object

# File 'lib/concurrent_worker.rb', line 353

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

#set_rcv_thread ⇒ Object

# File 'lib/concurrent_worker.rb', line 302

def set_rcv_thread
  Thread.new do
    Thread.handle_interrupt(Object => :never) do
      begin
        Thread.handle_interrupt(Object => :immediate) do
          loop do
            result = @ipc_channel.recv
            break if result == :worker_loop_finished
            raise result if result.kind_of?(Exception)

            call_result_callbacks(result)
          end
        end
      rescue
        Thread.pass
        raise $!
      ensure
        @req_counter.close
        @ipc_channel.close
        call_retired_callbacks
      end
    end
  end
end

#wait_cncr_proc ⇒ Object

# File 'lib/concurrent_worker.rb', line 358

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