jonnify·redis /redis-patterns/flow-control/groups/per-group-concurrency

R6.03.2 · Groups — dive

Per-group concurrency.

A per-group ceiling caps how much of the machine one tenant holds at once: a lane at its ceiling is de-ringed so the rota skips it, and re-ringed the moment a completion frees a permit. Round-robin keeps the rate of turns fair; the ceiling bounds the simultaneous in-flight work per tenant — a distributed semaphore, one per lane.

Grounding: echo/apps/echo_mq/lib/echo_mq/lanes.ex · EchoMQ.Lanes.limit/4 (the @glimit script) over emq:{q}:glimit (the cap) and emq:{q}:gactive (the live count); pause/3 / resume/3 over emq:{q}:paused. The de-ring runs in @glimit and at the @gclaim claim point.

A ceiling of 3, the permits spent down. Select a state to read whether the lane is on the ring. The semaphore is the ring membership: a free permit means the lane stays on the rota; none free means it is de-ringed.
glimit = 3 · gactive spent down → ring membership in flight in flight free 0 of 3on the ring 2 of 3on the ring 3 of 3de-ringed completionre-ringed on the ring = a permit is free · de-ringed = the permits are spent
readout · Select a state. While a permit is free the lane stays on the ring and the rota serves it. When gactive reaches glimit the lane is de-ringed and the rota skips it. A completion drops gactive and re-rings the lane — the semaphore is the ring membership.
§1

Two numbers per tenant

lanes.ex · glimit & gactive

Per-group concurrency rides two per-queue hashes, both keyed by the group id:

  • emq:{q}:glimit — the ceiling: how many of this tenant's jobs may be in flight at once.
  • emq:{q}:gactive — the live count: how many of this tenant's jobs are in flight right now.

The ceiling is configuration; the count is state. A claim that serves a tenant's job increments gactive; a completion, retry, or reap of that tenant's job decrements it. The ceiling is set with limit/4:

EchoMQ.Lanes.limit(conn, "orders", group, 5)   # this tenant: at most 5 in flight at once

This is a counting semaphore expressed as data. The cap is a hash field, not a key per tenant, so a queue with ten thousand tenants holds two hashes, not twenty thousand keys — and the whole tenant set shares the one {q} slot.

Two hashes carry the ceiling: glimit holds the cap, gactive the live in-flight count. A claim raises the count, a completion lowers it — a counting semaphore, one field per tenant.

§2

The de-ring is the enforcement

lanes.ex · the @glimit script

The ceiling is not checked at the claim and then ignored. It changes the membership of the ring: a lane at or above its ceiling is removed from rotation, so the rota never even offers it a turn. This is the @glimit script, verbatim — it writes the new cap and then reconciles the ring:

EchoMQ.Lanes · echo/apps/echo_mq — the @glimit script: set the ceiling, then de-ring or re-ringverbatim
redis.call('HSET', KEYS[1], ARGV[2], ARGV[3])
local act = tonumber(redis.call('HGET', KEYS[2], ARGV[2]) or '0')
if act >= tonumber(ARGV[3]) then
  redis.call('LREM', KEYS[3], 0, ARGV[2])              ← at the cap: de-ring
else
  local lane = ARGV[1] .. 'g:' .. ARGV[2] .. ':pending'
  if redis.call('SISMEMBER', KEYS[4], ARGV[2]) == 0 and redis.call('ZCARD', lane) > 0 and
     not redis.call('LPOS', KEYS[3], ARGV[2]) then
    redis.call('RPUSH', KEYS[3], ARGV[2])              ← serviceable: re-ring
    redis.call('LPUSH', KEYS[5], '1')
    redis.call('LTRIM', KEYS[5], 0, 63)
  end
end
return 1

HSET glimit group n writes the ceiling. Then it reads the live count: if the tenant already has n or more in flight, LREM ring 0 group removes the lane from the rota immediately — lowering a ceiling below the live count parks the tenant until it drains. Otherwise, if the lane is unpaused, has pending work, and is not already on the ring, RPUSH ring group returns it to rotation with a wake — raising a ceiling can re-admit a tenant the previous cap had de-ringed.

The same de-ring runs at the claim point. After serving a job, the claim increments gactive; if that increment reaches the ceiling, the lane is de-ringed on the spot:

# inside @gclaim, after serving the head:
local act = redis.call('HINCRBY', ARGV[1] .. 'gactive', g, 1)
local lim = redis.call('HGET', ARGV[1] .. 'glimit', g)
if lim and act >= tonumber(lim) then
  redis.call('LREM', KEYS[1], 0, g)   # reached the ceiling: de-ring
end

So a tenant at its ceiling is not on the ring at all, and the rota passes over it — no skip logic, no per-turn check. When a completion drops gactive below the ceiling, the transition re-rings the lane. The semaphore is the ring membership: in the rota means a permit is free, off the rota means the permits are spent.

The lane's two states, and what moves it between them. Select a state or an edge to read it. A lane is on the ring while a permit is free, de-ringed while the permits are spent.
on the ringa permit is free de-ringedpermits spent claim → gactive == glimit completion → gactive < glimit (re-ring + wake)
readout · Select a state or transition. A lane sits on the ring while a permit is free; a claim that brings gactive up to glimit de-rings it; de-ringed, the rota skips it; a completion drops gactive below the cap and re-rings the lane with a wake.

Notes on ValkeyHINCRBY is the atomic increment-and-return on a hash field, so two concurrent claims of the same tenant cannot both read the count as below the cap and both serve. LREM ring 0 group removes every occurrence of the group from the ring in one operation, and LPOS ring group is the membership test that keeps a re-ring from adding a duplicate — valkey.io/commands/hincrby.

§3

Parking a tenant without touching the rest

lanes.ex · pause/3 · resume/3

Two more controls act on a single lane and leave the rest of the queue alone. pause/3 removes a lane from the ring and records the group in emq:{q}:paused:

EchoMQ.Lanes · echo/apps/echo_mq — the @gpause script: mark paused, drop from the ringverbatim
redis.call('SADD', KEYS[1], ARGV[1])     ← mark the group paused
redis.call('LREM', KEYS[2], 0, ARGV[1])  ← drop the lane from the ring
return 1

The paused flag is why a re-ring guard always checks SISMEMBER paused first — a paused lane stays off the rota even when a completion frees a permit. resume/3 clears the flag and returns the lane to rotation if it is serviceable (unpaused, below its ceiling, nonempty), with a wake. The tenant's backlog and its in-flight work are untouched throughout — pause stops new turns, it does not cancel or drain anything.

This is the operations move a single shared queue otherwise cannot make: hold one tenant while the others keep running. A misbehaving tenant is paused, investigated, and resumed, with no effect on any other tenant's flow.

the pattern

A counting semaphore admits up to N concurrent holders and blocks the rest until a holder releases. The count of free permits is the gate.

its EchoMQ application

emq:{q}:gactive counts a tenant's in-flight jobs against the emq:{q}:glimit ceiling; EchoMQ.Lanes.limit/4 sets the cap and de-rings a lane at its ceiling, re-ringing it when a completion frees a permit — the semaphore is the lane's ring membership.

A pause drops one lane from the ring and flags the group, holding the tenant while every other lane runs; a resume returns it if serviceable. The backlog and in-flight work are untouched — pause stops turns, it does not cancel.

§4

The pattern, applied

echo/apps/codemojex

In codemojex — a Telegram emoji-guessing game on the same stack — a per-room ceiling caps how much of the scoring engine any one room holds at once: a room with a large backlog is served its turn in the rota, but never more than its ceiling of guesses in flight together (emq:{cm}:g:<PLR…>:pending behind a per-room glimit), so one room's burst cannot consume the workers a hundred other rooms need. The ceiling and the rota compose: the rota gives every room a turn, the ceiling bounds how much each holds at once.

/echomq/queue

The ceiling at the claim point

The Queue pillar wires the per-group ceiling into the dequeue path: the de-ring at the claim, the re-ring on completion, and the wake that returns a parked consumer to work.

 /bcs/bus/jobs-and-lanes

Pause and resume on one lane

Part B3.2: group-aware pause and resume act on a lane without touching the queue, so one tenant can be held while the others run.

 R2

The semaphore pattern

The per-group ceiling is a counting lock — permits checked against a cap. The coordination chapter builds the distributed-lock pattern the ceiling reuses.
§5

References

sources · related

Sources

  • Valkey — HINCRBY — the atomic increment-and-return on a hash field that moves the in-flight count without a race.
  • Valkey — HSET — write a group's ceiling into the glimit hash; the cap is a hash field, one per tenant.
  • Valkey — LREM — remove the group from the ring list in one operation — the de-ring that enforces the ceiling.
  • Valkey — LPOS — the membership test that keeps a re-ring from adding a duplicate lane to the rota.

Related in this course