Example: Benchmark Harness

The repo ships a load-test harness at cmd/worker. It dispatches 100,000 tasks against a real Redis, enables worker tuning, and reports throughput and memory. Use it to size your own deployment or to sanity check a Redis instance.

Requires a real Redis

Unlike the unit tests, this harness needs a live Redis. The miniredis test-double is not fast enough for meaningful throughput numbers.

The harness

package main

import (
    "context"
    "fmt"
    "log"
    "runtime"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/soroosh-tanzadeh/taskrunner/redisstream"
    "github.com/soroosh-tanzadeh/taskrunner/runner"
)

func main() {
    wg := sync.WaitGroup{}
    client := redis.NewClient(&redis.Options{
        Addr:        "127.0.0.1:6379",
        DB:          10,
        PoolSize:    500,
        PoolTimeout: time.Second * 10,
    })

    taskRunner := runner.NewTaskRunner(runner.TaskRunnerConfig{
        BatchSize:                100,
        ConsumerGroup:            "benchmark-group",
        MinWorkers:               5,
        MaxWorkers:               10000,
        DesiredWaitTime:          time.Millisecond * 100,
        DesiredWaitTimeTolerance: time.Millisecond * 50,
        TuningCooldownSeconds:    1,
        NumFetchers:              10,
        FailedTaskHandler:        func(ctx context.Context, t runner.TaskMessage, e error) error { return nil },
        LongQueueHook:            func(s runner.Stats) {},
        LongQueueThreshold:       time.Millisecond * 100,
    }, client, redisstream.NewRedisStreamMessageQueue(client, "benchmark", "benchmark", time.Second*5, true))

    taskRunner.RegisterTask(&runner.Task{
        Name:               "task",
        MaxRetry:           10,
        ReservationTimeout: time.Second,
        Action: func(ctx context.Context, payload any) error {
            defer wg.Done()
            time.Sleep(500 * time.Millisecond) // simulate work
            return nil
        },
    })

    numTasks := 100000
    client.FlushAll(context.Background())
    for i := 0; i < numTasks; i++ {
        wg.Add(1)
        taskRunner.Dispatch(context.TODO(), "task", nil)
    }

    start := time.Now()
    fmt.Println("Start Consuming")
    var mStart runtime.MemStats
    runtime.ReadMemStats(&mStart)

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go taskRunner.Start(ctx)

    wg.Wait()
    fmt.Println("End Consuming")

    var mEnd runtime.MemStats
    runtime.ReadMemStats(&mEnd)

    elapsed := time.Since(start)
    throughput := float64(numTasks) / elapsed.Seconds()
    memUsage := (mEnd.Alloc - mStart.Alloc) / 1024 // KB

    log.Printf("Processed %d tasks in %s", numTasks, elapsed)
    log.Printf("Throughput: %.2f tasks/second", throughput)
    log.Printf("Memory Usage: %d Mb", memUsage/1024)

    <-time.After(time.Second * 120)
    client.FlushDB(context.Background())
}

What it demonstrates

1. Aggressive tuning bounds. MaxWorkers: 10000 lets the pool scale up hard; DesiredWaitTime: 100ms is a tight target that forces the tuner to add workers as the queue grows.
2. Large batch size + many fetchers. BatchSize: 100, NumFetchers: 10 keeps the pool fed.
3. Generous Redis pool. PoolSize: 500 avoids connection-starvation at high concurrency.
4. Per-task timing. Each Action sleeps 500ms, so 100k tasks need either 50,000 workers or ~14 hours of serial work -- the tuner adds capacity until PredictedWatchTime falls in band.

Run it

# Start a Redis first, e.g.:
docker run -d -p 6379:6379 redis:7.4-bookworm

# Run the harness (takes a couple of minutes)
go run cmd/worker/main.go

Sample output shape:

Start Consuming
End Consuming
2026/... Processed 100000 tasks in 1m12s
2026/... Throughput: 1388.89 tasks/second
2026/... Memory Usage: 87 Mb

Your numbers will vary with CPU, Redis performance, and the simulated time.Sleep. The harness is most useful for relative comparisons (before / after a config change, or across Redis versions).

Profiling

For deeper investigation, add net/http/pprof to the harness and capture a CPU or allocation profile:

import _ "net/http/pprof"

func main() {
    go http.ListenAndServe("localhost:6060", nil)
    // ...rest of main...
}

Then:

# CPU profile (30s)
go tool pprof http://localhost:6060/debug/pprof/profile?seconds=30

# Allocation profile
go tool pprof http://localhost:6060/debug/pprof/heap

For the engine's own micro-benchmarks (no Redis round-trips), see Benchmarks and run go test -bench . ./runner/....