jonnifyechomq

Blocking, timers and the bus · dive 02

The blocking claim

A wake carries no job: it asks a parked worker to run its claim again, and the claim's own checks settle whether there is work. A path that makes work claimable without a wake leaves the worker to find out at its timeout.

There are two ways to wait for work. A worker can poll, sending its claim again and again, or it can park on a blocking command and let a producer wake it. EMQ.CLAIM does not block: with nothing to claim it answers at once. The Elixir library parks instead, on Valkey's own BLPOP over the queue's wake list, and the lane scripts push a token onto that list when a lane becomes serviceable. Follow one event from its transition to the claim it triggers, then read which transitions push a wake at all.

The native claim answers at once The wake list is a doorbell Which transitions ring it References

The native claim answers at once

EMQ.CLAIM reads the queue's pause flag, then peeks at the head of pending. A paused queue, an empty pending and a head with no row each end the call with an empty array:

// trafikey — EMQ.CLAIM: nothing to claim is an empty array, answered at once
pub(crate) fn claim(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // … shape, layout, the lease and the fence
    let (pending, active) = (s.key(1), s.key(2));
    let empty = || Ok(ValkeyValue::Array(Vec::new()));

    // … the queue's pause flag, read from its meta row as the caller
    let meta = q.meta();
    caller.open(&meta, Access::Read)?;
    if host::read(ctx, cmd::HGET, &[&meta, b"paused"])? != Reply::Null {
        return empty();
    }

    // Step 2, peeked: the member ZPOPMIN would pop (same order, same atomic command).
    let id = match host::read(ctx, cmd::ZRANGE, &[pending, b"0", b"0"])? {
        Reply::Array(head) => match head.into_iter().next() {
            None => return empty(),
            Some(Reply::Bulk(id)) => id,
            Some(_) => return Err(errors::emq(errors::HOST_REPLY)),
        },
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    };
    // … a head with no row is popped and answers empty too; a live head is leased and returned

No row of the registration table carries the blocking flag, and the module calls none of the host functions that park a client on a key. A raw client waiting for the next job therefore sends EMQ.CLAIM again after an empty reply: on the module's node, waiting is the client's loop.

An empty array is the native claim's whole answer to "no work"; the wait, if there is one, lives in the client.

The wake list is a doorbell

A standalone EchoMQ.Consumer ends every beat by parking on emq:{q}:wake, a list, with Valkey's blocking pop. The pop returns when a token arrives or when the beat runs out, and the consumer discards whatever it popped:

# echo_mq — EchoMQ.Consumer's park: Valkey's own blocking pop on the wake list, the beat as its timeout
defp park(s) do
  secs = :erlang.float_to_binary(s.beat_ms / 1000, decimals: 3)
  wake = Keyspace.queue_key(s.queue, "wake")
  _ = Connector.command(s.conn, ["BLPOP", wake, secs], s.beat_ms + 2_000)
  :ok
end

Either way the next beat runs: recovery, promotion, then the drain, which claims through the lane ring. The claim checks the queue's pause first and then rotates the ring one step, so a wake that arrives for work another consumer already took, or for a queue that is paused, ends in :empty and another park:

# echo_mq — EchoMQ.Lanes.claim/3, the drain's claim: the queue's pause first, then one turn of the ring
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if EchoMQ.Jobs.paused?(conn, queue) do
    :empty
  else
    keys = [
      Keyspace.queue_key(queue, "ring"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @gclaim2, keys, argv) do
      {:ok, []} -> :empty
      {:ok, [id, payload, att, group]} -> {:ok, {id, payload, att, group}}
      other -> other
    end
  end
end

The token is a doorbell, not a delivery: it holds the string 1 and names no job. Readiness only asks the waiter to check again, and the claim's own checks (the pause, the ring, the lane's head, the row) answer whether there is work. Send one event through the chain to see where it stops; the queue has one lane and one parked consumer.

From a transition to a claim · pick an event

transition@genqueue wake listLPUSH 1 parked popreturns now claima job EchoMQ.Lanes.enqueue/5
lane admission, idle lane · the group is unpaused, below its limit and off the ring
@genqueue writes the row, adds the job to its lane, rings the lane and pushes one token
the parked BLPOP pops the token and returns before the beat runs out
the next pass runs at once: recovery, promotion, then the drain leases the job through the ring
one token, one claim, one job

A wake says "look again", never "here is a job": the waiter's claim does the looking, and a wake that finds nothing costs one empty claim.

Which transitions ring it

The wake list is pushed from inside the scripts, in one place each: the branch that puts a lane back on the ring. Grouped admission is the clearest case. The script is EchoMQ.Lanes @genqueue, the script EchoMQ.Lanes.enqueue/5 runs:

-- EchoMQ.Lanes @genqueue — grouped admission: the wake is pushed only inside the branch that rings the lane
-- KEYS: 1 the row, 2 the lane, 3 the ring, 4 the paused groups, 5 glimit, 6 gactive, 7 the wake list
-- ARGV: 1 the job id, 2 the payload, 3 the group
-- the kind door: only a JOB-namespaced id is admitted
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- an existing row is a duplicate: no write, reply 0
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
redis.call('HSET', KEYS[1], 'state', 'pending', 'attempts', '0', 'payload', ARGV[2], 'group', ARGV[3])
redis.call('ZADD', KEYS[2], 0, ARGV[1])
-- the lane joins the ring only if its group is unpaused, below its limit and not ringed already
if redis.call('SISMEMBER', KEYS[4], ARGV[3]) == 0 then
  local lim = redis.call('HGET', KEYS[5], ARGV[3])
  local act = tonumber(redis.call('HGET', KEYS[6], ARGV[3]) or '0')
  if (not lim or act < tonumber(lim)) and not redis.call('LPOS', KEYS[3], ARGV[3]) then
    redis.call('RPUSH', KEYS[3], ARGV[3])
    -- the doorbell: one token, the list capped at 64 tokens
    redis.call('LPUSH', KEYS[7], '1')
    redis.call('LTRIM', KEYS[7], 0, 63)
  end
end
return 1

Every script that pushes a wake does it the same way, right after an RPUSH to the ring: grouped admission, promotion, the reaper and its one-group form, the counted sweep, completion, retry, reprocess, the removal of an active job, and the lane controls that resume a lane, raise or lift its limit, move a job to another group or repair the lane's active count. A flat branch pushes none. A flat enqueue, a flat promotion or recovery, a repeat firing and a flow release write the flat pending set and ring nothing; the lane drain does not serve that set, and flat work is taken by EchoMQ.Jobs.claim/3 or found by EchoMQ.BatchConsumer, which reads the pending depth on a cadence of its own. Pick a transition and a row to read its rule.

Which transitions push a wake · pick a cell

transitionflat rowlane row
admission
due promotion
lease recovery
counted sweep
completion
failure, retried
reprocess
repeat firing
flow release
admission, lane row · EchoMQ.Lanes @genqueue
adds the job to its lane, then rings the lane and pushes a token when the group is unpaused, its active count is below its limit (or it has none) and the lane is off the ring
a wake, when the lane joins the ring
seven transitions can push a wake, each only when a lane joins the ring; no flat branch pushes one

Today the doorbell rings for one event only, a lane returning to the ring; everything else that makes work claimable is found at the next beat or the next poll.

Pattern & implementation

the pattern (Redis Patterns Applied)

A worker parks on a blocking pop instead of polling, and a producer pushes a marker that wakes it; the worker then claims from the real queue. The blocking pop and The marker wake-up teach it.

the implementation (EchoMQ and trafikey)

emq:{q}:wake is the marker list, pushed only when a lane returns to the ring; EchoMQ.Consumer parks on it with the beat as timeout and claims through EchoMQ.Lanes.claim/3. EMQ.CLAIM never parks.

The last dive turns from the queue to the bus, where the waiting and the reclaim are built-ins too.

References

Sources

Related in this course