Class: Resque::Pool

Inherits:
Object
  • Object
show all
Extended by:
Logging
Includes:
Logging
Defined in:
lib/resque/pool.rb,
lib/resque/pool/cli.rb,
lib/resque/pool/killer.rb,
lib/resque/pool/logging.rb,
lib/resque/pool/version.rb,
lib/resque/pool/pooled_worker.rb,
lib/resque/pool/config_loaders/throttled.rb,
lib/resque/pool/config_loaders/file_or_hash_loader.rb

Defined Under Namespace

Modules: CLI, ConfigLoaders, Logging, PooledWorker Classes: Killer, QuitNowException

Constant Summary collapse

SIG_QUEUE_MAX_SIZE = 
5
DEFAULT_WORKER_INTERVAL = 
5
QUEUE_SIGS = 
[ :QUIT, :INT, :TERM, :USR1, :USR2, :CONT, :HUP, :WINCH, ]
CHUNK_SIZE = 
(16 * 1024)
VERSION = 
"0.7.1"

Constants included from Logging

Logging::PROCLINE_PREFIX

Class Attribute Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Logging

app, log, log_worker, procline, reopen_logs!

Constructor Details

#initialize(config_loader = nil) ⇒ Pool

Returns a new instance of Pool.



25
26
27
28
29
 
# File 'lib/resque/pool.rb', line 25

def initialize(config_loader=nil)
  init_config(config_loader)
  @workers = Hash.new { |workers, queues| workers[queues] = {} }
  procline "(initialized)"
end

Class Attribute Details

.app_name ⇒ Object

Returns the value of attribute app_name.



78
79
80
 
# File 'lib/resque/pool.rb', line 78

def app_name
  @app_name
end

.config_loader ⇒ Object

Returns the value of attribute config_loader.



78
79
80
 
# File 'lib/resque/pool.rb', line 78

def config_loader
  @config_loader
end

.kill_other_pools ⇒ Object

Returns the value of attribute kill_other_pools.



243
244
245
 
# File 'lib/resque/pool.rb', line 243

def kill_other_pools
  @kill_other_pools
end

.spawn_delay ⇒ Object

Returns the value of attribute spawn_delay.



78
79
80
 
# File 'lib/resque/pool.rb', line 78

def spawn_delay
  @spawn_delay
end

.term_behavior ⇒ Object

Returns the value of attribute term_behavior.



242
243
244
 
# File 'lib/resque/pool.rb', line 242

def term_behavior
  @term_behavior
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



21
22
23
 
# File 'lib/resque/pool.rb', line 21

def config
  @config
end

#config_loader ⇒ Object (readonly)

Returns the value of attribute config_loader.



22
23
24
 
# File 'lib/resque/pool.rb', line 22

def config_loader
  @config_loader
end

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



23
24
25
 
# File 'lib/resque/pool.rb', line 23

def workers
  @workers
end

Class Method Details

.create_configured ⇒ Object 



112
113
114
 
# File 'lib/resque/pool.rb', line 112

def self.create_configured
  Resque::Pool.new(config_loader)
end

.handle_winch=(bool) ⇒ Object 



87
88
89
 
# File 'lib/resque/pool.rb', line 87

def self.handle_winch=(bool)
  @handle_winch = bool
end

.handle_winch? ⇒ Boolean

Returns:

  • (Boolean) 


84
85
86
 
# File 'lib/resque/pool.rb', line 84

def self.handle_winch?
  @handle_winch ||= false
end

.hook(name) ⇒ Object

Config: hooks {{{



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/resque/pool.rb', line 33

def self.hook(name) # :nodoc:
  class_eval <<-CODE
    def self.#{name}(&block)
      @#{name} ||= []
      block ? (@#{name} << block) : @#{name}
    end

    def self.#{name}=(block)
      @#{name} = [block]
    end

    def call_#{name}!(*args)
      self.class.#{name}.each do |hook|
        hook.call(*args)
      end
    end
  CODE
end

.kill_other_pools! ⇒ Object 



91
92
93
94
 
# File 'lib/resque/pool.rb', line 91

def self.kill_other_pools!
  require 'resque/pool/killer'
  Resque::Pool::Killer.run
end

.run ⇒ Object 



105
106
107
108
109
110
 
# File 'lib/resque/pool.rb', line 105

def self.run
  if GC.respond_to?(:copy_on_write_friendly=)
    GC.copy_on_write_friendly = true
  end
  create_configured.start.join
end

.single_process_group ⇒ Object 



99
100
101
102
103
 
# File 'lib/resque/pool.rb', line 99

def self.single_process_group
  %w[yes y true t 1 okay sure please].include?(
    ENV["RESQUE_SINGLE_PGRP"].to_s.downcase
  )
end

.single_process_group=(bool) ⇒ Object 



96
97
98
 
# File 'lib/resque/pool.rb', line 96

def self.single_process_group=(bool)
  ENV["RESQUE_SINGLE_PGRP"] = !!bool ? "YES" : "NO"
end

Instance Method Details

#after_prefork ⇒ Object

:call-seq:

after_prefork do |worker| ... end   => add a hook
after_prefork << hook               => add a hook

after_prefork will run in workers before any jobs. Use these hooks e.g. to reload database connections to ensure that they’re not shared among workers.

:yields: worker



62
 
# File 'lib/resque/pool.rb', line 62

hook :after_prefork

#after_spawn ⇒ Object

:call-seq:

after_prefork do |worker, pid, workers| ... end  => add a hook
after_prefork << hook                            => add a hook

The ‘after_spawn` hooks will run in the master after spawning a new worker.

:yields: worker, pid, workers



73
 
# File 'lib/resque/pool.rb', line 73

hook :after_spawn

#all_known_queues ⇒ Object 



379
380
381
 
# File 'lib/resque/pool.rb', line 379

def all_known_queues
  config.keys | workers.keys
end

#all_pids ⇒ Object 



357
358
359
 
# File 'lib/resque/pool.rb', line 357

def all_pids
  workers.map {|q,workers| workers.keys }.flatten
end

#awaken_master ⇒ Object 



168
169
170
171
172
173
174
175
 
# File 'lib/resque/pool.rb', line 168

def awaken_master
  begin
    self_pipe.last.write_nonblock('.') # wakeup master process from select
  rescue Errno::EAGAIN, Errno::EINTR
    # pipe is full, master should wake up anyways
    retry
  end
end

#create_worker(queues) ⇒ Object 



424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
 
# File 'lib/resque/pool.rb', line 424

def create_worker(queues)
  queues = queues.to_s.split(',')
  worker = ::Resque::Worker.new(*queues)
  worker.pool_master_pid = Process.pid
  worker.term_timeout = (ENV['RESQUE_TERM_TIMEOUT'] || 4.0).to_f
  worker.term_child = ENV['TERM_CHILD']
  if worker.respond_to?(:run_at_exit_hooks=)
    # resque doesn't support this until 1.24, but we support 1.22
    worker.run_at_exit_hooks = ENV['RUN_AT_EXIT_HOOKS'] || false
  end
  if ENV['LOGGING'] || ENV['VERBOSE']
    worker.verbose = ENV['LOGGING'] || ENV['VERBOSE']
  end
  if ENV['VVERBOSE']
    worker.very_verbose = ENV['VVERBOSE']
  end
  worker
end

#delete_worker(pid) ⇒ Object

TODO: close any file descriptors connected to worker, if any



349
350
351
352
353
354
355
 
# File 'lib/resque/pool.rb', line 349

def delete_worker(pid)
  worker = nil
  workers.detect do |queues, pid_to_worker|
    worker = pid_to_worker.delete(pid)
  end
  worker
end

#environment ⇒ Object 



138
139
140
141
142
143
144
145
146
 
# File 'lib/resque/pool.rb', line 138

def environment
  if defined? RAILS_ENV
    RAILS_ENV
  elsif defined?(Rails) && Rails.respond_to?(:env)
    Rails.env
  else
    ENV['RACK_ENV'] || ENV['RAILS_ENV'] || ENV['RESQUE_ENV']
  end
end

#graceful_worker_shutdown!(signal) ⇒ Object 



257
258
259
260
261
262
263
264
265
 
# File 'lib/resque/pool.rb', line 257

def graceful_worker_shutdown!(signal)
  log "#{signal}: immediate shutdown (graceful worker shutdown)"
  if term_child
    signal_all_workers(:TERM)
  else
    signal_all_workers(:QUIT)
  end
  :break
end

#graceful_worker_shutdown_and_wait!(signal) ⇒ Object 



246
247
248
249
250
251
252
253
254
255
 
# File 'lib/resque/pool.rb', line 246

def graceful_worker_shutdown_and_wait!(signal)
  log "#{signal}: graceful shutdown, waiting for children"
  if term_child
    signal_all_workers(:TERM)
  else
    signal_all_workers(:QUIT)
  end
  reap_all_workers(0) # will hang until all workers are shutdown
  :break
end

#handle_sig_queue! ⇒ Object 



198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/resque/pool.rb', line 198

def handle_sig_queue!
  case signal = sig_queue.shift
  when :USR1, :USR2, :CONT
    log "#{signal}: sending to all workers"
    signal_all_workers(signal)
  when :HUP
    log "HUP: reset configuration and reload logfiles"
    reset_config
    Logging.reopen_logs!
    log "HUP: gracefully shutdown old children (which have old logfiles open)"
    if term_child
      signal_all_workers(:TERM)
    else
      signal_all_workers(:QUIT)
    end
    log "HUP: new children will inherit new logfiles"
    maintain_worker_count
  when :WINCH
    if self.class.handle_winch?
      log "WINCH: gracefully stopping all workers"
      @config = {}
      maintain_worker_count
    end
  when :QUIT
    if term_child
      shutdown_everything_now!(signal)
    else
      graceful_worker_shutdown_and_wait!(signal)
    end
  when :INT
    graceful_worker_shutdown!(signal)
  when :TERM
    case self.class.term_behavior
    when "graceful_worker_shutdown_and_wait"
      graceful_worker_shutdown_and_wait!(signal)
    when "graceful_worker_shutdown"
      graceful_worker_shutdown!(signal)
    else
      shutdown_everything_now!(signal)
    end
  end
end

#init_config(loader) ⇒ Object

}}} Config: store loader and load config {{{



119
120
121
122
123
124
125
126
127
 
# File 'lib/resque/pool.rb', line 119

def init_config(loader)
  case loader
  when String, Hash, nil
    @config_loader = ConfigLoaders::FileOrHashLoader.new(loader)
  else
    @config_loader = loader
  end
  load_config
end

#init_self_pipe! ⇒ Object 



157
158
159
160
161
 
# File 'lib/resque/pool.rb', line 157

def init_self_pipe!
  self_pipe.each { |io| io.close rescue nil }
  self_pipe.replace(IO.pipe)
  self_pipe.each { |io| io.fcntl(Fcntl::F_SETFD, Fcntl::FD_CLOEXEC) }
end

#init_sig_handlers! ⇒ Object 



163
164
165
166
 
# File 'lib/resque/pool.rb', line 163

def init_sig_handlers!
  QUEUE_SIGS.each { |sig| trap_deferred(sig) }
  trap(:CHLD)     { |_| awaken_master }
end

#join ⇒ Object 



300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
 
# File 'lib/resque/pool.rb', line 300

def join
  loop do
    reap_all_workers
    break if handle_sig_queue! == :break
    if sig_queue.empty?
      master_sleep
      load_config
      maintain_worker_count
    end
    procline("managing #{all_pids.inspect}")
  end
  procline("(shutting down)")
  #stop # gracefully shutdown all workers on our way out
  log "manager finished"
  #unlink_pid_safe(pid) if pid
end

#load_config ⇒ Object 



129
130
131
 
# File 'lib/resque/pool.rb', line 129

def load_config
  @config = config_loader.call(environment)
end

#maintain_worker_count ⇒ Object

}}} ???: maintain_worker_count, all_known_queues {{{



371
372
373
374
375
376
377
 
# File 'lib/resque/pool.rb', line 371

def maintain_worker_count
  all_known_queues.each do |queues|
    delta = worker_delta_for(queues)
    spawn_missing_workers_for(queues) if delta > 0
    quit_excess_workers_for(queues)   if delta < 0
  end
end

#master_sleep ⇒ Object 



317
318
319
320
321
322
323
324
 
# File 'lib/resque/pool.rb', line 317

def master_sleep
  begin
    ready = IO.select([self_pipe.first], nil, nil, 1) or return
    ready.first && ready.first.first or return
    loop { self_pipe.first.read_nonblock(CHUNK_SIZE) }
  rescue Errno::EAGAIN, Errno::EINTR
  end
end

#pids_for(queues) ⇒ Object 



405
406
407
 
# File 'lib/resque/pool.rb', line 405

def pids_for(queues)
  workers[queues].keys
end

#quit_excess_workers_for(queues) ⇒ Object 



394
395
396
397
398
399
 
# File 'lib/resque/pool.rb', line 394

def quit_excess_workers_for(queues)
  delta = -worker_delta_for(queues)
  pids_for(queues)[0...delta].each do |pid|
    Process.kill("QUIT", pid)
  end
end

#reap_all_workers(waitpid_flags = Process::WNOHANG) ⇒ Object

}}} worker process management {{{



329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
 
# File 'lib/resque/pool.rb', line 329

def reap_all_workers(waitpid_flags=Process::WNOHANG)
  @waiting_for_reaper = waitpid_flags == 0
  begin
    loop do
      # -1, wait for any child process
      wpid, status = Process.waitpid2(-1, waitpid_flags)
      break unless wpid

      if worker = delete_worker(wpid)
        log "Reaped resque worker[#{status.pid}] (status: #{status.exitstatus}) queues: #{worker.queues.join(",")}"
      else
        # this died before it could be killed, so it's not going to have any extra info
        log "Tried to reap worker [#{status.pid}], but it had already died. (status: #{status.exitstatus})"
      end
    end
  rescue Errno::ECHILD, QuitNowException
  end
end

#report_worker_pool_pids ⇒ Object 



292
293
294
295
296
297
298
 
# File 'lib/resque/pool.rb', line 292

def report_worker_pool_pids
  if workers.empty?
    log "Pool is empty"
  else
    log "Pool contains worker PIDs: #{all_pids.inspect}"
  end
end

#reset_config ⇒ Object 



133
134
135
136
 
# File 'lib/resque/pool.rb', line 133

def reset_config
  config_loader.reset! if config_loader.respond_to?(:reset!)
  load_config
end

#reset_sig_handlers! ⇒ Object 



194
195
196
 
# File 'lib/resque/pool.rb', line 194

def reset_sig_handlers!
  QUEUE_SIGS.each {|sig| trap(sig, "DEFAULT") }
end

#self_pipe ⇒ Object

Sig handlers and self pipe management {{{



152
 
# File 'lib/resque/pool.rb', line 152

def self_pipe; @self_pipe ||= [] end

#shutdown_everything_now!(signal) ⇒ Object 



267
268
269
270
271
272
273
274
275
 
# File 'lib/resque/pool.rb', line 267

def shutdown_everything_now!(signal)
  log "#{signal}: immediate shutdown (and immediate worker shutdown)"
  if term_child
    signal_all_workers(:QUIT)
  else
    signal_all_workers(:TERM)
  end
  :break
end

#sig_queue ⇒ Object 



153
 
# File 'lib/resque/pool.rb', line 153

def sig_queue; @sig_queue ||= [] end

#signal_all_workers(signal) ⇒ Object 



361
362
363
364
365
366
 
# File 'lib/resque/pool.rb', line 361

def signal_all_workers(signal)
  log "Sending #{signal} to all workers"
  all_pids.each do |pid|
    Process.kill signal, pid
  end
end

#spawn_missing_workers_for(queues) ⇒ Object

}}} methods that operate on a single grouping of queues {{{ perhaps this means a class is waiting to be extracted



387
388
389
390
391
392
 
# File 'lib/resque/pool.rb', line 387

def spawn_missing_workers_for(queues)
  worker_delta_for(queues).times do |nr|
    spawn_worker!(queues)
    sleep Resque::Pool.spawn_delay if Resque::Pool.spawn_delay
  end
end

#spawn_worker!(queues) ⇒ Object 



409
410
411
412
413
414
415
416
417
418
419
420
421
422
 
# File 'lib/resque/pool.rb', line 409

def spawn_worker!(queues)
  worker = create_worker(queues)
  pid = fork do
    Process.setpgrp unless Resque::Pool.single_process_group
    worker.worker_parent_pid = Process.pid
    log_worker "Starting worker #{worker}"
    call_after_prefork!(worker)
    reset_sig_handlers!
    #self_pipe.each {|io| io.close }
    worker.work(ENV['INTERVAL'] || DEFAULT_WORKER_INTERVAL) # interval, will block
  end
  workers[queues][pid] = worker
  call_after_spawn!(worker, pid, workers)
end

#start ⇒ Object

}}} start, join, and master sleep {{{



280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/resque/pool.rb', line 280

def start
  procline("(starting)")
  init_self_pipe!
  init_sig_handlers!
  maintain_worker_count
  procline("(started)")
  log "started manager"
  report_worker_pool_pids
  self.class.kill_other_pools! if self.class.kill_other_pools
  self
end

#term_child ⇒ Object 



154
 
# File 'lib/resque/pool.rb', line 154

def term_child; @term_child ||= ENV['TERM_CHILD'] end

#trap_deferred(signal) ⇒ Object

defer a signal for later processing in #join (master process)



179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/resque/pool.rb', line 179

def trap_deferred(signal)
  trap(signal) do |sig_nr|
    if @waiting_for_reaper && [:INT, :TERM].include?(signal)
      log "Recieved #{signal}: short circuiting QUIT waitpid"
      raise QuitNowException
    end
    if sig_queue.size < SIG_QUEUE_MAX_SIZE
      sig_queue << signal
      awaken_master
    else
      log "ignoring SIG#{signal}, queue=#{sig_queue.inspect}"
    end
  end
end

#worker_delta_for(queues) ⇒ Object 



401
402
403
 
# File 'lib/resque/pool.rb', line 401

def worker_delta_for(queues)
  config.fetch(queues, 0) - workers.fetch(queues, []).size
end