docs(run-engine): benchmark Redis CPU and memory vs concurrency-key cardinality
Adds a resource/cardinality arm to the CK virtual-time benchmark: a plan, a runnable harness (drives a real RunQueue against a dedicated Redis, flag OFF vs ON, and reads server-side INFO memory/cpu, MEMORY USAGE and OBJECT ENCODING; inert without CK_BENCH_REDIS_URL), and results from a local homelab. Findings: the :ckVtime ZSET is essentially a second copy of :ckIndex (~150 bytes per key), so memory is linear in cardinality (about +1.3MB for a 10k-key queue) and TTL-reclaimed; Redis CPU overhead is small and does not scale with cardinality (+5 to +12 percent on an identical workload, per-script cost flat at ~25 usec/call) because the dequeue window is fixed and the ZSET ops are O(log N); and ckVtime membership tracks ckIndex exactly under sustained churn.
This commit is contained in:
Wes Mason
+156
@@ -0,0 +1,156 @@
|
||||
# CK virtual-time scheduling: Redis CPU + memory vs cardinality test plan
|
||||
|
||||
Answers the review question: how do these changes affect the run-queue Redis
|
||||
CPU and memory, and how do both react as concurrency-key cardinality grows (e.g.
|
||||
a base queue that suddenly has 10k distinct concurrency keys)?
|
||||
|
||||
This is a cost/scaling plan, separate from the fairness A/B
|
||||
(`results-2026-07-27.md`). Same environment constraints: run on a single box, all
|
||||
numbers **relative** (flag OFF vs ON, identical load, same box), against a
|
||||
dedicated throwaway Redis.
|
||||
|
||||
## What the change adds to Redis (grounded in the branch)
|
||||
|
||||
Per base queue, flag ON adds:
|
||||
|
||||
- `:ckVtime`, a ZSET whose members are the exact same full CK-variant queue-name
|
||||
strings the existing `:ckIndex` ZSET already holds, each with an 8-byte double
|
||||
score. So it is effectively a second copy of `ckIndex`'s membership.
|
||||
- `:ckVtimeFloor`, a STRING holding one number. Negligible.
|
||||
|
||||
Both are GC'd from the dequeue path when a variant drains, carry a 24h TTL, and
|
||||
live under the base queue's `{org}` hash tag. The per-call dequeue scan window is
|
||||
`maxCount * windowMultiplier` (default 30) and does **not** grow with cardinality;
|
||||
the cardinality-sensitive operations are the ZSET writes/reads (`ZADD` `NX`,
|
||||
`ZSCORE`, `ZRANGE` by rank, `ZRANGE 0 0`), which are O(log N) on the `ckVtime`
|
||||
skiplist.
|
||||
|
||||
## Hypotheses
|
||||
|
||||
1. **Memory grows linearly with cardinality, adding roughly one `ckIndex`-sized
|
||||
ZSET per base queue.** Incremental `used_memory` under ON minus OFF should
|
||||
track `cardinality x per-member cost`, and `MEMORY USAGE :ckVtime` should be
|
||||
close to `MEMORY USAGE :ckIndex` for the same queue (same members, one extra
|
||||
double score). At 10k keys on one queue this is a low single-digit MB for that
|
||||
queue, bounded and TTL-reclaimed. There is a one-step jump at the
|
||||
listpack->skiplist encoding boundary (128 entries by default).
|
||||
2. **Redis CPU per operation grows sub-linearly (about O(log cardinality)), not
|
||||
linearly.** The fixed 30-entry window scan dominates the vtime-specific work
|
||||
and does not change with N; the ZSET ops add a `log N` term. So dequeue/enqueue
|
||||
`usec_per_call` should rise only mildly from 100 to 10k keys, and the ON/OFF
|
||||
`usec_per_call` ratio should stay roughly flat across the sweep.
|
||||
3. **Tombstone drift stays bounded under high-cardinality churn.** `ack`, TTL
|
||||
expiry, and DLQ drain a variant without removing it from `ckVtime` (documented
|
||||
in `CK_VTIME_KNOWN_LIMITATIONS.md`). Under churn where variants drain via those
|
||||
paths rather than a vtime dequeue, `ckVtime` may transiently exceed `ckIndex`,
|
||||
but it should self-heal (next vtime pass GCs empties) or expire (24h TTL), so
|
||||
`size(ckVtime) / size(ckIndex)` stays bounded and does not grow without limit.
|
||||
|
||||
## Scenarios
|
||||
|
||||
All on a single base queue (worst case for one queue's ZSETs), dedicated Redis,
|
||||
flag OFF then ON with identical load.
|
||||
|
||||
- **Cardinality sweep (memory).** Enqueue N distinct concurrency keys, one
|
||||
message each, N in {100, 1_000, 10_000, 50_000}. Measure the Redis memory
|
||||
footprint at rest for each N, OFF vs ON. This isolates the storage cost with no
|
||||
dequeue activity.
|
||||
- **Steady-state load (CPU).** At each N, after building cardinality, run a fixed
|
||||
60s workload of enqueue + batched dequeue (`maxCount 10`) + ack at a capped
|
||||
concurrency, so keys are continuously served and re-registered. Measure Redis
|
||||
CPU and per-command time over the window, OFF vs ON.
|
||||
- **Churn / tombstone (memory under adversarial drain).** Build 10k keys, then
|
||||
drain them via `ack` (not via vtime dequeue) while enqueuing new keys, for a
|
||||
fixed duration. Sample `size(ckVtime)` and `size(ckIndex)` over time and confirm
|
||||
the ratio stays bounded (self-heal + TTL), not monotonically growing.
|
||||
|
||||
## Metrics and collection (exact commands)
|
||||
|
||||
Use a second plain Redis client for measurement so it does not perturb the
|
||||
harness. `redis-cli` shown; the harness issues the same via ioredis.
|
||||
|
||||
Memory:
|
||||
|
||||
- Totals: `INFO memory` -> `used_memory`, `used_memory_dataset`. Delta ON vs OFF
|
||||
at each N is the incremental footprint.
|
||||
- Per structure (exact bytes): `MEMORY USAGE {org...}:queue:<base>:ckIndex` and
|
||||
`MEMORY USAGE {org...}:queue:<base>:ckVtime`. Report both and the ratio.
|
||||
- Encoding: `OBJECT ENCODING <ckVtime key>` (listpack vs skiplist) at each N, to
|
||||
mark the transition.
|
||||
|
||||
CPU:
|
||||
|
||||
- Process CPU over the load window: `INFO cpu` -> `used_cpu_user` +
|
||||
`used_cpu_sys`, sampled before and after the fixed 60s load; the delta is Redis
|
||||
CPU-seconds consumed. Divide by op count for CPU-per-op.
|
||||
- Per-command time: `CONFIG RESETSTAT` before the window, then `INFO commandstats`
|
||||
after -> `cmdstat_zadd`, `cmdstat_zrange`, `cmdstat_zrangebyscore`,
|
||||
`cmdstat_zscore`, `cmdstat_get`, `cmdstat_set`, `cmdstat_expire`
|
||||
(`calls`, `usec`, `usec_per_call`). Compare OFF vs ON.
|
||||
- Cross-check (optional, OS-level): `pidstat -p <redis-pid> 1` over the window, or
|
||||
`redis-cli --latency` / `--latency-history` for command latency.
|
||||
|
||||
Derived:
|
||||
|
||||
- Memory vs N curve (expect linear, slope ~ per-member bytes), OFF and ON.
|
||||
- CPU-per-op vs N curve (expect flat-ish / log), OFF/ON ratio.
|
||||
- `size(ckVtime)/size(ckIndex)` over time in the churn scenario (expect bounded).
|
||||
|
||||
## A/B procedure
|
||||
|
||||
1. Dedicated throwaway Redis; `FLUSHDB` between arms and between cardinality
|
||||
points. Warm up once.
|
||||
2. For each N in the sweep, for each arm (OFF, ON):
|
||||
- Build N keys (enqueue N distinct concurrency keys).
|
||||
- Snapshot memory (`used_memory`, `MEMORY USAGE` of both ZSETs, `OBJECT
|
||||
ENCODING`).
|
||||
- `CONFIG RESETSTAT`; run the fixed 60s steady load; snapshot `INFO cpu` and
|
||||
`INFO commandstats`.
|
||||
- Record, then `FLUSHDB`.
|
||||
3. N trials of the load window per point; report median for CPU (memory at rest is
|
||||
near-deterministic).
|
||||
4. Run the churn scenario once per arm at N = 10k.
|
||||
|
||||
Only the OFF-vs-ON delta and the shape of the growth curves are reported; absolute
|
||||
throughput on a single box is not prod scale.
|
||||
|
||||
## Results template
|
||||
|
||||
### Memory at rest, per cardinality (single base queue)
|
||||
|
||||
| keys (N) | used_memory OFF | used_memory ON | delta | MEMORY USAGE ckIndex | MEMORY USAGE ckVtime | ckVtime/ckIndex | ckVtime encoding |
|
||||
| --- | --- | --- | --- | --- | --- | --- | --- |
|
||||
| 100 | | | | | | | |
|
||||
| 1,000 | | | | | | | |
|
||||
| 10,000 | | | | | | | |
|
||||
| 50,000 | | | | | | | |
|
||||
|
||||
### Redis CPU under 60s steady load, per cardinality
|
||||
|
||||
| keys (N) | CPU-sec OFF | CPU-sec ON | delta | dequeue usec/call OFF | dequeue usec/call ON | delta | total redis calls OFF/ON |
|
||||
| --- | --- | --- | --- | --- | --- | --- | --- |
|
||||
| 100 | | | | | | | |
|
||||
| 1,000 | | | | | | | |
|
||||
| 10,000 | | | | | | | |
|
||||
|
||||
### Tombstone drift (N=10k, drain via ack)
|
||||
|
||||
| elapsed | size(ckIndex) | size(ckVtime) | ratio |
|
||||
| --- | --- | --- | --- |
|
||||
| 0s | | | |
|
||||
| 30s | | | |
|
||||
| 60s | | | |
|
||||
|
||||
## Harness
|
||||
|
||||
Extend the existing micro-benchmark
|
||||
(`../../src/run-queue/bench/ckMicroBench.bench.test.ts`), which already drives a
|
||||
real `RunQueue` against an external Redis and reads `INFO commandstats`. Add a
|
||||
resource/cardinality mode that: builds N variants, snapshots `used_memory` +
|
||||
`MEMORY USAGE` of the base queue's `ckIndex`/`ckVtime` + `OBJECT ENCODING`, runs a
|
||||
fixed-duration steady load while sampling `INFO cpu`/`commandstats`, and emits the
|
||||
tables above. The churn scenario reuses the same enqueue/ack primitives with a
|
||||
drain-by-ack loop and periodic `ZCARD` sampling of both ZSETs.
|
||||
|
||||
Reuse the same dedicated Redis and the OFF-vs-ON constructor-flag pattern, so this
|
||||
arm, like the micro-benchmark, needs no webapp or redeploy.
|
||||
@@ -0,0 +1,82 @@
|
||||
# CK virtual-time scheduling: Redis CPU + memory vs cardinality (2026-07-28)
|
||||
|
||||
Answers the review question: how do these changes affect the run-queue Redis CPU
|
||||
and memory, and how do both react as concurrency-key cardinality grows?
|
||||
|
||||
Run on a **local homelab box, not production**, against a dedicated Redis
|
||||
configured for measurement (no RDB/AOF in the sampling windows, `maxmemory 0`,
|
||||
zset listpack thresholds at their defaults so the encoding boundary sits at 128).
|
||||
All numbers are **relative** (flag OFF vs ON, identical load, same box); absolute
|
||||
throughput is not prod scale. Method: `2026-07-28-ck-vtime-resource-cardinality-plan.md`.
|
||||
Server-side metrics (`INFO memory`/`cpu`/`commandstats`, `MEMORY USAGE`,
|
||||
`OBJECT ENCODING`) are RTT-independent.
|
||||
|
||||
## Memory at rest (single base queue, one queued message per key)
|
||||
|
||||
| keys (N) | used_memory OFF | used_memory ON | ON-OFF delta | ckIndex | ckVtime | ckVtime/ckIndex | ckVtime encoding |
|
||||
| --- | --- | --- | --- | --- | --- | --- | --- |
|
||||
| 100 | 1.84 MB | 1.93 MB | +0.09 MB | 6.9 KB | 6.2 KB | 0.89 | listpack |
|
||||
| 1,000 | 2.42 MB | 2.58 MB | +0.15 MB | 131 KB | 131 KB | 1.00 | skiplist |
|
||||
| 10,000 | 9.12 MB | 10.4 MB | +1.26 MB | 1.42 MB | 1.42 MB | 1.00 | skiplist |
|
||||
| 50,000 | 39.1 MB | 45.6 MB | +6.51 MB | 7.55 MB | 7.54 MB | 1.00 | skiplist |
|
||||
|
||||
The `:ckVtime` ZSET is the whole added footprint, and it is essentially a second
|
||||
copy of `:ckIndex`: same members (the full CK-variant queue names), one extra
|
||||
8-byte score, so `MEMORY USAGE(ckVtime) ~= MEMORY USAGE(ckIndex)` once past the
|
||||
listpack boundary. Cost is linear in cardinality at roughly **150 bytes per
|
||||
concurrency key** on top of the index the queue already keeps: about +1.3 MB for
|
||||
a queue with 10k keys, +6.5 MB at 50k. It is bounded by live cardinality (entries
|
||||
are GC'd when a variant drains) and expires on the 24h state TTL. The
|
||||
`used_memory` delta tracks the direct `MEMORY USAGE(ckVtime)` figure to within
|
||||
allocator noise. The listpack->skiplist transition lands between 100 and 1,000
|
||||
keys as expected.
|
||||
|
||||
## Redis CPU under an identical workload (2,000 rounds, ~42k script calls)
|
||||
|
||||
Same logical workload both arms (enqueue + batched dequeue + ack), so the
|
||||
`evalsha` call count is identical and the difference is pure vtime overhead. Note
|
||||
the RunQueue Lua runs as `EVALSHA`, so per-`redis.call` costs inside a script are
|
||||
not separable in `commandstats`; the reportable signals are total Redis CPU and
|
||||
aggregate `evalsha` time per call.
|
||||
|
||||
| keys (N) | CPU-sec OFF | CPU-sec ON | delta | overhead | evalsha usec/call OFF | evalsha usec/call ON |
|
||||
| --- | --- | --- | --- | --- | --- | --- |
|
||||
| 100 | 1.34 | 1.49 | +0.15 | +12% | 22.3 | 26.2 |
|
||||
| 1,000 | 1.34 | 1.44 | +0.10 | +7% | 22.2 | 24.8 |
|
||||
| 10,000 | 1.38 | 1.45 | +0.07 | +5% | 23.8 | 25.8 |
|
||||
|
||||
CPU overhead does not grow with cardinality, it shrinks: +12% at 100 keys down to
|
||||
+5% at 10k. Per-script cost stays roughly flat (about +2 to +4 usec/call, ~25
|
||||
usec/call at every cardinality), which is what the design predicts: the pass-1
|
||||
dequeue window is fixed (`maxCount * multiplier`, default 30) and independent of
|
||||
N, and the added ZSET ops are O(log N), so `log2(10000) ~= 13` adds a negligible
|
||||
constant. The overhead falls as a percentage because that fixed per-call cost is
|
||||
amortised over more work as the queue grows.
|
||||
|
||||
## Membership under sustained churn (N = 10,000, flag ON)
|
||||
|
||||
60 rounds of continuous registration + drain (fresh keys enqueued while others
|
||||
are served and acked), holding cardinality at 10k:
|
||||
|
||||
| round | ckIndex card | ckVtime card | ratio |
|
||||
| --- | --- | --- | --- |
|
||||
| 0 | 10,000 | 10,000 | 1.0 |
|
||||
| 20 | 10,000 | 10,000 | 1.0 |
|
||||
| 40 | 10,000 | 10,000 | 1.0 |
|
||||
| 59 | 10,000 | 10,000 | 1.0 |
|
||||
|
||||
`ckVtime` membership tracks `ckIndex` exactly throughout: no tombstone
|
||||
accumulation, no unbounded growth. The bounded/self-healing drift the limitations
|
||||
doc calls out (ack/TTL/DLQ draining a variant without a vtime GC) stays reclaimed
|
||||
by the next vtime pass and the state TTL.
|
||||
|
||||
## Bottom line for the cardinality question
|
||||
|
||||
- **Memory** grows linearly with concurrency-key cardinality, adding one
|
||||
`ckIndex`-sized ZSET per base queue (~150 B/key): a low-single-digit MB even at
|
||||
10k keys on one queue, bounded by live cardinality and TTL-reclaimed. A sudden
|
||||
10k-key queue costs about +1.3 MB for that queue.
|
||||
- **CPU** overhead is small and does not scale with cardinality: ~+5 to +12% on
|
||||
an identical workload, per-script cost flat at ~+2 to +4 usec/call regardless of
|
||||
N, because the dequeue scan window is fixed and the ZSET ops are O(log N).
|
||||
- **No runaway state**: `ckVtime` never outgrows `ckIndex` under sustained churn.
|
||||
@@ -0,0 +1,390 @@
|
||||
import { createRedisClient } from "@internal/redis";
|
||||
import { trace } from "@internal/tracing";
|
||||
import { Logger } from "@trigger.dev/core/logger";
|
||||
import { Decimal } from "@trigger.dev/database";
|
||||
import { mkdirSync, writeFileSync } from "node:fs";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js";
|
||||
import { RunQueue } from "../index.js";
|
||||
import { RunQueueFullKeyProducer } from "../keyProducer.js";
|
||||
import type { InputPayload } from "../types.js";
|
||||
|
||||
// CK virtual-time RESOURCE arm: Redis CPU + memory vs concurrency-key cardinality.
|
||||
//
|
||||
// Answers "how do these changes affect the run-queue Redis CPU/memory, and how do
|
||||
// both react as cardinality grows (e.g. 10k concurrency keys on one base queue)".
|
||||
// Drives a real RunQueue against an EXTERNAL dedicated Redis, flag OFF vs ON on
|
||||
// identical load, and reads server-side metrics (INFO memory/cpu/commandstats,
|
||||
// MEMORY USAGE, OBJECT ENCODING) that are unaffected by client<->server RTT.
|
||||
//
|
||||
// Inert unless CK_BENCH_REDIS_URL is set. FLUSHALLs the target between points, so
|
||||
// point it ONLY at a dedicated throwaway store (a redis-bench lab store).
|
||||
//
|
||||
// export CK_BENCH_REDIS_URL="$(lab store url ckbench1)"
|
||||
// pnpm exec vitest run src/run-queue/bench/ckResourceBench.bench.test.ts
|
||||
//
|
||||
// Env knobs:
|
||||
// CK_RES_MEM_CARDS=100,1000,10000,50000 memory-at-rest sweep
|
||||
// CK_RES_CPU_CARDS=100,1000,10000 cpu-under-load sweep
|
||||
// CK_RES_LOAD_OPS=8000 load rounds per cpu point
|
||||
// CK_RES_CONCURRENCY=64 client concurrency (beats RTT)
|
||||
// CK_RES_CHURN_CARD=10000 churn/tombstone cardinality
|
||||
// CK_RES_CHURN_ROUNDS=60 churn sample rounds
|
||||
// CK_BENCH_OUT=./bench-results
|
||||
//
|
||||
// NOTE: the RunQueue Lua commands run via EVALSHA, so redis.call() ops inside a
|
||||
// script do NOT show up as separate cmdstat_* lines; they roll up under evalsha.
|
||||
// The reportable CPU signals are therefore total used_cpu over an identical
|
||||
// workload and aggregate evalsha usec_per_call, not a per-Redis-command split.
|
||||
|
||||
const REDIS_URL = process.env.CK_BENCH_REDIS_URL;
|
||||
const MEM_CARDS = (process.env.CK_RES_MEM_CARDS ?? "100,1000,10000,50000")
|
||||
.split(",")
|
||||
.map((s) => +s.trim());
|
||||
const CPU_CARDS = (process.env.CK_RES_CPU_CARDS ?? "100,1000,10000")
|
||||
.split(",")
|
||||
.map((s) => +s.trim());
|
||||
const LOAD_OPS = +(process.env.CK_RES_LOAD_OPS ?? "8000");
|
||||
const CONCURRENCY = +(process.env.CK_RES_CONCURRENCY ?? "64");
|
||||
const CHURN_CARD = +(process.env.CK_RES_CHURN_CARD ?? "10000");
|
||||
const CHURN_ROUNDS = +(process.env.CK_RES_CHURN_ROUNDS ?? "60");
|
||||
const OUT_DIR = process.env.CK_BENCH_OUT ?? "./bench-results";
|
||||
|
||||
const keys = new RunQueueFullKeyProducer();
|
||||
|
||||
const testOptions = {
|
||||
name: "rq",
|
||||
tracer: trace.getTracer("rq"),
|
||||
workers: 1,
|
||||
defaultEnvConcurrency: 1_000_000,
|
||||
logger: new Logger("RunQueue", "error"),
|
||||
retryOptions: {
|
||||
maxAttempts: 5,
|
||||
factor: 1.1,
|
||||
minTimeoutInMs: 100,
|
||||
maxTimeoutInMs: 1_000,
|
||||
randomize: true,
|
||||
},
|
||||
keys,
|
||||
};
|
||||
|
||||
const env = {
|
||||
id: "e1234",
|
||||
type: "PRODUCTION" as const,
|
||||
maximumConcurrencyLimit: 1_000_000,
|
||||
concurrencyLimitBurstFactor: new Decimal(1.0),
|
||||
project: { id: "p1234" },
|
||||
organization: { id: "o1234" },
|
||||
};
|
||||
|
||||
function conn() {
|
||||
const u = new URL(REDIS_URL!);
|
||||
return {
|
||||
host: u.hostname,
|
||||
port: Number(u.port || "6379"),
|
||||
password: decodeURIComponent(u.password || "") || undefined,
|
||||
username: decodeURIComponent(u.username || "") || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
function createQueue(keyPrefix: string, vtimeEnabled: boolean) {
|
||||
const c = conn();
|
||||
return new RunQueue({
|
||||
...testOptions,
|
||||
masterQueueConsumersDisabled: true,
|
||||
workerOptions: { disabled: true },
|
||||
ckVirtualTimeScheduling: { enabled: vtimeEnabled },
|
||||
queueSelectionStrategy: new FairQueueSelectionStrategy({ redis: { keyPrefix, ...c }, keys }),
|
||||
redis: { keyPrefix, ...c },
|
||||
});
|
||||
}
|
||||
|
||||
function makeMessage(o: Partial<InputPayload> = {}): InputPayload {
|
||||
return {
|
||||
runId: "r1",
|
||||
taskIdentifier: "task/my-task",
|
||||
orgId: "o1234",
|
||||
projectId: "p1234",
|
||||
environmentId: "e1234",
|
||||
environmentType: "PRODUCTION",
|
||||
queue: "task/my-task",
|
||||
timestamp: Date.now(),
|
||||
attempt: 0,
|
||||
...o,
|
||||
};
|
||||
}
|
||||
|
||||
// bounded-concurrency runner (beats the ~3.6ms workstation->box RTT)
|
||||
async function pool(n: number, count: number, fn: (i: number) => Promise<void>) {
|
||||
let i = 0;
|
||||
await Promise.all(
|
||||
Array.from({ length: Math.min(n, count) }, async () => {
|
||||
while (i < count) {
|
||||
const idx = i++;
|
||||
await fn(idx);
|
||||
}
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
// ---- server-side metric helpers (separate no-prefix admin client) ----
|
||||
function admin() {
|
||||
return createRedisClient(conn(), { onError: () => {} });
|
||||
}
|
||||
function infoField(info: string, key: string): number {
|
||||
const line = info.split("\n").find((l) => l.startsWith(key + ":"));
|
||||
return line ? Number(line.split(":")[1]) : NaN;
|
||||
}
|
||||
async function usedMemory(a: any) {
|
||||
return infoField(await a.info("memory"), "used_memory");
|
||||
}
|
||||
async function usedCpu(a: any) {
|
||||
const i = await a.info("cpu");
|
||||
return infoField(i, "used_cpu_user") + infoField(i, "used_cpu_sys");
|
||||
}
|
||||
function evalsha(info: string) {
|
||||
const line = info.split("\n").find((l) => l.startsWith("cmdstat_evalsha:"));
|
||||
if (!line) return { calls: 0, usec: 0, usecPerCall: 0 };
|
||||
const g = (k: string) => Number(line.match(new RegExp(`${k}=([0-9.]+)`))?.[1] ?? 0);
|
||||
return { calls: g("calls"), usec: g("usec"), usecPerCall: g("usec_per_call") };
|
||||
}
|
||||
|
||||
// ---- build N distinct concurrency keys (one queued message each) ----
|
||||
async function buildCardinality(queue: RunQueue, n: number) {
|
||||
const t0 = Date.now() - 500_000;
|
||||
await pool(CONCURRENCY, n, async (i) => {
|
||||
await queue.enqueueMessage({
|
||||
env,
|
||||
message: makeMessage({ runId: `r-${i}`, concurrencyKey: `k${i}`, timestamp: t0 + i }),
|
||||
workerQueue: env.id,
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function findKey(a: any, prefix: string, suffix: string): Promise<string | null> {
|
||||
const found = await a.keys(`${prefix}*:${suffix}`);
|
||||
return found[0] ?? null;
|
||||
}
|
||||
|
||||
// ---- the bench ----
|
||||
describe.runIf(!!REDIS_URL)("CK virtual-time resource + cardinality benchmark", () => {
|
||||
it(
|
||||
"measures Redis memory and CPU vs cardinality, flag OFF vs ON",
|
||||
{ timeout: 60 * 60_000 },
|
||||
async () => {
|
||||
const a = admin();
|
||||
const report: any = { generatedAtMs: Date.now(), memory: [], cpu: [], churn: null };
|
||||
|
||||
// ---------- memory at rest ----------
|
||||
for (const n of MEM_CARDS) {
|
||||
const row: any = { cardinality: n };
|
||||
for (const on of [false, true]) {
|
||||
await a.flushall();
|
||||
await a.call("CONFIG", "RESETSTAT");
|
||||
const prefix = `ckres:mem:${n}:${on ? "on" : "off"}:`;
|
||||
const q = createQueue(prefix, on);
|
||||
try {
|
||||
await q.updateEnvConcurrencyLimits(env);
|
||||
await buildCardinality(q, n);
|
||||
const arm = on ? "on" : "off";
|
||||
row[`used_memory_${arm}`] = await usedMemory(a);
|
||||
const ckIndexKey = await findKey(a, prefix, "ckIndex");
|
||||
const ckVtimeKey = await findKey(a, prefix, "ckVtime");
|
||||
row[`ckIndex_bytes_${arm}`] = ckIndexKey
|
||||
? await a.call("MEMORY", "USAGE", ckIndexKey)
|
||||
: null;
|
||||
row[`ckIndex_card_${arm}`] = ckIndexKey ? await a.zcard(ckIndexKey) : 0;
|
||||
if (on) {
|
||||
row.ckVtime_bytes = ckVtimeKey ? await a.call("MEMORY", "USAGE", ckVtimeKey) : null;
|
||||
row.ckVtime_card = ckVtimeKey ? await a.zcard(ckVtimeKey) : 0;
|
||||
row.ckVtime_encoding = ckVtimeKey
|
||||
? await a.call("OBJECT", "ENCODING", ckVtimeKey)
|
||||
: null;
|
||||
// ckVtime must mirror ckIndex membership when built via the slow path
|
||||
expect(row.ckVtime_card).toBe(row.ckIndex_card_on);
|
||||
}
|
||||
} finally {
|
||||
await q.quit();
|
||||
}
|
||||
}
|
||||
row.used_memory_delta = row.used_memory_on - row.used_memory_off;
|
||||
row.ckVtime_over_ckIndex =
|
||||
row.ckIndex_bytes_on && row.ckVtime_bytes
|
||||
? +(row.ckVtime_bytes / row.ckIndex_bytes_on).toFixed(2)
|
||||
: null;
|
||||
report.memory.push(row);
|
||||
// eslint-disable-next-line no-console
|
||||
console.log(
|
||||
`[ckres] mem N=${n}: used_memory delta=${row.used_memory_delta}B ckVtime=${row.ckVtime_bytes}B (${row.ckVtime_encoding})`
|
||||
);
|
||||
}
|
||||
|
||||
// ---------- CPU under identical load ----------
|
||||
for (const n of CPU_CARDS) {
|
||||
const row: any = { cardinality: n };
|
||||
for (const on of [false, true]) {
|
||||
await a.flushall();
|
||||
const prefix = `ckres:cpu:${n}:${on ? "on" : "off"}:`;
|
||||
const q = createQueue(prefix, on);
|
||||
try {
|
||||
await q.updateEnvConcurrencyLimits(env);
|
||||
await buildCardinality(q, n);
|
||||
const shard = keys.masterQueueShardForEnvironment(env.id, 2);
|
||||
|
||||
await a.call("CONFIG", "RESETSTAT");
|
||||
const cpu0 = await usedCpu(a);
|
||||
const t0wall = Date.now();
|
||||
|
||||
// identical workload both arms: LOAD_OPS rounds, each round enqueues
|
||||
// 10 fresh messages (rotating keys, keeps N populated) and does one
|
||||
// batched dequeue (maxCount 10) + acks the served set.
|
||||
let served = 0;
|
||||
await pool(CONCURRENCY, LOAD_OPS, async (i) => {
|
||||
for (let j = 0; j < 10; j++) {
|
||||
await q.enqueueMessage({
|
||||
env,
|
||||
message: makeMessage({
|
||||
runId: `L-${i}-${j}`,
|
||||
concurrencyKey: `k${(i * 10 + j) % n}`,
|
||||
timestamp: Date.now(),
|
||||
}),
|
||||
workerQueue: env.id,
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
}
|
||||
const msgs = await q.testDequeueFromMasterQueue(shard, env.id, 10);
|
||||
served += msgs.length;
|
||||
for (const m of msgs) {
|
||||
await q.acknowledgeMessage(env.organization.id, m.messageId, {
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
const cpu1 = await usedCpu(a);
|
||||
const es = evalsha(await a.info("commandstats"));
|
||||
const arm = on ? "on" : "off";
|
||||
row[`cpu_sec_${arm}`] = +(cpu1 - cpu0).toFixed(3);
|
||||
row[`evalsha_calls_${arm}`] = es.calls;
|
||||
row[`evalsha_usec_per_call_${arm}`] = +es.usecPerCall.toFixed(2);
|
||||
row[`wall_ms_${arm}`] = Date.now() - t0wall;
|
||||
row[`served_${arm}`] = served;
|
||||
} finally {
|
||||
await q.quit();
|
||||
}
|
||||
}
|
||||
row.cpu_sec_delta = +(row.cpu_sec_on - row.cpu_sec_off).toFixed(3);
|
||||
row.cpu_overhead_pct = row.cpu_sec_off
|
||||
? Math.round(((row.cpu_sec_on - row.cpu_sec_off) / row.cpu_sec_off) * 100)
|
||||
: null;
|
||||
report.cpu.push(row);
|
||||
// eslint-disable-next-line no-console
|
||||
console.log(
|
||||
`[ckres] cpu N=${n}: cpu OFF=${row.cpu_sec_off}s ON=${row.cpu_sec_on}s (${row.cpu_overhead_pct}%) evalsha usec/call OFF=${row.evalsha_usec_per_call_off} ON=${row.evalsha_usec_per_call_on}`
|
||||
);
|
||||
}
|
||||
|
||||
// ---------- churn: ckVtime membership stays bounded vs ckIndex ----------
|
||||
{
|
||||
await a.flushall();
|
||||
const prefix = `ckres:churn:on:`;
|
||||
const q = createQueue(prefix, true);
|
||||
const samples: any[] = [];
|
||||
try {
|
||||
await q.updateEnvConcurrencyLimits(env);
|
||||
await buildCardinality(q, CHURN_CARD);
|
||||
const shard = keys.masterQueueShardForEnvironment(env.id, 2);
|
||||
const ckIndexKey = (await findKey(a, prefix, "ckIndex"))!;
|
||||
const ckVtimeKey = (await findKey(a, prefix, "ckVtime"))!;
|
||||
let nextKey = CHURN_CARD;
|
||||
for (let r = 0; r < CHURN_ROUNDS; r++) {
|
||||
// hold cardinality: each iteration enqueues one FRESH key and drains
|
||||
// one message (maxCount 1), so registration and GC churn continuously
|
||||
// while total membership stays ~CHURN_CARD.
|
||||
await pool(CONCURRENCY, 200, async () => {
|
||||
await q.enqueueMessage({
|
||||
env,
|
||||
message: makeMessage({
|
||||
runId: `C-${nextKey}`,
|
||||
concurrencyKey: `k${nextKey++}`,
|
||||
timestamp: Date.now(),
|
||||
}),
|
||||
workerQueue: env.id,
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
const msgs = await q.testDequeueFromMasterQueue(shard, env.id, 1);
|
||||
for (const m of msgs) {
|
||||
await q.acknowledgeMessage(env.organization.id, m.messageId, {
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
}
|
||||
});
|
||||
if (r % 10 === 0 || r === CHURN_ROUNDS - 1) {
|
||||
const ckIndexCard = await a.zcard(ckIndexKey);
|
||||
const ckVtimeCard = await a.zcard(ckVtimeKey);
|
||||
samples.push({
|
||||
round: r,
|
||||
ckIndex: ckIndexCard,
|
||||
ckVtime: ckVtimeCard,
|
||||
ratio: ckIndexCard ? +(ckVtimeCard / ckIndexCard).toFixed(2) : null,
|
||||
});
|
||||
}
|
||||
}
|
||||
report.churn = { cardinality: CHURN_CARD, rounds: CHURN_ROUNDS, samples };
|
||||
// bounded: ckVtime never wildly exceeds ckIndex (allow generous 3x for transient tombstones)
|
||||
for (const s of samples) if (s.ratio !== null) expect(s.ratio).toBeLessThanOrEqual(3);
|
||||
} finally {
|
||||
await q.quit();
|
||||
}
|
||||
}
|
||||
|
||||
await a.flushall();
|
||||
await a.quit().catch(() => {});
|
||||
|
||||
mkdirSync(OUT_DIR, { recursive: true });
|
||||
writeFileSync(`${OUT_DIR}/ck-resource-results.json`, JSON.stringify(report, null, 2));
|
||||
writeFileSync(`${OUT_DIR}/ck-resource-results.md`, renderMarkdown(report));
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
function renderMarkdown(r: any): string {
|
||||
const L: string[] = [];
|
||||
const kb = (b: number) => (b == null ? "n/a" : (b / 1024).toFixed(1) + "KB");
|
||||
L.push(`# CK virtual-time resource + cardinality results`, "");
|
||||
L.push(
|
||||
`Redis \`${(process.env.CK_BENCH_REDIS_URL || "").replace(/:[^:@/]*@/, ":***@")}\`. Relative OFF-vs-ON on one box; not prod scale.`,
|
||||
""
|
||||
);
|
||||
L.push(`## Memory at rest (single base queue, one message per key)`, "");
|
||||
L.push(
|
||||
`| keys (N) | used_memory OFF | used_memory ON | delta | ckIndex bytes | ckVtime bytes | ckVtime/ckIndex | ckVtime encoding |`
|
||||
);
|
||||
L.push(`| --- | --- | --- | --- | --- | --- | --- | --- |`);
|
||||
for (const m of r.memory)
|
||||
L.push(
|
||||
`| ${m.cardinality} | ${kb(m.used_memory_off)} | ${kb(m.used_memory_on)} | ${kb(m.used_memory_delta)} | ${kb(m.ckIndex_bytes_on)} | ${kb(m.ckVtime_bytes)} | ${m.ckVtime_over_ckIndex ?? "n/a"} | ${m.ckVtime_encoding ?? "n/a"} |`
|
||||
);
|
||||
L.push(
|
||||
"",
|
||||
`## Redis CPU under identical workload (${process.env.CK_RES_LOAD_OPS ?? "8000"} rounds)`,
|
||||
""
|
||||
);
|
||||
L.push(
|
||||
`| keys (N) | CPU-sec OFF | CPU-sec ON | delta | overhead | evalsha usec/call OFF | evalsha usec/call ON | evalsha calls OFF/ON |`
|
||||
);
|
||||
L.push(`| --- | --- | --- | --- | --- | --- | --- | --- |`);
|
||||
for (const c of r.cpu)
|
||||
L.push(
|
||||
`| ${c.cardinality} | ${c.cpu_sec_off} | ${c.cpu_sec_on} | ${c.cpu_sec_delta} | ${c.cpu_overhead_pct}% | ${c.evalsha_usec_per_call_off} | ${c.evalsha_usec_per_call_on} | ${c.evalsha_calls_off}/${c.evalsha_calls_on} |`
|
||||
);
|
||||
if (r.churn) {
|
||||
L.push("", `## Tombstone / membership under churn (N=${r.churn.cardinality}, flag ON)`, "");
|
||||
L.push(`| round | ckIndex card | ckVtime card | ratio |`, `| --- | --- | --- | --- |`);
|
||||
for (const s of r.churn.samples)
|
||||
L.push(`| ${s.round} | ${s.ckIndex} | ${s.ckVtime} | ${s.ratio} |`);
|
||||
}
|
||||
L.push("");
|
||||
return L.join("\n");
|
||||
}
|
||||
Reference in New Issue
Block a user