9. Building Your Own Load Generators

Off-the-shelf tools cover most needs, but building a minimal generator yourself teaches you exactly what "requests/sec" costs at the socket level, and is sometimes necessary for protocol-specific or highly custom scenarios (§6.3's idempotency race test, for instance, needs custom assertion logic no generic tool provides).

9.1 C — raw TCP client with precise per-request latency (epoll)

Demonstrates connection-per-request cost (§6.6, IoT scenario) using raw sockets, TCP_NODELAY (§2.4), and clock_gettime(CLOCK_MONOTONIC) for latency measurement unaffected by wall-clock adjustments.

// tcp_bench.c — minimal HTTP/1.0 GET load generator, connection-per-request
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/tcp.h>

static double now_ms(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec * 1000.0 + ts.tv_nsec / 1e6;
}

static double do_request(const char *ip, int port, const char *path) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    int one = 1;
    setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); // disable Nagle, see 2.4

    struct sockaddr_in addr = {0};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    inet_pton(AF_INET, ip, &addr.sin_addr);

    double t0 = now_ms();
    if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
        perror("connect");
        close(fd);
        return -1;
    }

    char req[512];
    int n = snprintf(req, sizeof(req),
        "GET %s HTTP/1.0\r\nHost: %s\r\nConnection: close\r\n\r\n", path, ip);
    write(fd, req, n);

    char buf[4096];
    ssize_t r;
    while ((r = read(fd, buf, sizeof(buf))) > 0) { /* drain response */ }
    double elapsed = now_ms() - t0;

    close(fd);
    return elapsed;
}

int main(int argc, char **argv) {
    if (argc < 5) {
        fprintf(stderr, "usage: %s <ip> <port> <path> <num_requests>\n", argv[0]);
        return 1;
    }
    const char *ip = argv[1];
    int port = atoi(argv[2]);
    const char *path = argv[3];
    int total = atoi(argv[4]);

    double *latencies = malloc(sizeof(double) * total);
    int ok = 0;
    double sum = 0, max_lat = 0;

    for (int i = 0; i < total; i++) {
        double lat = do_request(ip, port, path);
        if (lat >= 0) {
            latencies[ok++] = lat;
            sum += lat;
            if (lat > max_lat) max_lat = lat;
        }
    }

    // sort for percentiles
    for (int i = 0; i < ok - 1; i++)
        for (int j = i + 1; j < ok; j++)
            if (latencies[j] < latencies[i]) {
                double tmp = latencies[i]; latencies[i] = latencies[j]; latencies[j] = tmp;
            }

    printf("requests=%d ok=%d\n", total, ok);
    printf("avg=%.2fms p50=%.2fms p95=%.2fms p99=%.2fms max=%.2fms\n",
        sum / ok,
        latencies[(int)(ok * 0.50)],
        latencies[(int)(ok * 0.95)],
        latencies[(int)(ok * 0.99)],
        max_lat);

    free(latencies);
    return 0;
}

gcc -O2 -o tcp_bench tcp_bench.c
./tcp_bench 127.0.0.1 8080 /api/health 1000

This is intentionally sequential (one connection at a time) to isolate per-connection setup cost. For concurrent connection-per-request load, fork worker processes or add an epoll event loop managing many sockets in parallel — which is exactly what wrk does internally.

9.2 Python — asyncio open-loop load generator with latency histogram

Models a proper open-loop generator (§3.4): requests are scheduled at a fixed Poisson rate regardless of response time, avoiding coordinated omission.

# async_load.py
import asyncio
import time
import random
import statistics
import aiohttp

TARGET_URL = "http://localhost:8080/api/products/1"
TARGET_RATE_PER_SEC = 200      # open-loop arrival rate
DURATION_SEC = 30

latencies = []
errors = 0

async def fire_request(session: aiohttp.ClientSession):
    global errors
    start = time.perf_counter()
    try:
        async with session.get(TARGET_URL, timeout=aiohttp.ClientTimeout(total=5)) as resp:
            await resp.read()
            elapsed = (time.perf_counter() - start) * 1000
            if resp.status == 200:
                latencies.append(elapsed)
            else:
                errors += 1
    except Exception:
        errors += 1

async def generator(session: aiohttp.ClientSession):
    end_time = time.perf_counter() + DURATION_SEC
    tasks = []
    while time.perf_counter() < end_time:
        # schedule immediately; don't await -> true open-loop, overlapping in-flight requests
        tasks.append(asyncio.create_task(fire_request(session)))
        # Poisson inter-arrival, see 4.2
        await asyncio.sleep(random.expovariate(TARGET_RATE_PER_SEC))
    await asyncio.gather(*tasks)

def percentile(data, p):
    data = sorted(data)
    k = int(len(data) * p)
    return data[min(k, len(data) - 1)]

async def main():
    connector = aiohttp.TCPConnector(limit=0)  # no artificial cap; let the OS/server bound it
    async with aiohttp.ClientSession(connector=connector) as session:
        await generator(session)

    total = len(latencies) + errors
    print(f"requests={total} success={len(latencies)} errors={errors}")
    if latencies:
        print(f"avg={statistics.mean(latencies):.2f}ms "
              f"p50={percentile(latencies,0.50):.2f}ms "
              f"p95={percentile(latencies,0.95):.2f}ms "
              f"p99={percentile(latencies,0.99):.2f}ms "
              f"max={max(latencies):.2f}ms")

if __name__ == "__main__":
    asyncio.run(main())

pip install aiohttp
python async_load.py

9.3 Go — worker-pool HTTP load generator (closed-loop, concurrency-controlled)

Idiomatic Go concurrency pattern: N goroutines pulling from a shared job channel, results aggregated on a separate channel — the pattern behind most Go-based load tools (hey, vegeta).

// main.go
package main

import (
	"fmt"
	"net/http"
	"os"
	"sort"
	"strconv"
	"sync"
	"time"
)

type result struct {
	latency time.Duration
	err     error
	status  int
}

func worker(client *http.Client, url string, jobs <-chan int, results chan<- result, wg *sync.WaitGroup) {
	defer wg.Done()
	for range jobs {
		start := time.Now()
		resp, err := client.Get(url)
		lat := time.Since(start)
		if err != nil {
			results <- result{latency: lat, err: err}
			continue
		}
		resp.Body.Close()
		results <- result{latency: lat, status: resp.StatusCode}
	}
}

func main() {
	url := os.Args[1]
	concurrency, _ := strconv.Atoi(os.Args[2])
	totalRequests, _ := strconv.Atoi(os.Args[3])

	// reuse connections: keep-alive transport tuned for high concurrency (see 5.3)
	transport := &http.Transport{
		MaxIdleConns:        concurrency * 2,
		MaxIdleConnsPerHost: concurrency * 2,
		IdleConnTimeout:     90 * time.Second,
	}
	client := &http.Client{Transport: transport, Timeout: 10 * time.Second}

	jobs := make(chan int, totalRequests)
	results := make(chan result, totalRequests)
	var wg sync.WaitGroup

	start := time.Now()
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go worker(client, url, jobs, results, &wg)
	}
	for i := 0; i < totalRequests; i++ {
		jobs <- i
	}
	close(jobs)

	go func() { wg.Wait(); close(results) }()

	var latencies []float64
	errs, ok := 0, 0
	for r := range results {
		if r.err != nil || r.status >= 500 {
			errs++
		} else {
			ok++
			latencies = append(latencies, r.latency.Seconds()*1000)
		}
	}
	elapsed := time.Since(start)

	sort.Float64s(latencies)
	pct := func(p float64) float64 {
		if len(latencies) == 0 {
			return 0
		}
		idx := int(float64(len(latencies)) * p)
		if idx >= len(latencies) {
			idx = len(latencies) - 1
		}
		return latencies[idx]
	}

	fmt.Printf("total=%d ok=%d errors=%d duration=%s throughput=%.1f req/s\n",
		totalRequests, ok, errs, elapsed, float64(ok)/elapsed.Seconds())
	fmt.Printf("p50=%.2fms p95=%.2fms p99=%.2fms\n", pct(0.50), pct(0.95), pct(0.99))
}

go run main.go http://localhost:8080/api/products/1 200 50000
# args: url, concurrency, total requests

9.4 Rust — tokio async generator with hdrhistogram for accurate percentiles

Rust's async ecosystem (tokio + reqwest) plus hdrhistogram (the same technique HdrHistogram/wrk2 use internally) gives accurate, low-overhead percentile tracking without keeping every raw sample in memory.

// Cargo.toml
// [dependencies]
// tokio = { version = "1", features = ["full"] }
// reqwest = { version = "0.12", features = ["rustls-tls"] }
// hdrhistogram = "7"
// futures = "0.3"

use hdrhistogram::Histogram;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    let url = std::env::args().nth(1).unwrap();
    let concurrency: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let total: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    let client = reqwest::Client::builder()
        .pool_max_idle_per_host(concurrency)
        .build()
        .unwrap();

    let sem = Arc::new(Semaphore::new(concurrency));   // bound in-flight requests
    let hist = Arc::new(Mutex::new(Histogram::<u64>::new(3).unwrap())); // 3 sig figs
    let errors = Arc::new(std::sync::atomic::AtomicUsize::new(0));

    let start = Instant::now();
    let mut handles = Vec::with_capacity(total);

    for _ in 0..total {
        let permit = sem.clone().acquire_owned().await.unwrap();
        let client = client.clone();
        let url = url.clone();
        let hist = hist.clone();
        let errors = errors.clone();

        handles.push(tokio::spawn(async move {
            let t0 = Instant::now();
            match client.get(&url).send().await {
                Ok(resp) if resp.status().is_success() => {
                    let _ = resp.bytes().await;
                    let micros = t0.elapsed().as_micros() as u64;
                    hist.lock().unwrap().record(micros).unwrap();
                }
                _ => {
                    errors.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }
            drop(permit);
        }));
    }

    for h in handles {
        let _ = h.await;
    }

    let elapsed = start.elapsed();
    let h = hist.lock().unwrap();
    let to_ms = |v: u64| v as f64 / 1000.0;

    println!(
        "total={} errors={} duration={:.2}s throughput={:.1} req/s",
        total,
        errors.load(std::sync::atomic::Ordering::Relaxed),
        elapsed.as_secs_f64(),
        h.len() as f64 / elapsed.as_secs_f64()
    );
    println!(
        "p50={:.2}ms p95={:.2}ms p99={:.2}ms p99.9={:.2}ms max={:.2}ms",
        to_ms(h.value_at_quantile(0.50)),
        to_ms(h.value_at_quantile(0.95)),
        to_ms(h.value_at_quantile(0.99)),
        to_ms(h.value_at_quantile(0.999)),
        to_ms(h.max())
    );
}

cargo run --release -- http://localhost:8080/api/products/1 500 100000

The Semaphore-bounded concurrency model here is deliberately closed-loop (§3.4) — it caps in-flight requests at concurrency. To convert this into an open-loop generator, replace the semaphore-gated spawn loop with a tokio::time::interval ticking at the target rate, spawning a task per tick without waiting on a permit (mirroring the Python asyncio example in §9.2).