Class: PerfectQueue::Multiprocess::ForkProcessor

Inherits:
Object
  • Object
show all
Defined in:
lib/perfectqueue/multiprocess/fork_processor.rb

Instance Method Summary collapse

Constructor Details

#initialize(runner, processor_id, config) ⇒ ForkProcessor

Returns a new instance of ForkProcessor.



23
24
25
26
27
28
29
30
31
32
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 23

def initialize(runner, processor_id, config)
  @runner = runner
  @processor_id = processor_id

  @stop = false
  @cpm = nil
  @last_fork_time = 0

  restart(false, config)
end

Instance Method Details

#join ⇒ Object 



97
98
99
100
101
102
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 97

def join
  while !try_join
    sleep (@child_kill_interval+1) / 2  # TODO
  end
  self
end

#keepalive ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 56

def keepalive
  if @stop
    try_join
    return
  end

  if c = @cpm
    if c.killing_status != true
      # don't check status if killing status is immediate-killing
      begin
        # receive heartbeat
        keptalive = c.check_heartbeat(@child_heartbeat_limit)
        if !keptalive
          @log.error "Heartbeat broke out. Restarting child process id=#{@processor_id} pid=#{c.pid}."
          c.start_killing(true)
        end
      rescue EOFError
        @log.error "Heartbeat pipe is closed. Restarting child process id=#{@processor_id} pid=#{c.pid}."
        c.start_killing(true, @child_heartbeat_kill_delay)
      rescue
        @log.error "Unknown error: #{$!.class}: #{$!}: Restarting child process id=#{@processor_id} pid=#{c.pid}."
        $!.backtrace.each {|bt| @log.warn "\t#{bt}" }
        c.start_killing(true, @child_heartbeat_kill_delay)
      end
    end

    try_join
  end

  unless @cpm
    begin
      @cpm = fork_child
    rescue
      @log.error "Failed to fork child process id=#{@processor_id}: #{$!.class}: #{$!}"
      $!.backtrace.each {|bt| @log.warn "\t#{bt}" }
    end
  end

  nil
end

#logrotated ⇒ Object 



104
105
106
107
108
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 104

def logrotated
  if c = @cpm
    c.send_signal(:CONT)
  end
end

#restart(immediate, config) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 34

def restart(immediate, config)
  @child_heartbeat_limit = config[:child_heartbeat_limit] || 60.0
  @child_kill_interval = config[:child_kill_interval] || 2.0
  @child_graceful_kill_limit = config[:child_graceful_kill_limit] || nil
  @child_fork_frequency_limit = config[:child_fork_frequency_limit] || 5.0
  @child_heartbeat_kill_delay = config[:child_heartbeat_kill_delay] || 10
  @log = config[:logger]
  @config = config  # for child process

  if c = @cpm
    c.start_killing(immediate)
  end
end

#stop(immediate) ⇒ Object 



48
49
50
51
52
53
54
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 48

def stop(immediate)
  @stop = true
  if c = @cpm
    c.start_killing(immediate)
  end
  self
end