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.



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

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.



126
127
128
 
# File 'lib/wurk/queue_slot.rb', line 126

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

#hold(token, slot_key) ⇒ Object 



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

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

#size ⇒ Object 



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

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

#snapshot ⇒ Object 



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

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