Class: Wurk::QueueSlot::Held

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

Overview

Which slots this process is holding right now, so the heartbeat can extend them on the beat it was already sending.

A process-wide singleton rather than fetcher state, for the reason Processor::WORK_STATE is one: the reader is the Heartbeat, which lives a layer above the capsules and would otherwise have to walk them looking for fetchers that may not exist. Keyed by holder token, which names one claim (claim_token) — a thread runs one job at a time, so the map holds one entry per busy thread and is empty for every install that caps nothing. The exception is the window a failed ACK flush opens: a thread whose finished job's release has not landed yet appears twice until it does, which keeps that hold alive rather than letting it be reused early.

Nothing here is authoritative: Redis is. The ledger only decides what gets refreshed, so an entry lost to a crash costs a hold its TTL, never correctness.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Held

Returns a new instance of Held.



104
105
106
107
 
# File 'lib/wurk/queue_slot.rb', line 104

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

Instance Method Details

#drop(token, slot_key) ⇒ Object

Drops the named hold and only the named hold — the slot key is checked as well as the token, so a release that arrives late can never stop the heartbeat refreshing a hold that is actually live.



116
117
118
 
# File 'lib/wurk/queue_slot.rb', line 116

def drop(token, slot_key)
  @lock.synchronize { @held.delete(token) if @held[token] == slot_key }
end

#hold(token, slot_key) ⇒ Object 



109
110
111
 
# File 'lib/wurk/queue_slot.rb', line 109

def hold(token, slot_key)
  @lock.synchronize { @held[token] = slot_key }
end

#size ⇒ Object 



124
125
126
 
# File 'lib/wurk/queue_slot.rb', line 124

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

#snapshot ⇒ Object 



120
121
122
 
# File 'lib/wurk/queue_slot.rb', line 120

def snapshot
  @lock.synchronize { @held.dup }
end