Benchmark

"""flexiq throughput benchmark.

Measures:
1. Enqueue throughput (jobs/sec) using batch insert
2. Processing throughput (jobs/sec) with N workers
3. End-to-end latency
"""

import os
import threading
import time

from flexiq import Queue

# ── Configuration ────────────────────────────────────────

NUM_JOBS = 10_000
NUM_WORKERS = os.cpu_count() or 4
DB_PATH = ":memory:"  # In-memory for pure speed test

queue = Queue(db_path=DB_PATH, workers=NUM_WORKERS)

@queue.task()
def noop(x):
    """Minimal task — measures framework overhead."""
    return x

@queue.task()
def cpu_light(x):
    """Light CPU work — string formatting."""
    return f"processed-{x}-{'x' * 100}"

# ── Benchmark Functions ──────────────────────────────────

def bench_enqueue(task, n):
    """Measure batch enqueue throughput."""
    args_list = [(i,) for i in range(n)]

    start = time.perf_counter()
    task.map(args_list)
    elapsed = time.perf_counter() - start

    rate = n / elapsed
    print(f"  Enqueued {n:,} jobs in {elapsed:.2f}s ({rate:,.0f} jobs/s)")

def bench_process(n, baseline, timeout=120):
    """Measure processing throughput by polling stats until n jobs drain.

    Waiting on a single job handle isn't enough — jobs don't finish in
    enqueue order, so the counter (against a baseline from before this
    round) is the only reliable "all done" signal.
    """
    start = time.perf_counter()
    deadline = start + timeout
    while time.perf_counter() < deadline:
        stats = queue.stats()
        if stats["failed"] or stats["dead"]:
            print(f"  Aborting — jobs failed during the run: {stats}")
            return
        if stats["completed"] - baseline >= n:
            elapsed = time.perf_counter() - start
            print(f"  Processed {n:,} jobs in {elapsed:.2f}s ({n / elapsed:,.0f} jobs/s)")
            return
        time.sleep(0.02)
    print(f"  Timed out! Stats: {queue.stats()}")

def bench_latency(task, samples=100):
    """Measure single-job round-trip latency."""
    latencies = []
    for i in range(samples):
        start = time.perf_counter()
        job = task.delay(i)
        job.result(timeout=10)
        latencies.append(time.perf_counter() - start)

    avg = sum(latencies) / len(latencies)
    p50 = sorted(latencies)[len(latencies) // 2]
    p99 = sorted(latencies)[int(len(latencies) * 0.99)]
    print(f"  Latency (n={samples}): avg={avg*1000:.1f}ms p50={p50*1000:.1f}ms p99={p99*1000:.1f}ms")

# ── Main ─────────────────────────────────────────────────

def main():
    print(f"flexiq benchmark")
    print(f"  Workers: {NUM_WORKERS}")
    print(f"  Jobs:    {NUM_JOBS:,}")
    print(f"  DB:      {DB_PATH}")
    print()

    # Stage the first batch BEFORE the worker starts, so the enqueue rate
    # doesn't include worker/database contention — same methodology as the
    # other-language benchmarks on this page.
    print("── noop task (framework overhead) ──")
    bench_enqueue(noop, NUM_JOBS)

    worker_thread = threading.Thread(target=queue.run_worker, daemon=True)
    worker_thread.start()
    bench_process(NUM_JOBS, baseline=0)
    print()

    # The queue is fully drained here, so the running worker only
    # idle-polls during this enqueue — negligible contention.
    print("── cpu_light task ──")
    baseline = queue.stats()["completed"]
    bench_enqueue(cpu_light, NUM_JOBS)
    bench_process(NUM_JOBS, baseline=baseline)
    print()

    print("── single-job latency ──")
    bench_latency(noop)
    print()

    stats = queue.stats()
    print(f"Final stats: {stats}")

if __name__ == "__main__":
    main()

import { Queue } from "@byteveda/flexiq";

const N = 50_000;
// Fresh database per run — leftover rows from a previous run would satisfy
// the completed-count below early and skew the latency sample.
const queue = new Queue({ dbPath: `bench-${process.pid}.db` });

queue.task("noop", (_n: number) => undefined);

// 1. Enqueue throughput — stage N jobs in batches of 1,000.
const t0 = Date.now();
for (let i = 0; i < N; i += 1_000) {
  queue.enqueueMany("noop", Array.from({ length: 1_000 }, (_, k) => ({ args: [i + k] })));
}
const enqueueMs = Date.now() - t0;

// 2. Processing throughput — drain the backlog, polling stats with a
// deadline so a failed job or stuck worker can't hang the benchmark.
const t1 = Date.now();
const deadline = t1 + 120_000;
const worker = queue.runWorker({ queues: ["default"], batchSize: 64, channelCapacity: 512 });
let processMs = 0;
try {
  while (true) {
    const stats = await queue.stats();
    if (stats.failed > 0 || stats.dead > 0) {
      throw new Error(`jobs failed during the run: ${JSON.stringify(stats)}`);
    }
    if (stats.completed >= N) {
      processMs = Date.now() - t1;
      break;
    }
    if (Date.now() > deadline) {
      throw new Error(`drain timed out: ${JSON.stringify(stats)}`);
    }
    await new Promise((r) => setTimeout(r, 20));
  }
} finally {
  worker.stop();
}

// 3. End-to-end latency — created → completed across a sample of jobs.
const sample = await queue.listJobs({ status: "complete", limit: 1_000 });
const latencies = sample
  .filter((j) => j.completedAt)
  .map((j) => (j.completedAt as number) - j.createdAt)
  .sort((a, b) => a - b);
const p50 = latencies[Math.floor(latencies.length * 0.5)];
const p99 = latencies[Math.floor(latencies.length * 0.99)];

console.log(`enqueue:  ${Math.round(N / (enqueueMs / 1000))} jobs/s`);
console.log(`process:  ${Math.round(N / (processMs / 1000))} jobs/s`);
console.log(`latency:  p50 ${p50}ms  p99 ${p99}ms`);

import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.model.Job;
import org.byteveda.flexiq.model.JobFilter;
import org.byteveda.flexiq.model.JobStatus;
import org.byteveda.flexiq.model.QueueStats;
import org.byteveda.flexiq.task.Task;
import org.byteveda.flexiq.worker.Worker;

public final class Benchmark {
    private static final int N = 50_000;
    private static final Task<Integer> NOOP = Task.of("noop", Integer.class);

    public static void main(String[] args) throws Exception {
        // Fresh database per run — a reused file skews stats with old rows.
        Path dir = Files.createTempDirectory("flexiq-bench");
        try (FlexiQ flexiq = FlexiQ.builder().sqlite(dir.resolve("bench.db").toString()).open()) {

            // 1. Enqueue throughput — stage N jobs in batches of 1,000.
            long t0 = System.currentTimeMillis();
            for (int i = 0; i < N; i += 1_000) {
                List<Integer> batch = new ArrayList<>(1_000);
                for (int k = 0; k < 1_000; k++) {
                    batch.add(i + k);
                }
                flexiq.enqueueMany(NOOP, batch);
            }
            long enqueueMs = System.currentTimeMillis() - t0;

            // 2. Processing throughput — drain the backlog, polling stats with a
            // deadline so a failed job or stuck worker can't hang the benchmark.
            long t1 = System.currentTimeMillis();
            long deadline = t1 + 120_000;
            long processMs;
            try (Worker worker = flexiq.worker()
                    .handle(NOOP, n -> null)
                    .concurrency(8)
                    .batchSize(64)
                    .channelCapacity(512)
                    .start()) {
                QueueStats stats = flexiq.stats();
                while (stats.completed < N) {
                    if (stats.failed > 0 || stats.dead > 0) {
                        throw new IllegalStateException(
                                "jobs failed during the run: failed=" + stats.failed + " dead=" + stats.dead);
                    }
                    if (System.currentTimeMillis() > deadline) {
                        throw new IllegalStateException("drain timed out with " + stats.completed + "/" + N + " completed");
                    }
                    Thread.sleep(20);
                    stats = flexiq.stats();
                }
                processMs = System.currentTimeMillis() - t1;
            }

            // 3. End-to-end latency — created → completed across a sample of jobs.
            List<Long> latencies = flexiq
                    .listJobs(JobFilter.builder().status(JobStatus.COMPLETE).limit(1_000).build())
                    .stream()
                    .filter(job -> job.completedAt != null)
                    .map(job -> job.completedAt - job.createdAt)
                    .sorted()
                    .toList();
            long p50 = latencies.get((int) (latencies.size() * 0.5));
            long p99 = latencies.get((int) (latencies.size() * 0.99));

            System.out.printf("enqueue:  %d jobs/s%n", Math.round(N / (enqueueMs / 1000.0)));
            System.out.printf("process:  %d jobs/s%n", Math.round(N / (processMs / 1000.0)));
            System.out.printf("latency:  p50 %dms  p99 %dms%n", p50, p99);
        }
    }
}

python benchmark.py

node benchmark.ts

java -cp app.jar Benchmark

flexiq benchmark
  Workers: 8
  Jobs:    10,000
  DB:      :memory:

── noop task (framework overhead) ──
  Enqueued 10,000 jobs in 0.18s (55,556 jobs/s)
  Processed 10,000 jobs in 2.41s (4,149 jobs/s)

── cpu_light task ──
  Enqueued 10,000 jobs in 0.19s (52,632 jobs/s)
  Processed 10,000 jobs in 2.53s (3,953 jobs/s)

── single-job latency ──
  Latency (n=100): avg=1.2ms p50=1.1ms p99=3.4ms

Final stats: {'pending': 0, 'running': 0, 'completed': 20100, 'failed': 0, 'dead': 0, 'cancelled': 0}

enqueue:  120000 jobs/s
process:  35000 jobs/s
latency:  p50 8ms  p99 42ms

enqueue:  110000 jobs/s
process:  30000 jobs/s
latency:  p50 9ms  p99 45ms