diff --git a/internal/vllmbench/http_load.go b/internal/vllmbench/http_load.go index 8e3258d..eacdc31 100644 --- a/internal/vllmbench/http_load.go +++ b/internal/vllmbench/http_load.go @@ -114,8 +114,9 @@ type openAIChoice struct { } type openAIMessage struct { - Role string `json:"role,omitempty"` - Content string `json:"content,omitempty"` + Role string `json:"role,omitempty"` + Content string `json:"content,omitempty"` + ReasoningContent string `json:"reasoning_content,omitempty"` } type openAIUsage struct { @@ -610,18 +611,25 @@ func (stream *httpStreamResult) consume(body io.Reader) *httpLoadFailure { } } stream.completedAt = time.Now().UTC() - if err := scanner.Err(); err != nil { - return newHTTPLoadFailure("response_read", "", err.Error(), stream.completedAt, nil) - } - // EOF before [DONE] is a truncated stream: recording it as completed - // would silently keep partial output and fallback token counts. - if !terminated { + return stream.finalize(terminated, scanner.Err()) +} + +// finalize validates a fully-read stream. EOF before [DONE] is a truncated +// stream (recording it would keep partial output). A stream that finished +// cleanly is valid even with zero streamed tokens — a 1-token prefill point +// or a reasoning model that emitted only a final message; only a stream with +// neither any token nor a finish reason is malformed. +func (stream *httpStreamResult) finalize(terminated bool, scanErr error) *httpLoadFailure { + switch { + case scanErr != nil: + return newHTTPLoadFailure("response_read", "", scanErr.Error(), stream.completedAt, nil) + case !terminated: return newHTTPLoadFailure("response_read", "", "stream ended before [DONE] terminator", stream.completedAt, nil) + case stream.firstTokenAt == nil && stream.finishReason == "": + return newHTTPLoadFailure("response_shape", "", "stream produced neither tokens nor a finish reason", stream.completedAt, nil) + default: + return nil } - if stream.firstTokenAt == nil { - return newHTTPLoadFailure("response_shape", "", "stream produced no completion content", stream.completedAt, nil) - } - return nil } func ssePayload(line string) (string, bool) { @@ -669,11 +677,18 @@ func (stream *httpStreamResult) applyChoice(choice openAIStreamChoice) { stream.content.WriteString(text) } -// streamChoiceText extracts the token text of one stream choice: chat chunks -// carry a delta message, completions chunks carry plain text. +// streamChoiceText extracts the token text of one stream choice. Chat chunks +// carry a delta message; reasoning models stream their first tokens as +// reasoning_content before any content, so both count for TTFT/ITL timing. +// Completions chunks carry plain text. func streamChoiceText(choice openAIStreamChoice) string { - if choice.Delta != nil && choice.Delta.Content != "" { - return choice.Delta.Content + if choice.Delta != nil { + if choice.Delta.Content != "" { + return choice.Delta.Content + } + if choice.Delta.ReasoningContent != "" { + return choice.Delta.ReasoningContent + } } return choice.Text } @@ -687,15 +702,24 @@ func (stream *httpStreamResult) applyToSample(sample RequestSample, request Cano if stream.firstByteAt != nil { sample.FirstByteMillis = stream.firstByteAt.Sub(sample.StartedAt).Seconds() * 1000 } - sample.TTFTMillis = stream.firstTokenAt.Sub(sample.StartedAt).Seconds() * 1000 sample.PromptTokens = usageInt(stream.usage.PromptTokens, request.InputTokensExpected) - sample.CompletionTokens = usageInt(stream.usage.CompletionTokens, request.OutputTokensExpected) + // Completion tokens fall back to the observed streamed-chunk count, never + // to the requested output length: a backend that omits usage on an empty + // stream must not be credited phantom output tokens. + sample.CompletionTokens = usageInt(stream.usage.CompletionTokens, stream.tokenChunks) sample.TotalTokens = usageInt(stream.usage.TotalTokens, sample.PromptTokens+sample.CompletionTokens) - if sample.CompletionTokens > 1 { - sample.TPOTMillis = stream.completedAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(sample.CompletionTokens-1) - } - if stream.tokenChunks > 1 { - sample.ITLMeanMillis = stream.lastTokenAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(stream.tokenChunks-1) + // firstTokenAt is nil only for a clean finish that streamed no token + // (accepted above); such a point contributes no TTFT/TPOT/ITL, which is + // honest rather than a fabricated zero. Token counts are assigned first + // so TPOT sees the real completion-token count. + if stream.firstTokenAt != nil { + sample.TTFTMillis = stream.firstTokenAt.Sub(sample.StartedAt).Seconds() * 1000 + if sample.CompletionTokens > 1 { + sample.TPOTMillis = stream.completedAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(sample.CompletionTokens-1) + } + if stream.tokenChunks > 1 { + sample.ITLMeanMillis = stream.lastTokenAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(stream.tokenChunks-1) + } } sample.ResponseSHA256 = sha256Hex([]byte(stream.content.String())) sample.ResponseMetadata = streamResponseMetadata(stream) diff --git a/internal/vllmbench/http_stream_test.go b/internal/vllmbench/http_stream_test.go index 3a0e43d..5c8951d 100644 --- a/internal/vllmbench/http_stream_test.go +++ b/internal/vllmbench/http_stream_test.go @@ -97,6 +97,9 @@ func TestStreamingBenchmarkMeasuresTTFT(t *testing.T) { if sample.ITLMeanMillis <= 0 { t.Fatalf("sample ITL = %.3f, want positive from chunk gaps", sample.ITLMeanMillis) } + if sample.TPOTMillis <= 0 { + t.Fatalf("sample TPOT = %.3f, want positive for a multi-token completion", sample.TPOTMillis) + } if sample.PromptTokens != 32 || sample.CompletionTokens != 4 { t.Fatalf("sample tokens = %d/%d, want 32/4 from stream usage", sample.PromptTokens, sample.CompletionTokens) } @@ -136,21 +139,66 @@ func TestNonStreamingBenchmarkRecordsNoTTFT(t *testing.T) { } } -func TestStreamWithNoContentFailsShape(t *testing.T) { - server := sseTestServer(t, 0, 0, 0) +func TestStreamWithNeitherTokenNorFinishFailsShape(t *testing.T) { + // A stream that ends ([DONE]) having emitted neither a token nor a + // finish_reason is genuinely malformed and must fail. (A clean finish + // with zero content is valid — see TestStreamingFinishWithNoContentCompletes.) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fl := w.(http.Flusher) + fmt.Fprint(w, "data: {\"id\":\"x\",\"choices\":[{\"delta\":{}}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + fl.Flush() + })) defer server.Close() result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(server.URL, nil)) if err != nil { t.Fatal(err) } if result.Failed != 2 { - t.Fatalf("failed = %d, want 2 for empty streams", result.Failed) + t.Fatalf("failed = %d, want 2 for streams with no token and no finish reason", result.Failed) } if result.RequestSamples[0].ErrorType != "response_shape" { t.Fatalf("error type = %q, want response_shape", result.RequestSamples[0].ErrorType) } } +func TestStreamWithoutUsageDoesNotFabricateTokens(t *testing.T) { + // A backend that ignores stream_options.include_usage: content streams + // but no usage chunk arrives. Completion tokens must reflect observed + // chunks, never the requested output length (which would inflate + // throughput). Empty-but-finished streams must land at zero. + makeServer := func(contentChunks int) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fl := w.(http.Flusher) + for i := 0; i < contentChunks; i++ { + fmt.Fprintf(w, "data: {\"id\":\"u\",\"choices\":[{\"delta\":{\"content\":\"t\"}}]}\n\n") + } + fmt.Fprint(w, "data: {\"id\":\"u\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + fl.Flush() + })) + } + // request.OutputTokensExpected is 8 (RandomOutputLen); assert we never see it. + for _, tc := range []struct{ chunks, wantTokens int }{{0, 0}, {3, 3}} { + srv := makeServer(tc.chunks) + result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(srv.URL, nil)) + srv.Close() + if err != nil { + t.Fatalf("chunks=%d: %v", tc.chunks, err) + } + if result.Completed != 2 { + t.Fatalf("chunks=%d: completed=%d, want 2", tc.chunks, result.Completed) + } + for _, s := range result.RequestSamples { + if s.CompletionTokens != tc.wantTokens { + t.Fatalf("chunks=%d: completion_tokens=%d, want %d (observed chunks, not the requested 8)", tc.chunks, s.CompletionTokens, tc.wantTokens) + } + } + } +} + func TestExtraBodyCannotFlipStreaming(t *testing.T) { client := openAIHTTPClient{ profile: Profile{Model: "m"}, @@ -294,3 +342,73 @@ func TestStreamErrorChunkFailsRequest(t *testing.T) { t.Fatalf("error = %q/%q, want server_error/boom", sample.ErrorType, sample.ErrorMessage) } } + +func TestStreamingReasoningContentCountsForTTFT(t *testing.T) { + // A reasoning model streams its first tokens as reasoning_content, then + // a normal content token, then finishes. TTFT must be observed from the + // first reasoning token, not lost. + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fl := w.(http.Flusher) + time.Sleep(20 * time.Millisecond) + fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n") + fl.Flush() + time.Sleep(5 * time.Millisecond) + fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n") + fl.Flush() + fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n") + fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[],\"usage\":{\"prompt_tokens\":32,\"completion_tokens\":2,\"total_tokens\":34}}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + fl.Flush() + })) + defer server.Close() + result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(server.URL, nil)) + if err != nil { + t.Fatal(err) + } + if result.Completed != 2 || result.TTFTSource != TTFTSourceStream { + t.Fatalf("completed=%d ttft_source=%q, want 2 / stream", result.Completed, result.TTFTSource) + } + for _, s := range result.RequestSamples { + if s.TTFTMillis <= 0 { + t.Fatalf("TTFT=%.1f, want reasoning_content to seed TTFT", s.TTFTMillis) + } + if s.ITLMeanMillis <= 0 { + t.Fatalf("ITL=%.3f, want gap between reasoning and content tokens", s.ITLMeanMillis) + } + } +} + +func TestStreamingFinishWithNoContentCompletes(t *testing.T) { + // A 1-token prefill / empty-generation stream that finishes cleanly with + // zero streamed tokens must complete, not fail response_shape. It carries + // no TTFT (honest), and the request still counts as completed. + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fl := w.(http.Flusher) + fmt.Fprint(w, "data: {\"id\":\"p1\",\"choices\":[{\"delta\":{},\"finish_reason\":\"length\"}]}\n\n") + fl.Flush() + fmt.Fprint(w, "data: {\"id\":\"p1\",\"choices\":[],\"usage\":{\"prompt_tokens\":16000,\"completion_tokens\":1,\"total_tokens\":16001}}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + fl.Flush() + })) + defer server.Close() + result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(server.URL, nil)) + if err != nil { + t.Fatal(err) + } + if result.Completed != 2 || result.Failed != 0 { + t.Fatalf("completed/failed=%d/%d, want 2/0: a clean finish with no content is valid", result.Completed, result.Failed) + } + for _, s := range result.RequestSamples { + if s.Status != "completed" || !s.Streamed { + t.Fatalf("sample status/streamed=%s/%v, want completed/true", s.Status, s.Streamed) + } + if s.TTFTMillis != 0 { + t.Fatalf("TTFT=%.1f, want 0 (no token was streamed)", s.TTFTMillis) + } + } + if result.TTFTSource != "" { + t.Fatalf("ttft_source=%q, want empty when no point produced a streamed token", result.TTFTSource) + } +}