Class: MiniScheduler::Manager::Runner
- Inherits:
-
Object
- Object
- MiniScheduler::Manager::Runner
- Defined in:
- lib/mini_scheduler/manager.rb
Instance Method Summary collapse
- #attempts(max_attempts) ⇒ Object
- #current_worker_thread_id ⇒ Object
- #enq(klass) ⇒ Object
- #ensure_worker_threads ⇒ Object
- #hostname ⇒ Object
-
#initialize(manager) ⇒ Runner
constructor
A new instance of Runner.
- #keep_alive(*ids) ⇒ Object
- #process_queue ⇒ Object
- #repair_queue ⇒ Object
- #reschedule_orphans ⇒ Object
- #set_current_worker_thread_id! ⇒ Object
- #stop! ⇒ Object
- #wait_till_done ⇒ Object
- #worker_loop ⇒ Object
- #worker_thread_ids ⇒ Object
Constructor Details
#initialize(manager) ⇒ Runner
Returns a new instance of Runner.
8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
# File 'lib/mini_scheduler/manager.rb', line 8 def initialize(manager) @stopped = false @mutex = Mutex.new @queue = Queue.new @manager = manager @hostname = manager.hostname @recovery_thread = Thread.new do while !@stopped sleep 60 @mutex.synchronize do repair_queue reschedule_orphans ensure_worker_threads end end end @keep_alive_thread = Thread.new do while !@stopped @mutex.synchronize do keep_alive end sleep (@manager.keep_alive_duration / 2) end end ensure_worker_threads end |
Instance Method Details
#attempts(max_attempts) ⇒ Object
193 194 195 196 197 198 199 200 201 202 203 |
# File 'lib/mini_scheduler/manager.rb', line 193 def attempts(max_attempts) attempt = 0 begin yield rescue attempt += 1 raise if attempt >= max_attempts sleep Random.rand retry end end |
#current_worker_thread_id ⇒ Object
82 83 84 |
# File 'lib/mini_scheduler/manager.rb', line 82 def current_worker_thread_id Thread.current[:mini_scheduler_worker_thread_id] end |
#enq(klass) ⇒ Object
177 178 179 |
# File 'lib/mini_scheduler/manager.rb', line 177 def enq(klass) @queue << klass end |
#ensure_worker_threads ⇒ Object
55 56 57 58 59 60 61 62 63 |
# File 'lib/mini_scheduler/manager.rb', line 55 def ensure_worker_threads @threads ||= [] @threads.delete_if { |t| !t.alive? } (@manager.workers - @threads.size).times do @threads << Thread.new { worker_loop } end rescue => ex MiniScheduler.handle_job_exception(ex, message: "Error during MiniScheduler ensure_worker_threads") end |
#hostname ⇒ Object
78 79 80 |
# File 'lib/mini_scheduler/manager.rb', line 78 def hostname @hostname end |
#keep_alive(*ids) ⇒ Object
37 38 39 40 41 |
# File 'lib/mini_scheduler/manager.rb', line 37 def keep_alive(*ids) @manager.keep_alive(*ids) rescue => ex MiniScheduler.handle_job_exception(ex, message: "Error during MiniScheduler keep_alive") end |
#process_queue ⇒ Object
94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
# File 'lib/mini_scheduler/manager.rb', line 94 def process_queue klass = @queue.deq # hack alert, I need to both deq and set @running atomically. @running = true return if !klass failed = false start = Time.now.to_f info = @mutex.synchronize { @manager.schedule_info(klass) } stat = nil error = nil begin info.prev_result = "RUNNING" info.current_owner = current_worker_thread_id @mutex.synchronize { info.write! } if @manager.enable_stats stat = MiniScheduler::Stat.create!( name: klass.to_s, hostname: hostname, pid: Process.pid, started_at: Time.now, live_slots_start: GC.stat[:heap_live_slots] ) end klass.new.perform rescue => e MiniScheduler.handle_job_exception(e, message: "Error while running a scheduled job", job: { "class" => klass }) error = "#{e.class}: #{e.} #{e.backtrace.join("\n")}" failed = true end duration = ((Time.now.to_f - start) * 1000).to_i info.prev_duration = duration info.prev_result = failed ? "FAILED" : "OK" info.current_owner = nil if stat stat.update!( duration_ms: duration, live_slots_finish: GC.stat[:heap_live_slots], success: !failed, error: error ) MiniScheduler.job_ran&.call(stat) end attempts(3) do @mutex.synchronize { info.write! } end ensure @running = false if defined?(ActiveRecord::Base) ActiveRecord::Base.connection_handler.clear_active_connections! end end |
#repair_queue ⇒ Object
43 44 45 46 47 |
# File 'lib/mini_scheduler/manager.rb', line 43 def repair_queue @manager.repair_queue rescue => ex MiniScheduler.handle_job_exception(ex, message: "Error during MiniScheduler repair_queue") end |
#reschedule_orphans ⇒ Object
49 50 51 52 53 |
# File 'lib/mini_scheduler/manager.rb', line 49 def reschedule_orphans @manager.reschedule_orphans! rescue => ex MiniScheduler.handle_job_exception(ex, message: "Error during MiniScheduler reschedule_orphans") end |
#set_current_worker_thread_id! ⇒ Object
86 87 88 |
# File 'lib/mini_scheduler/manager.rb', line 86 def set_current_worker_thread_id! Thread.current[:mini_scheduler_worker_thread_id] = "#{@manager.identity_key}:thread_#{SecureRandom.alphanumeric(10)}" end |
#stop! ⇒ Object
152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
# File 'lib/mini_scheduler/manager.rb', line 152 def stop! return if @stopped @mutex.synchronize do @stopped = true @keep_alive_thread.kill @recovery_thread.kill @keep_alive_thread.join @recovery_thread.join enq(nil) kill_thread = Thread.new do sleep 0.5 @threads.each(&:kill) end @threads.each(&:join) kill_thread.kill kill_thread.join end end |
#wait_till_done ⇒ Object
181 182 183 184 185 186 187 188 189 190 191 |
# File 'lib/mini_scheduler/manager.rb', line 181 def wait_till_done while !@queue.empty? && !(@queue.num_waiting > 0) sleep 0.001 end # this is a hack, but is only used for test anyway # if tests fail that depend on this we are forced to increase it. sleep 0.010 while @running sleep 0.001 end end |
#worker_loop ⇒ Object
65 66 67 68 69 70 71 72 73 74 75 76 |
# File 'lib/mini_scheduler/manager.rb', line 65 def worker_loop set_current_worker_thread_id! keep_alive(current_worker_thread_id) while !@stopped begin process_queue rescue => ex MiniScheduler.handle_job_exception(ex, message: "Error during MiniScheduler worker_loop") break # Data could be in a bad state - stop the thread end end end |
#worker_thread_ids ⇒ Object
90 91 92 |
# File 'lib/mini_scheduler/manager.rb', line 90 def worker_thread_ids @threads.filter(&:alive?).filter_map { |t| t[:mini_scheduler_worker_thread_id] } end |