Class: PerfectQueue::Engine

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(runner, config) ⇒ Engine

Returns a new instance of Engine.



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/perfectqueue/engine.rb', line 22

def initialize(runner, config)
  @runner = runner

  @finish_flag = BlockingFlag.new

  processor_type = config[:processor_type] || :process
  case processor_type.to_sym
  when :process
    @processor_class = Multiprocess::ForkProcessor
  when :thread
    @processor_class = Multiprocess::ThreadProcessor
  else
    raise ConfigError, "Unknown processor_type: #{config[:processor_type].inspect}"
  end

  @processors = []
  restart(false, config)
end

Instance Attribute Details

#processors ⇒ Object (readonly)

Returns the value of attribute processors.



41
42
43
 
# File 'lib/perfectqueue/engine.rb', line 41

def processors
  @processors
end

Instance Method Details

#join ⇒ Object 



96
97
98
99
 
# File 'lib/perfectqueue/engine.rb', line 96

def join
  @processors.each {|c| c.join }
  self
end

#logrotated ⇒ Object 



113
114
115
 
# File 'lib/perfectqueue/engine.rb', line 113

def logrotated
  @processors.each {|c| c.logrotated }
end

#replace(immediate, command = [$0]+ARGV) ⇒ Object 



106
107
108
109
110
111
 
# File 'lib/perfectqueue/engine.rb', line 106

def replace(immediate, command=[$0]+ARGV)
  return if @replaced_pid
  stop(immediate)
  @replaced_pid = Process.spawn(*command)
  self
end

#restart(immediate, config) ⇒ Object 



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/perfectqueue/engine.rb', line 43

def restart(immediate, config)
  return nil if @finish_flag.set?

  # TODO connection check

  @log = config[:logger] || Logger.new(STDERR)

  num_processors = config[:processors] || 1

  # scaling
  extra = num_processors - @processors.length
  if extra > 0
    extra.times do
      @processors << @processor_class.new(@runner, @processors.size+1, config)
    end
  elsif extra < 0
    (-extra).times do
      c = @processors.shift
      c.stop(immediate)
      c.join
    end
    extra = 0
  end

  @processors[0..(-extra-1)].each {|c|
    c.restart(immediate, config)
  }

  @child_keepalive_interval = (config[:child_keepalive_interval] || config[:child_heartbeat_interval] || 2).to_i

  self
end

#run ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/perfectqueue/engine.rb', line 76

def run
  @processors.each {|c|
    c.keepalive
    # add wait time before starting processors to avoid
    # a spike of the number of concurrent connections.
    sleep rand*2  # upto 2 second, average 1 seoncd
  }
  until @finish_flag.set?
    @processors.each {|c| c.keepalive }
    @finish_flag.wait(@child_keepalive_interval)
  end
  join
end

#shutdown(immediate) ⇒ Object 



101
102
103
104
 
# File 'lib/perfectqueue/engine.rb', line 101

def shutdown(immediate)
  stop(immediate)
  join
end

#stop(immediate) ⇒ Object 



90
91
92
93
94
 
# File 'lib/perfectqueue/engine.rb', line 90

def stop(immediate)
  @processors.each {|c| c.stop(immediate) }
  @finish_flag.set!
  self
end