Class: Wurk::Processor::SharedWorkState

Inherits:

Object

• Object
• Wurk::Processor::SharedWorkState

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

• #clear ⇒ Object
• #delete(tid) ⇒ Object
• #dup ⇒ Object
• #initialize ⇒ SharedWorkState constructor

  A new instance of SharedWorkState.

• #set(tid, hash) ⇒ Object
• #size ⇒ Object
• #track(tid, hash) ⇒ Object

  RAII: publish for the duration of the block, always retract.

Constructor Details

#initialize ⇒ SharedWorkState

Returns a new instance of SharedWorkState.

# File 'lib/wurk/processor.rb', line 144

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

Instance Method Details

#clear ⇒ Object

# File 'lib/wurk/processor.rb', line 178

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

#delete(tid) ⇒ Object

# File 'lib/wurk/processor.rb', line 153

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

#dup ⇒ Object

# File 'lib/wurk/processor.rb', line 170

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

#set(tid, hash) ⇒ Object

# File 'lib/wurk/processor.rb', line 149

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

#size ⇒ Object

# File 'lib/wurk/processor.rb', line 174

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.

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

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