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/file_or_hash_loader.rb

Defined Under Namespace

Modules: CLI, Logging, PooledWorker Classes: FileOrHashLoader, 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.0.3"

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.



62
63
64
65
66
 
# File 'lib/resque/pool.rb', line 62

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.



115
116
117
 
# File 'lib/resque/pool.rb', line 115

def app_name
  @app_name
end

.config_loader ⇒ Object

Returns the value of attribute config_loader.



115
116
117
 
# File 'lib/resque/pool.rb', line 115

def config_loader
  @config_loader
end

.kill_other_pools ⇒ Object

Returns the value of attribute kill_other_pools.



280
281
282
 
# File 'lib/resque/pool.rb', line 280

def kill_other_pools
  @kill_other_pools
end

.spawn_delay ⇒ Object

Returns the value of attribute spawn_delay.



115
116
117
 
# File 'lib/resque/pool.rb', line 115

def spawn_delay
  @spawn_delay
end

.term_behavior ⇒ Object

Returns the value of attribute term_behavior.



279
280
281
 
# File 'lib/resque/pool.rb', line 279

def term_behavior
  @term_behavior
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



58
59
60
 
# File 'lib/resque/pool.rb', line 58

def config
  @config
end

#config_loader ⇒ Object (readonly)

Returns the value of attribute config_loader.



59
60
61
 
# File 'lib/resque/pool.rb', line 59

def config_loader
  @config_loader
end

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



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

def workers
  @workers
end

Class Method Details

.create_configured ⇒ Object 



149
150
151
 
# File 'lib/resque/pool.rb', line 149

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

.handle_winch=(bool) ⇒ Object 



124
125
126
 
# File 'lib/resque/pool.rb', line 124

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

.handle_winch? ⇒ Boolean

Returns:

  • (Boolean) 


121
122
123
 
# File 'lib/resque/pool.rb', line 121

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

.hook(name) ⇒ Object

Config: hooks {{{



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/resque/pool.rb', line 70

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 



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

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

.run ⇒ Object 



142
143
144
145
146
147
 
# File 'lib/resque/pool.rb', line 142

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 



136
137
138
139
140
 
# File 'lib/resque/pool.rb', line 136

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 



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

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



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

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



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

hook :after_spawn

#all_known_queues ⇒ Object 



416
417
418
 
# File 'lib/resque/pool.rb', line 416

def all_known_queues
  config.keys | workers.keys
end

#all_pids ⇒ Object 



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

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

#awaken_master ⇒ Object 



205
206
207
208
209
210
211
212
 
# File 'lib/resque/pool.rb', line 205

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 



461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
 
# File 'lib/resque/pool.rb', line 461

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



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

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

#environment ⇒ Object 



175
176
177
178
179
180
181
182
183
 
# File 'lib/resque/pool.rb', line 175

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 



294
295
296
297
298
299
300
301
302
 
# File 'lib/resque/pool.rb', line 294

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 



283
284
285
286
287
288
289
290
291
292
 
# File 'lib/resque/pool.rb', line 283

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 



235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
 
# File 'lib/resque/pool.rb', line 235

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 {{{



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

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

#init_self_pipe! ⇒ Object 



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

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 



200
201
202
203
 
# File 'lib/resque/pool.rb', line 200

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

#join ⇒ Object 



337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
 
# File 'lib/resque/pool.rb', line 337

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 



166
167
168
 
# File 'lib/resque/pool.rb', line 166

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

#maintain_worker_count ⇒ Object

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



408
409
410
411
412
413
414
 
# File 'lib/resque/pool.rb', line 408

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 



354
355
356
357
358
359
360
361
 
# File 'lib/resque/pool.rb', line 354

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 



442
443
444
 
# File 'lib/resque/pool.rb', line 442

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

#quit_excess_workers_for(queues) ⇒ Object 



431
432
433
434
435
436
 
# File 'lib/resque/pool.rb', line 431

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 {{{



366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
 
# File 'lib/resque/pool.rb', line 366

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 



329
330
331
332
333
334
335
 
# File 'lib/resque/pool.rb', line 329

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 



170
171
172
173
 
# File 'lib/resque/pool.rb', line 170

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

#reset_sig_handlers! ⇒ Object 



231
232
233
 
# File 'lib/resque/pool.rb', line 231

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

#self_pipe ⇒ Object

Sig handlers and self pipe management {{{



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

def self_pipe; @self_pipe ||= [] end

#shutdown_everything_now!(signal) ⇒ Object 



304
305
306
307
308
309
310
311
312
 
# File 'lib/resque/pool.rb', line 304

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 



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

def sig_queue; @sig_queue ||= [] end

#signal_all_workers(signal) ⇒ Object 



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

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



424
425
426
427
428
429
 
# File 'lib/resque/pool.rb', line 424

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 



446
447
448
449
450
451
452
453
454
455
456
457
458
459
 
# File 'lib/resque/pool.rb', line 446

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 {{{



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

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 



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

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

#trap_deferred(signal) ⇒ Object

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



216
217
218
219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/resque/pool.rb', line 216

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 



438
439
440
 
# File 'lib/resque/pool.rb', line 438

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