Class: Wurk::Processor::SharedWorkState

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/processor.rb

Overview

tid → { queue:, payload:, run_at: } for every Processor currently running a job. Read by Heartbeat each beat to publish into Redis.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ SharedWorkState

Returns a new instance of SharedWorkState.



129
130
131
132
 
# File 'lib/wurk/processor.rb', line 129

def initialize
  @work = {}
  @lock = ::Mutex.new
end

Instance Method Details

#clear ⇒ Object 



163
164
165
 
# File 'lib/wurk/processor.rb', line 163

def clear
  @lock.synchronize { @work.clear }
end

#delete(tid) ⇒ Object 



138
139
140
 
# File 'lib/wurk/processor.rb', line 138

def delete(tid)
  @lock.synchronize { @work.delete(tid) }
end

#dup ⇒ Object 



155
156
157
 
# File 'lib/wurk/processor.rb', line 155

def dup
  @lock.synchronize { @work.dup }
end

#set(tid, hash) ⇒ Object 



134
135
136
 
# File 'lib/wurk/processor.rb', line 134

def set(tid, hash)
  @lock.synchronize { @work[tid] = hash }
end

#size ⇒ Object 



159
160
161
 
# File 'lib/wurk/processor.rb', line 159

def size
  @lock.synchronize { @work.size }
end

#track(tid, hash) ⇒ Object

RAII: publish for the duration of the block, always retract. The write lives inside this method's ensure frame on purpose — with the set one line above a begin, an async raise landing in that gap (a host timeout's Thread#raise, say) escapes the ensure and strands the entry for the life of the process: the payload String stays reachable and every heartbeat reports the thread as busy.



148
149
150
151
152
153
 
# File 'lib/wurk/processor.rb', line 148

def track(tid, hash)
  set(tid, hash)
  yield
ensure
  delete(tid)
end