Class: Resque::Worker

Inherits:
Object
  • Object
show all
Includes:
Plugins::MultiJobForks::RssReader
Defined in:
lib/resque-multi-job-forks.rb

Constant Summary

Constants included from Plugins::MultiJobForks::RssReader

Plugins::MultiJobForks::RssReader::LINUX, Plugins::MultiJobForks::RssReader::PS_CMD, Plugins::MultiJobForks::RssReader::UNITS

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Plugins::MultiJobForks::RssReader

#rss, #rss_linux, #rss_posix

Instance Attribute Details

#jobs_per_fork ⇒ Object

Returns the value of attribute jobs_per_fork.



9
10
11
 
# File 'lib/resque-multi-job-forks.rb', line 9

def jobs_per_fork
  @jobs_per_fork
end

#jobs_processed ⇒ Object (readonly)

Returns the value of attribute jobs_processed.



11
12
13
 
# File 'lib/resque-multi-job-forks.rb', line 11

def jobs_processed
  @jobs_processed
end

#memory_threshold ⇒ Object

Returns the value of attribute memory_threshold.



10
11
12
 
# File 'lib/resque-multi-job-forks.rb', line 10

def memory_threshold
  @memory_threshold
end

#seconds_per_fork ⇒ Object

Returns the value of attribute seconds_per_fork.



8
9
10
 
# File 'lib/resque-multi-job-forks.rb', line 8

def seconds_per_fork
  @seconds_per_fork
end

Class Method Details

.multi_jobs_per_fork? ⇒ Boolean

Returns:

  • (Boolean) 


13
14
15
 
# File 'lib/resque-multi-job-forks.rb', line 13

def self.multi_jobs_per_fork?
  ENV["DISABLE_MULTI_JOBS_PER_FORK"].nil?
end

Instance Method Details

#fork_hijacked? ⇒ Boolean

Returns:

  • (Boolean) 


95
96
97
 
# File 'lib/resque-multi-job-forks.rb', line 95

def fork_hijacked?
  @release_fork_limit
end

#fork_job_limit ⇒ Object 



118
119
120
 
# File 'lib/resque-multi-job-forks.rb', line 118

def fork_job_limit
  jobs_per_fork.nil? ? Time.now.to_f + seconds_per_fork : jobs_per_fork
end

#fork_job_limit_reached? ⇒ Boolean

Returns:

  • (Boolean) 


122
123
124
 
# File 'lib/resque-multi-job-forks.rb', line 122

def fork_job_limit_reached?
  fork_job_limit_remaining <= 0 || fork_job_over_memory_threshold?
end

#fork_job_limit_remaining ⇒ Object 



126
127
128
 
# File 'lib/resque-multi-job-forks.rb', line 126

def fork_job_limit_remaining
  jobs_per_fork.nil? ? @release_fork_limit - Time.now.to_f : jobs_per_fork - @jobs_processed
end

#fork_job_over_memory_threshold? ⇒ Boolean

Returns:

  • (Boolean) 


142
143
144
 
# File 'lib/resque-multi-job-forks.rb', line 142

def fork_job_over_memory_threshold?
  !!(memory_threshold && rss > memory_threshold)
end

#hijack_fork ⇒ Object 



99
100
101
102
103
104
105
106
107
 
# File 'lib/resque-multi-job-forks.rb', line 99

def hijack_fork
  log 'hijack fork.'
  @suppressed_fork_hooks = [Resque.after_fork, Resque.before_fork]
  Resque.after_fork = Resque.before_fork = nil
  @release_fork_limit = fork_job_limit
  @jobs_processed = 0
  @cant_fork = true
  trap('QUIT') { shutdown }
end

#is_parent_process? ⇒ Boolean

Returns:

  • (Boolean) 


86
87
88
 
# File 'lib/resque-multi-job-forks.rb', line 86

def is_parent_process?
  @child || @pid == Process.pid
end

#minutes_per_fork ⇒ Object 



134
135
136
 
# File 'lib/resque-multi-job-forks.rb', line 134

def minutes_per_fork
  ENV['MINUTES_PER_FORK'].nil? ? 1 : ENV['MINUTES_PER_FORK'].to_i
end

#pause_processing_with_multi_job_forks ⇒ Object Also known as: pause_processing 



49
50
51
52
 
# File 'lib/resque-multi-job-forks.rb', line 49

def pause_processing_with_multi_job_forks
  shutdown_child if is_parent_process?
  pause_processing_without_multi_job_forks
end

#perform_with_multi_job_forks(job = nil) ⇒ Object Also known as: perform 



26
27
28
29
30
31
 
# File 'lib/resque-multi-job-forks.rb', line 26

def perform_with_multi_job_forks(job = nil)
  trap('QUIT') { shutdown } unless fork_hijacked?
  perform_without_multi_job_forks(job)
  hijack_fork unless fork_hijacked?
  @jobs_processed += 1
end

#reconnect_with_multi_job_forks ⇒ Object Also known as: reconnect

Reconnect only once



64
65
66
67
68
69
 
# File 'lib/resque-multi-job-forks.rb', line 64

def reconnect_with_multi_job_forks
  unless @reconnected
    reconnect_without_multi_job_forks
    @reconnected = true
  end
end

#release_and_exit! ⇒ Object 



90
91
92
93
 
# File 'lib/resque-multi-job-forks.rb', line 90

def release_and_exit!
  release_fork if fork_hijacked?
  run_at_exit_hooks ? exit : exit!(true)
end

#release_fork ⇒ Object 



109
110
111
112
113
114
115
116
 
# File 'lib/resque-multi-job-forks.rb', line 109

def release_fork
  log "jobs processed by child: #{jobs_processed}; rss: #{rss}"
  run_hook :before_child_exit, self
  Resque.after_fork, Resque.before_fork = *@suppressed_fork_hooks
  @release_fork_limit = @jobs_processed = @cant_fork = nil
  log 'hijack over, counter terrorists win.'
  @shutdown = true unless $TESTING
end

#shutdown_child ⇒ Object

Need to tell the child to shutdown since it might be looping performing multiple jobs per fork. The QUIT signal normally does a graceful shutdown, and is re-registered in children (term_child normally unregisters it).



77
78
79
80
81
82
83
84
 
# File 'lib/resque-multi-job-forks.rb', line 77

def shutdown_child
  begin
    log! "multi_jobs_per_fork: Sending QUIT signal to #{@child}"
    Process.kill('QUIT', @child)
  rescue Errno::ESRCH
    nil
  end
end

#shutdown_with_multi_job_forks ⇒ Object Also known as: shutdown 



42
43
44
45
 
# File 'lib/resque-multi-job-forks.rb', line 42

def shutdown_with_multi_job_forks
  shutdown_child if is_parent_process?
  shutdown_without_multi_job_forks
end

#shutdown_with_multi_job_forks? ⇒ Boolean Also known as: shutdown?

Returns:

  • (Boolean) 


35
36
37
38
 
# File 'lib/resque-multi-job-forks.rb', line 35

def shutdown_with_multi_job_forks?
  release_fork if fork_hijacked? && (fork_job_limit_reached? || @shutdown)
  shutdown_without_multi_job_forks?
end

#work_with_multi_job_forks(*args) ⇒ Object Also known as: work 



18
19
20
21
22
 
# File 'lib/resque-multi-job-forks.rb', line 18

def work_with_multi_job_forks(*args)
  pid # forces @pid to be set in the parent
  work_without_multi_job_forks(*args)
  release_and_exit! unless is_parent_process?
end

#working_on_with_worker_registration(job) ⇒ Object Also known as: working_on 



56
57
58
59
 
# File 'lib/resque-multi-job-forks.rb', line 56

def working_on_with_worker_registration(job)
  register_worker
  working_on_without_worker_registration(job)
end