From: Willy Tarreau Date: Fri, 2 Oct 2026 12:08:59 +0000 (+0200) Subject: DEV: sse-latency: add a latency measuring tool for SSE response streams X-Git-Tag: v3.5-dev9~215 X-Git-Url: https://git.kaiwu.me/?a=commitdiff_plain;h=345aa2b73e90bc6dc3a97c9160b1309463874d7e;p=haproxy.git DEV: sse-latency: add a latency measuring tool for SSE response streams Common HTTP load generation tools are not convenient to measure latency through haproxy when it comes to server-sent events, because the measurements do not apply to the partial contents that SSE are. These are particularly important for AI environments where agents need to minimize latency. This commit introduces a new tool that is both a client and a server, and that is meant to measure an intermediary's added latency. For this, it connects to itself to measure its own inflicted latency (loopback test) then does it again through the configured proxy, and reports both as well as the traversal time. The principle is to note the timestamp at which each SSE is sent, and measure the difference when received by the client, since both sides are the same process and use the same clock reference. The output shows the reference time (loopback), the gateway time, and the gateway overhead as the difference between the two. The tool supports adding more clients, doing more requests, delaying the first token response, and setting the inter-token latency. An example haproxy config is provided (gateway.cfg) that measures both with and without option http-no-delay. Without http-no-delay (+ ~100ms, max 200 due to Nagle): === loopback (direct to listen address): 1 requests, 1 ok, 0 failed, 200 tokens === TTFT (client) n=1 mean=0.04ms min=0.04ms p50=0.04ms p95=0.04ms p99=0.04ms max=0.04ms TTFT emit->recv(1st) n=1 mean=0.00ms min=0.00ms p50=0.00ms p95=0.00ms p99=0.00ms max=0.00ms inter-token (client) n=199 mean=5.00ms min=4.07ms p50=5.13ms p95=5.33ms p99=5.36ms max=5.38ms emit->recv (tokens) n=199 mean=0.04ms min=0.01ms p50=0.03ms p95=0.07ms p99=0.18ms max=0.19ms === gateway: 1 requests, 1 ok, 0 failed, 1000 tokens === TTFT (client) n=1 mean=207.17ms min=207.17ms p50=207.17ms p95=207.17ms p99=207.17ms max=207.17ms TTFT emit->recv(1st) n=1 mean=206.86ms min=206.86ms p50=206.86ms p95=206.86ms p99=206.86ms max=206.86ms inter-token (client) n=999 mean=4.80ms min=0.00ms p50=0.00ms p95=0.00ms p99=207.98ms max=208.97ms emit->recv (tokens) n=999 mean=104.04ms min=1.02ms p50=103.88ms p95=197.67ms p99=205.89ms max=208.07ms === gateway overhead (gateway minus loopback; negative = loopback noise) === emit->recv (1st token) mean +206.851ms p50 +206.851ms p95 +206.851ms p99 +206.851ms emit->recv (tokens) mean +104.002ms p50 +103.851ms p95 +197.599ms p99 +205.709ms TTFT (client) mean +207.122ms p50 +207.122ms p95 +207.122ms p99 +207.122ms inter-token (client) mean -0.205ms p50 -5.128ms p95 -5.331ms p99 +202.624ms With http-no-delay (+ ~40 microseconds max): === loopback (direct to listen address): 1 requests, 1 ok, 0 failed, 200 tokens === TTFT (client) n=1 mean=0.05ms min=0.05ms p50=0.05ms p95=0.05ms p99=0.05ms max=0.05ms TTFT emit->recv(1st) n=1 mean=0.01ms min=0.01ms p50=0.01ms p95=0.01ms p99=0.01ms max=0.01ms inter-token (client) n=199 mean=5.00ms min=4.03ms p50=5.06ms p95=5.27ms p99=5.34ms max=5.59ms emit->recv (tokens) n=199 mean=0.04ms min=0.01ms p50=0.03ms p95=0.07ms p99=0.18ms max=0.49ms === gateway: 1 requests, 1 ok, 0 failed, 1000 tokens === TTFT (client) n=1 mean=0.18ms min=0.18ms p50=0.18ms p95=0.18ms p99=0.18ms max=0.18ms TTFT emit->recv(1st) n=1 mean=0.03ms min=0.03ms p50=0.03ms p95=0.03ms p99=0.03ms max=0.03ms inter-token (client) n=999 mean=5.00ms min=4.04ms p50=5.10ms p95=5.31ms p99=5.37ms max=5.45ms emit->recv (tokens) n=999 mean=0.07ms min=0.03ms p50=0.07ms p95=0.13ms p99=0.20ms max=0.28ms === gateway overhead (gateway minus loopback; negative = loopback noise) === emit->recv (1st token) mean +0.026ms p50 +0.026ms p95 +0.026ms p99 +0.026ms emit->recv (tokens) mean +0.034ms p50 +0.036ms p95 +0.059ms p99 +0.029ms TTFT (client) mean +0.135ms p50 +0.135ms p95 +0.135ms p99 +0.135ms inter-token (client) mean -0.002ms p50 +0.042ms p95 +0.041ms p99 +0.035ms --- diff --git a/Makefile b/Makefile index c20ed3005..3904277db 100644 --- a/Makefile +++ b/Makefile @@ -1068,7 +1068,8 @@ IGNORE_OPTS=help install install-man install-doc install-bin \ uninstall clean tags cscope tar git-tar version update-version \ opts reg-tests reg-tests-help unit-tests admin/halog/halog dev/flags/flags \ dev/haring/haring dev/ncpu/ncpu dev/poll/poll dev/tcploop/tcploop \ - dev/term_events/term_events dev/gdb/pm-from-core dev/gdb/libs-from-core + dev/term_events/term_events dev/gdb/pm-from-core dev/gdb/libs-from-core \ + dev/sse-latency/sse-latency ifneq ($(TARGET),) ifeq ($(filter $(firstword $(MAKECMDGOALS)),$(IGNORE_OPTS)),) @@ -1163,6 +1164,9 @@ dev/qpack/decode: dev/qpack/decode.o dev/tcploop/tcploop: $(cmd_MAKE) -C dev/tcploop tcploop CC='$(CC)' OPTIMIZE='$(COPTS)' V='$(V)' +dev/sse-latency/sse-latency: + $(cmd_MAKE) -C dev/sse-latency sse-latency CC='$(CC)' OPTIMIZE='$(COPTS)' V='$(V)' + dev/udp/udp-perturb: dev/udp/udp-perturb.o $(cmd_LD) $(ARCH_FLAGS) $(LDFLAGS) -o $@ $^ $(LDOPTS) @@ -1170,7 +1174,8 @@ dev/term_events/term_events: dev/term_events/term_events.o $(cmd_LD) $(ARCH_FLAGS) $(LDFLAGS) -o $@ $^ $(LDOPTS) # rebuild it every time -.PHONY: src/version.c dev/ncpu/ncpu dev/poll/poll dev/tcploop/tcploop +.PHONY: src/version.c dev/ncpu/ncpu dev/poll/poll dev/tcploop/tcploop \ + dev/sse-latency/sse-latency src/calltrace.o: src/calltrace.c $(DEP) $(cmd_CC) $(TRACE_COPTS) -c -o $@ $< @@ -1245,6 +1250,7 @@ distclean: clean $(Q)rm -f dev/haring/haring dev/ncpu/ncpu{,.so} dev/poll/poll dev/tcploop/tcploop $(Q)rm -f dev/hpack/decode dev/hpack/gen-enc dev/hpack/gen-rht $(Q)rm -f dev/qpack/decode dev/gdb/pm-from-core dev/gdb/libs-from-core + $(Q)rm -f dev/sse-latency/sse-latency tags: $(Q)find src include \( -name '*.c' -o -name '*.h' \) -print0 | \ diff --git a/dev/sse-latency/Makefile b/dev/sse-latency/Makefile new file mode 100644 index 000000000..01cc9fa29 --- /dev/null +++ b/dev/sse-latency/Makefile @@ -0,0 +1,13 @@ +include ../../include/make/verbose.mk + +CC = gcc +OPTIMIZE = -O2 -g +DEFINE = +INCLUDE = +OBJS = sse-latency + +sse-latency: sse-latency.c + $(cmd_CC) $(OPTIMIZE) $(DEFINE) $(INCLUDE) -o $@ $^ + +clean: + rm -f $(OBJS) *.[oas] *~ diff --git a/dev/sse-latency/README.txt b/dev/sse-latency/README.txt new file mode 100644 index 000000000..dbc597125 --- /dev/null +++ b/dev/sse-latency/README.txt @@ -0,0 +1,109 @@ +sse-latency - SSE latency measurement for AI gateways +===================================================== + +sse-latency is a single process that acts as BOTH sides of an OpenAI-like +SSE conversation, sharing one process-local clock so that it can measure the +exact delay each token suffers when going through a gateway (haproxy or any +other OpenAI-like proxy): + + SERVER side: listens on -L and exposes POST /v1/chat/completions (plus + /v1/models). It streams standard chat.completion.chunk SSE events over + HTTP/1.1 chunked encoding. Tokens are 4-letter words; defaults: 1000 tokens + per response, no delay before the first token (aka thinking), 5 ms between + tokens. The number of tokens and both delays are overridable per-request + via the JSON body ("pi_tokens", "pi_think_ms", "pi_delay_ms"), which real + gateways silently ignore. + + CLIENT side: sends standard streaming chat completion requests to -t (the + gateway under test), reads the SSE stream (handling chunked encoding when + the gateway uses it) and records: + - TTFT: time from request fully written to first content token received + - inter-token gaps: arrival time deltas between consecutive tokens + - emission delay: arrival_time - pi_ts of each chunk + +Since both sides share the same clock, every SSE chunk carries an extra +"pi_ts" field with the emission time in nanoseconds since the epoch +(CLOCK_REALTIME), taken at the moment the chunk is appended to the socket. +This makes it possible to measure the true one-way delay through the gateway +instead of just end-to-end timing. + +By default a loopback calibration batch runs first (client connects directly +to the listening address), then the gateway batch; the difference between +both is reported as the estimated gateway overhead. Negative values are +normal, they just mean the loopback run happened to be noisier. The +calibration is deliberately kept short (at most max(3, concurrency) +requests of at most 200 tokens each): it only measures the noise floor, it +doesn't need the full load. Long batches announce their expected duration +before starting, and a progress line is displayed once per second while a +batch is running (on terminals only). + +Usage +----- + +The default listening port is :8090, so that the gateway under test can +conveniently listen on the usual port 8080 and forward to this tool: + + ./sse-latency -L :8090 \ # where our dummy backend listens + -t 127.0.0.1:8080 \ # the gateway (haproxy frontend) + -r 20 -c 5 # 20 requests, 5 in parallel + -n 1000 # tokens (no think delay, 5ms/token by default) + +Options: + -L <[ip:]port> listen address of the dummy OpenAI-like server (def :8090) + -t <[ip:]port> gateway address; empty = server only + -r number of requests (def 1) + -c concurrency (def 1) + -n tokens per response (def 1000) + -T think delay before first token (def 0) + -d inter-token delay (def 5, float ok) + -w per-request timeout (def 300000) + -l disable the loopback calibration batch + -q quiet, no per-request lines + -v verbose + -h help + +Reported metrics +---------------- + TTFT (client) request fully sent -> first token received + TTFT emit->recv(1st) one-way delay of the first token + inter-token (client) gaps between consecutive token arrivals + emit->recv (tokens) one-way delay of every token + gateway overhead gateway batch minus loopback batch + +Interpretation notes: +--------------------- + - emit->recv constant and small (loopback: ~0.02 ms) -> the gateway + flushes SSE immediately. + - Bursts in inter-token (many 0 ms gaps + one large gap) -> the gateway + buffers chunks before flushing. + - TTFT (client) ~= think + emit->recv(1st) sanity-checks the run. + +haproxy results (3.5-dev8, loopback, 5x5 concurrency) +----------------------------------------------------- +Without any tuning, SSE responses are buffered in the kernel buffers (bursts of +~20 tokens every ~200 ms, ~100 ms mean per-token delay, first token up to +105 +ms): + + global + daemon + defaults + mode http + option http-no-delay + timeout client 5m + timeout server 5m + timeout connect 5s + frontend fe + bind 127.0.0.1:8080 + default_backend be + backend be + server sse-latency 127.0.0.1:8090 + +Adding "option http-no-delay" to the defaults section disables the payload +corking in the kernel and produces the lowest latency, as can be seen below +where this tool shows that haproxy adds roughly 0.04ms: + + === gateway overhead (gateway minus loopback; negative = loopback noise) === + emit->recv (1st token) mean +0.026ms p50 +0.026ms p99 +0.026ms + emit->recv (tokens) mean +0.034ms p50 +0.036ms p99 +0.029ms + TTFT (client) mean +0.135ms p50 +0.135ms p99 +0.135ms + inter-token (client) mean -0.002ms p50 +0.042ms p99 +0.035ms diff --git a/dev/sse-latency/gateway.cfg b/dev/sse-latency/gateway.cfg new file mode 100644 index 000000000..cc57d77e8 --- /dev/null +++ b/dev/sse-latency/gateway.cfg @@ -0,0 +1,14 @@ +defaults + mode http + timeout connect 5s + timeout client 30s + timeout server 30s + option forwardfor + # option http-no-delay + +frontend ai-front + bind *:8080 + default_backend ai-back + +backend ai-back + server sse-local 127.0.0.1:8090 diff --git a/dev/sse-latency/sse-latency.c b/dev/sse-latency/sse-latency.c new file mode 100644 index 000000000..9e760a280 --- /dev/null +++ b/dev/sse-latency/sse-latency.c @@ -0,0 +1,1604 @@ +/* + * sse-latency.c - SSE benchmark tool for AI gateways, both client and server. + * + * Copyright (C) 2026 Willy Tarreau + * + * Permission is hereby granted, free of charge, to any person obtaining + * a copy of this software and associated documentation files (the + * "Software"), to deal in the Software without restriction, including + * without limitation the rights to use, copy, modify, merge, publish, + * distribute, sublicense, and/or sell copies of the Software, and to + * permit persons to whom the Software is furnished to do so, subject to + * the following conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +/* + * A single process plays both roles at once, sharing one process-local clock: + * + * SERVER side: listens on -L and exposes an OpenAI-like + * POST /v1/chat/completions endpoint that streams chat.completion.chunk SSE + * events over HTTP/1.1 chunked encoding. Each token is a 4-letter word. The + * number of tokens, the pre-first-token ("thinking") delay and the + * inter-token delay are configurable and overridable per-request via the + * JSON body ("pi_tokens", "pi_think_ms", "pi_delay_ms"). Each chunk carries + * an extra "pi_ts" field with the emission time in nanoseconds since the + * epoch (CLOCK_REALTIME), taken at the moment the chunk is appended to the + * socket, so that the client can measure the one-way delay through the + * gateway using the same clock. + * + * CLIENT side: connects to -t (the gateway), sends an OpenAI-like streaming + * chat completion request, then reads the SSE stream (transparently handling + * chunked encoding if the gateway uses it) and records: + * - TTFT: time from request fully written to first content token received + * - inter-token gaps: arrival time deltas between consecutive tokens + * - emission delay: arrival_time - pi_ts of each chunk + * + * By default a loopback calibration batch is run first (client connects + * directly to the listening address), then the gateway batch, and the + * difference between both is reported as the estimated gateway overhead. + */ + +#define _GNU_SOURCE /* for POLLRDHUP, memmem, strcasestr */ +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifndef POLLRDHUP +#define POLLRDHUP 0 +#endif + +#ifndef MSG_NOSIGNAL +#define MSG_NOSIGNAL 0 +#endif + +/* ------------------------------------------------------------------ */ +/* Types */ +/* ------------------------------------------------------------------ */ + +/* growable array of doubles, values in milliseconds */ +struct dlist { + double *v; + size_t n; + size_t cap; +}; + +struct batch_stats { + struct dlist ttft; /* per request, ms */ + struct dlist emit_ttft; /* per request, ms */ + struct dlist gaps; /* per token, ms */ + struct dlist emits; /* per token, ms */ + int nb_req; /* number of requests in this batch */ + long ok_req, fail_req, tok_count; +}; + +enum { + ROLE_SERVER, + ROLE_CLIENT, +}; + +/* server states */ +enum { + CS_READ_REQ, /* reading the HTTP request */ + CS_EMIT, /* emitting tokens on schedule */ + CS_FLUSH_CLOSE, /* draining then close */ +}; + +/* client chunked-decoding states */ +enum { + CC_HEAD, /* reading response headers */ + CC_RAW, /* no chunked encoding: raw stream till close */ + CC_CHUNK_LINE, /* chunked: reading the chunk size line */ + CC_CHUNK_DATA, /* chunked: reading chunk payload */ + CC_CHUNK_CRLF, /* chunked: reading the CRLF after payload */ + CC_LAST_CHUNK, /* chunked: after the last (0) chunk, discard */ +}; + +struct batch_ctx { + struct sockaddr_storage target; + char host[64]; /* target address as a string, for Host: */ + int total, started, done, failed; + long tokens_hint; /* per-request pi_tokens override, 0 = none */ +}; + +struct conn { + int fd; + int role; + int state; + + char *rbuf; size_t rlen, rcap; /* raw receive buffer */ + char *wbuf; size_t wlen, wpos, wcap; + char *sbuf; size_t slen, scap; /* SSE line reassembly */ + + /* server */ + long tokens_left; + long delay_us; /* inter-token delay */ + uint64_t next_wake_us; /* monotonic deadline of next emission */ + long chunk_index; + char id[48]; + + /* client */ + int idx; /* request index within the batch */ + int connecting; + int cstate; /* CC_* */ + int chunked; + int status_ok; + int head_ok; + uint64_t csize, cpos; /* current chunked chunk being decoded */ + uint64_t deadline_us; /* monotonic overall timeout */ + uint64_t t0_us; /* monotonic start, for wall time */ + uint64_t sent_ns; /* realtime ns when request fully written */ + uint64_t last_arrival_ns; + double ttft_ms; + double emit_ttft_ms; + long tok_recv; + int ok; + char err[128]; +}; + +/* ------------------------------------------------------------------ */ +/* Options and globals */ +/* ------------------------------------------------------------------ */ + +static const char *listen_str = ":8090"; +static const char *target_str = NULL; /* NULL = server only */ +static int nb_requests = 1; +static int concurrency = 1; +static long nb_tokens = 1000; +static double think_ms = 0.0; +static double delay_ms = 5.0; +static long timeout_ms = 300000; +static int use_loopback = 1; +static int quiet; +static int verbose; + +static int listener_fd = -1; + +static struct batch_ctx *cur_batch; +static struct batch_stats *cur_stats; +static const char *ctx_tag; /* "loop" or "gw", for per-req lines */ +static uint64_t prog_tokens; /* tokens received in current batch */ +static uint64_t batch_t0_us; +static int progress_shown;/* a progress line is on stderr */ + +static struct conn **conns; +static int nbconns, conncap; + +/* ------------------------------------------------------------------ */ +/* Small helpers */ +/* ------------------------------------------------------------------ */ + +/* display the message and exit with the code */ +__attribute__((noreturn)) void die(int code, const char *format, ...) +{ + va_list args; + + if (format) { + va_start(args, format); + vfprintf(stderr, format, args); + va_end(args); + } + exit(code); +} + +/* display the usage message and exit with the code */ +__attribute__((noreturn)) void usage(int code, const char *arg0) +{ + die(code, + "Usage : %s [options]*\n" + "\n" + "SSE benchmark for AI gateways: runs an OpenAI-like SSE server\n" + "and an SSE client in the same process, sharing one clock.\n" + "\n" + "options :\n" + " -L <[ip:]port> : address to bind the dummy OpenAI-like SSE server to\n" + " (default :8090)\n" + " -t <[ip:]port> : gateway address to send requests to (default: none,\n" + " server only)\n" + " -r : number of client requests to send (default 1)\n" + " -c : number of parallel client requests (default 1)\n" + " -n : tokens per response (default 1000)\n" + " -T : delay before the first token, aka thinking (default 0)\n" + " -d : delay between tokens (default 5, float ok)\n" + " -w : per-request timeout (default 300000)\n" + " -l : disable the loopback calibration batch\n" + " -q : quiet, don't print per-request lines\n" + " -v : verbose\n" + " -h : this help\n" + "\n" + "Example against a gateway on 127.0.0.1:8080 forwarding to this tool's\n" + "server listening on :8090 :\n" + " %s -L :8090 -t 127.0.0.1:8080 -r 20 -c 5\n" + "\n" + "Metrics reported :\n" + " TTFT (client) request fully sent -> first token received\n" + " TTFT emit->recv(1st) one-way delay of the first token\n" + " inter-token (client) gaps between consecutive token arrivals\n" + " emit->recv (tokens) one-way delay of every token\n" + " gateway overhead gateway batch minus loopback batch\n" + "", arg0, arg0); +} + +/* erases the progress line if one is currently displayed, so that real + * output doesn't get glued to it. + */ +static void progress_clear(void) +{ + if (progress_shown) { + fprintf(stderr, "\r\33[K"); + progress_shown = 0; + } +} + +/* current time in nanoseconds since the epoch, for the pi_ts fields */ +static inline uint64_t now_ns_real(void) +{ + struct timespec ts; + + clock_gettime(CLOCK_REALTIME, &ts); + return ts.tv_sec * 1000000000ULL + ts.tv_nsec; +} + +/* current monotonic time in microseconds, for scheduling */ +static inline uint64_t now_us_mono(void) +{ + struct timespec ts; + + clock_gettime(CLOCK_MONOTONIC, &ts); + return ts.tv_sec * 1000000ULL + ts.tv_nsec / 1000; +} + +/* case-insensitive search of a NUL-terminated needle within at most + * bytes of (which is NOT necessarily NUL-terminated). + */ +static const char *ci_find(const char *hay, size_t haylen, const char *needle) +{ + size_t nlen = strlen(needle); + size_t i, j; + + if (nlen > haylen) + return NULL; + + for (i = 0; i + nlen <= haylen; i++) { + for (j = 0; j < nlen; j++) + if (tolower((unsigned char)hay[i + j]) != tolower((unsigned char)needle[j])) + break; + if (j == nlen) + return hay + i; + } + return NULL; +} + +/* converts str in the form [[||]:]port to struct + * sockaddr_storage. Returns < 0 in case of error. + */ +static int addr_to_ss(const char *str, struct sockaddr_storage *ss) +{ + char *str_copy, *port_str; + int port; + + memset(ss, 0, sizeof(*ss)); + + str_copy = strdup(str); + if (!str_copy) + die(1, "out of memory\n"); + + /* look for the addr/port delimiter, it's the last colon. If there's no + * colon, it's 0:. + */ + if ((port_str = strrchr(str_copy, ':')) == NULL) { + port = atoi(str_copy); + if (port < 0 || port > 65535) { + fprintf(stderr, "Missing/invalid port number: '%s'\n", str); + goto fail; + } + + ss->ss_family = AF_INET; + ((struct sockaddr_in *)ss)->sin_port = htons(port); + ((struct sockaddr_in *)ss)->sin_addr.s_addr = INADDR_ANY; + goto done; + } + + *port_str++ = 0; + + if (strrchr(str_copy, ':') != NULL) { + /* IPv6 address contains ':' */ + ss->ss_family = AF_INET6; + ((struct sockaddr_in6 *)ss)->sin6_port = htons(atoi(port_str)); + + if (!inet_pton(ss->ss_family, str_copy, &((struct sockaddr_in6 *)ss)->sin6_addr)) { + fprintf(stderr, "Invalid server address: '%s'\n", str); + goto fail; + } + } + else { + ss->ss_family = AF_INET; + ((struct sockaddr_in *)ss)->sin_port = htons(atoi(port_str)); + + if (*str_copy == '*' || *str_copy == '\0') { /* INADDR_ANY */ + ((struct sockaddr_in *)ss)->sin_addr.s_addr = INADDR_ANY; + goto done; + } + + if (!inet_pton(ss->ss_family, str_copy, &((struct sockaddr_in *)ss)->sin_addr)) { + struct hostent *he = gethostbyname(str_copy); + + if (he == NULL) { + fprintf(stderr, "Invalid server name: '%s'\n", str); + goto fail; + } + ((struct sockaddr_in *)ss)->sin_addr = *(struct in_addr *)*he->h_addr_list; + } + } + done: + free(str_copy); + return 0; + fail: + free(str_copy); + return -1; +} + +static int addr_eq(const struct sockaddr_storage *a, const struct sockaddr_storage *b) +{ + if (a->ss_family != b->ss_family) + return 0; + + if (a->ss_family == AF_INET) { + const struct sockaddr_in *x = (const struct sockaddr_in *)a; + const struct sockaddr_in *y = (const struct sockaddr_in *)b; + + return x->sin_port == y->sin_port && + x->sin_addr.s_addr == y->sin_addr.s_addr; + } + + if (a->ss_family == AF_INET6) { + const struct sockaddr_in6 *x = (const struct sockaddr_in6 *)a; + const struct sockaddr_in6 *y = (const struct sockaddr_in6 *)b; + + return x->sin6_port == y->sin6_port && + !memcmp(&x->sin6_addr, &y->sin6_addr, sizeof(x->sin6_addr)); + } + + return 0; +} + +/* replaces the wildcard address with loopback, for addresses used as + * connection destinations (a bare ":port" or "port" argument). + */ +static void addr_force_loopback(struct sockaddr_storage *ss) +{ + if (ss->ss_family == AF_INET) { + struct sockaddr_in *sin = (struct sockaddr_in *)ss; + + if (sin->sin_addr.s_addr == INADDR_ANY) + sin->sin_addr.s_addr = htonl(INADDR_LOOPBACK); + } + else if (ss->ss_family == AF_INET6) { + struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)ss; + + if (!memcmp(&sin6->sin6_addr, &in6addr_any, sizeof(sin6->sin6_addr))) + sin6->sin6_addr = in6addr_loopback; + } +} + +static const char *addr_to_str(const struct sockaddr_storage *ss) +{ + static char buf[INET6_ADDRSTRLEN + 8]; + + if (ss->ss_family == AF_INET) { + const struct sockaddr_in *s = (const struct sockaddr_in *)ss; + char tmp[INET_ADDRSTRLEN]; + + inet_ntop(AF_INET, &s->sin_addr, tmp, sizeof(tmp)); + snprintf(buf, sizeof(buf), "%s:%d", tmp, ntohs(s->sin_port)); + } + else { + const struct sockaddr_in6 *s = (const struct sockaddr_in6 *)ss; + char tmp[INET6_ADDRSTRLEN]; + + inet_ntop(AF_INET6, &s->sin6_addr, tmp, sizeof(tmp)); + snprintf(buf, sizeof(buf), "%s:%d", tmp, ntohs(s->sin6_port)); + } + return buf; +} + +/* ------------------------------------------------------------------ */ +/* Statistics */ +/* ------------------------------------------------------------------ */ + +static void dl_push(struct dlist *l, double x) +{ + if (l->n == l->cap) { + l->cap = l->cap ? l->cap * 2 : 256; + l->v = realloc(l->v, l->cap * sizeof(*l->v)); + if (!l->v) + die(1, "out of memory\n"); + } + l->v[l->n++] = x; +} + +static int dbl_cmp(const void *a, const void *b) +{ + double x = *(const double *)a, y = *(const double *)b; + + if (x < y) + return -1; + if (x > y) + return 1; + return 0; +} + +/* returns a sorted copy of the list, or NULL if empty */ +static double *dl_sorted(const struct dlist *l) +{ + double *v; + + if (!l->n) + return NULL; + + v = malloc(l->n * sizeof(*v)); + if (!v) + die(1, "out of memory\n"); + memcpy(v, l->v, l->n * sizeof(*v)); + qsort(v, l->n, sizeof(*v), dbl_cmp); + return v; +} + +static void stats_line(const char *name, const struct dlist *l) +{ + double *v, sum = 0, mean; + size_t i; + + if (!l->n) { + printf("%-24s n=0\n", name); + return; + } + + v = dl_sorted(l); + for (i = 0; i < l->n; i++) + sum += v[i]; + mean = sum / l->n; + + printf("%-24s n=%zu mean=%.2fms min=%.2fms p50=%.2fms p95=%.2fms p99=%.2fms max=%.2fms\n", + name, l->n, mean, v[0], + v[(size_t)((l->n - 1) * 0.50)], + v[(size_t)((l->n - 1) * 0.95)], + v[(size_t)((l->n - 1) * 0.99)], + v[l->n - 1]); + free(v); +} + +/* prints the difference of relative to (gw minus loop). + * Negative values just mean the loopback run happened to be noisier. + */ +static void delta_line(const char *name, const struct dlist *gw, const struct dlist *lo) +{ + double *vg, *vl, sg = 0, sl = 0; + size_t i; + + if (!gw->n || !lo->n) { + printf("%-24s n/a (no samples)\n", name); + return; + } + + vg = dl_sorted(gw); + vl = dl_sorted(lo); + for (i = 0; i < gw->n; i++) + sg += vg[i]; + for (i = 0; i < lo->n; i++) + sl += vl[i]; + + printf("%-24s mean %+.3fms p50 %+.3fms p95 %+.3fms p99 %+.3fms\n", + name, + sg / gw->n - sl / lo->n, + vg[(size_t)((gw->n - 1) * 0.50)] - vl[(size_t)((lo->n - 1) * 0.50)], + vg[(size_t)((gw->n - 1) * 0.95)] - vl[(size_t)((lo->n - 1) * 0.95)], + vg[(size_t)((gw->n - 1) * 0.99)] - vl[(size_t)((lo->n - 1) * 0.99)]); + free(vg); + free(vl); +} + +static void print_batch(const char *title, struct batch_stats *st) +{ + printf("\n=== %s: %d requests, %ld ok, %ld failed, %ld tokens ===\n", + title, st->nb_req, st->ok_req, st->fail_req, st->tok_count); + stats_line("TTFT (client)", &st->ttft); + stats_line("TTFT emit->recv(1st)", &st->emit_ttft); + stats_line("inter-token (client)", &st->gaps); + stats_line("emit->recv (tokens)", &st->emits); +} + +/* ------------------------------------------------------------------ */ +/* Connection basics */ +/* ------------------------------------------------------------------ */ + +static void conn_append(struct conn *c, const char *data, size_t len) +{ + if (c->wlen + len > c->wcap) { + size_t cap = c->wcap; + + while (c->wlen + len > cap) + cap = cap ? cap * 2 : 8192; + c->wbuf = realloc(c->wbuf, cap); + if (!c->wbuf) + die(1, "out of memory\n"); + c->wcap = cap; + } + memcpy(c->wbuf + c->wlen, data, len); + c->wlen += len; +} + +/* tries to write pending data. Returns -1 on write error (caller closes). */ +static int conn_flush(struct conn *c) +{ + while (c->wpos < c->wlen) { + ssize_t ret = send(c->fd, c->wbuf + c->wpos, c->wlen - c->wpos, MSG_NOSIGNAL); + + if (ret > 0) { + c->wpos += ret; + continue; + } + if (ret < 0 && errno == EINTR) + continue; + if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) + break; + return -1; + } + + if (c->wpos == c->wlen) { + c->wpos = c->wlen = 0; + + /* for clients, this is the "request fully sent" instant */ + if (c->role == ROLE_CLIENT && !c->sent_ns) + c->sent_ns = now_ns_real(); + } + else if (c->wpos > 65536) { + /* compact the pending data */ + memmove(c->wbuf, c->wbuf + c->wpos, c->wlen - c->wpos); + c->wlen -= c->wpos; + c->wpos = 0; + } + return 0; +} + +/* close and release the connection. marks a client-side failure. */ +static void conn_close(struct conn *c, const char *err) +{ + int i; + + if (err && c->role == ROLE_CLIENT && !c->err[0]) + snprintf(c->err, sizeof(c->err), "%s", err); + + if (c->role == ROLE_CLIENT && cur_batch) { + struct batch_ctx *ctx = cur_batch; + uint64_t wall_us = now_us_mono() - c->t0_us; + + ctx->done++; + if (c->ok) { + cur_stats->ok_req++; + cur_stats->tok_count += c->tok_recv; + } else { + ctx->failed++; + cur_stats->fail_req++; + } + + if (!quiet) { + progress_clear(); + if (c->ok) + printf("%s %3d: ok ttft=%9.2fms emit-delay(first)=%9.2fms tokens=%ld wall=%.3fs\n", + ctx_tag, c->idx, c->ttft_ms, c->emit_ttft_ms, + c->tok_recv, wall_us / 1000000.0); + else + printf("%s %3d: FAILED: %s\n", ctx_tag, c->idx, c->err); + } + } + + close(c->fd); + free(c->rbuf); + free(c->wbuf); + free(c->sbuf); + + /* remove from the list */ + for (i = 0; i < nbconns; i++) { + if (conns[i] == c) { + conns[i] = conns[--nbconns]; + break; + } + } + free(c); +} + +static struct conn *conn_new(int fd, int role) +{ + struct conn *c; + + c = calloc(1, sizeof(*c)); + if (!c) + die(1, "out of memory\n"); + + c->fd = fd; + c->role = role; + c->state = role == ROLE_SERVER ? CS_READ_REQ : CS_EMIT; + c->cstate = CC_HEAD; + + if (nbconns == conncap) { + conncap = conncap ? conncap * 2 : 16; + conns = realloc(conns, conncap * sizeof(*conns)); + if (!conns) + die(1, "out of memory\n"); + } + conns[nbconns++] = c; + return c; +} + +/* flush pending output, close the connection if it failed or if a server + * connection reached the end of its response. Returns -1 if closed. + */ +static int conn_output(struct conn *c) +{ + if (conn_flush(c) < 0) { + conn_close(c, NULL); + return -1; + } + if (c->role == ROLE_SERVER && c->state == CS_FLUSH_CLOSE && + c->wpos == c->wlen) { + conn_close(c, NULL); + return -1; + } + return 0; +} + +/* ------------------------------------------------------------------ */ +/* Server side: SSE emission */ +/* ------------------------------------------------------------------ */ + +/* appends one SSE event wrapped in its own HTTP chunk */ +static void sse_append(struct conn *c, const char *payload, size_t plen) +{ + char hdr[16]; + int hlen; + + hlen = snprintf(hdr, sizeof(hdr), "%zx\r\n", plen + 6 + 2); + conn_append(c, hdr, hlen); + conn_append(c, "data: ", 6); + conn_append(c, payload, plen); + conn_append(c, "\n\n\r\n", 4); +} + +/* 4-letter pseudo-word for token index */ +static void token_text(long i, char out[4]) +{ + const char letters[] = "abcdefghijklmnopqrstuvwxyz"; + int j; + + for (j = 0; j < 4; j++) + out[j] = letters[(i * 7 + j * 13) % 26]; +} + +/* builds and queues one content chunk; timestamped at append time */ +static void server_queue_token(struct conn *c) +{ + char json[512]; + char tok[4]; + uint64_t pi_ts = now_ns_real(); + int jl; + + token_text(c->chunk_index++, tok); + + jl = snprintf(json, sizeof(json), + "{\"id\":\"%s\",\"object\":\"chat.completion.chunk\"," + "\"created\":%llu,\"pi_ts\":%llu,\"model\":\"sse-latency-1\"," + "\"choices\":[{\"index\":0," + "\"delta\":{\"content\":\"%c%c%c%c\"}," + "\"finish_reason\":null}]}", + c->id, (unsigned long long)(pi_ts / 1000000000ULL), + (unsigned long long)pi_ts, + tok[0], tok[1], tok[2], tok[3]); + + sse_append(c, json, jl); +} + +/* queues the final chunk, [DONE] and the terminating empty HTTP chunk */ +static void server_queue_finish(struct conn *c) +{ + char json[512]; + uint64_t pi_ts = now_ns_real(); + int jl; + + jl = snprintf(json, sizeof(json), + "{\"id\":\"%s\",\"object\":\"chat.completion.chunk\"," + "\"created\":%llu,\"pi_ts\":%llu,\"model\":\"sse-latency-1\"," + "\"choices\":[{\"index\":0,\"delta\":{}," + "\"finish_reason\":\"stop\"}]}", + c->id, (unsigned long long)(pi_ts / 1000000000ULL), + (unsigned long long)pi_ts); + + sse_append(c, json, jl); + sse_append(c, "[DONE]", 6); + conn_append(c, "0\r\n\r\n", 5); + c->state = CS_FLUSH_CLOSE; +} + +/* processes a complete HTTP request: extracts the hints, queues the + * response head + role chunk, and schedules the first token. + * Returns -1 if the connection was closed. + */ +static int server_process_request(struct conn *c) +{ + char head[512], json[512]; + const char *body, *p; + size_t head_end, body_len = 0; + long tokens = nb_tokens; + double think = think_ms, delay = delay_ms; + uint64_t pi_ts; + int hl, jl; + + /* find end of headers */ + p = memmem(c->rbuf, c->rlen, "\r\n\r\n", 4); + if (!p) + return 0; /* not yet */ + head_end = p - c->rbuf + 4; + + /* only POST is supported */ + if (c->rlen < 4 || strncmp(c->rbuf, "POST", 4)) { + static const char notfound[] = + "HTTP/1.1 404 Not Found\r\n" + "Content-Length: 0\r\n" + "Connection: close\r\n" + "\r\n"; + + conn_append(c, notfound, sizeof(notfound) - 1); + c->state = CS_FLUSH_CLOSE; + return conn_output(c); + } + + /* find Content-Length within the headers */ + if ((p = ci_find(c->rbuf, head_end, "\nContent-Length:"))) + body_len = strtoul(p + 15, NULL, 10); + + /* wait for the whole body */ + if (c->rlen < head_end + body_len) + return 0; + + body = c->rbuf + head_end; + + /* make sure the body is NUL-terminated for the hint parsers below */ + if (c->rcap < c->rlen + 1) { + c->rcap = c->rlen + 1; + c->rbuf = realloc(c->rbuf, c->rcap); + if (!c->rbuf) + die(1, "out of memory\n"); + } + c->rbuf[c->rlen] = 0; + + /* per-request overrides, ignored by real gateways */ + if ((p = strstr(body, "\"pi_tokens\":"))) + tokens = strtoul(p + 12, NULL, 10); + if ((p = strstr(body, "\"pi_think_ms\":"))) + think = strtod(p + 14, NULL); + if ((p = strstr(body, "\"pi_delay_ms\":"))) + delay = strtod(p + 14, NULL); + + if (tokens < 0) + tokens = 0; + + /* response id, shared by all chunks of this response */ + snprintf(c->id, sizeof(c->id), "chatcmpl-sse-latency-%llu", + (unsigned long long)(now_ns_real() % 100000000000ULL)); + + /* response head + role chunk */ + hl = snprintf(head, sizeof(head), + "HTTP/1.1 200 OK\r\n" + "Content-Type: text/event-stream\r\n" + "Cache-Control: no-cache\r\n" + "Connection: close\r\n" + "X-Accel-Buffering: no\r\n" + "Transfer-Encoding: chunked\r\n" + "\r\n"); + conn_append(c, head, hl); + + pi_ts = now_ns_real(); + jl = snprintf(json, sizeof(json), + "{\"id\":\"%s\",\"object\":\"chat.completion.chunk\"," + "\"created\":%llu,\"pi_ts\":%llu,\"model\":\"sse-latency-1\"," + "\"choices\":[{\"index\":0," + "\"delta\":{\"role\":\"assistant\"}," + "\"finish_reason\":null}]}", + c->id, (unsigned long long)(pi_ts / 1000000000ULL), + (unsigned long long)pi_ts); + sse_append(c, json, jl); + + c->tokens_left = tokens; + c->delay_us = (long)(delay * 1000.0); + c->next_wake_us = now_us_mono() + (long)(think * 1000.0); + c->state = CS_EMIT; + + /* we don't expect anything more from the client; drop the request */ + free(c->rbuf); + c->rbuf = NULL; + c->rlen = c->rcap = 0; + + if (verbose) + fprintf(stderr, "srv %d: %ld tokens, think %.0fms, delay %.0fms\n", + c->fd, tokens, think, delay); + + return conn_output(c); +} + +/* ------------------------------------------------------------------ */ +/* Client side */ +/* ------------------------------------------------------------------ */ + +static void client_spawn(struct batch_ctx *ctx) +{ + struct sockaddr_storage *ss = &ctx->target; + struct conn *c; + char req[1024]; + char body[256]; + socklen_t sslen; + int fd, ret, rl, bl; + + /* the calibration batch shortens its responses via the pi_tokens hint, + * which our own server honors and real gateways silently ignore. + */ + if (ctx->tokens_hint > 0) + bl = snprintf(body, sizeof(body), + "{\"model\":\"sse-latency-1\",\"stream\":true," + "\"messages\":[{\"role\":\"user\",\"content\":\"Say 4-letter words\"}]," + "\"pi_tokens\":%ld}", ctx->tokens_hint); + else + bl = snprintf(body, sizeof(body), + "{\"model\":\"sse-latency-1\",\"stream\":true," + "\"messages\":[{\"role\":\"user\",\"content\":\"Say 4-letter words\"}]}"); + + fd = socket(ss->ss_family, SOCK_STREAM, 0); + if (fd < 0) + die(1, "socket(): %s\n", strerror(errno)); + + fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK); + { + int one = 1; + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + } + + rl = snprintf(req, sizeof(req), + "POST /v1/chat/completions HTTP/1.1\r\n" + "Host: %s\r\n" + "Content-Type: application/json\r\n" + "Accept: text/event-stream\r\n" + "Content-Length: %d\r\n" + "Connection: close\r\n" + "\r\n" + "%s", + ctx->host, bl, body); + + c = conn_new(fd, ROLE_CLIENT); + c->idx = ctx->started++; + c->connecting = 1; + c->t0_us = now_us_mono(); + c->deadline_us = c->t0_us + (uint64_t)timeout_ms * 1000; + conn_append(c, req, rl); + + sslen = ss->ss_family == AF_INET6 ? sizeof(struct sockaddr_in6) + : sizeof(struct sockaddr_in); + ret = connect(fd, (const struct sockaddr *)ss, sslen); + if (ret < 0 && errno != EINPROGRESS) { + snprintf(c->err, sizeof(c->err), "connect(): %s", strerror(errno)); + conn_close(c, NULL); + return; + } + + if (verbose) + fprintf(stderr, "cl %d: connecting (req %d)\n", fd, c->idx); +} + +/* processes one complete SSE line (NUL-terminated in c->sbuf). + * Returns -1 if the connection was closed. + */ +static int client_line(struct conn *c, char *line) +{ + char *data, *p; + + /* strip trailing CR */ + p = line + strlen(line); + if (p > line && p[-1] == '\r') + p[-1] = 0; + + if (strncmp(line, "data:", 5)) + return 0; /* not a data line */ + + data = line + 5; + while (*data == ' ' || *data == '\t') + data++; + + if (!strcmp(data, "[DONE]")) { + c->ok = c->tok_recv > 0; + if (!c->ok && !c->err[0]) + snprintf(c->err, sizeof(c->err), "no content tokens received"); + conn_close(c, NULL); + return -1; + } + + /* extract pi_ts */ + p = strstr(data, "\"pi_ts\":"); + if (!p) + return 0; /* unknown frame, ignore */ + { + uint64_t pi_ts = strtoull(p + 8, NULL, 10); + uint64_t arrival = now_ns_real(); + double delta_ms = (double)(arrival - pi_ts) / 1e6; + + /* a content token? */ + p = strstr(data, "\"content\":\""); + if (!p || p[11] == '"') + return 0; /* role or finish chunk */ + + c->tok_recv++; + prog_tokens++; + + if (c->tok_recv == 1) { + c->ttft_ms = c->sent_ns ? + (double)(arrival - c->sent_ns) / 1e6 : -1.0; + c->emit_ttft_ms = delta_ms; + if (cur_stats) { + dl_push(&cur_stats->ttft, c->ttft_ms); + dl_push(&cur_stats->emit_ttft, delta_ms); + } + } else { + if (cur_stats) { + dl_push(&cur_stats->gaps, + (double)(arrival - c->last_arrival_ns) / 1e6); + dl_push(&cur_stats->emits, delta_ms); + } + } + c->last_arrival_ns = arrival; + } + return 0; +} + +/* feeds decoded payload bytes to the SSE line reassembly buffer. + * Returns -1 if the connection was closed. + */ +static int client_sse_feed(struct conn *c, const char *data, size_t len) +{ + while (len) { + char *nl = memchr(data, '\n', len); + size_t got; + int ret; + + if (!nl) { + /* no complete line, keep the remainder */ + if (c->slen + len > c->scap) { + c->scap = c->slen + len + 256; + c->sbuf = realloc(c->sbuf, c->scap); + if (!c->sbuf) + die(1, "out of memory\n"); + } + memcpy(c->sbuf + c->slen, data, len); + c->slen += len; + break; + } + + got = nl - data; /* length without the \n */ + + if (c->slen + got + 1 > c->scap) { + c->scap = c->slen + got + 1; + c->sbuf = realloc(c->sbuf, c->scap); + if (!c->sbuf) + die(1, "out of memory\n"); + } + memcpy(c->sbuf + c->slen, data, got); + c->sbuf[c->slen + got] = 0; + + ret = client_line(c, c->sbuf); + c->slen = 0; + if (ret < 0) + return -1; + + data += got + 1; + len -= got + 1; + } + return 0; +} + +/* consumes bytes from c->rbuf according to the transfer encoding. + * Returns -1 if the connection was closed. + */ +static int client_feed(struct conn *c) +{ + size_t pos = 0; + + while (pos < c->rlen) { + size_t avail = c->rlen - pos; + size_t take; + char *nl; + + switch (c->cstate) { + case CC_RAW: + if (client_sse_feed(c, c->rbuf + pos, avail) < 0) + return -1; + pos = c->rlen; + break; + + case CC_CHUNK_LINE: + nl = memchr(c->rbuf + pos, '\n', avail); + if (!nl) + return 0; /* wait for more */ + { + uint64_t v = 0; + char *q; + + for (q = c->rbuf + pos; q < nl; q++) { + int h; + + if (*q == '\r' || *q == ' ' || *q == '\t') + continue; + if (*q >= '0' && *q <= '9') + h = *q - '0'; + else if (*q >= 'a' && *q <= 'f') + h = *q - 'a' + 10; + else if (*q >= 'A' && *q <= 'F') + h = *q - 'A' + 10; + else + break; /* chunk extension, ignore */ + v = v * 16 + h; + } + c->csize = v; + c->cpos = 0; + c->cstate = v ? CC_CHUNK_DATA : CC_LAST_CHUNK; + } + pos = nl - c->rbuf + 1; + break; + + case CC_CHUNK_DATA: + take = avail < c->csize - c->cpos ? avail : c->csize - c->cpos; + if (take) { + if (client_sse_feed(c, c->rbuf + pos, take) < 0) + return -1; + pos += take; + c->cpos += take; + } + if (c->cpos == c->csize) + c->cstate = CC_CHUNK_CRLF; + break; + + case CC_CHUNK_CRLF: + if (avail < 2) + return 0; /* wait for more */ + pos += 2; /* swallow "\r\n" */ + c->cstate = CC_CHUNK_LINE; + break; + + case CC_LAST_CHUNK: + /* discard trailers till EOF; we're done */ + pos = c->rlen; + if (c->ok) { + conn_close(c, NULL); + return -1; + } + break; + + default: + return 0; + } + } + + /* everything consumed */ + c->rlen = 0; + return 0; +} + +/* processes bytes accumulated in c->rbuf on a client connection. + * Returns -1 if the connection was closed. + */ +static int client_process(struct conn *c) +{ + const char *p; + + if (!c->head_ok) { + size_t head_end; + + p = memmem(c->rbuf, c->rlen, "\r\n\r\n", 4); + if (!p) { + if (c->rlen > 65536) { + conn_close(c, "response headers too large"); + return -1; + } + return 0; /* wait for more */ + } + head_end = p - c->rbuf + 4; + + /* validate the status line directly from the head: at this point + * the whole head is present, so the first line is complete. + */ + if (c->rlen < 12 || strncmp(c->rbuf, "HTTP/", 5) != 0 || + !isdigit((unsigned char)c->rbuf[5]) || c->rbuf[9] != '2') { + if (c->rlen >= 12) + snprintf(c->err, sizeof(c->err), "status %.32s", c->rbuf); + else + snprintf(c->err, sizeof(c->err), "short response"); + conn_close(c, NULL); + return -1; + } + c->status_ok = 1; + + /* detect chunked encoding within the head */ + c->chunked = ci_find(c->rbuf, head_end, "chunked") != NULL; + c->head_ok = 1; + c->cstate = c->chunked ? CC_CHUNK_LINE : CC_RAW; + + /* move the payload bytes to the front */ + c->rlen -= head_end; + memmove(c->rbuf, c->rbuf + head_end, c->rlen); + } + + return client_feed(c); +} + +/* ------------------------------------------------------------------ */ +/* Socket readiness handling */ +/* ------------------------------------------------------------------ */ + +/* returns -1 if the connection was closed */ +static int conn_read(struct conn *c) +{ + ssize_t ret; + + if (c->rlen == c->rcap) { + c->rcap = c->rcap ? c->rcap * 2 : 16384; + c->rbuf = realloc(c->rbuf, c->rcap); + if (!c->rbuf) + die(1, "out of memory\n"); + } + + ret = recv(c->fd, c->rbuf + c->rlen, c->rcap - c->rlen, 0); + if (ret > 0) { + c->rlen += ret; + + if (c->role == ROLE_CLIENT) + return client_process(c); + + if (c->state == CS_READ_REQ) + return server_process_request(c); + + /* unexpected data after the request, discard */ + c->rlen = 0; + return 0; + } + if (ret == 0) { + /* EOF */ + if (c->role == ROLE_CLIENT && !c->ok) { + conn_close(c, "connection closed before [DONE]"); + return -1; + } + conn_close(c, NULL); + return -1; + } + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) + return 0; + + conn_close(c, c->role == ROLE_CLIENT ? "recv error" : NULL); + return -1; +} + +static void do_accept(void) +{ + while (1) { + int fd = accept(listener_fd, NULL, NULL); + struct conn *c; + int one = 1; + + if (fd < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) + return; + die(1, "accept(): %s\n", strerror(errno)); + } + + fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK); + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + + c = conn_new(fd, ROLE_SERVER); + snprintf(c->id, sizeof(c->id), "chatcmpl-sse-latency-%llu", + (unsigned long long)(now_ns_real() % 100000000000ULL)); + + if (verbose) + fprintf(stderr, "srv %d: accepted\n", fd); + } +} + +/* ------------------------------------------------------------------ */ +/* Event loop */ +/* ------------------------------------------------------------------ */ + +/* the main poll loop. If is NULL, runs forever (server-only mode), + * otherwise runs until the batch is complete. + */ +static void event_loop(struct batch_ctx *ctx) +{ + struct pollfd *pfds = NULL; + struct conn **pmap = NULL; + int cap = 0; + uint64_t last_prog = 0; + + while (1) { + uint64_t now = now_us_mono(); + int timeout = 1000; + int n, i, nb, nb_active; + + if (ctx && ctx->done >= ctx->total) + return; + + /* spawn clients to fill the concurrency */ + if (ctx) { + nb_active = 0; + for (i = 0; i < nbconns; i++) + if (conns[i]->role == ROLE_CLIENT) + nb_active++; + while (ctx->started < ctx->total && nb_active < concurrency) { + client_spawn(ctx); + nb_active++; + } + } + + /* build the pollfd map */ + nb = nbconns; + if (cap < nb + 1) { + cap = nb + 16; + pfds = realloc(pfds, cap * sizeof(*pfds)); + pmap = realloc(pmap, cap * sizeof(*pmap)); + if (!pfds || !pmap) + die(1, "out of memory\n"); + } + + n = 0; + pfds[n].fd = listener_fd; + pfds[n].events = POLLIN; + pfds[n].revents = 0; + pmap[n] = NULL; + n++; + + for (i = 0; i < nb; i++) { + struct conn *c = conns[i]; + short ev = POLLIN; + + if (c->connecting) + ev = POLLOUT; /* connect() completion */ + else if (c->wpos < c->wlen) + ev |= POLLOUT; + + pfds[n].fd = c->fd; + pfds[n].events = ev; + pfds[n].revents = 0; + pmap[n] = c; + n++; + + /* compute the timeout from server wakeups and client deadlines */ + if (c->role == ROLE_SERVER && c->state == CS_EMIT) { + uint64_t left = c->next_wake_us > now ? c->next_wake_us - now : 0; + int ms = (int)((left + 999) / 1000); + + if (ms < timeout) + timeout = ms; + } + if (c->role == ROLE_CLIENT) { + uint64_t left = c->deadline_us > now ? c->deadline_us - now : 0; + int ms = (int)((left + 999) / 1000); + + if (ms < timeout) + timeout = ms; + } + } + + n = poll(pfds, n, timeout); + if (n < 0) { + if (errno == EINTR) + n = 0; + else + die(1, "poll(): %s\n", strerror(errno)); + } + + now = now_us_mono(); + + /* process poll results; the listener is entry 0. Note that the + * conn list may have changed in between, so we iterate over the + * snapshot only. + */ + if (n > 0 && (pfds[0].revents & POLLIN)) + do_accept(); + + for (i = 1; i <= nb; i++) { + struct conn *c = pmap[i]; + short re = pfds[i].revents; + + if (!re || !c) + continue; + + if (c->connecting && (re & POLLOUT)) { + int err = 0; + socklen_t len = sizeof(err); + + getsockopt(c->fd, SOL_SOCKET, SO_ERROR, &err, &len); + if (err) { + snprintf(c->err, sizeof(c->err), "connect(): %s", strerror(err)); + conn_close(c, NULL); + continue; + } + c->connecting = 0; + if (conn_output(c) < 0) + continue; + } + + if (re & POLLOUT) { + if (conn_output(c) < 0) + continue; + } + + if (re & (POLLIN | POLLERR | POLLHUP | POLLRDHUP)) { + if (conn_read(c) < 0) + continue; + } + } + + /* now emit scheduled tokens for all server conns */ + for (i = 0; i < nbconns; i++) { + struct conn *c = conns[i]; + + if (c->role != ROLE_SERVER || c->state != CS_EMIT) + continue; + + while (c->next_wake_us <= now) { + if (c->wcap - c->wlen < 1024) { + /* out of room, retry once drained */ + c->next_wake_us = now + 1000; + break; + } + if (c->tokens_left > 0) { + c->tokens_left--; + server_queue_token(c); + c->next_wake_us += c->delay_us; + } else { + server_queue_finish(c); + } + if (conn_output(c) < 0) + break; + } + } + + /* enforce client deadlines */ + for (i = 0; i < nbconns; i++) { + struct conn *c = conns[i]; + + if (c->role != ROLE_CLIENT) + continue; + if (now > c->deadline_us) { + snprintf(c->err, sizeof(c->err), "timeout after %ldms", timeout_ms); + conn_close(c, NULL); + i--; /* list shrank */ + } + } + + /* progress line, once per second */ + if (ctx && isatty(2) && now - last_prog >= 1000000) { + int in_flight = 0; + + for (i = 0; i < nbconns; i++) + if (conns[i]->role == ROLE_CLIENT) + in_flight++; + + fprintf(stderr, "\r streaming: %llus elapsed, %d/%d done, %d in flight, %llu tokens received", + (unsigned long long)((now - batch_t0_us) / 1000000), + ctx->done, ctx->total, in_flight, + (unsigned long long)prog_tokens); + progress_shown = 1; + last_prog = now; + } + } +} + +/* ------------------------------------------------------------------ */ +/* Batch runner */ +/* ------------------------------------------------------------------ */ + +static struct batch_stats *run_batch(struct sockaddr_storage *target, const char *tag, + int nb_req, long tokens_hint) +{ + struct batch_ctx ctx; + struct batch_stats *st; + uint64_t per_req_ms, total_ms; + long eff_tokens; + + memset(&ctx, 0, sizeof(ctx)); + ctx.target = *target; + ctx.total = nb_req; + ctx.tokens_hint = tokens_hint; + snprintf(ctx.host, sizeof(ctx.host), "%s", addr_to_str(target)); + + st = calloc(1, sizeof(*st)); + if (!st) + die(1, "out of memory\n"); + st->nb_req = nb_req; + + cur_batch = &ctx; + cur_stats = st; + ctx_tag = tag; + prog_tokens = 0; + batch_t0_us = now_us_mono(); + + if (tokens_hint > 0) + fprintf(stderr, "%s: calibrating the noise floor with %d request(s) of %ld tokens to %s (concurrency=%d, think=%.0fms, delay=%.0fms) ...\n", + tag, nb_req, tokens_hint, ctx.host, concurrency, think_ms, delay_ms); + else + fprintf(stderr, "%s: sending %d request(s) to %s (concurrency=%d, tokens=%ld, think=%.0fms, delay=%.0fms) ...\n", + tag, nb_req, ctx.host, concurrency, nb_tokens, think_ms, delay_ms); + + /* announce the expected duration of long batches, so that they don't + * look stuck or surprising. + */ + eff_tokens = tokens_hint > 0 ? tokens_hint : nb_tokens; + per_req_ms = (uint64_t)(think_ms + delay_ms * (long long)(eff_tokens > 0 ? eff_tokens - 1 : 0)); + total_ms = per_req_ms * ((nb_req + concurrency - 1) / concurrency); + if (total_ms > 30000) + fprintf(stderr, " (expect ~%llumin%02llus for this batch)\n", + (unsigned long long)(total_ms / 60000), + (unsigned long long)((total_ms % 60000) / 1000)); + + event_loop(&ctx); + + cur_batch = NULL; + cur_stats = NULL; + ctx_tag = NULL; + + progress_clear(); + + return st; +} + +/* ------------------------------------------------------------------ */ +/* main */ +/* ------------------------------------------------------------------ */ + +int main(int argc, char **argv) +{ + struct sockaddr_storage listen_ss, target_ss, loop_ss; + struct batch_stats *loop_stats = NULL, *gw_stats; + const char *arg0 = argv[0]; + int has_loop, calib_req; + long calib_tokens; + + signal(SIGPIPE, SIG_IGN); + + /* keep the per-request lines visible even when stdout is redirected */ + setvbuf(stdout, NULL, _IOLBF, 0); + + while (argc > 1) { + const char *arg; + + argc--; argv++; + arg = *argv; + + if (*arg != '-') + usage(1, arg0); + + switch (arg[1]) { + case 'L': if (argc < 2) usage(1, arg0); listen_str = *++argv; argc--; break; + case 't': if (argc < 2) usage(1, arg0); target_str = *++argv; argc--; break; + case 'r': if (argc < 2) usage(1, arg0); nb_requests = atoi(*++argv); argc--; break; + case 'c': if (argc < 2) usage(1, arg0); concurrency = atoi(*++argv); argc--; break; + case 'n': if (argc < 2) usage(1, arg0); nb_tokens = atol(*++argv); argc--; break; + case 'T': if (argc < 2) usage(1, arg0); think_ms = atof(*++argv); argc--; break; + case 'd': if (argc < 2) usage(1, arg0); delay_ms = atof(*++argv); argc--; break; + case 'w': if (argc < 2) usage(1, arg0); timeout_ms = atol(*++argv); argc--; break; + case 'l': use_loopback = 0; break; + case 'q': quiet = 1; break; + case 'v': verbose++; break; + case 'h': usage(0, arg0); break; + default : usage(1, arg0); break; + } + } + + if (nb_requests < 1) + nb_requests = 1; + if (concurrency < 1) + concurrency = 1; + if (concurrency > nb_requests) + concurrency = nb_requests; + + if (addr_to_ss(listen_str, &listen_ss) < 0) + usage(1, arg0); + + listener_fd = socket(listen_ss.ss_family, SOCK_STREAM, 0); + if (listener_fd < 0) + die(1, "socket(): %s\n", strerror(errno)); + { + int one = 1; + setsockopt(listener_fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); + } + fcntl(listener_fd, F_SETFL, fcntl(listener_fd, F_GETFL, 0) | O_NONBLOCK); + + if (bind(listener_fd, (struct sockaddr *)&listen_ss, sizeof(listen_ss)) < 0) + die(1, "bind(%s): %s\n", listen_str, strerror(errno)); + + if (listen(listener_fd, 1024) < 0) + die(1, "listen(): %s\n", strerror(errno)); + + fprintf(stderr, "dummy OpenAI-like SSE server listening on %s (POST /v1/chat/completions)\n", + addr_to_str(&listen_ss)); + + if (!target_str) { + fprintf(stderr, "no -t given: running as server only. Ctrl-C to stop.\n"); + event_loop(NULL); + return 0; + } + + if (addr_to_ss(target_str, &target_ss) < 0) + usage(1, arg0); + + /* a bare port or a wildcard address means loopback for a destination */ + addr_force_loopback(&target_ss); + + /* the loopback calibration always connects to a local address */ + loop_ss = listen_ss; + addr_force_loopback(&loop_ss); + + has_loop = use_loopback && !addr_eq(&loop_ss, &target_ss); + if (!has_loop && use_loopback) + fprintf(stderr, "note: -t equals -L, skipping loopback calibration\n"); + + if (has_loop) { + /* The calibration only establishes the noise floor, it doesn't + * need the full load: a few requests with shorter responses are + * enough. At least one full concurrency wave is used so that + * the measurements are made in comparable conditions. + */ + calib_req = concurrency > 3 ? concurrency : 3; + if (calib_req > nb_requests) + calib_req = nb_requests; + calib_tokens = nb_tokens < 200 ? nb_tokens : 200; + + loop_stats = run_batch(&loop_ss, "loop", calib_req, calib_tokens); + } + + gw_stats = run_batch(&target_ss, "gw", nb_requests, 0); + + if (has_loop) + print_batch("loopback (direct to listen address)", loop_stats); + print_batch("gateway", gw_stats); + + if (has_loop) { + printf("\n=== gateway overhead (gateway minus loopback; negative = loopback noise) ===\n"); + delta_line("emit->recv (1st token)", &gw_stats->emit_ttft, &loop_stats->emit_ttft); + delta_line("emit->recv (tokens)", &gw_stats->emits, &loop_stats->emits); + delta_line("TTFT (client)", &gw_stats->ttft, &loop_stats->ttft); + delta_line("inter-token (client)", &gw_stats->gaps, &loop_stats->gaps); + } + + return gw_stats->fail_req > 0 ? 1 : 0; +}