Skip to content

Commit 845c792

Browse files
committed
perf(run-engine): stop re-registering ck variants already in the fair order
A concurrency-gated candidate cost two Redis calls: the SCARD that discovers it is gated, and a ZADD NX to make sure it is in the fair order. For anything pass 1 selected the second is a guaranteed no-op, because pass 1 draws its candidates from the ckVtime zset and being in that scan is what registration means. tryServe now takes a knownRegistered flag, so a gated visit from pass 1 costs only the SCARD, the same as the flag-off command. Pass 2 candidates can genuinely be unregistered, and ones outside pass 1's scanned prefix are indistinguishable from those, so both are collected and settled after pass 2 by a single variadic ZADD NX. Its return value is the count it inserted, so the idle-tag correction that stops a drained variant reclaiming full credit only runs when something actually registered. In the steady state nothing does and the whole batch costs that one call. Measured on the fully-gated shape, saturated (generator co-located with Redis, 1M invocations per arm, Redis CPU/wall 0.978-0.987, no state drift): N=1000 flag-off 21.05 before 76.63 (+55.6) after 54.56 (+33.5) -22.1us, 40% N=10000 flag-off 21.09 before 77.99 (+56.9) after 57.52 (+36.4) -20.5us, 36% A batched ZMSCORE reads the same information and measured better, -24.9us and -24.7us for 45% and 43%. It was rejected anyway: it would have been the first thing in this file to require Redis 6.2, and about 3 to 4 usec is a fair price for not raising the floor. The variadic ZADD NX is one call either way; it costs more because thirty skiplist lookups on the write path are dearer than thirty reads. Worth noting the saving per call removed is nearer 0.4 usec than the 0.33 usec measured previously on small keys, because the call being removed is a write against a zset holding thousands of members. Fully gated is the worst case by construction: a dequeue that serves exits earlier. Per-key concurrency limits are ordinary on ck queues, so it is worth having.
1 parent 1d96b41 commit 845c792

1 file changed

Lines changed: 41 additions & 17 deletions

File tree

  • internal-packages/run-engine/src/run-queue

internal-packages/run-engine/src/run-queue/index.ts

Lines changed: 41 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -5403,10 +5403,14 @@ local minServableTag = nil
54035403
local results = {}
54045404
local dequeuedCount = 0
54055405
local attempted = {}
5406+
local gatedPending = nil
54065407
54075408
-- Per-candidate serve. Body is dequeueMessagesFromCkQueueTracked's per-candidate
5408-
-- block, verbatim, with the marked NEW lines added.
5409-
local function tryServe(ckQueueName, mayRaiseFloor)
5409+
-- block, verbatim, with the marked NEW lines added. knownRegistered means the caller can
5410+
-- vouch the candidate is a ckVtime member (pass 1's scan read it out of that set, and
5411+
-- nothing removes a member this call has not served), so the gated branch below can skip
5412+
-- its registration check for it.
5413+
local function tryServe(ckQueueName, mayRaiseFloor, knownRegistered)
54105414
attempted[ckQueueName] = true
54115415
local fullQueueKey = keyPrefix .. ckQueueName
54125416
-- NEW: the tag this call wrote back, if it served. Site A below reuses it rather than
@@ -5528,19 +5532,15 @@ local function tryServe(ckQueueName, mayRaiseFloor)
55285532
-- held. Unregistered here means it reached ckIndex without ever passing through a
55295533
-- vtime-aware write: a backlog queued before the flag went on, an enqueue from an
55305534
-- instance that still has it off, or a ckVtime that expired while ckIndex lived, which
5531-
-- are the same cases pass 2's discovery exists to repair. NX, so a variant that is
5532-
-- already registered keeps the tag it earned, which is the usual case and costs one op.
5533-
-- The TTL only needs setting when this actually registered something, since that is the
5534-
-- path that can recreate a ckVtime key which expired out from under a live ckIndex.
5535-
-- NEW: registers at max(floor, remembered idle tag), same rule as the enqueue path, so
5536-
-- a variant that drained under the gate does not come back with full credit. Ordered
5537-
-- the same way too: the ZADD NX decides whether the idle lookup is worth doing at all.
5538-
if redis.call('ZADD', ckVtimeKey, 'NX', tostring(floor), ckQueueName) == 1 then
5539-
local gateIdle = redis.call('ZSCORE', ckVtimeIdleKey, ckQueueName)
5540-
if gateIdle and tonumber(gateIdle) > floor then
5541-
redis.call('ZADD', ckVtimeKey, 'XX', gateIdle, ckQueueName)
5542-
end
5543-
redis.call('EXPIRE', ckVtimeKey, stateTtl)
5535+
-- are the same cases pass 2's discovery exists to repair.
5536+
-- NEW: a knownRegistered candidate needs none of that, so its gated visit costs just
5537+
-- the SCARD above, same as the flag-off command. The rest are collected and resolved
5538+
-- after pass 2 by one variadic ZADD NX, where the rare genuinely unregistered candidate
5539+
-- registers at max(floor, remembered idle tag), same rule as the enqueue path, so a
5540+
-- variant that drained under the gate does not come back with full credit.
5541+
if not knownRegistered then
5542+
if gatedPending == nil then gatedPending = {} end
5543+
table.insert(gatedPending, ckQueueName)
55445544
end
55455545
end
55465546
end
@@ -5564,7 +5564,7 @@ end
55645564
local windowBudget = window
55655565
for _, ckQueueName in ipairs(vtimeCandidates) do
55665566
if dequeuedCount >= actualMaxCount or windowBudget <= 0 then break end
5567-
if tryServe(ckQueueName, true) ~= 'notReady' then
5567+
if tryServe(ckQueueName, true, true) ~= 'notReady' then
55685568
windowBudget = windowBudget - 1
55695569
end
55705570
end
@@ -5584,7 +5584,7 @@ local discovered = nil
55845584
for _, ckQueueName in ipairs(ckQueues) do
55855585
if not attempted[ckQueueName] then
55865586
if dequeuedCount < actualMaxCount then
5587-
tryServe(ckQueueName, false)
5587+
tryServe(ckQueueName, false, registered[ckQueueName])
55885588
elseif not registered[ckQueueName] then
55895589
-- Collected into one variadic ZADD: discovery costs at most a single op per
55905590
-- call however many variants it registers. Skipping attempted matters:
@@ -5600,6 +5600,30 @@ if discovered ~= nil then
56005600
redis.call('ZADD', unpack(discovered))
56015601
end
56025602
5603+
-- One variadic ZADD NX settles the whole batch: it registers the genuinely unregistered
5604+
-- and no-ops the rest, and its return value is the count it added. In the steady state
5605+
-- that count is zero, so a fully gated scan costs exactly this one call and nothing else.
5606+
-- The idle-tag correction is only reachable when something actually registered, which is
5607+
-- rare, so the ZSCOREs it needs are not on the hot path. ZADD NX rather than ZMSCORE
5608+
-- deliberately: the batched read would cost the same one call but would make this the
5609+
-- first thing in the file to require Redis 6.2.
5610+
if gatedPending ~= nil then
5611+
local gatedArgs = {ckVtimeKey, 'NX'}
5612+
for _, ckQueueName in ipairs(gatedPending) do
5613+
table.insert(gatedArgs, tostring(floor))
5614+
table.insert(gatedArgs, ckQueueName)
5615+
end
5616+
if redis.call('ZADD', unpack(gatedArgs)) > 0 then
5617+
for _, ckQueueName in ipairs(gatedPending) do
5618+
local gateIdle = redis.call('ZSCORE', ckVtimeIdleKey, ckQueueName)
5619+
if gateIdle and tonumber(gateIdle) > floor then
5620+
redis.call('ZADD', ckVtimeKey, 'XX', gateIdle, ckQueueName)
5621+
end
5622+
end
5623+
redis.call('EXPIRE', ckVtimeKey, stateTtl)
5624+
end
5625+
end
5626+
56035627
-- NEW: persist floor and refresh TTLs. A call that served nothing writes nothing: the two
56045628
-- things this block would persist are both re-derivable, since minServableTag is only set
56055629
-- inside a successful serve and pass 2's discovery only runs once the batch is full, so

0 commit comments

Comments
 (0)