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.2"

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, is_log?, log, log_worker, procline, reopen_logs!

Constructor Details

#initialize(config_loader = nil) ⇒ Pool

Returns a new instance of Pool.



65
66
67
68
69
 
# File 'lib/resque/pool.rb', line 65

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.



118
119
120
 
# File 'lib/resque/pool.rb', line 118

def app_name
  @app_name
end

.config_loader ⇒ Object

Returns the value of attribute config_loader.



118
119
120
 
# File 'lib/resque/pool.rb', line 118

def config_loader
  @config_loader
end

.kill_other_pools ⇒ Object

Returns the value of attribute kill_other_pools.



283
284
285
 
# File 'lib/resque/pool.rb', line 283

def kill_other_pools
  @kill_other_pools
end

.spawn_delay ⇒ Object

Returns the value of attribute spawn_delay.



118
119
120
 
# File 'lib/resque/pool.rb', line 118

def spawn_delay
  @spawn_delay
end

.term_behavior ⇒ Object

Returns the value of attribute term_behavior.



282
283
284
 
# File 'lib/resque/pool.rb', line 282

def term_behavior
  @term_behavior
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



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

def config
  @config
end

#config_loader ⇒ Object (readonly)

Returns the value of attribute config_loader.



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

def config_loader
  @config_loader
end

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



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

def workers
  @workers
end

Class Method Details

.create_configured ⇒ Object 



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

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

.handle_winch=(bool) ⇒ Object 



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

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

.handle_winch? ⇒ Boolean

Returns:

  • (Boolean) 


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

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

.hook(name) ⇒ Object

Config: hooks {{{



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

def self.hook(name) # :nodoc:
  class_eval "    def self.\#{name}(&block)\n      @\#{name} ||= []\n      block ? (@\#{name} << block) : @\#{name}\n    end\n\n    def self.\#{name}=(block)\n      @\#{name} = [block]\n    end\n\n    def call_\#{name}!(*args)\n      self.class.\#{name}.each do |hook|\n        hook.call(*args)\n      end\n    end\n  CODE\nend\n"

.kill_other_pools! ⇒ Object 



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

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

.run ⇒ Object 



145
146
147
148
149
150
 
# File 'lib/resque/pool.rb', line 145

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 



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

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 



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

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



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

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



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

hook :after_spawn

#all_known_queues ⇒ Object 



419
420
421
 
# File 'lib/resque/pool.rb', line 419

def all_known_queues
  config.keys | workers.keys
end

#all_pids ⇒ Object 



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

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

#awaken_master ⇒ Object 



208
209
210
211
212
213
214
215
 
# File 'lib/resque/pool.rb', line 208

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 



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

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



389
390
391
392
393
394
395
 
# File 'lib/resque/pool.rb', line 389

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

#environment ⇒ Object 



178
179
180
181
182
183
184
185
186
 
# File 'lib/resque/pool.rb', line 178

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 



297
298
299
300
301
302
303
304
305
 
# File 'lib/resque/pool.rb', line 297

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 



286
287
288
289
290
291
292
293
294
295
 
# File 'lib/resque/pool.rb', line 286

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 



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
277
278
279
 
# File 'lib/resque/pool.rb', line 238

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



159
160
161
162
163
164
165
166
167
 
# File 'lib/resque/pool.rb', line 159

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 



197
198
199
200
201
 
# File 'lib/resque/pool.rb', line 197

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 



203
204
205
206
 
# File 'lib/resque/pool.rb', line 203

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

#join ⇒ Object 



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

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 



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

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

#maintain_worker_count ⇒ Object

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



411
412
413
414
415
416
417
 
# File 'lib/resque/pool.rb', line 411

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 



357
358
359
360
361
362
363
364
 
# File 'lib/resque/pool.rb', line 357

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 



445
446
447
 
# File 'lib/resque/pool.rb', line 445

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

#quit_excess_workers_for(queues) ⇒ Object 



434
435
436
437
438
439
 
# File 'lib/resque/pool.rb', line 434

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



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

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 



332
333
334
335
336
337
338
 
# File 'lib/resque/pool.rb', line 332

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 



173
174
175
176
 
# File 'lib/resque/pool.rb', line 173

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

#reset_sig_handlers! ⇒ Object 



234
235
236
 
# File 'lib/resque/pool.rb', line 234

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

#self_pipe ⇒ Object

Sig handlers and self pipe management {{{



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

def self_pipe; @self_pipe ||= [] end

#shutdown_everything_now!(signal) ⇒ Object 



307
308
309
310
311
312
313
314
315
 
# File 'lib/resque/pool.rb', line 307

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 



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

def sig_queue; @sig_queue ||= [] end

#signal_all_workers(signal) ⇒ Object 



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

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



427
428
429
430
431
432
 
# File 'lib/resque/pool.rb', line 427

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 



449
450
451
452
453
454
455
456
457
458
459
460
461
462
 
# File 'lib/resque/pool.rb', line 449

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



320
321
322
323
324
325
326
327
328
329
330
 
# File 'lib/resque/pool.rb', line 320

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 



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

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

#trap_deferred(signal) ⇒ Object

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



219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/resque/pool.rb', line 219

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 



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

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