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
33
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 23

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

  require 'fcntl'
  @stop = false
  @cpm = nil
  @last_fork_time = 0

  restart(false, config)
end

Instance Method Details

#join ⇒ Object 



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

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

#keepalive ⇒ Object 



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
96
 
# File 'lib/perfectqueue/multiprocess/fork_processor.rb', line 57

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 



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

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

#restart(immediate, config) ⇒ Object 



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

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 



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

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