From df6e307cc77fd395477eac6ec4480ead6b293751 Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Mon, 21 Sep 2026 08:47:34 +0800 Subject: [PATCH] fix(keys): re-claim the expired slot in place instead of allocating a new one When a key's index slot is reclaimed from the shared dict, add() allocated a new slot number and bumped delete_count so that every other worker would drop the stale slot on its next sync(). Both are avoidable: the slot number is free, so the key can simply take it back. Re-claiming in place keeps the key's identity, which removes the condition the duplicate-metric fix was written for -- a key can no longer occupy two slots -- and has three further effects: * key_count stops growing on expiry churn. It only ever grows, and sync_range(0, key_count) walks every slot it ever handed out, so each re-allocation makes every later full sync more expensive, permanently. * delete_count stops moving, so no worker is pushed into a full sync_range(0, key_count) from its request path. With several workers on one dict that scan is O(key_count) shared-dict reads each, all contending for the same shm mutex. * that full sync is itself what wiped index[key] for every slot whose node had ttl-expired but not yet been reclaimed, so those keys were re-allocated too. One bump therefore cost far more than one slot. expire() only reports "not found" once the node is physically gone. A node that has merely ttl-expired is resurrected by expire() and keeps its slot, so this branch is reached only after flush_expired() (or an LRU eviction) has removed it, and the slot is genuinely free. Races are handled by relying on dict:add() being atomic. A peer that re-claims the slot first makes add() return "exists"; the value is then compared against the key before the slot is accepted. If it cannot be re-claimed at all -- out of memory, or the slot holds a different key because key_count was evicted and restarted -- the old behaviour of allocating a new slot still applies, and list() keeps reporting the key once from the slot index points at. Measured on OpenResty with 10 workers sharing a 512m dict, 144k slots, a 4000 combination rotation expiring and returning, and flush_expired() once a second: total CPU key_count delete_count duplicates before 421% (peak 970%) +1368 268 0 after 13% (peak 40%) +0 0 0 testExpiredReAddNoDuplicate now asserts the slot is re-claimed, key_count does not grow and delete_count stays unset, and that the second worker stays correct without being told anything. testExpiredReAddFallbackNoDuplicate covers the fallback path, where the slot is stolen between expire() and add(). --- prometheus_keys.lua | 40 +++++++++++++++------- prometheus_test.lua | 83 +++++++++++++++++++++++++++++++++++++-------- 2 files changed, 97 insertions(+), 26 deletions(-) diff --git a/prometheus_keys.lua b/prometheus_keys.lua index faeaa66..c5b42cd 100644 --- a/prometheus_keys.lua +++ b/prometheus_keys.lua @@ -156,22 +156,38 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime) local ok, err = self.dict:expire(self.key_prefix .. self.index[key], exptime) if not ok then if err == "not found" then - -- The slot already expired in the shared dict. Drop the stale - -- local state and bump delete_count so other workers do a full - -- sync and reclaim the slot; without this the old slot lingers in - -- their local self.keys while the metric is re-added at a new slot, - -- desynchronizing the index and causing duplicate metric emission. - -- The dict slot is already gone (expire returned "not found"), so - -- there is no slot to clear here. + -- The slot was reclaimed from the shared dict (expire() only + -- reports "not found" once the node is physically gone; a merely + -- ttl-expired node is resurrected by expire() and keeps its slot). + -- Re-claim the SAME slot number rather than allocating a new one. + -- The key keeps its identity, so no peer's index[key] goes stale, + -- key_count does not grow, and a key can never occupy two slots -- + -- which is what produced the duplicate metrics this branch was + -- written for. Nothing has to be broadcast either, so delete_count + -- stays put and no worker is forced into a sync_range(0, key_count) + -- on its request path. local idx = self.index[key] + local ok2, err2, forcible2 = + self.dict:add(self.key_prefix .. idx, key, exptime) + if ok2 or (err2 == "exists" + and self.dict:get(self.key_prefix .. idx) == key) then + -- either we re-claimed it, or a peer re-claimed it for us + if exptime and exptime > 0 then + self.expire_keys[idx] = true + end + if forcible2 then + return (err_msg_lru_eviction .. "; key index: re-claimed slot: idx=" + .. self.key_prefix .. idx .. ", key=" .. key) + end + break + end + -- The slot could not be re-claimed (out of memory, or it now holds + -- a different key because key_count was evicted and restarted). + -- Fall back to allocating a new slot below; list() still reports + -- the key once, from the slot index currently points at. self.index[key] = nil self.keys[idx] = nil self.expire_keys[idx] = nil - self.deleted = self.deleted + 1 - local _, incr_err, forcible = self.dict:incr(self.delete_count, 1, 0) - if incr_err or forcible then - return incr_err or err_msg_lru_eviction - end expired = true else -- Unexpected expire error: the slot may still be live, so leave it diff --git a/prometheus_test.lua b/prometheus_test.lua index b485b0a..0dcce31 100644 --- a/prometheus_test.lua +++ b/prometheus_test.lua @@ -848,8 +848,8 @@ function TestKeyIndex:testExpiredReAddNoDuplicate() luaunit.assertEquals(self.key_index.index["expkey"], 1) -- A second worker sharing the same shared dict syncs the initial state, so it - -- now holds slot 1 in its local self.keys/index. This is the worker that the - -- delete_count bump must later force to re-sync and reclaim the stale slot. + -- now holds slot 1 in its local self.keys/index. It must stay correct across + -- the re-add without having to be told anything. local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1) worker2:sync() luaunit.assertEquals(worker2.index["expkey"], 1) @@ -861,30 +861,85 @@ function TestKeyIndex:testExpiredReAddNoDuplicate() sleep(2) luaunit.assertEquals(self.dict:get("_prefix_key_1"), nil) - -- Re-adding the now-expired key takes the expired branch and allocates slot 2. + -- Re-adding the now-expired key re-claims slot 1 in place. err = self.key_index:add("expkey", "eviction_err", 1) luaunit.assertEquals(err, nil) - luaunit.assertEquals(self.dict:get("_prefix_key_2"), "expkey") + luaunit.assertEquals(self.dict:get("_prefix_key_1"), "expkey") + luaunit.assertEquals(self.key_index.index["expkey"], 1) - -- delete_count must have been bumped on the expired re-add path so that - -- other workers do a full sync and drop the stale slot. - luaunit.assertEquals(self.dict:get("_prefix_delete_count"), 1) + -- No new slot was allocated, so key_count does not grow ... + luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1) + -- ... and nothing has to be broadcast, so no worker is forced into a full + -- sync_range(0, key_count) on its request path. + luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil) -- list() must report the key exactly once, not twice. local keys = self.key_index:list() luaunit.assertEquals(#keys, 1) luaunit.assertEquals(keys[1], "expkey") - -- The second worker must converge: its next sync() sees the bumped - -- delete_count, does a full sync, drops the stale slot 1 and picks up slot 2. - -- Without the delete_count bump it would keep slot 1 forever and list() the - -- key twice. - worker2:sync() - luaunit.assertEquals(worker2.keys[1], nil) - luaunit.assertEquals(worker2.index["expkey"], 2) + -- The second worker never went stale: the key kept its slot number, so its + -- local index is still correct and list() reports the key once, with or + -- without an intervening sync(). local keys2 = worker2:list() luaunit.assertEquals(#keys2, 1) luaunit.assertEquals(keys2[1], "expkey") + worker2:sync() + luaunit.assertEquals(worker2.index["expkey"], 1) + luaunit.assertEquals(#worker2:list(), 1) +end + + +-- A key whose slot cannot be re-claimed must still be listed exactly once. +-- The slot is stolen between expire() reporting it gone and add() trying to +-- re-claim it, which is the only way another key can end up on that slot +-- number; add() must then fall back to allocating a new one. +function TestKeyIndex:testExpiredReAddFallbackNoDuplicate() + local err = self.key_index:add("expkey", "eviction_err", 1) + luaunit.assertEquals(err, nil) + self.key_index:sync() + luaunit.assertEquals(self.key_index.index["expkey"], 1) + + local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1) + worker2:sync() + luaunit.assertEquals(worker2.index["expkey"], 1) + + sleep(2) + luaunit.assertEquals(self.dict:get("_prefix_key_1"), nil) + + -- Simulate a peer writing a different key onto slot 1 in the window between + -- our expire() and our add(). + local real_expire = self.dict.expire + self.dict.expire = function(dict_self, k, exptime) + local ok, expire_err = real_expire(dict_self, k, exptime) + if not ok and k == "_prefix_key_1" then + dict_self:set(k, "someone_else") + end + return ok, expire_err + end + + err = self.key_index:add("expkey", "eviction_err", 1) + self.dict.expire = real_expire + + luaunit.assertEquals(err, nil) + luaunit.assertEquals(self.dict:get("_prefix_key_1"), "someone_else") + luaunit.assertEquals(self.dict:get("_prefix_key_2"), "expkey") + luaunit.assertEquals(self.key_index.index["expkey"], 2) + + local seen = 0 + for _, k in ipairs(self.key_index:list()) do + if k == "expkey" then seen = seen + 1 end + end + luaunit.assertEquals(seen, 1) + + -- worker2 still points at slot 1; list() must not emit the key from a slot + -- the index no longer points at. + worker2:sync() + local seen2 = 0 + for _, k in ipairs(worker2:list()) do + if k == "expkey" then seen2 = seen2 + 1 end + end + luaunit.assertEquals(seen2, 1) end -- remove_expired_keys() must physically reclaim the shared-dict space of