diff --git a/Cargo.lock b/Cargo.lock index 815a8ef..202e436 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -93,18 +93,19 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] name = "bitflags" -version = "2.11.0" +version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" +checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" [[package]] name = "blasthttp" -version = "0.3.2" +version = "0.4.0" dependencies = [ "brotli", "bytes", "clap", "flate2", + "futures", "http", "http-body-util", "hyper", @@ -151,9 +152,9 @@ checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" [[package]] name = "cc" -version = "1.2.58" +version = "1.2.61" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e1e928d4b69e3077709075a938a05ffbedfa53a84c8f766efbf8220bb1ff60e1" +checksum = "d16d90359e986641506914ba71350897565610e87ce0ad9e6f28569db3dd5c6d" dependencies = [ "find-msvc-tools", "shlex", @@ -167,9 +168,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "clap" -version = "4.6.0" +version = "4.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b193af5b67834b676abd72466a96c1024e6a6ad978a1f484bd90b85c94041351" +checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" dependencies = [ "clap_builder", "clap_derive", @@ -189,9 +190,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.6.0" +version = "4.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1110bd8a634a1ab8cb04345d8d878267d57c3cf1b38d91b71af6686408bbca6a" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" dependencies = [ "heck", "proc-macro2", @@ -372,9 +373,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.16.1" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" [[package]] name = "heck" @@ -467,9 +468,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.13.1" +version = "2.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "45a8a2b9cb3e0b0c1803dbb0758ffac5de2f425b23c28f518faabd9d805342ff" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", "hashbrown", @@ -504,9 +505,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "libc" -version = "0.2.184" +version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" [[package]] name = "memchr" @@ -564,9 +565,9 @@ checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" [[package]] name = "openssl" -version = "0.10.76" +version = "0.10.78" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "951c002c75e16ea2c65b8c7e4d3d51d5530d8dfa7d060b4776828c88cfb18ecf" +checksum = "f38c4372413cdaaf3cc79dd92d29d7d9f5ab09b51b10dded508fb90bb70b9222" dependencies = [ "bitflags", "cfg-if", @@ -590,9 +591,9 @@ dependencies = [ [[package]] name = "openssl-sys" -version = "0.9.112" +version = "0.9.114" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb" +checksum = "13ce1245cd07fcc4cfdb438f7507b0c7e4f3849a69fd84d52374c66d83741bb6" dependencies = [ "cc", "libc", @@ -614,9 +615,9 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" [[package]] name = "pkg-config" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" [[package]] name = "portable-atomic" @@ -856,9 +857,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.50.0" +version = "1.52.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "27ad5e34374e03cfffefc301becb44e9dc3c17584f414349ebe29ed26661822d" +checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6" dependencies = [ "bytes", "libc", @@ -871,9 +872,9 @@ dependencies = [ [[package]] name = "tokio-macros" -version = "2.6.1" +version = "2.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c55a2eff8b69ce66c84f85e1da1c233edc36ceb85a2058d11b0d6a3c7e7569c" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 08548ea..b30183c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "blasthttp" -version = "0.3.2" +version = "0.4.0" edition = "2024" description = "Offensive-first HTTP library with Python bindings" license = "GPL-3.0" @@ -20,6 +20,9 @@ python = ["dep:pyo3", "dep:pyo3-async-runtimes", "pyo3/extension-module"] # Async runtime tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "net", "io-util"] } +# Stream combinators (buffer_unordered for concurrent batch dispatch) +futures = "0.3" + # Derive macros for h2 module error types thiserror = "2" diff --git a/pyproject.toml b/pyproject.toml index b5bdc45..1f4bee6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "maturin" [project] name = "blasthttp" -version = "0.3.2" +version = "0.4.0" description = "Offensive-first HTTP library" license = "GPL-3.0" requires-python = ">=3.10" diff --git a/scripts/bench/Makefile b/scripts/bench/Makefile new file mode 100644 index 0000000..6c3581b --- /dev/null +++ b/scripts/bench/Makefile @@ -0,0 +1,27 @@ +CC ?= cc +CFLAGS ?= -O2 -Wall -Wextra -Wno-unused-parameter +LDFLAGS ?= + +CURL_CFLAGS := $(shell pkg-config --cflags libcurl) +CURL_LIBS := $(shell pkg-config --libs libcurl) + +BIN_DIR := ../../target/bench + +all: $(BIN_DIR)/server $(BIN_DIR)/client-go $(BIN_DIR)/client-c + +$(BIN_DIR): + mkdir -p $(BIN_DIR) + +$(BIN_DIR)/server: server.go | $(BIN_DIR) + go build -o $@ server.go + +$(BIN_DIR)/client-go: client.go | $(BIN_DIR) + go build -o $@ client.go + +$(BIN_DIR)/client-c: client.c | $(BIN_DIR) + $(CC) $(CFLAGS) $(CURL_CFLAGS) -o $@ client.c $(CURL_LIBS) $(LDFLAGS) + +clean: + rm -f $(BIN_DIR)/server $(BIN_DIR)/client-go $(BIN_DIR)/client-c + +.PHONY: all clean diff --git a/scripts/bench/client.c b/scripts/bench/client.c new file mode 100644 index 0000000..e996fe8 --- /dev/null +++ b/scripts/bench/client.c @@ -0,0 +1,173 @@ +/* + * HTTP benchmark client using libcurl's multi interface. + * + * Usage: client + * Output: JSON lines to stdout, one per completed request: + * {"url":"...","status":200} or {"url":"...","error":"..."} + * + * Pattern: pre-allocate `workers` easy handles, add them all to a multi + * handle, then drive curl_multi_perform in a loop. When an easy handle + * finishes (CURLMSG_DONE), print its result and refill it with the next + * URL from the queue. Connection reuse happens automatically via libcurl's + * internal connection cache on the multi handle. + */ + +#define _POSIX_C_SOURCE 200809L +#include +#include +#include +#include + +struct slot { + CURL *eh; + char url[2048]; + char errbuf[CURL_ERROR_SIZE]; +}; + +static size_t discard_cb(char *p, size_t size, size_t nmemb, void *ud) { + (void)p; (void)ud; + return size * nmemb; +} + +/* Escape a string for inclusion as a JSON string value. Writes into dst + * (size n) and null-terminates. Handles ", \, and control chars. */ +static void json_escape(char *dst, size_t n, const char *src) { + size_t i = 0; + for (const unsigned char *s = (const unsigned char *)src; *s && i + 7 < n; s++) { + unsigned char c = *s; + if (c == '"' || c == '\\') { + dst[i++] = '\\'; dst[i++] = (char)c; + } else if (c == '\n') { dst[i++] = '\\'; dst[i++] = 'n'; } + else if (c == '\r') { dst[i++] = '\\'; dst[i++] = 'r'; } + else if (c == '\t') { dst[i++] = '\\'; dst[i++] = 't'; } + else if (c < 0x20) { + i += (size_t)snprintf(dst + i, n - i, "\\u%04x", c); + } else { + dst[i++] = (char)c; + } + } + dst[i] = 0; +} + +static void configure(CURL *eh, struct slot *s, const char *url) { + strncpy(s->url, url, sizeof(s->url) - 1); + s->url[sizeof(s->url) - 1] = 0; + s->errbuf[0] = 0; + curl_easy_setopt(eh, CURLOPT_URL, s->url); + curl_easy_setopt(eh, CURLOPT_WRITEFUNCTION, discard_cb); + curl_easy_setopt(eh, CURLOPT_NOSIGNAL, 1L); + curl_easy_setopt(eh, CURLOPT_TIMEOUT, 10L); + curl_easy_setopt(eh, CURLOPT_CONNECTTIMEOUT, 10L); + curl_easy_setopt(eh, CURLOPT_ERRORBUFFER, s->errbuf); + curl_easy_setopt(eh, CURLOPT_PRIVATE, s); + /* No redirects to match the other clients' defaults. */ + curl_easy_setopt(eh, CURLOPT_FOLLOWLOCATION, 0L); +} + +int main(int argc, char **argv) { + if (argc != 3) { + fprintf(stderr, "usage: %s \n", argv[0]); + return 2; + } + const char *urls_path = argv[1]; + int workers = atoi(argv[2]); + if (workers <= 0) { + fprintf(stderr, "invalid workers: %s\n", argv[2]); + return 2; + } + + /* Slurp URLs into memory — benchmark driver writes small files. */ + FILE *f = fopen(urls_path, "r"); + if (!f) { perror("fopen"); return 1; } + size_t cap = 1024, n = 0; + char **urls = malloc(cap * sizeof(char *)); + char line[4096]; + while (fgets(line, sizeof(line), f)) { + size_t len = strlen(line); + while (len && (line[len-1] == '\n' || line[len-1] == '\r')) line[--len] = 0; + if (!len || line[0] == '#') continue; + if (n == cap) { cap *= 2; urls = realloc(urls, cap * sizeof(char *)); } + urls[n++] = strdup(line); + } + fclose(f); + + curl_global_init(CURL_GLOBAL_DEFAULT); + CURLM *mh = curl_multi_init(); + /* Size the connection cache to match concurrency so every worker can hold + * its own keepalive connection. Default is 10, which would serialize us. */ + curl_multi_setopt(mh, CURLMOPT_MAXCONNECTS, (long)workers); + + struct slot *slots = calloc((size_t)workers, sizeof(struct slot)); + size_t next_url = 0; + long in_flight = 0; + + int initial = workers < (int)n ? workers : (int)n; + for (int i = 0; i < initial; i++) { + slots[i].eh = curl_easy_init(); + configure(slots[i].eh, &slots[i], urls[next_url++]); + curl_multi_add_handle(mh, slots[i].eh); + in_flight++; + } + + char escaped[8192]; + int still_running = 0; + do { + curl_multi_perform(mh, &still_running); + + int msgs_left = 0; + CURLMsg *msg; + while ((msg = curl_multi_info_read(mh, &msgs_left))) { + if (msg->msg != CURLMSG_DONE) continue; + + CURL *eh = msg->easy_handle; + struct slot *s = NULL; + curl_easy_getinfo(eh, CURLINFO_PRIVATE, &s); + + if (msg->data.result == CURLE_OK) { + long status = 0; + curl_easy_getinfo(eh, CURLINFO_RESPONSE_CODE, &status); + json_escape(escaped, sizeof(escaped), s->url); + printf("{\"url\":\"%s\",\"status\":%ld}\n", escaped, status); + } else { + const char *em = s->errbuf[0] ? s->errbuf + : curl_easy_strerror(msg->data.result); + char url_esc[4096], err_esc[4096]; + json_escape(url_esc, sizeof(url_esc), s->url); + json_escape(err_esc, sizeof(err_esc), em); + printf("{\"url\":\"%s\",\"error\":\"%s\"}\n", url_esc, err_esc); + } + + curl_multi_remove_handle(mh, eh); + if (next_url < n) { + /* Refill this slot with the next URL. Reset before reuse so + * per-request state (err buffer, response code) is clean. */ + curl_easy_reset(eh); + configure(eh, s, urls[next_url++]); + curl_multi_add_handle(mh, eh); + } else { + curl_easy_cleanup(eh); + s->eh = NULL; + in_flight--; + } + } + + if (still_running || in_flight > 0) { + /* Block until there's something to do. */ + int numfds = 0; + curl_multi_poll(mh, NULL, 0, 1000, &numfds); + } + } while (still_running > 0 || in_flight > 0); + + for (int i = 0; i < workers; i++) { + if (slots[i].eh) { + curl_multi_remove_handle(mh, slots[i].eh); + curl_easy_cleanup(slots[i].eh); + } + } + free(slots); + for (size_t i = 0; i < n; i++) free(urls[i]); + free(urls); + curl_multi_cleanup(mh); + curl_global_cleanup(); + return 0; +} diff --git a/scripts/bench/client.go b/scripts/bench/client.go new file mode 100644 index 0000000..ac73eff --- /dev/null +++ b/scripts/bench/client.go @@ -0,0 +1,117 @@ +// HTTP benchmark client using Go's net/http stdlib. +// Worker-pool pattern: W goroutines pulling URLs from a channel, shared +// http.Client with connection pool sized to W so each worker can hold its +// own idle connection between requests. +// +// Usage: client +// Output: one JSON line per completed request to stdout: +// {"url":"...","status":200} or {"url":"...","error":"..."} +package main + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "os" + "runtime" + "strconv" + "strings" + "sync" + "time" +) + +type result struct { + URL string `json:"url"` + Status int `json:"status,omitempty"` + Error string `json:"error,omitempty"` +} + +func main() { + if len(os.Args) != 3 { + fmt.Fprintf(os.Stderr, "usage: %s \n", os.Args[0]) + os.Exit(2) + } + urlsPath := os.Args[1] + workers, err := strconv.Atoi(os.Args[2]) + if err != nil || workers <= 0 { + fmt.Fprintf(os.Stderr, "invalid workers: %v\n", os.Args[2]) + os.Exit(2) + } + + runtime.GOMAXPROCS(runtime.NumCPU()) + + f, err := os.Open(urlsPath) + if err != nil { + fmt.Fprintf(os.Stderr, "open urls: %v\n", err) + os.Exit(1) + } + defer f.Close() + + transport := &http.Transport{ + MaxIdleConns: workers, + MaxIdleConnsPerHost: workers, + MaxConnsPerHost: workers, + IdleConnTimeout: 60 * time.Second, + DialContext: (&net.Dialer{ + Timeout: 10 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + } + client := &http.Client{ + Transport: transport, + Timeout: 10 * time.Second, + } + + urlCh := make(chan string, workers*2) + resultCh := make(chan result, workers*2) + + var workerWg sync.WaitGroup + for i := 0; i < workers; i++ { + workerWg.Add(1) + go func() { + defer workerWg.Done() + for url := range urlCh { + resp, err := client.Get(url) + if err != nil { + resultCh <- result{URL: url, Error: err.Error()} + continue + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + resultCh <- result{URL: url, Status: resp.StatusCode} + } + }() + } + + // Writer goroutine: serializes JSON output to stdout. + var writerWg sync.WaitGroup + writerWg.Add(1) + go func() { + defer writerWg.Done() + out := bufio.NewWriter(os.Stdout) + defer out.Flush() + enc := json.NewEncoder(out) + for r := range resultCh { + enc.Encode(&r) + } + }() + + // Feed URLs. + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 64*1024), 1024*1024) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + urlCh <- line + } + close(urlCh) + + workerWg.Wait() + close(resultCh) + writerWg.Wait() +} diff --git a/scripts/bench/server.go b/scripts/bench/server.go new file mode 100644 index 0000000..db8368e --- /dev/null +++ b/scripts/bench/server.go @@ -0,0 +1,33 @@ +// Minimal HTTP benchmark target server. +// Serves a fixed ~1KB body as fast as possible on 127.0.0.1:8080. +package main + +import ( + "flag" + "log" + "net/http" + "runtime" + "strings" +) + +func main() { + addr := flag.String("addr", "127.0.0.1:8080", "listen address") + flag.Parse() + + runtime.GOMAXPROCS(runtime.NumCPU()) + + body := []byte(strings.Repeat("A", 1024)) + + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain") + w.Header().Set("Content-Length", "1024") + w.Write(body) + }) + + log.Printf("bench server listening on http://%s/", *addr) + srv := &http.Server{Addr: *addr, Handler: mux} + if err := srv.ListenAndServe(); err != nil { + log.Fatal(err) + } +} diff --git a/scripts/benchmark.py b/scripts/benchmark.py new file mode 100644 index 0000000..4c41aae --- /dev/null +++ b/scripts/benchmark.py @@ -0,0 +1,462 @@ +#!/usr/bin/env python3 +""" +Benchmark comparing blasthttp against Go / C / Python HTTP clients. + +Drives a workload of N requests at fixed worker count W against a local +HTTP server, mirroring the style of blastdns/scripts/benchmark.py. + +Every implementation runs against the same workload and the same +effective concurrency cap: + - blasthttp-cli : target/release/blasthttp -l urls.txt -c W + - blasthttp-python : BlastHTTP().request_batch(configs, concurrency=W) + - python-httpx : W asyncio tasks pulling from asyncio.Queue + - go-stdlib : W goroutines pulling from a channel (subprocess) + - c-libcurl : W libcurl easy handles on a multi handle (subprocess) + +Requires: + - target/release/blasthttp (cargo build --release) + - target/bench/{server,client-go,client-c} (make -C scripts/bench) + - Python deps: httpx, tabulate, uvloop, and the blasthttp module +""" + +import argparse +import asyncio +import contextlib +import json +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +import httpx +import uvloop +from tabulate import tabulate + +from blasthttp import BatchConfig, BlastHTTP + + +REPO_ROOT = Path(__file__).resolve().parent.parent +BENCH_DIR = REPO_ROOT / "target" / "bench" +BLASTHTTP_CLI = REPO_ROOT / "target" / "release" / "blasthttp" +BENCH_SERVER = BENCH_DIR / "server" +BENCH_GO_CLIENT = BENCH_DIR / "client-go" +BENCH_C_CLIENT = BENCH_DIR / "client-c" + + +# ============================================================================= +# Bundled Go server lifecycle +# ============================================================================= + + +def _wait_for_port(host, port, timeout=5.0): + deadline = time.perf_counter() + timeout + while time.perf_counter() < deadline: + with contextlib.suppress(OSError): + with socket.create_connection((host, port), timeout=0.25): + return True + time.sleep(0.05) + return False + + +@contextlib.contextmanager +def local_server(addr): + """Start the bundled Go server on `addr` (host:port); stop it on exit.""" + if not BENCH_SERVER.exists(): + raise RuntimeError( + f"Server binary not found at {BENCH_SERVER}. " + f"Run: make -C {BENCH_DIR.relative_to(REPO_ROOT).parent}/bench" + ) + host, port = addr.split(":") + port = int(port) + proc = subprocess.Popen( + [str(BENCH_SERVER), "-addr", addr], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + if not _wait_for_port(host, port, timeout=5.0): + proc.terminate() + raise RuntimeError(f"Server never became ready on {addr}") + yield f"http://{addr}/" + finally: + proc.terminate() + try: + proc.wait(timeout=5.0) + except subprocess.TimeoutExpired: + proc.kill() + + +# ============================================================================= +# Subprocess helper +# ============================================================================= + + +def _count_json_results(stdout): + """Count success/failure lines in a JSON-per-line stdout blob. + + Success lines have a numeric "status"; failure lines have "error". + """ + success = 0 + errors = 0 + for line in stdout.splitlines(): + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if "error" in obj: + errors += 1 + else: + success += 1 + return success, errors + + +def _run_subprocess_benchmark(binary, urls, workers): + """Write URLs to a temp file, run `binary `, parse JSON.""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".txt", delete=False, dir=REPO_ROOT + ) as urls_file: + urls_file.write("\n".join(urls)) + urls_file.write("\n") + urls_path = urls_file.name + + try: + start = time.perf_counter() + result = subprocess.run( + [str(binary), urls_path, str(workers)], + capture_output=True, + text=True, + ) + total_time = time.perf_counter() - start + + if result.returncode != 0: + raise RuntimeError( + f"{binary} exited {result.returncode}: {result.stderr[-500:]}" + ) + success, errors = _count_json_results(result.stdout) + qps = len(urls) / total_time if total_time > 0 else 0 + return total_time, qps, success, errors + finally: + Path(urls_path).unlink(missing_ok=True) + + +# ============================================================================= +# blasthttp CLI +# ============================================================================= + + +def benchmark_blasthttp_cli(urls, workers, rate_limit=None): + if not BLASTHTTP_CLI.exists(): + raise RuntimeError(f"{BLASTHTTP_CLI} missing (cargo build --release)") + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".txt", delete=False, dir=REPO_ROOT + ) as urls_file: + urls_file.write("\n".join(urls)) + urls_file.write("\n") + urls_path = urls_file.name + + try: + argv = [str(BLASTHTTP_CLI), "-l", urls_path, "-c", str(workers)] + if rate_limit is not None: + argv += ["--rate-limit", str(rate_limit)] + + start = time.perf_counter() + result = subprocess.run(argv, capture_output=True, text=True) + total_time = time.perf_counter() - start + + if result.returncode != 0: + raise RuntimeError(f"blasthttp failed: {result.stderr[-500:]}") + + # blasthttp emits rich JSON per response — collapse to success/error. + success = errors = 0 + for line in result.stdout.splitlines(): + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if "error" in obj: + errors += 1 + else: + success += 1 + qps = len(urls) / total_time if total_time > 0 else 0 + return total_time, qps, success, errors + finally: + Path(urls_path).unlink(missing_ok=True) + + +# ============================================================================= +# blasthttp Python +# ============================================================================= + + +async def benchmark_blasthttp_python(urls, workers, rate_limit=None): + client = BlastHTTP() + configs = [BatchConfig(url) for url in urls] + + start = time.perf_counter() + results = await client.request_batch( + configs, concurrency=workers, rate_limit=rate_limit + ) + total_time = time.perf_counter() - start + + success = sum(1 for r in results if r.success) + errors = len(results) - success + qps = len(urls) / total_time if total_time > 0 else 0 + return total_time, qps, success, errors + + +async def benchmark_blasthttp_python_stream(urls, workers, rate_limit=None): + client = BlastHTTP() + configs = [BatchConfig(url) for url in urls] + + success = 0 + errors = 0 + start = time.perf_counter() + async for batch in client.request_batch_stream( + configs, concurrency=workers, rate_limit=rate_limit + ): + for r in batch: + if r.success: + success += 1 + else: + errors += 1 + total_time = time.perf_counter() - start + + qps = len(urls) / total_time if total_time > 0 else 0 + return total_time, qps, success, errors + + +# ============================================================================= +# httpx +# ============================================================================= + + +async def _httpx_worker(client, queue, counts): + while True: + url = await queue.get() + if url is None: + queue.task_done() + break + try: + resp = await client.get(url) + await resp.aread() + if 200 <= resp.status_code < 400: + counts[0] += 1 + else: + counts[1] += 1 + except Exception: + counts[1] += 1 + queue.task_done() + + +async def benchmark_httpx(urls, workers): + limits = httpx.Limits( + max_connections=workers, + max_keepalive_connections=workers, + ) + timeout = httpx.Timeout(10.0) + async with httpx.AsyncClient(limits=limits, timeout=timeout) as client: + queue = asyncio.Queue(maxsize=workers * 2) + counts = [0, 0] # [success, errors] + + start = time.perf_counter() + worker_tasks = [ + asyncio.create_task(_httpx_worker(client, queue, counts)) + for _ in range(workers) + ] + + for url in urls: + await queue.put(url) + for _ in range(workers): + await queue.put(None) + + await queue.join() + await asyncio.gather(*worker_tasks) + + total_time = time.perf_counter() - start + qps = len(urls) / total_time if total_time > 0 else 0 + return total_time, qps, counts[0], counts[1] + + +# ============================================================================= +# Go & C subprocess wrappers +# ============================================================================= + + +def benchmark_go_stdlib(urls, workers): + if not BENCH_GO_CLIENT.exists(): + raise RuntimeError(f"{BENCH_GO_CLIENT} missing (make -C scripts/bench)") + return _run_subprocess_benchmark(BENCH_GO_CLIENT, urls, workers) + + +def benchmark_c_libcurl(urls, workers): + if not BENCH_C_CLIENT.exists(): + raise RuntimeError(f"{BENCH_C_CLIENT} missing (make -C scripts/bench)") + return _run_subprocess_benchmark(BENCH_C_CLIENT, urls, workers) + + +# ============================================================================= +# Output +# ============================================================================= + + +def print_table(results, baseline): + baseline_qps = results.get(baseline, (0, 1, 0, 0))[1] if baseline in results else 0 + + rows = [] + for name, (total_time, qps, success, errors) in sorted( + results.items(), key=lambda x: -x[1][1] + ): + multiplier = (qps / baseline_qps) if baseline_qps > 0 else 0 + rows.append( + [ + name, + f"{total_time:.3f}s", + f"{qps:,.0f}", + f"{success:,}", + f"{errors:,}", + f"{multiplier:.2f}x" if baseline_qps > 0 else "-", + ] + ) + + headers = ["Library", "Time", "QPS", "Success", "Failed", f"vs {baseline}"] + print(tabulate(rows, headers=headers, tablefmt="github")) + + +def generate_urls(num, target, pattern): + """Build a URL list. Default pattern = same URL N times (maximum reuse).""" + if pattern == "same": + return [target] * num + elif pattern == "unique": + # Each request hits / + sep = "" if target.endswith("/") else "/" + return [f"{target}{sep}{i}" for i in range(num)] + else: + raise ValueError(f"unknown URL pattern: {pattern}") + + +# ============================================================================= +# Main +# ============================================================================= + +ENGINES = [ + "blasthttp-cli", + "blasthttp-cli-200k", + "blasthttp-python", + "blasthttp-python-200k", + "blasthttp-python-stream", + "blasthttp-python-stream-200k", + "httpx", + "go", + "c", +] + +RATE_LIMITED_RPS = 200_000 + + +async def main(): + parser = argparse.ArgumentParser(description="Benchmark blasthttp vs Go/C/Python HTTP clients") + parser.add_argument("-n", "--num-queries", type=int, default=20_000, help="Number of requests") + parser.add_argument("-w", "--num-workers", type=int, default=100, help="Concurrent workers") + parser.add_argument( + "--target", + default=None, + help="Target URL. If unset, a bundled Go server is started on 127.0.0.1:8080.", + ) + parser.add_argument( + "--pattern", + choices=["same", "unique"], + default="same", + help="URL pattern: 'same' URL N times, or 'unique' paths /0../N-1", + ) + parser.add_argument("--only", choices=ENGINES, help="Run only one engine") + parser.add_argument( + "--baseline", + choices=ENGINES, + default="httpx", + help="Baseline engine for the multiplier column", + ) + args = parser.parse_args() + + # Set up target (bundled server unless user supplied --target). + server_ctx = contextlib.nullcontext(args.target) + if args.target is None: + server_ctx = local_server("127.0.0.1:8080") + + with server_ctx as target: + urls = generate_urls(args.num_queries, target, args.pattern) + + print("## HTTP Client Benchmark") + print() + print(f"- **Requests:** {args.num_queries:,}") + print(f"- **Workers:** {args.num_workers}") + print(f"- **Target:** {target}") + print(f"- **URL pattern:** {args.pattern}") + print() + + results = {} + + def run(name): + return args.only is None or args.only == name + + if run("blasthttp-cli"): + print("Running blasthttp-cli...", file=sys.stderr, flush=True) + results["blasthttp-cli"] = benchmark_blasthttp_cli(urls, args.num_workers) + + if run("blasthttp-cli-200k"): + print("Running blasthttp-cli-200k...", file=sys.stderr, flush=True) + results["blasthttp-cli-200k"] = benchmark_blasthttp_cli( + urls, args.num_workers, rate_limit=RATE_LIMITED_RPS + ) + + if run("blasthttp-python"): + print("Running blasthttp-python...", file=sys.stderr, flush=True) + results["blasthttp-python"] = await benchmark_blasthttp_python( + urls, args.num_workers + ) + + if run("blasthttp-python-200k"): + print("Running blasthttp-python-200k...", file=sys.stderr, flush=True) + results["blasthttp-python-200k"] = await benchmark_blasthttp_python( + urls, args.num_workers, rate_limit=RATE_LIMITED_RPS + ) + + if run("blasthttp-python-stream"): + print("Running blasthttp-python-stream...", file=sys.stderr, flush=True) + results["blasthttp-python-stream"] = await benchmark_blasthttp_python_stream( + urls, args.num_workers + ) + + if run("blasthttp-python-stream-200k"): + print("Running blasthttp-python-stream-200k...", file=sys.stderr, flush=True) + results["blasthttp-python-stream-200k"] = await benchmark_blasthttp_python_stream( + urls, args.num_workers, rate_limit=RATE_LIMITED_RPS + ) + + if run("httpx"): + print("Running httpx...", file=sys.stderr, flush=True) + results["httpx"] = await benchmark_httpx(urls, args.num_workers) + + if run("go"): + print("Running go-stdlib...", file=sys.stderr, flush=True) + results["go"] = benchmark_go_stdlib(urls, args.num_workers) + + if run("c"): + print("Running c-libcurl...", file=sys.stderr, flush=True) + results["c"] = benchmark_c_libcurl(urls, args.num_workers) + + print() + print("### Results") + print() + print_table(results, baseline=args.baseline) + + +if __name__ == "__main__": + uvloop.install() + asyncio.run(main()) diff --git a/src/batch.rs b/src/batch.rs index c59ea2d..25a0b55 100644 --- a/src/batch.rs +++ b/src/batch.rs @@ -2,6 +2,8 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; +use futures::stream::{self, Stream, StreamExt}; + use crate::client::{ClientError, HttpClient}; use crate::config::RequestConfig; use crate::response::Response; @@ -109,23 +111,20 @@ impl RateLimiter { // ── Batch dispatch ──────────────────────────────────────────────── -pub async fn send_batch( - client: Arc, - configs: Vec, - concurrency: usize, - rate_limit: Option, - shared_limiter: Option>, -) -> Vec { - let semaphore = Arc::new(tokio::sync::Semaphore::new(concurrency)); - // When both a client-level and per-call rate limit are set, use the more - // restrictive (lower RPS) of the two. This lets modules enforce a tighter - // rate than the global without overriding the global for other callers. - let limiter = match (shared_limiter, rate_limit) { +/// Pick the effective rate limiter for a batch call. +/// +/// When both a client-level and per-call rate limit are set, use the more +/// restrictive (lower RPS) of the two. This lets modules enforce a tighter +/// rate than the global without overriding the global for other callers. +fn merge_limiters( + shared: Option>, + per_call: Option, +) -> Option> { + match (shared, per_call) { (Some(shared), Some(per_call_rps)) => { let shared_interval = shared.interval(); let per_call_interval = Duration::from_secs_f64(1.0 / per_call_rps); if per_call_interval > shared_interval { - // Per-call limit is more restrictive (slower), use it Some(Arc::new(RateLimiter::new(per_call_rps))) } else { Some(shared) @@ -134,7 +133,18 @@ pub async fn send_batch( (Some(shared), None) => Some(shared), (None, Some(rps)) => Some(Arc::new(RateLimiter::new(rps))), (None, None) => None, - }; + } +} + +pub async fn send_batch( + client: Arc, + configs: Vec, + concurrency: usize, + rate_limit: Option, + shared_limiter: Option>, +) -> Vec { + let semaphore = Arc::new(tokio::sync::Semaphore::new(concurrency)); + let limiter = merge_limiters(shared_limiter, rate_limit); let mut handles = Vec::new(); for config in configs { @@ -168,6 +178,77 @@ pub async fn send_batch( results } +/// Streaming variant of `send_batch`. Yields `BatchResult`s in completion +/// order (out-of-dispatch-order) as each request finishes, so a slow request +/// doesn't block faster peers that follow it in the input list. +/// +/// Architecture: a driver task spawns one tokio task per request and pipes +/// completed `BatchResult`s into an unbounded mpsc channel; the returned +/// stream is the receiver end. Each request runs as its own spawned task so +/// HTTP work keeps progressing while the consumer (e.g. Python) is busy +/// iterating a returned batch — the same in-flight model as `send_batch`. +/// +/// `buffer_unordered` is intentionally avoided: its inner futures only make +/// progress while the stream is being polled. While Python iterates a +/// 1000-item batch, no one polls the stream, so 100 in-flight HTTP futures +/// would stall — measured at ~3.7× throughput regression. blastdns gets +/// away with `buffer_unordered` because its actual work runs on persistent +/// worker tasks queued via crossfire; the stream just multiplexes oneshot +/// waits. blasthttp has no such workers, so we spawn per request. +/// +/// Spawning has to happen on the runtime, but `send_batch_stream` is called +/// from a synchronous PyO3 constructor that isn't itself on a tokio task. +/// `stream::once(async { ... }).flatten()` defers the driver-spawn into the +/// stream's first poll, which happens inside `PyBatchResultIterator`'s +/// `__anext__` (a `future_into_py` block running on the tokio runtime). +/// +/// Concurrency is gated *before* spawn by a semaphore acquire on the driver, +/// so at most `concurrency` requests are in-flight at any time. Rate-limit +/// acquire happens before the semaphore so dispatch pacing matches +/// `send_batch`. In-flight tasks are NOT cancelled if the consumer drops +/// the stream — they run to completion and their sends fail silently. This +/// also matches `send_batch`. +pub fn send_batch_stream( + client: Arc, + configs: Vec, + concurrency: usize, + rate_limit: Option, + shared_limiter: Option>, +) -> impl Stream + Send + 'static { + let limiter = merge_limiters(shared_limiter, rate_limit); + let semaphore = Arc::new(tokio::sync::Semaphore::new(concurrency)); + + stream::once(async move { + let (tx, rx) = futures::channel::mpsc::unbounded::(); + + tokio::spawn(async move { + for config in configs { + if let Some(ref l) = limiter { + l.acquire().await; + } + let permit = match semaphore.clone().acquire_owned().await { + Ok(p) => p, + Err(_) => break, + }; + let client = client.clone(); + let tx = tx.clone(); + tokio::spawn(async move { + let _permit = permit; + let url = config.url.clone(); + let result = client.send(&config).await; + let _ = tx.unbounded_send(BatchResult { url, result }); + }); + } + // Driver's `tx` clone drops here. Channel closes once every + // per-request task's `tx` clone also drops (i.e. all sends + // done), signaling stream end to the consumer. + }); + + rx + }) + .flatten() +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/main.rs b/src/main.rs index 5a302f3..a2e85a9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -32,6 +32,10 @@ struct Cli { #[arg(short = 'c', long, default_value = "50")] concurrency: usize, + /// Rate limit in requests per second (batch mode; unlimited by default) + #[arg(long = "rate-limit")] + rate_limit: Option, + /// Enable TLS certificate validation (off by default) #[arg(long)] verify: bool, @@ -179,7 +183,7 @@ async fn run_batch(cli: &Cli, file_path: &str) { .collect(); let client = Arc::new(HyperClient::new()); - let results = batch::send_batch(client, configs, cli.concurrency, None, None).await; + let results = batch::send_batch(client, configs, cli.concurrency, cli.rate_limit, None).await; for r in results { match r.result { diff --git a/src/python.rs b/src/python.rs index f453ca8..ad02c37 100644 --- a/src/python.rs +++ b/src/python.rs @@ -3,12 +3,17 @@ // Kept separate from Rust structs so the Python API can diverge freely // (e.g. complex request builders for Phase 4 raw byte control). -use pyo3::exceptions::PyRuntimeError; +use futures::stream::{Stream, StreamExt}; +use pyo3::exceptions::{PyRuntimeError, PyStopAsyncIteration}; use pyo3::prelude::*; use pyo3_async_runtimes::tokio::future_into_py; +use std::pin::Pin; use std::sync::Arc; +use std::time::Duration; +use tokio::sync::Mutex as TokioMutex; +use tokio::time::Instant; -use crate::batch::{self, RateLimiter}; +use crate::batch::{self, BatchResult, RateLimiter}; use crate::client::HttpClient; use crate::client::hyper::HyperClient; use crate::client::raw; @@ -257,6 +262,86 @@ impl PyBatchResult { } } +fn to_py_batch_result(r: BatchResult) -> PyBatchResult { + let (response, error) = match r.result { + Ok(resp) => (Some(resp), None), + Err(e) => (None, Some(e.message)), + }; + PyBatchResult { + url: r.url, + response, + error, + } +} + +// ── Streaming batch iterator ────────────────────────────────────── + +/// Async iterator exposed to Python for `request_batch_stream`. Each +/// `__anext__` drains the underlying stream into a batch (up to 1000 +/// items or 200ms — whichever comes first) and returns the batch as a +/// `list[BatchResult]`. Callers iterate with: +/// +/// async for batch in client.request_batch_stream(configs): +/// for r in batch: +/// ... +/// +/// Two reasons for batching at this boundary: +/// 1. Throughput. Each `__anext__` is a full Python↔Rust round-trip +/// (`future_into_py`, GIL release/reacquire, asyncio scheduling). +/// At 100k+ QPS, paying that per result caps us roughly an order +/// of magnitude below non-streaming. Batching ~1000 amortizes it. +/// 2. Streaming latency. The 200ms timeout is the actual streaming +/// property: even when results trickle in slowly, partial batches +/// flush after 200ms so the consumer is never starved. +/// +/// Delicate bits (mirrors blastdns's PyBatchIterator — changing these +/// can deadlock Python's event loop or leak tasks): +/// • TokioMutex, not std::sync::Mutex: the guard crosses .await. +/// • future_into_py releases the GIL while polling; do NOT +/// Python::attach inside the loop. +/// • PyStopAsyncIteration is only raised when a NEW __anext__ call +/// finds both the stream empty AND the batch empty. If the stream +/// ends mid-batch, return what we have and let the next call raise. +#[pyclass(name = "BatchResultIterator")] +pub struct PyBatchResultIterator { + inner: Arc + Send>>>>, +} + +#[pymethods] +impl PyBatchResultIterator { + fn __aiter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> { + slf + } + + fn __anext__<'py>(&self, py: Python<'py>) -> PyResult> { + let inner = Arc::clone(&self.inner); + + future_into_py(py, async move { + let mut stream = inner.lock().await; + let mut batch: Vec = Vec::new(); + let start = Instant::now(); + let timeout = Duration::from_millis(200); + + loop { + if batch.len() >= 1000 || (!batch.is_empty() && start.elapsed() >= timeout) { + return Ok(batch); + } + + match stream.next().await { + Some(r) => batch.push(to_py_batch_result(r)), + None => { + if batch.is_empty() { + return Err(PyStopAsyncIteration::new_err("end of stream")); + } else { + return Ok(batch); + } + } + } + } + }) + } +} + // ── Main client class ───────────────────────────────────────────── #[pyclass] @@ -401,25 +486,50 @@ impl BlastHTTP { ) .await; - let py_results: Vec = results - .into_iter() - .map(|r| { - let (response, error) = match r.result { - Ok(resp) => (Some(resp), None), - Err(e) => (None, Some(e.message)), - }; - PyBatchResult { - url: r.url, - response, - error, - } - }) - .collect(); + let py_results: Vec = + results.into_iter().map(to_py_batch_result).collect(); Ok(py_results) }) } + /// Streaming variant of request_batch. Returns an async iterator that + /// yields `list[BatchResult]` chunks as requests complete, in + /// completion order — a slow request doesn't block faster peers + /// behind it. Each chunk holds up to 1000 results or 200ms worth, + /// whichever fills first; partial chunks flush on the timeout so the + /// consumer is never starved when results trickle in. + /// + /// Iterate as: + /// + /// async for batch in client.request_batch_stream(configs): + /// for r in batch: + /// ... + #[pyo3(signature = (configs, concurrency=50, rate_limit=None))] + fn request_batch_stream( + &self, + configs: Vec, + concurrency: usize, + rate_limit: Option, + ) -> PyResult { + let request_configs: Vec = configs + .into_iter() + .map(|c| c.into_request_config()) + .collect(); + + let stream = batch::send_batch_stream( + self.client.clone(), + request_configs, + concurrency, + rate_limit, + self.rate_limiter.clone(), + ); + + Ok(PyBatchResultIterator { + inner: Arc::new(TokioMutex::new(Box::pin(stream))), + }) + } + /// Download a URL directly to a local file. /// Returns the file path on success. /// max_size: maximum bytes to download (None = no limit, uses default 10MB) @@ -1255,6 +1365,7 @@ fn blasthttp(m: &Bound<'_, PyModule>) -> PyResult<()> { // so Python can reference them for type hints / isinstance checks m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/tests/python/test_request_batch_stream.py b/tests/python/test_request_batch_stream.py new file mode 100644 index 0000000..3834a71 --- /dev/null +++ b/tests/python/test_request_batch_stream.py @@ -0,0 +1,147 @@ +"""Pytest-asyncio tests for `BlastHTTP.request_batch_stream`. + +Exercises the streaming batch iterator (`PyBatchResultIterator.__anext__`) +against a local asyncio HTTP server that delays responses based on the URL +path so we can drive completion timing and verify the 200ms timeout flushes +partial batches. +""" +import asyncio +import re + +import blasthttp +import pytest_asyncio + + +@pytest_asyncio.fixture +async def delay_server(): + """Minimal HTTP/1.1 server on an ephemeral port. + + Responds 200 OK after sleeping `ms` milliseconds, where `ms` is parsed + from a `GET /delay/` request line. Sends `Connection: close` so we + don't have to deal with keep-alive bookkeeping. + + Yields the port; tears down on cleanup. + """ + async def handle(reader, writer): + try: + data = b"" + while b"\r\n\r\n" not in data: + chunk = await reader.read(4096) + if not chunk: + return + data += chunk + request_line = data.split(b"\r\n", 1)[0].decode("ascii", "replace") + m = re.match(r"[A-Z]+ /delay/(\d+) ", request_line) + ms = int(m.group(1)) if m else 0 + if ms > 0: + await asyncio.sleep(ms / 1000) + body = b"ok" + response = ( + b"HTTP/1.1 200 OK\r\n" + b"Content-Length: " + str(len(body)).encode() + b"\r\n" + b"Connection: close\r\n" + b"\r\n" + body + ) + writer.write(response) + await writer.drain() + finally: + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + + server = await asyncio.start_server(handle, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + try: + yield port + finally: + server.close() + await server.wait_closed() + + +async def test_stream_yields_batches_of_batchresults(delay_server): + """Smoke test: each yielded item is a `list[BatchResult]` and totals match.""" + port = delay_server + client = blasthttp.BlastHTTP() + configs = [blasthttp.BatchConfig(f"http://127.0.0.1:{port}/delay/0") for _ in range(10)] + + total = 0 + async for batch in client.request_batch_stream(configs, concurrency=10): + assert isinstance(batch, list) + assert all(isinstance(r, blasthttp.BatchResult) for r in batch) + total += len(batch) + + assert total == 10 + + +async def test_stream_completes_in_completion_order(delay_server): + """A slow request doesn't block faster peers behind it in the input list.""" + port = delay_server + client = blasthttp.BlastHTTP() + # Slow first, then fast — completion order must invert dispatch order. + urls = [f"http://127.0.0.1:{port}/delay/300"] + [f"http://127.0.0.1:{port}/delay/0"] * 5 + configs = [blasthttp.BatchConfig(u) for u in urls] + + completed = [] + async for batch in client.request_batch_stream(configs, concurrency=6): + for r in batch: + completed.append(r.url) + + # The 5 fast requests must complete before the single slow one. + assert completed[-1].endswith("/delay/300"), f"got order: {completed}" + + +async def test_stream_timeout_flushes_partial_batches(delay_server): + """The 200ms `__anext__` timeout splits results into separate batches + when completions come in temporally distinct clusters. + + Setup: 60 requests in three completion clusters — 20 immediate, + 20 at ~300ms, 20 at ~600ms. All dispatched concurrently. + + Expected batch boundaries (mirrors blastdns's test_batch_timeout_triggers): + - __anext__ #1: 20 fast results pile up at ~T+5ms; loop awaits; + first slow result arrives at ~T+300ms; push (batch=21); elapsed + ≥ 200ms → flush → batch of 21. + - __anext__ #2 starts at ~T+300ms: 19 remaining cluster-B results + arrive immediately; loop awaits; first cluster-C result at + ~T+600ms; push (batch=20); elapsed ≥ 200ms → flush → batch of 20. + - __anext__ #3 starts at ~T+600ms: 19 remaining cluster-C results + arrive immediately; stream ends → batch of 19. + """ + port = delay_server + client = blasthttp.BlastHTTP() + + urls = ( + [f"http://127.0.0.1:{port}/delay/0"] * 20 + + [f"http://127.0.0.1:{port}/delay/300"] * 20 + + [f"http://127.0.0.1:{port}/delay/600"] * 20 + ) + configs = [blasthttp.BatchConfig(u) for u in urls] + + batch_sizes = [] + async for batch in client.request_batch_stream(configs, concurrency=60): + batch_sizes.append(len(batch)) + + assert sum(batch_sizes) == 60, f"expected 60 results, got {batch_sizes}" + assert batch_sizes == [21, 20, 19], ( + f"expected timeout to flush as [21, 20, 19], got {batch_sizes}" + ) + + +async def test_stream_no_timeout_under_load(delay_server): + """When results arrive faster than the 200ms timeout, the iterator + should drain into one batch (or hit the 1000-item ceiling), not + fragment uselessly. Verifies we don't have a pathological flush + every poll.""" + port = delay_server + client = blasthttp.BlastHTTP() + configs = [blasthttp.BatchConfig(f"http://127.0.0.1:{port}/delay/0") for _ in range(50)] + + batch_sizes = [] + async for batch in client.request_batch_stream(configs, concurrency=50): + batch_sizes.append(len(batch)) + + assert sum(batch_sizes) == 50 + # 50 fast results in well under 200ms — should land in a single batch. + assert len(batch_sizes) == 1, f"expected 1 batch, got sizes {batch_sizes}" diff --git a/uv.lock b/uv.lock deleted file mode 100644 index c130b6c..0000000 --- a/uv.lock +++ /dev/null @@ -1,8 +0,0 @@ -version = 1 -revision = 3 -requires-python = ">=3.9" - -[[package]] -name = "blasthttp" -version = "0.1.0" -source = { editable = "." }