Comments, doc comments, test names, error and log messages, CLI help, README, Makefile, CI step names and the design document are now all in English. No behaviour changes; the full suite passes under the race detector. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
b7417e1a69
commit
28029eb43b
21 files changed
+910
-869
No files matched your search
@@ -16,15 +16,15 @@ jobs:
|
||||
go-version: "1.26"
|
||||
cache: true
|
||||
|
||||
- name: Biçim kontrolü
|
||||
- name: Check formatting
|
||||
run: |
|
||||
out=$(gofmt -l .)
|
||||
if [ -n "$out" ]; then echo "gofmt gerekli:"; echo "$out"; exit 1; fi
|
||||
if [ -n "$out" ]; then echo "gofmt needed on:"; echo "$out"; exit 1; fi
|
||||
|
||||
- name: Vet
|
||||
run: go vet ./...
|
||||
|
||||
- name: Test (race detector ile)
|
||||
- name: Test with the race detector
|
||||
run: go test -race ./...
|
||||
|
||||
- name: Build
|
||||
|
||||
+1
-1
@@ -20,7 +20,7 @@ linters:
|
||||
|
||||
exclusions:
|
||||
rules:
|
||||
# Testlerde okunabilirlik uğruna gevşetilenler.
|
||||
# Relaxed in tests for readability.
|
||||
- path: _test\.go
|
||||
linters: [bodyclose, noctx, errorlint]
|
||||
|
||||
|
||||
@@ -23,9 +23,9 @@ fmt:
|
||||
clean:
|
||||
rm -rf bin dist
|
||||
|
||||
# Modül yolunu değiştirmek için: make rename NEW=git.ornek.com/ekip/llmgw
|
||||
# Change the module path: make rename NEW=git.example.com/team/llmgw
|
||||
rename:
|
||||
@test -n "$(NEW)" || (echo "kullanim: make rename NEW=<yeni/modul/yolu>"; exit 1)
|
||||
@test -n "$(NEW)" || (echo "usage: make rename NEW=<new/module/path>"; exit 1)
|
||||
go mod edit -module=$(NEW)
|
||||
grep -rl '$(MODULE)' --include='*.go' . | xargs sed -i '' 's|$(MODULE)|$(NEW)|g'
|
||||
sed -i '' 's|^MODULE := .*|MODULE := $(NEW)|' Makefile
|
||||
|
||||
@@ -1,58 +1,62 @@
|
||||
# llmgw
|
||||
|
||||
Çoklu InfiniBand fabric ve heterojen GPU için tek binary LLM gateway.
|
||||
A single-binary LLM gateway for clusters with multiple InfiniBand fabrics and
|
||||
heterogeneous GPUs.
|
||||
|
||||
Kendi GPU sunucularında çalışan LLM'leri (vLLM / SGLang) tek bir OpenAI uyumlu
|
||||
uç noktadan sunar. Tasarım dokümanı: [llmgw-plan.md](llmgw-plan.md).
|
||||
It serves LLMs running on your own GPU servers (vLLM / SGLang) behind one
|
||||
OpenAI-compatible endpoint. Design document: [llmgw-plan.md](llmgw-plan.md).
|
||||
|
||||
## Durum
|
||||
## Status
|
||||
|
||||
| Faz | Kapsam | Durum |
|
||||
| Phase | Scope | State |
|
||||
|---|---|---|
|
||||
| 0 | İskelet + sim-engine | ✅ |
|
||||
| 1 | Statik router (filter/score/pick, streaming proxy) | ⬜ |
|
||||
| 1.5 | Admission control (öncelik bantları) | ⬜ |
|
||||
| 2 | Agent: GPU/IB keşfi, tek-node engine | ⬜ |
|
||||
| 3 | Control plane: katalog, scheduler, reconciler | ⬜ |
|
||||
| 4 | Multi-node & fabric farkındalığı | ⬜ |
|
||||
| 5 | Tenancy, prefix affinity, bench | ⬜ |
|
||||
| 0 | Skeleton and sim-engine | done |
|
||||
| 1 | Static router (filter/score/pick, streaming proxy) | in progress |
|
||||
| 1.5 | Admission control (priority bands) | todo |
|
||||
| 2 | Agent: GPU/IB discovery, single-node engines | todo |
|
||||
| 3 | Control plane: catalog, scheduler, reconciler | todo |
|
||||
| 4 | Multi-node placement and fabric awareness | todo |
|
||||
| 5 | Tenancy, prefix affinity, benchmarking | todo |
|
||||
|
||||
## Kurulum
|
||||
## Building
|
||||
|
||||
```sh
|
||||
make build # bin/llmgw
|
||||
make test # birim testleri
|
||||
make race # yarış dedektörüyle
|
||||
make test # unit tests
|
||||
make race # with the race detector
|
||||
```
|
||||
|
||||
## sim-engine
|
||||
|
||||
Gerçek bir inference engine'in **router açısından önemli** davranışlarını simüle
|
||||
eder; GPU olmadan geliştirme ve test yapabilmek için. Modellediği şeyler:
|
||||
Simulates the inference-engine behaviours **the router cares about**, so the
|
||||
router can be developed and tested without GPUs. What it models:
|
||||
|
||||
- continuous batching (sınırlı eşzamanlı dizi sayısı)
|
||||
- KV cache blok muhasebesi; blok yetmediğinde kuyruğa alma (head-of-line blocking dahil)
|
||||
- otomatik prefix cache (zincirlenmiş blok hash'leri + LRU) ve cache hit'in TTFT'ye etkisi
|
||||
- batch büyüdükçe token başına gecikmenin artması
|
||||
- istemci koptuğunda isteğin batch'ten düşürülüp blokların bırakılması
|
||||
- vLLM metrik adlarıyla `/metrics`
|
||||
- continuous batching (bounded number of concurrent sequences)
|
||||
- KV cache block accounting, queueing once blocks run out (head-of-line
|
||||
blocking included)
|
||||
- automatic prefix caching (chained block hashes plus an LRU) and its effect
|
||||
on time to first token
|
||||
- per-token latency growing with batch size
|
||||
- dropping a request and freeing its blocks when the client disconnects
|
||||
- `/metrics` using vLLM's metric names
|
||||
|
||||
```sh
|
||||
./bin/llmgw sim-engine --listen :8000 --model llama-3.3-70b
|
||||
|
||||
curl -N localhost:8000/v1/chat/completions \
|
||||
-H 'content-type: application/json' \
|
||||
-d '{"messages":[{"role":"user","content":"merhaba"}],"max_tokens":8,"stream":true}'
|
||||
-d '{"messages":[{"role":"user","content":"hello"}],"max_tokens":8,"stream":true}'
|
||||
```
|
||||
|
||||
Yük/gecikme davranışını ayarlamak için: `--max-num-seqs`, `--num-gpu-blocks`,
|
||||
`--prefill-tps`, `--decode-tps`, `--batch-contention`, `--prefix-cache-blocks`.
|
||||
Kaos testleri için: `--fail-rate`, deterministik koşu için `--seed`.
|
||||
Tune load and latency behaviour with `--max-num-seqs`, `--num-gpu-blocks`,
|
||||
`--prefill-tps`, `--decode-tps`, `--batch-contention` and
|
||||
`--prefix-cache-blocks`. For chaos testing use `--fail-rate`, and `--seed` for
|
||||
deterministic runs.
|
||||
|
||||
### Engine metrik kontratı
|
||||
### The engine metric contract
|
||||
|
||||
Router'ın yönlendirme yapabilmesi için bir engine'in vermesi **zorunlu** üç metrik
|
||||
(plan §7.6). sim-engine bunları gerçek vLLM adlarıyla yayınlar:
|
||||
For the router to make routing decisions an engine **must** expose these three
|
||||
metrics (plan §7.6). sim-engine publishes them under vLLM's real names:
|
||||
|
||||
```
|
||||
vllm:num_requests_running
|
||||
@@ -60,10 +64,17 @@ vllm:num_requests_waiting
|
||||
vllm:kv_cache_usage_perc
|
||||
```
|
||||
|
||||
## Modül yolu
|
||||
Adding support for a new engine means satisfying that contract. Metric names
|
||||
drift between versions, so the name mapping is configurable.
|
||||
|
||||
`gitea.mynodes.xyz/ilker/llmgw`. Değiştirmek için:
|
||||
## Module path
|
||||
|
||||
`gitea.mynodes.xyz/ilker/llmgw`. To change it:
|
||||
|
||||
```sh
|
||||
make rename NEW=git.ornek.com/ekip/llmgw
|
||||
make rename NEW=git.example.com/team/llmgw
|
||||
```
|
||||
|
||||
## Language
|
||||
|
||||
Code, comments, tests, commit messages and documentation are in English.
|
||||
+3
-3
@@ -1,5 +1,5 @@
|
||||
// Command llmgw, tek binary halinde router, control plane, agent ve yardımcı
|
||||
// araçları barındırır. Hangi rolde çalışacağı alt komutla belirlenir.
|
||||
// Command llmgw bundles the router, control plane, agent and supporting tools
|
||||
// into a single binary. The subcommand selects which role it runs as.
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -11,7 +11,7 @@ import (
|
||||
|
||||
func main() {
|
||||
if err := cli.Root().Execute(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "hata:", err)
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
// Package cli, llmgw'nin komut satırı yüzeyini kurar.
|
||||
// Package cli builds llmgw's command line surface.
|
||||
package cli
|
||||
|
||||
import (
|
||||
@@ -17,13 +17,13 @@ type globalFlags struct {
|
||||
|
||||
var global globalFlags
|
||||
|
||||
// Root, kök komutu ve tüm alt komutları döner.
|
||||
// Root returns the root command with all subcommands attached.
|
||||
func Root() *cobra.Command {
|
||||
root := &cobra.Command{
|
||||
Use: "llmgw",
|
||||
Short: "Çoklu IB fabric ve heterojen GPU için LLM gateway",
|
||||
Long: "llmgw, kendi GPU sunucularında çalışan LLM'leri tek bir OpenAI uyumlu\n" +
|
||||
"uç noktadan sunar. Tek binary, alt komutla farklı roller.",
|
||||
Short: "LLM gateway for multi-fabric InfiniBand and heterogeneous GPU clusters",
|
||||
Long: "llmgw serves LLMs running on your own GPU servers behind a single\n" +
|
||||
"OpenAI-compatible endpoint. One binary, different roles per subcommand.",
|
||||
Version: version.Version,
|
||||
SilenceUsage: true,
|
||||
SilenceErrors: true,
|
||||
@@ -33,8 +33,8 @@ func Root() *cobra.Command {
|
||||
}
|
||||
|
||||
pf := root.PersistentFlags()
|
||||
pf.StringVar(&global.logLevel, "log-level", "info", "log seviyesi: debug|info|warn|error")
|
||||
pf.StringVar(&global.logFormat, "log-format", "text", "log biçimi: text|json")
|
||||
pf.StringVar(&global.logLevel, "log-level", "info", "log level: debug|info|warn|error")
|
||||
pf.StringVar(&global.logFormat, "log-format", "text", "log format: text|json")
|
||||
|
||||
root.AddCommand(
|
||||
newServerCmd(),
|
||||
|
||||
+19
-19
@@ -20,10 +20,10 @@ func newSimEngineCmd() *cobra.Command {
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "sim-engine",
|
||||
Short: "OpenAI uyumlu sahte inference engine (test için)",
|
||||
Long: "Gerçek bir engine'in router açısından önemli davranışlarını simüle eder:\n" +
|
||||
"continuous batching, KV blok muhasebesi, prefix cache, vLLM formatında /metrics.\n" +
|
||||
"Router'ı GPU olmadan test etmek için kullanılır.",
|
||||
Short: "OpenAI-compatible fake inference engine (for testing)",
|
||||
Long: "Simulates the engine behaviours that matter to the router:\n" +
|
||||
"continuous batching, KV block accounting, prefix caching, and vLLM-shaped\n" +
|
||||
"/metrics. Lets the router be developed and tested without GPUs.",
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
log := logger()
|
||||
eng := sim.New(cfg)
|
||||
@@ -39,7 +39,7 @@ func newSimEngineCmd() *cobra.Command {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Info("sim-engine dinliyor",
|
||||
log.Info("sim-engine listening",
|
||||
"addr", ln.Addr().String(),
|
||||
"model", cfg.Model,
|
||||
"max_num_seqs", cfg.MaxNumSeqs,
|
||||
@@ -60,7 +60,7 @@ func newSimEngineCmd() *cobra.Command {
|
||||
}
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
log.Info("kapanıyor")
|
||||
log.Info("shutting down")
|
||||
shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
return srv.Shutdown(shutCtx)
|
||||
@@ -69,19 +69,19 @@ func newSimEngineCmd() *cobra.Command {
|
||||
}
|
||||
|
||||
f := cmd.Flags()
|
||||
f.StringVar(&listen, "listen", ":8000", "dinlenecek adres")
|
||||
f.StringVar(&cfg.Model, "model", cfg.Model, "sunulan model adı")
|
||||
f.IntVar(&cfg.MaxNumSeqs, "max-num-seqs", cfg.MaxNumSeqs, "eşzamanlı dizi sayısı (batch kapasitesi)")
|
||||
f.IntVar(&cfg.BlockSize, "block-size", cfg.BlockSize, "KV blok başına token")
|
||||
f.IntVar(&cfg.NumGPUBlocks, "num-gpu-blocks", cfg.NumGPUBlocks, "toplam KV cache bloğu")
|
||||
f.IntVar(&cfg.PrefixCacheBlocks, "prefix-cache-blocks", cfg.PrefixCacheBlocks, "prefix cache blok kapasitesi (0=kapalı)")
|
||||
f.Float64Var(&cfg.PrefillTokensPerSec, "prefill-tps", cfg.PrefillTokensPerSec, "prefill hızı (token/sn, engine geneli)")
|
||||
f.Float64Var(&cfg.DecodeTokensPerSec, "decode-tps", cfg.DecodeTokensPerSec, "tek dizi decode hızı (token/sn)")
|
||||
f.Float64Var(&cfg.BatchContention, "batch-contention", cfg.BatchContention, "batch'teki her ek dizinin yavaşlatma katsayısı")
|
||||
f.IntVar(&cfg.DefaultOutputTokens, "default-output-tokens", cfg.DefaultOutputTokens, "max_tokens verilmezse üretilecek token")
|
||||
f.DurationVar(&cfg.Tick, "tick", cfg.Tick, "simülasyon adımı")
|
||||
f.Float64Var(&cfg.FailRate, "fail-rate", 0, "isteklerin yapay hata oranı (kaos testi)")
|
||||
f.Int64Var(&cfg.Seed, "seed", 0, "rastgelelik tohumu (0=zaman)")
|
||||
f.StringVar(&listen, "listen", ":8000", "address to listen on")
|
||||
f.StringVar(&cfg.Model, "model", cfg.Model, "model name to serve")
|
||||
f.IntVar(&cfg.MaxNumSeqs, "max-num-seqs", cfg.MaxNumSeqs, "concurrent sequences (continuous batch capacity)")
|
||||
f.IntVar(&cfg.BlockSize, "block-size", cfg.BlockSize, "tokens per KV cache block")
|
||||
f.IntVar(&cfg.NumGPUBlocks, "num-gpu-blocks", cfg.NumGPUBlocks, "total KV cache blocks")
|
||||
f.IntVar(&cfg.PrefixCacheBlocks, "prefix-cache-blocks", cfg.PrefixCacheBlocks, "prefix cache capacity in blocks (0 disables)")
|
||||
f.Float64Var(&cfg.PrefillTokensPerSec, "prefill-tps", cfg.PrefillTokensPerSec, "prefill rate in tokens/sec, shared engine-wide")
|
||||
f.Float64Var(&cfg.DecodeTokensPerSec, "decode-tps", cfg.DecodeTokensPerSec, "single-sequence decode rate in tokens/sec")
|
||||
f.Float64Var(&cfg.BatchContention, "batch-contention", cfg.BatchContention, "per-extra-sequence decode slowdown factor")
|
||||
f.IntVar(&cfg.DefaultOutputTokens, "default-output-tokens", cfg.DefaultOutputTokens, "tokens generated when max_tokens is absent")
|
||||
f.DurationVar(&cfg.Tick, "tick", cfg.Tick, "simulation step")
|
||||
f.Float64Var(&cfg.FailRate, "fail-rate", 0, "fraction of requests failed artificially (chaos testing)")
|
||||
f.Int64Var(&cfg.Seed, "seed", 0, "random seed (0 = time-based)")
|
||||
|
||||
return cmd
|
||||
}
|
||||
+12
-12
@@ -6,19 +6,19 @@ import (
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
// errNotImplemented, henüz yazılmamış fazların komutları için.
|
||||
var errNotImplemented = errors.New("bu komut henüz uygulanmadı (bkz. llmgw-plan.md §11)")
|
||||
// errNotImplemented marks commands belonging to phases not yet built.
|
||||
var errNotImplemented = errors.New("not implemented yet (see llmgw-plan.md §11)")
|
||||
|
||||
func newServerCmd() *cobra.Command {
|
||||
var configPath string
|
||||
cmd := &cobra.Command{
|
||||
Use: "server",
|
||||
Short: "Router + control plane (Faz 1-3)",
|
||||
Short: "Router and control plane (phases 1-3)",
|
||||
RunE: func(_ *cobra.Command, _ []string) error {
|
||||
return errNotImplemented
|
||||
},
|
||||
}
|
||||
cmd.Flags().StringVar(&configPath, "config", "server.yaml", "konfigürasyon dosyası")
|
||||
cmd.Flags().StringVar(&configPath, "config", "server.yaml", "configuration file")
|
||||
return cmd
|
||||
}
|
||||
|
||||
@@ -26,26 +26,26 @@ func newAgentCmd() *cobra.Command {
|
||||
var configPath string
|
||||
cmd := &cobra.Command{
|
||||
Use: "agent",
|
||||
Short: "GPU node'unda çalışan agent (Faz 2)",
|
||||
Short: "Node agent running on each GPU host (phase 2)",
|
||||
RunE: func(_ *cobra.Command, _ []string) error {
|
||||
return errNotImplemented
|
||||
},
|
||||
}
|
||||
cmd.Flags().StringVar(&configPath, "config", "agent.yaml", "konfigürasyon dosyası")
|
||||
cmd.Flags().StringVar(&configPath, "config", "agent.yaml", "configuration file")
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newCtlCmd() *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: "ctl",
|
||||
Short: "Yönetim CLI'ı (Faz 3)",
|
||||
Short: "Administrative CLI (phase 3)",
|
||||
}
|
||||
for _, sub := range []struct{ use, short string }{
|
||||
{"nodes", "node'ları listele"},
|
||||
{"models", "model kataloğunu göster"},
|
||||
{"replicas", "replikaları listele"},
|
||||
{"explain-placement", "bir modelin neden yerleştirilemediğini açıkla"},
|
||||
{"bench", "profil kapasitesini ölç (§10)"},
|
||||
{"nodes", "list nodes"},
|
||||
{"models", "show the model catalog"},
|
||||
{"replicas", "list replicas"},
|
||||
{"explain-placement", "explain why a model could not be placed"},
|
||||
{"bench", "measure profile capacity (§10)"},
|
||||
} {
|
||||
cmd.AddCommand(&cobra.Command{
|
||||
Use: sub.use,
|
||||
|
||||
+3
-3
@@ -1,4 +1,4 @@
|
||||
// Package obs, log ve metrik kurulumunu tek yerden yapar.
|
||||
// Package obs sets up logging and metrics in one place.
|
||||
package obs
|
||||
|
||||
import (
|
||||
@@ -7,13 +7,13 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Options, log kurulumunu belirler.
|
||||
// Options controls logger setup.
|
||||
type Options struct {
|
||||
Level string // debug, info, warn, error
|
||||
Format string // text, json
|
||||
}
|
||||
|
||||
// Setup, slog varsayılan logger'ını kurar ve onu döner.
|
||||
// Setup installs the default slog logger and returns it.
|
||||
func Setup(o Options) *slog.Logger {
|
||||
var lvl slog.Level
|
||||
switch strings.ToLower(o.Level) {
|
||||
|
||||
+22
-22
@@ -6,17 +6,17 @@ import (
|
||||
"hash/fnv"
|
||||
)
|
||||
|
||||
// BlockHash, KV cache'in muhasebe birimi olan bir bloğun kimliği.
|
||||
// BlockHash identifies one KV cache block, the unit of cache accounting.
|
||||
type BlockHash uint64
|
||||
|
||||
// BlockHashes, token dizisi için zincirlenmiş blok hash'leri üretir.
|
||||
// BlockHashes computes chained block hashes over a token sequence.
|
||||
//
|
||||
// Zincirleme (h_i = hash(h_{i-1}, block_i)) vLLM'in yaptığının aynısıdır ve
|
||||
// şart: blok kimliği sadece kendi içeriğini değil, kendisinden önce gelen tüm
|
||||
// blokları da temsil etmeli. Aksi halde "aynı cümleyi farklı bağlamda içeren"
|
||||
// iki istek yanlışlıkla cache hit sayılır.
|
||||
// Chaining (h_i = hash(h_{i-1}, block_i)) mirrors what vLLM does and is
|
||||
// required: a block's identity must represent not only its own contents but
|
||||
// every block before it. Otherwise two requests containing the same sentence
|
||||
// in different contexts would falsely count as a cache hit.
|
||||
//
|
||||
// Son eksik blok yok sayılır — yarım blok cache'lenmez.
|
||||
// A trailing partial block is ignored — half blocks are never cached.
|
||||
func BlockHashes(tokens []Token, blockSize int) []BlockHash {
|
||||
if blockSize <= 0 || len(tokens) < blockSize {
|
||||
return nil
|
||||
@@ -39,13 +39,12 @@ func BlockHashes(tokens []Token, blockSize int) []BlockHash {
|
||||
return out
|
||||
}
|
||||
|
||||
// PrefixCache, engine'in otomatik prefix cache'ini modelleyen blok bazlı LRU.
|
||||
// PrefixCache models an engine's automatic prefix cache as a block-level LRU.
|
||||
//
|
||||
// Gerçek engine'de bu bir radix tree'dir; ama yönlendirme kalitesini ölçmek için
|
||||
// önemli olan davranış "hangi blokların hâlâ cache'te olduğu"dur ve onu LRU
|
||||
// doğru modeller.
|
||||
// A real engine uses a radix tree, but what matters for measuring routing
|
||||
// quality is which blocks are still resident, and LRU models that faithfully.
|
||||
type PrefixCache struct {
|
||||
capacity int // blok cinsinden
|
||||
capacity int // in blocks
|
||||
ll *list.List
|
||||
m map[BlockHash]*list.Element
|
||||
|
||||
@@ -53,8 +52,8 @@ type PrefixCache struct {
|
||||
hits int64
|
||||
}
|
||||
|
||||
// NewPrefixCache, verilen blok kapasitesinde bir cache kurar.
|
||||
// capacity <= 0 ise cache devre dışıdır (her zaman miss).
|
||||
// NewPrefixCache creates a cache holding at most capacity blocks.
|
||||
// A capacity of zero or less disables the cache (every lookup misses).
|
||||
func NewPrefixCache(capacity int) *PrefixCache {
|
||||
return &PrefixCache{
|
||||
capacity: capacity,
|
||||
@@ -63,14 +62,14 @@ func NewPrefixCache(capacity int) *PrefixCache {
|
||||
}
|
||||
}
|
||||
|
||||
// Enabled, cache'in açık olup olmadığını söyler.
|
||||
// Enabled reports whether the cache is active.
|
||||
func (c *PrefixCache) Enabled() bool { return c != nil && c.capacity > 0 }
|
||||
|
||||
// Lookup, hashes dizisinin cache'te bulunan **kesintisiz ön ek** uzunluğunu
|
||||
// blok cinsinden döner ve bulunan blokları LRU'da tazeler.
|
||||
// Lookup returns the length, in blocks, of the longest **contiguous prefix**
|
||||
// of hashes present in the cache, refreshing those blocks in the LRU.
|
||||
//
|
||||
// Kesintisiz olması önemli: ortadaki bir blok düşmüşse ondan sonrası da
|
||||
// yeniden hesaplanmak zorundadır, çünkü attention önceki tüm bloklara bağlıdır.
|
||||
// Contiguity matters: if a block in the middle was evicted, everything after
|
||||
// it must be recomputed too, because attention depends on all preceding blocks.
|
||||
func (c *PrefixCache) Lookup(hashes []BlockHash) int {
|
||||
if !c.Enabled() {
|
||||
return 0
|
||||
@@ -89,7 +88,8 @@ func (c *PrefixCache) Lookup(hashes []BlockHash) int {
|
||||
return n
|
||||
}
|
||||
|
||||
// Insert, blokları cache'e yazar (LRU'nun başına) ve kapasiteyi aşarsa en eskileri atar.
|
||||
// Insert stores blocks at the front of the LRU, evicting the oldest ones once
|
||||
// capacity is exceeded.
|
||||
func (c *PrefixCache) Insert(hashes []BlockHash) {
|
||||
if !c.Enabled() {
|
||||
return
|
||||
@@ -111,7 +111,7 @@ func (c *PrefixCache) Insert(hashes []BlockHash) {
|
||||
}
|
||||
}
|
||||
|
||||
// Stats, cache sorgu ve isabet sayılarını döner (blok cinsinden).
|
||||
// Stats returns cumulative query and hit counts, in blocks.
|
||||
func (c *PrefixCache) Stats() (queries, hits int64) {
|
||||
if c == nil {
|
||||
return 0, 0
|
||||
@@ -119,7 +119,7 @@ func (c *PrefixCache) Stats() (queries, hits int64) {
|
||||
return c.queries, c.hits
|
||||
}
|
||||
|
||||
// Len, cache'teki blok sayısı.
|
||||
// Len returns the number of blocks currently cached.
|
||||
func (c *PrefixCache) Len() int {
|
||||
if !c.Enabled() {
|
||||
return 0
|
||||
|
||||
+27
-27
@@ -2,76 +2,76 @@ package sim
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestBlockHashesZincirleme(t *testing.T) {
|
||||
// Aynı içerik farklı ön ekten sonra geliyorsa blok hash'i farklı olmalı.
|
||||
func TestBlockHashesAreChained(t *testing.T) {
|
||||
// The same content following a different prefix must hash differently.
|
||||
a := BlockHashes(Tokenize("AAAAAAAABBBBBBBB"), 2)
|
||||
b := BlockHashes(Tokenize("CCCCCCCCBBBBBBBB"), 2)
|
||||
if len(a) != 2 || len(b) != 2 {
|
||||
t.Fatalf("beklenen 2 blok, alınan a=%d b=%d", len(a), len(b))
|
||||
t.Fatalf("expected 2 blocks each, got a=%d b=%d", len(a), len(b))
|
||||
}
|
||||
if a[1] == b[1] {
|
||||
t.Error("farklı ön ekten sonra gelen aynı blok aynı hash'i verdi; zincirleme çalışmıyor")
|
||||
t.Error("identical block after different prefixes hashed the same; chaining is broken")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlockHashesOnEkKararli(t *testing.T) {
|
||||
// Blok sınırına hizalı bir ön ek, uzun metinde aynı blok hash'lerini vermeli.
|
||||
// Hizalama şart: son yarım token farklı olursa onu içeren blok da farklı
|
||||
// çıkar — gerçek tokenizer'larda da olan sınır etkisi, bu yüzden prefix
|
||||
// cache hit'i her zaman tam blok cinsinden ölçülür.
|
||||
const aligned = "merhaba " // 8 bayt = 2 token = blockSize 2 ile 1 tam blok
|
||||
func TestBlockHashesArePrefixStable(t *testing.T) {
|
||||
// A prefix aligned to a block boundary must hash identically in a longer
|
||||
// text. Alignment matters: a differing trailing partial token changes the
|
||||
// block containing it — the same boundary effect real tokenizers have,
|
||||
// which is why cache hits are always measured in whole blocks.
|
||||
const aligned = "merhaba " // 8 bytes = 2 tokens = 1 full block at blockSize 2
|
||||
short := BlockHashes(Tokenize(aligned), 2)
|
||||
long := BlockHashes(Tokenize(aligned+"dunya nasilsin"), 2)
|
||||
if len(short) != 1 {
|
||||
t.Fatalf("beklenen 1 blok, alınan %d", len(short))
|
||||
t.Fatalf("expected 1 block, got %d", len(short))
|
||||
}
|
||||
if short[0] != long[0] {
|
||||
t.Fatal("hizalı ön ek blok hash'i değişti; ön ek kararlı değil")
|
||||
t.Fatal("aligned prefix produced a different block hash; not prefix stable")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBlockHashesYarimBlokAtilir(t *testing.T) {
|
||||
// 3 token, blockSize 2 -> 1 tam blok
|
||||
func TestBlockHashesDropPartialBlock(t *testing.T) {
|
||||
// 3 tokens at blockSize 2 -> 1 full block
|
||||
if got := len(BlockHashes(make([]Token, 3), 2)); got != 1 {
|
||||
t.Errorf("beklenen 1 blok, alınan %d", got)
|
||||
t.Errorf("expected 1 block, got %d", got)
|
||||
}
|
||||
if got := len(BlockHashes(make([]Token, 1), 2)); got != 0 {
|
||||
t.Errorf("tam blok yokken blok üretildi: %d", got)
|
||||
t.Errorf("produced %d blocks with no full block available", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrefixCacheKesintisizOnEk(t *testing.T) {
|
||||
func TestPrefixCacheReturnsContiguousPrefix(t *testing.T) {
|
||||
c := NewPrefixCache(10)
|
||||
h := []BlockHash{1, 2, 3, 4}
|
||||
c.Insert(h)
|
||||
|
||||
if n := c.Lookup(h); n != 4 {
|
||||
t.Fatalf("tam eşleşme beklendi, alınan %d", n)
|
||||
t.Fatalf("expected a full match, got %d", n)
|
||||
}
|
||||
// Ortası eksik: ilk blok hit, sonrası miss olmalı.
|
||||
// With a gap in the middle only the first block counts.
|
||||
if n := c.Lookup([]BlockHash{1, 99, 3}); n != 1 {
|
||||
t.Errorf("kesintisiz ön ek 1 olmalıydı, alınan %d", n)
|
||||
t.Errorf("contiguous prefix should have been 1, got %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrefixCacheLRUTahliye(t *testing.T) {
|
||||
func TestPrefixCacheEvictsLRU(t *testing.T) {
|
||||
c := NewPrefixCache(2)
|
||||
c.Insert([]BlockHash{1, 2})
|
||||
c.Lookup([]BlockHash{1}) // 1'i tazele
|
||||
c.Insert([]BlockHash{3}) // kapasite aşıldı, en eski (2) düşmeli
|
||||
c.Lookup([]BlockHash{1}) // refresh 1
|
||||
c.Insert([]BlockHash{3}) // over capacity: the oldest (2) must go
|
||||
|
||||
if n := c.Lookup([]BlockHash{1}); n != 1 {
|
||||
t.Error("taze blok tahliye edildi")
|
||||
t.Error("recently used block was evicted")
|
||||
}
|
||||
if n := c.Lookup([]BlockHash{2}); n != 0 {
|
||||
t.Error("en eski blok tahliye edilmeliydi")
|
||||
t.Error("least recently used block should have been evicted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrefixCacheKapali(t *testing.T) {
|
||||
func TestPrefixCacheDisabled(t *testing.T) {
|
||||
c := NewPrefixCache(0)
|
||||
c.Insert([]BlockHash{1, 2})
|
||||
if n := c.Lookup([]BlockHash{1}); n != 0 {
|
||||
t.Errorf("kapalı cache hit verdi: %d", n)
|
||||
t.Errorf("disabled cache reported a hit: %d", n)
|
||||
}
|
||||
}
|
||||
+69
-70
@@ -1,12 +1,12 @@
|
||||
// Package sim, OpenAI uyumlu sahte bir inference engine'i barındırır.
|
||||
// Package sim provides an OpenAI-compatible fake inference engine.
|
||||
//
|
||||
// Amacı router'ı GPU olmadan gerçekçi koşullarda test edebilmek. Gerçek bir
|
||||
// engine'in router açısından önemli olan davranışlarını modeller:
|
||||
// - continuous batching (sınırlı eşzamanlı dizi sayısı)
|
||||
// - KV cache blok muhasebesi ve blok yetmediğinde kuyruğa alma
|
||||
// - otomatik prefix cache (blok bazlı LRU) ve cache hit'in TTFT'ye etkisi
|
||||
// - batch büyüdükçe token başına gecikmenin artması
|
||||
// - vLLM formatında /metrics
|
||||
// Its purpose is to let the router be tested under realistic conditions
|
||||
// without GPUs. It models the real-engine behaviours the router cares about:
|
||||
// - continuous batching (bounded number of concurrent sequences)
|
||||
// - KV cache block accounting, queueing once blocks run out
|
||||
// - automatic prefix caching (block-level LRU) and its effect on TTFT
|
||||
// - per-token latency growing as the batch grows
|
||||
// - /metrics in vLLM's format
|
||||
package sim
|
||||
|
||||
import (
|
||||
@@ -19,40 +19,40 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// Config, simüle edilen engine'in davranışını belirler.
|
||||
// Config controls the simulated engine's behaviour.
|
||||
type Config struct {
|
||||
Model string
|
||||
|
||||
// MaxNumSeqs, aynı anda çalışabilecek dizi sayısı (continuous batch kapasitesi).
|
||||
// MaxNumSeqs is how many sequences may run concurrently (batch capacity).
|
||||
MaxNumSeqs int
|
||||
// BlockSize, bir KV cache bloğundaki token sayısı.
|
||||
// BlockSize is the number of tokens per KV cache block.
|
||||
BlockSize int
|
||||
// NumGPUBlocks, toplam KV cache bloğu.
|
||||
// NumGPUBlocks is the total number of KV cache blocks.
|
||||
NumGPUBlocks int
|
||||
// PrefixCacheBlocks, prefix cache'in tutacağı blok sayısı. 0 = kapalı.
|
||||
// PrefixCacheBlocks is the prefix cache capacity in blocks. 0 disables it.
|
||||
PrefixCacheBlocks int
|
||||
|
||||
// PrefillTokensPerSec, engine genelinde paylaşılan prefill hızı.
|
||||
// PrefillTokensPerSec is the prefill rate, shared engine-wide.
|
||||
PrefillTokensPerSec float64
|
||||
// DecodeTokensPerSec, tek dizi çalışırken token üretim hızı.
|
||||
// DecodeTokensPerSec is the generation rate for a single running sequence.
|
||||
DecodeTokensPerSec float64
|
||||
// BatchContention, batch'teki her ek dizinin decode hızını ne kadar
|
||||
// yavaşlattığı. Etkin hız = DecodeTokensPerSec / (1 + BatchContention*(n-1)).
|
||||
// BatchContention is how much each additional sequence slows decoding:
|
||||
// effective rate = DecodeTokensPerSec / (1 + BatchContention*(n-1)).
|
||||
BatchContention float64
|
||||
|
||||
// DefaultOutputTokens, istek max_tokens vermediğinde üretilecek token sayısı.
|
||||
// DefaultOutputTokens is generated when a request omits max_tokens.
|
||||
DefaultOutputTokens int
|
||||
// Tick, simülasyon adımı. Küçük olması çözünürlüğü artırır.
|
||||
// Tick is the simulation step; smaller values increase resolution.
|
||||
Tick time.Duration
|
||||
|
||||
// FailRate, isteklerin hangi oranda 500 ile reddedileceği (kaos testi için).
|
||||
// FailRate is the fraction of requests rejected artificially, for chaos tests.
|
||||
FailRate float64
|
||||
// Seed, deterministik testler için.
|
||||
// Seed makes runs deterministic.
|
||||
Seed int64
|
||||
}
|
||||
|
||||
// Default, makul varsayılanlarla bir Config döner.
|
||||
// Kabaca 8xH100 üstünde 70B bir modelin davranışına benzer ölçekler.
|
||||
// Default returns a configuration with sensible values, roughly in the range
|
||||
// of a 70B model on 8xH100.
|
||||
func Default() Config {
|
||||
return Config{
|
||||
Model: "sim-model",
|
||||
@@ -95,34 +95,34 @@ func (c *Config) normalize() {
|
||||
}
|
||||
}
|
||||
|
||||
// EventKind, akıştaki olay türü.
|
||||
// EventKind is the type of an event in the stream.
|
||||
type EventKind int
|
||||
|
||||
const (
|
||||
// EventToken, üretilen bir token.
|
||||
// EventToken carries one generated token.
|
||||
EventToken EventKind = iota
|
||||
// EventDone, üretimin normal bitişi.
|
||||
// EventDone marks normal completion.
|
||||
EventDone
|
||||
// EventError, üretim sırasında oluşan hata.
|
||||
// EventError reports a failure during generation.
|
||||
EventError
|
||||
)
|
||||
|
||||
// Event, engine'in istemciye ilettiği tek bir olay.
|
||||
// Event is a single event delivered to the caller.
|
||||
type Event struct {
|
||||
Kind EventKind
|
||||
Text string
|
||||
Err error
|
||||
// Result yalnızca EventDone'da doludur; prompt/cache/completion sayılarını taşır.
|
||||
// Result is set only on EventDone; it carries the final token counts.
|
||||
Result *Result
|
||||
}
|
||||
|
||||
// Prompt, engine'e verilen istek.
|
||||
// Prompt is a request handed to the engine.
|
||||
type Prompt struct {
|
||||
Text string
|
||||
MaxTokens int
|
||||
}
|
||||
|
||||
// Result, istek bittiğinde okunabilen özet (Done olayından sonra geçerli).
|
||||
// Result summarizes a finished request; valid from the Done event onwards.
|
||||
type Result struct {
|
||||
PromptTokens int
|
||||
CachedTokens int
|
||||
@@ -130,12 +130,12 @@ type Result struct {
|
||||
}
|
||||
|
||||
var (
|
||||
// ErrTooLong, isteğin KV cache'e hiç sığmadığını belirtir.
|
||||
ErrTooLong = errors.New("istek KV cache kapasitesine sığmıyor")
|
||||
// ErrInjected, FailRate ile üretilen yapay hata.
|
||||
ErrInjected = errors.New("yapay hata (sim fail_rate)")
|
||||
// ErrShutdown, engine kapanırken bekleyen istekler için.
|
||||
ErrShutdown = errors.New("engine kapanıyor")
|
||||
// ErrTooLong means the request does not fit in the KV cache at all.
|
||||
ErrTooLong = errors.New("request does not fit in KV cache capacity")
|
||||
// ErrInjected is the artificial failure produced by FailRate.
|
||||
ErrInjected = errors.New("injected failure (sim fail_rate)")
|
||||
// ErrShutdown is returned to pending requests while the engine stops.
|
||||
ErrShutdown = errors.New("engine shutting down")
|
||||
)
|
||||
|
||||
type req struct {
|
||||
@@ -164,9 +164,9 @@ type req struct {
|
||||
res *Result
|
||||
}
|
||||
|
||||
// flush, biriken olayları bloke olmadan kanala aktarır.
|
||||
// Bloke olmamak şart: tick döngüsü kilidi tutarken yavaş bir istemci
|
||||
// tüm engine'i durdurmamalı.
|
||||
// flush moves buffered events into the channel without blocking.
|
||||
// Not blocking is essential: the tick loop holds the engine lock, and one slow
|
||||
// client must not stall every other request.
|
||||
func (r *req) flush() {
|
||||
for len(r.pending) > 0 {
|
||||
select {
|
||||
@@ -183,7 +183,7 @@ func (r *req) emit(ev Event) {
|
||||
r.flush()
|
||||
}
|
||||
|
||||
// Engine, simüle edilen inference engine.
|
||||
// Engine is the simulated inference engine.
|
||||
type Engine struct {
|
||||
cfg Config
|
||||
|
||||
@@ -203,8 +203,7 @@ type Engine struct {
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// New, verilen konfigürasyonla bir engine kurar ve simülasyon döngüsünü başlatır.
|
||||
// Kapatmak için Close çağrılmalı.
|
||||
// New creates an engine and starts its simulation loop. Call Close to stop it.
|
||||
func New(cfg Config) *Engine {
|
||||
cfg.normalize()
|
||||
seed := cfg.Seed
|
||||
@@ -215,7 +214,7 @@ func New(cfg Config) *Engine {
|
||||
cfg: cfg,
|
||||
freeBlocks: cfg.NumGPUBlocks,
|
||||
cache: NewPrefixCache(cfg.PrefixCacheBlocks),
|
||||
rng: rand.New(rand.NewSource(seed)), //nolint:gosec // simülasyon
|
||||
rng: rand.New(rand.NewSource(seed)), //nolint:gosec // simulation only
|
||||
m: newMetrics(),
|
||||
stop: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
@@ -224,10 +223,10 @@ func New(cfg Config) *Engine {
|
||||
return e
|
||||
}
|
||||
|
||||
// Config, engine'in aktif konfigürasyonunu döner.
|
||||
// Config returns the engine's effective configuration.
|
||||
func (e *Engine) Config() Config { return e.cfg }
|
||||
|
||||
// Close, simülasyon döngüsünü durdurur ve bekleyen istekleri hataya düşürür.
|
||||
// Close stops the simulation loop and fails any pending requests.
|
||||
func (e *Engine) Close() {
|
||||
e.stopOnce.Do(func() {
|
||||
close(e.stop)
|
||||
@@ -235,8 +234,8 @@ func (e *Engine) Close() {
|
||||
})
|
||||
}
|
||||
|
||||
// Submit, isteği kuyruğa alır ve olay kanalını döner.
|
||||
// Kanal, üretim bitince veya hata oluşunca kapanır.
|
||||
// Submit enqueues a request and returns its event channel. The channel closes
|
||||
// when generation finishes or fails.
|
||||
func (e *Engine) Submit(ctx context.Context, p Prompt) (<-chan Event, error) {
|
||||
if e.cfg.FailRate > 0 {
|
||||
e.mu.Lock()
|
||||
@@ -310,9 +309,9 @@ func (e *Engine) run() {
|
||||
}
|
||||
}
|
||||
|
||||
// reap, istemcisi vazgeçmiş istekleri hem kuyruktan hem batch'ten düşürür.
|
||||
// Router'ın iptal propagasyonunu test edebilmek için bu davranış şart:
|
||||
// istemci koptuğunda GPU boşa çalışmamalı.
|
||||
// reap drops requests whose client has gone away, from both the queue and the
|
||||
// running batch. The router's cancellation propagation cannot be tested
|
||||
// without this: when a client disconnects, the GPU must stop working.
|
||||
func (e *Engine) reap() {
|
||||
e.waiting = filterInPlace(e.waiting, func(r *req) bool {
|
||||
if r.ctx.Err() == nil {
|
||||
@@ -338,7 +337,7 @@ func (e *Engine) schedule() {
|
||||
r := e.waiting[0]
|
||||
need := e.blocksFor(r.promptTokens + r.outputTokens)
|
||||
if need > e.freeBlocks {
|
||||
// Head-of-line blocking: gerçek engine de böyle davranır.
|
||||
// Head-of-line blocking; real engines behave the same way.
|
||||
break
|
||||
}
|
||||
|
||||
@@ -362,7 +361,7 @@ func (e *Engine) advance(dt float64) {
|
||||
return
|
||||
}
|
||||
|
||||
// Prefill bütçesi engine genelinde paylaşılır ve FIFO dağıtılır.
|
||||
// The prefill budget is shared engine-wide and handed out FIFO.
|
||||
budget := e.cfg.PrefillTokensPerSec * dt
|
||||
for _, r := range e.running {
|
||||
if budget <= 0 {
|
||||
@@ -376,7 +375,7 @@ func (e *Engine) advance(dt float64) {
|
||||
budget -= used
|
||||
}
|
||||
|
||||
// Decode hızı batch büyüdükçe düşer.
|
||||
// Decoding slows down as the batch grows.
|
||||
n := 0
|
||||
for _, r := range e.running {
|
||||
if r.prefillLeft <= 0 {
|
||||
@@ -420,9 +419,9 @@ func (e *Engine) finish(r *req) {
|
||||
r.finished = true
|
||||
e.freeBlocks += r.blocks
|
||||
|
||||
// Üretilen çıktı da cache'e girer: bir sonraki turda konuşma geçmişi
|
||||
// olarak geri geldiğinde hit vermeli. Çok turlu sohbeti modellemenin
|
||||
// tek yolu bu.
|
||||
// The generated output enters the cache too: when it comes back as
|
||||
// conversation history on the next turn it must hit. That is the only way
|
||||
// to model multi-turn chat.
|
||||
full := Tokenize(r.prompt.Text + generatedText(r.emitted))
|
||||
e.cache.Insert(BlockHashes(full, e.cfg.BlockSize))
|
||||
|
||||
@@ -469,7 +468,19 @@ func (e *Engine) shutdown() {
|
||||
e.waiting, e.running, e.draining = nil, nil, nil
|
||||
}
|
||||
|
||||
// Stats, router'ın scrape edeceği anlık durumu döner.
|
||||
// Stats returns the instantaneous state the router would scrape.
|
||||
type Stats struct {
|
||||
Running int
|
||||
Waiting int
|
||||
KVCacheUsage float64
|
||||
BlockSize int
|
||||
NumGPUBlocks int
|
||||
PrefixQueries int64
|
||||
PrefixHits int64
|
||||
CachedBlocks int
|
||||
}
|
||||
|
||||
// Stats returns a snapshot of the engine's current state.
|
||||
func (e *Engine) Stats() Stats {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
@@ -487,18 +498,6 @@ func (e *Engine) Stats() Stats {
|
||||
}
|
||||
}
|
||||
|
||||
// Stats, engine'in anlık durumu.
|
||||
type Stats struct {
|
||||
Running int
|
||||
Waiting int
|
||||
KVCacheUsage float64
|
||||
BlockSize int
|
||||
NumGPUBlocks int
|
||||
PrefixQueries int64
|
||||
PrefixHits int64
|
||||
CachedBlocks int
|
||||
}
|
||||
|
||||
func filterInPlace(s []*req, keep func(*req) bool) []*req {
|
||||
out := s[:0]
|
||||
for _, r := range s {
|
||||
|
||||
+39
-42
@@ -7,7 +7,7 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// testConfig, testlerin hızlı ve deterministik koşması için hızlandırılmış ayarlar.
|
||||
// testConfig speeds the simulation up so tests stay fast and deterministic.
|
||||
func testConfig() Config {
|
||||
c := Default()
|
||||
c.Tick = time.Millisecond
|
||||
@@ -18,7 +18,7 @@ func testConfig() Config {
|
||||
return c
|
||||
}
|
||||
|
||||
// drain, kanaldaki tüm token'ları toplar ve bitişi bekler.
|
||||
// drain collects every token from the channel and waits for completion.
|
||||
func drain(t *testing.T, ch <-chan Event) (text string, done bool) {
|
||||
t.Helper()
|
||||
var b strings.Builder
|
||||
@@ -35,36 +35,36 @@ func drain(t *testing.T, ch <-chan Event) (text string, done bool) {
|
||||
case EventDone:
|
||||
done = true
|
||||
case EventError:
|
||||
t.Fatalf("beklenmeyen hata: %v", ev.Err)
|
||||
t.Fatalf("unexpected error: %v", ev.Err)
|
||||
}
|
||||
case <-deadline:
|
||||
t.Fatal("zaman aşımı: engine token üretmedi")
|
||||
t.Fatal("timed out waiting for tokens")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineTemelUretim(t *testing.T) {
|
||||
func TestEngineGeneratesTokens(t *testing.T) {
|
||||
e := New(testConfig())
|
||||
defer e.Close()
|
||||
|
||||
ch, err := e.Submit(t.Context(), Prompt{Text: "merhaba dunya", MaxTokens: 10})
|
||||
ch, err := e.Submit(t.Context(), Prompt{Text: "hello world", MaxTokens: 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
text, done := drain(t, ch)
|
||||
if !done {
|
||||
t.Error("EventDone gelmedi")
|
||||
t.Error("no EventDone received")
|
||||
}
|
||||
if got := len(strings.Fields(text)); got != 10 {
|
||||
t.Errorf("beklenen 10 token, alınan %d (%q)", got, text)
|
||||
t.Errorf("expected 10 tokens, got %d (%q)", got, text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineIptalBloklariSerbestBirakir(t *testing.T) {
|
||||
// Router'ın iptal propagasyonunu test edebilmesi için kritik davranış:
|
||||
// istemci koptuğunda engine isteği düşürmeli ve KV bloklarını bırakmalı.
|
||||
func TestCancellationReleasesBlocks(t *testing.T) {
|
||||
// Essential for testing the router's cancellation propagation: when the
|
||||
// client goes away the engine must drop the request and free its KV blocks.
|
||||
cfg := testConfig()
|
||||
cfg.DecodeTokensPerSec = 20 // uzun sürsün ki iptal ederken çalışıyor olsun
|
||||
cfg.DecodeTokensPerSec = 20 // slow, so it is still running when cancelled
|
||||
e := New(cfg)
|
||||
defer e.Close()
|
||||
|
||||
@@ -76,7 +76,7 @@ func TestEngineIptalBloklariSerbestBirakir(t *testing.T) {
|
||||
|
||||
waitFor(t, time.Second, func() bool { return e.Stats().Running == 1 })
|
||||
if usage := e.Stats().KVCacheUsage; usage <= 0 {
|
||||
t.Fatal("çalışan istek KV bloğu ayırmadı")
|
||||
t.Fatal("running request reserved no KV blocks")
|
||||
}
|
||||
|
||||
cancel()
|
||||
@@ -86,20 +86,19 @@ func TestEngineIptalBloklariSerbestBirakir(t *testing.T) {
|
||||
return s.Running == 0 && s.KVCacheUsage == 0
|
||||
})
|
||||
|
||||
// Kanal kapanmalı ki istemci tarafı bloke kalmasın.
|
||||
// The channel must close so the client side never blocks.
|
||||
select {
|
||||
case _, open := <-ch:
|
||||
if open {
|
||||
// token kalmış olabilir; kapanana kadar tüket
|
||||
for range ch { //nolint:revive // kanalı boşalt
|
||||
for range ch { //nolint:revive // drain remaining events
|
||||
}
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("iptal sonrası kanal kapanmadı")
|
||||
t.Fatal("channel stayed open after cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineBatchDolunkaKuyrugaAlir(t *testing.T) {
|
||||
func TestRequestsQueueWhenBatchIsFull(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.MaxNumSeqs = 2
|
||||
cfg.DecodeTokensPerSec = 20
|
||||
@@ -115,56 +114,54 @@ func TestEngineBatchDolunkaKuyrugaAlir(t *testing.T) {
|
||||
waitFor(t, time.Second, func() bool { return e.Stats().Running == 2 })
|
||||
s := e.Stats()
|
||||
if s.Running != 2 {
|
||||
t.Errorf("running %d, MaxNumSeqs=2 aşıldı", s.Running)
|
||||
t.Errorf("running=%d exceeds MaxNumSeqs=2", s.Running)
|
||||
}
|
||||
if s.Waiting != 3 {
|
||||
t.Errorf("waiting %d, beklenen 3", s.Waiting)
|
||||
t.Errorf("waiting=%d, expected 3", s.Waiting)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineKVBlogunaSigmayanIstekReddedilir(t *testing.T) {
|
||||
func TestOversizedRequestIsRejected(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.NumGPUBlocks = 4
|
||||
cfg.BlockSize = 16 // toplam 64 token kapasite
|
||||
cfg.BlockSize = 16 // 64 tokens of capacity in total
|
||||
e := New(cfg)
|
||||
defer e.Close()
|
||||
|
||||
_, err := e.Submit(t.Context(), Prompt{Text: strings.Repeat("x", 10000), MaxTokens: 100})
|
||||
if err != ErrTooLong {
|
||||
t.Errorf("ErrTooLong beklendi, alınan %v", err)
|
||||
t.Errorf("expected ErrTooLong, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnginePrefixCacheIkinciIstegiHizlandirir(t *testing.T) {
|
||||
func TestPrefixCacheSpeedsUpRepeatedPrompt(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.PrefillTokensPerSec = 2000 // prefill maliyeti ölçülebilir olsun
|
||||
cfg.PrefillTokensPerSec = 2000 // make prefill cost measurable
|
||||
cfg.DecodeTokensPerSec = 5000
|
||||
e := New(cfg)
|
||||
defer e.Close()
|
||||
|
||||
prompt := Prompt{Text: strings.Repeat("ortak sistem promptu. ", 200), MaxTokens: 5}
|
||||
prompt := Prompt{Text: strings.Repeat("shared system prompt. ", 200), MaxTokens: 5}
|
||||
|
||||
first := timeToFirstToken(t, e, prompt)
|
||||
// İlk istek bitip cache'e yazıldıktan sonra aynısını tekrar gönder.
|
||||
// Send the same prompt again now that the first one populated the cache.
|
||||
second := timeToFirstToken(t, e, prompt)
|
||||
|
||||
if second >= first {
|
||||
t.Errorf("prefix cache TTFT'yi düşürmedi: ilk=%v ikinci=%v", first, second)
|
||||
t.Errorf("prefix cache did not reduce TTFT: first=%v second=%v", first, second)
|
||||
}
|
||||
|
||||
s := e.Stats()
|
||||
if s.PrefixHits == 0 {
|
||||
t.Error("prefix cache hit sayacı artmadı")
|
||||
if e.Stats().PrefixHits == 0 {
|
||||
t.Error("prefix cache hit counter did not advance")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnginePrefixCacheKapaliykenHitYok(t *testing.T) {
|
||||
func TestNoHitsWhenPrefixCacheDisabled(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.PrefixCacheBlocks = 0
|
||||
e := New(cfg)
|
||||
defer e.Close()
|
||||
|
||||
p := Prompt{Text: strings.Repeat("ayni prompt. ", 50), MaxTokens: 3}
|
||||
p := Prompt{Text: strings.Repeat("same prompt. ", 50), MaxTokens: 3}
|
||||
for range 3 {
|
||||
ch, err := e.Submit(t.Context(), p)
|
||||
if err != nil {
|
||||
@@ -173,11 +170,11 @@ func TestEnginePrefixCacheKapaliykenHitYok(t *testing.T) {
|
||||
drain(t, ch)
|
||||
}
|
||||
if h := e.Stats().PrefixHits; h != 0 {
|
||||
t.Errorf("cache kapalıyken %d hit raporlandı", h)
|
||||
t.Errorf("disabled cache reported %d hits", h)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineKapanisBekleyenleriBosaltir(t *testing.T) {
|
||||
func TestShutdownDrainsPendingRequests(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.DecodeTokensPerSec = 1
|
||||
e := New(cfg)
|
||||
@@ -197,23 +194,23 @@ func TestEngineKapanisBekleyenleriBosaltir(t *testing.T) {
|
||||
return
|
||||
}
|
||||
case <-deadline:
|
||||
t.Fatal("kapanışta kanal kapanmadı")
|
||||
t.Fatal("channel stayed open after shutdown")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestEngineYapayHata(t *testing.T) {
|
||||
func TestInjectedFailure(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.FailRate = 1.0
|
||||
e := New(cfg)
|
||||
defer e.Close()
|
||||
|
||||
if _, err := e.Submit(t.Context(), Prompt{Text: "abc"}); err != ErrInjected {
|
||||
t.Errorf("ErrInjected beklendi, alınan %v", err)
|
||||
t.Errorf("expected ErrInjected, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// --- yardımcılar ---
|
||||
// --- helpers ---
|
||||
|
||||
func timeToFirstToken(t *testing.T, e *Engine, p Prompt) time.Duration {
|
||||
t.Helper()
|
||||
@@ -229,7 +226,7 @@ func timeToFirstToken(t *testing.T, e *Engine, p Prompt) time.Duration {
|
||||
}
|
||||
}
|
||||
if ttft == 0 {
|
||||
t.Fatal("hiç token üretilmedi")
|
||||
t.Fatal("no tokens were generated")
|
||||
}
|
||||
return ttft
|
||||
}
|
||||
@@ -243,5 +240,5 @@ func waitFor(t *testing.T, limit time.Duration, cond func() bool) {
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatalf("koşul %v içinde sağlanmadı", limit)
|
||||
t.Fatalf("condition not met within %v", limit)
|
||||
}
|
||||
@@ -8,13 +8,13 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// ttftBuckets ve tpotBuckets, vLLM'in kullandıklarına yakın saniye sınırları.
|
||||
// ttftBuckets and tpotBuckets use second boundaries close to vLLM's own.
|
||||
var (
|
||||
ttftBuckets = []float64{0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, 20, 40, 80}
|
||||
tpotBuckets = []float64{0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1, 2.5}
|
||||
)
|
||||
|
||||
// histogram, Prometheus uyumlu basit bir kümülatif histogram.
|
||||
// histogram is a minimal Prometheus-compatible cumulative histogram.
|
||||
type histogram struct {
|
||||
bounds []float64
|
||||
counts []uint64
|
||||
@@ -36,7 +36,7 @@ func (h *histogram) observe(v float64) {
|
||||
}
|
||||
}
|
||||
|
||||
// write, histogramı Prometheus text formatında yazar (kümülatif le bucket'ları).
|
||||
// write renders the histogram in Prometheus text format (cumulative le buckets).
|
||||
func (h *histogram) write(w io.Writer, name, labels string) {
|
||||
var cum uint64
|
||||
for i, b := range h.bounds {
|
||||
@@ -81,12 +81,12 @@ func (m *metrics) observeTPOT(d time.Duration) {
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
// WriteMetrics, engine durumunu vLLM'in metrik adlarıyla Prometheus text
|
||||
// formatında yazar.
|
||||
// WriteMetrics renders the engine's state in Prometheus text format using
|
||||
// vLLM's metric names.
|
||||
//
|
||||
// Metrik adları §7.6'daki engine kontratıdır; router bunlara göre parse eder.
|
||||
// Gerçek vLLM sürümleriyle aynı adları kullanmak şart — sim-engine'in varlık
|
||||
// sebebi router'ın gerçek engine'e karşı davranışını taklit etmek.
|
||||
// Those names are the engine contract from plan §7.6; the router parses
|
||||
// against them. Matching real vLLM names exactly is the point: sim-engine
|
||||
// exists to stand in for a real engine as far as the router can tell.
|
||||
func (e *Engine) WriteMetrics(w io.Writer) {
|
||||
s := e.Stats()
|
||||
lbl := fmt.Sprintf("model_name=%q", e.cfg.Model)
|
||||
@@ -127,7 +127,7 @@ func (e *Engine) WriteMetrics(w io.Writer) {
|
||||
fmt.Fprintf(w, "# TYPE vllm:request_success_total counter\n")
|
||||
fmt.Fprintf(w, "vllm:request_success_total{%s,finished_reason=\"length\"} %d\n", lbl, e.m.finished.Load())
|
||||
|
||||
// Simülasyona özgü (gerçek vLLM'de yok; kaos testlerinde işe yarar).
|
||||
// Simulation-specific (absent in real vLLM; useful for chaos tests).
|
||||
fmt.Fprintf(w, "# TYPE sim:requests_cancelled_total counter\n")
|
||||
fmt.Fprintf(w, "sim:requests_cancelled_total{%s} %d\n", lbl, e.m.cancelled.Load())
|
||||
fmt.Fprintf(w, "# TYPE sim:requests_failed_total counter\n")
|
||||
|
||||
+16
-16
@@ -9,14 +9,14 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// Server, Engine'i OpenAI uyumlu bir HTTP yüzeyiyle sarar.
|
||||
// Server wraps an Engine in an OpenAI-compatible HTTP surface.
|
||||
type Server struct {
|
||||
eng *Engine
|
||||
log *slog.Logger
|
||||
mux *http.ServeMux
|
||||
}
|
||||
|
||||
// NewServer, engine için HTTP yüzeyini kurar.
|
||||
// NewServer builds the HTTP surface for an engine.
|
||||
func NewServer(eng *Engine, log *slog.Logger) *Server {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
@@ -32,7 +32,7 @@ func NewServer(eng *Engine, log *slog.Logger) *Server {
|
||||
|
||||
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mux.ServeHTTP(w, r) }
|
||||
|
||||
// --- istek/yanıt tipleri (yalnızca gereken alanlar) ---
|
||||
// --- request/response types (only the fields we need) ---
|
||||
|
||||
type streamOptions struct {
|
||||
IncludeUsage bool `json:"include_usage"`
|
||||
@@ -64,15 +64,15 @@ type usage struct {
|
||||
PromptTokens int `json:"prompt_tokens"`
|
||||
CompletionTokens int `json:"completion_tokens"`
|
||||
TotalTokens int `json:"total_tokens"`
|
||||
// OpenAI uyumlu cache raporlaması; prefix cache etkisini istemciye gösterir.
|
||||
// OpenAI-compatible cache reporting; surfaces the prefix cache to clients.
|
||||
PromptTokensDetails struct {
|
||||
CachedTokens int `json:"cached_tokens"`
|
||||
} `json:"prompt_tokens_details"`
|
||||
}
|
||||
|
||||
// text, mesaj içeriğini düz metne indirger.
|
||||
// İçerik string ya da parça dizisi olabilir; sim için ikisini de kaba şekilde
|
||||
// metne çeviriyoruz, çünkü tek ihtiyacımız deterministik bir token sayısı.
|
||||
// text reduces a message's content to plain text.
|
||||
// Content may be a string or an array of parts; the simulation coerces both
|
||||
// crudely, since all it needs is a deterministic token count.
|
||||
func (m chatMessage) text() string {
|
||||
var s string
|
||||
if err := json.Unmarshal(m.Content, &s); err == nil {
|
||||
@@ -92,7 +92,7 @@ func promptFromMessages(msgs []chatMessage) string {
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// --- handler'lar ---
|
||||
// --- handlers ---
|
||||
|
||||
func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
|
||||
var req chatRequest
|
||||
@@ -168,7 +168,7 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) {
|
||||
s.eng.WriteMetrics(w)
|
||||
}
|
||||
|
||||
// --- ortak akış ---
|
||||
// --- shared request flow ---
|
||||
|
||||
type servePlan struct {
|
||||
prompt Prompt
|
||||
@@ -241,7 +241,7 @@ func (s *Server) bufferedResponse(w http.ResponseWriter, ch <-chan Event, p serv
|
||||
func (s *Server) streamResponse(w http.ResponseWriter, r *http.Request, ch <-chan Event, p servePlan, id string, created int64) {
|
||||
flusher, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
writeError(w, http.StatusInternalServerError, "server_error", "streaming desteklenmiyor")
|
||||
writeError(w, http.StatusInternalServerError, "server_error", "streaming not supported")
|
||||
return
|
||||
}
|
||||
h := w.Header()
|
||||
@@ -256,7 +256,7 @@ func (s *Server) streamResponse(w http.ResponseWriter, r *http.Request, ch <-cha
|
||||
for {
|
||||
select {
|
||||
case <-r.Context().Done():
|
||||
// İstemci koptu. Engine reap() ile isteği batch'ten düşürecek.
|
||||
// Client disconnected; the engine's reap() drops it from the batch.
|
||||
return
|
||||
case ev, open := <-ch:
|
||||
if !open {
|
||||
@@ -291,7 +291,7 @@ func (s *Server) streamResponse(w http.ResponseWriter, r *http.Request, ch <-cha
|
||||
func (s *Server) writeChunk(w http.ResponseWriter, v any) {
|
||||
b, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
s.log.Error("chunk serileştirilemedi", "err", err)
|
||||
s.log.Error("failed to marshal chunk", "err", err)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(w, "data: %s\n\n", b)
|
||||
@@ -319,7 +319,7 @@ func chunk(id string, p servePlan, created int64, text *string, finish string, u
|
||||
out := map[string]any{
|
||||
"id": id, "object": p.chunkObject, "created": created, "model": p.model,
|
||||
}
|
||||
// Usage chunk'ında choices boş dizi olur (OpenAI davranışı).
|
||||
// The usage chunk carries an empty choices array (OpenAI behaviour).
|
||||
if u != nil {
|
||||
out["choices"] = []map[string]any{}
|
||||
out["usage"] = u
|
||||
@@ -329,9 +329,9 @@ func chunk(id string, p servePlan, created int64, text *string, finish string, u
|
||||
return out
|
||||
}
|
||||
|
||||
// usage, mümkünse engine'in raporladığı gerçek sayıları kullanır.
|
||||
// res nil olabilir (istek iptal edildiyse veya hata aldıysa); o durumda
|
||||
// prompt token sayısı metinden hesaplanır ve cached_tokens bilinmez.
|
||||
// usage prefers the counts the engine reported. res may be nil (cancelled or
|
||||
// failed request), in which case prompt tokens are derived from the text and
|
||||
// cached_tokens is unknown.
|
||||
func (s *Server) usage(p Prompt, completion int, res *Result) usage {
|
||||
var u usage
|
||||
u.PromptTokens = CountTokens(p.Text)
|
||||
|
||||
+45
-44
@@ -50,7 +50,7 @@ func TestChatCompletionsBuffered(t *testing.T) {
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("durum %d", resp.StatusCode)
|
||||
t.Fatalf("status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
var out struct {
|
||||
@@ -75,17 +75,17 @@ func TestChatCompletionsBuffered(t *testing.T) {
|
||||
t.Fatalf("choices=%d", len(out.Choices))
|
||||
}
|
||||
if n := len(strings.Fields(out.Choices[0].Message.Content)); n != 7 {
|
||||
t.Errorf("beklenen 7 token, alınan %d", n)
|
||||
t.Errorf("expected 7 tokens, got %d", n)
|
||||
}
|
||||
if out.Usage.CompletionTokens != 7 {
|
||||
t.Errorf("usage.completion_tokens=%d", out.Usage.CompletionTokens)
|
||||
}
|
||||
if out.Usage.TotalTokens != out.Usage.PromptTokens+out.Usage.CompletionTokens {
|
||||
t.Error("usage toplamı tutarsız")
|
||||
t.Error("usage totals are inconsistent")
|
||||
}
|
||||
}
|
||||
|
||||
// sseEvent, SSE akışından çözülmüş tek bir olay.
|
||||
// sseEvent is one decoded event from an SSE stream.
|
||||
type sseEvent struct {
|
||||
raw string
|
||||
data map[string]any
|
||||
@@ -107,7 +107,7 @@ func readSSE(t *testing.T, body *bufio.Reader) []sseEvent {
|
||||
ev := sseEvent{raw: payload}
|
||||
if payload != "[DONE]" {
|
||||
if err := json.Unmarshal([]byte(payload), &ev.data); err != nil {
|
||||
t.Fatalf("SSE yükü çözülemedi: %v (%q)", err, payload)
|
||||
t.Fatalf("could not decode SSE payload: %v (%q)", err, payload)
|
||||
}
|
||||
}
|
||||
out = append(out, ev)
|
||||
@@ -132,10 +132,10 @@ func TestChatCompletionsStreaming(t *testing.T) {
|
||||
|
||||
events := readSSE(t, bufio.NewReader(resp.Body))
|
||||
if len(events) == 0 {
|
||||
t.Fatal("hiç SSE olayı gelmedi")
|
||||
t.Fatal("no SSE events received")
|
||||
}
|
||||
if last := events[len(events)-1]; last.raw != "[DONE]" {
|
||||
t.Errorf("akış [DONE] ile bitmedi: %q", last.raw)
|
||||
t.Errorf("stream did not end with [DONE]: %q", last.raw)
|
||||
}
|
||||
|
||||
var contentTokens, usageChunks, finishChunks int
|
||||
@@ -163,17 +163,17 @@ func TestChatCompletionsStreaming(t *testing.T) {
|
||||
}
|
||||
|
||||
if contentTokens != 5 {
|
||||
t.Errorf("beklenen 5 içerik chunk'ı, alınan %d", contentTokens)
|
||||
t.Errorf("expected 5 content chunks, got %d", contentTokens)
|
||||
}
|
||||
if finishChunks != 1 {
|
||||
t.Errorf("beklenen 1 finish_reason chunk'ı, alınan %d", finishChunks)
|
||||
t.Errorf("expected 1 finish_reason chunk, got %d", finishChunks)
|
||||
}
|
||||
if usageChunks != 1 {
|
||||
t.Errorf("include_usage istendi ama %d usage chunk'ı geldi", usageChunks)
|
||||
t.Errorf("include_usage was requested but got %d usage chunks", usageChunks)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamingUsageIstenmezseGonderilmez(t *testing.T) {
|
||||
func TestStreamingOmitsUsageUnlessRequested(t *testing.T) {
|
||||
srv := newTestServer(t, testConfig())
|
||||
|
||||
resp := postJSON(t, srv.URL+"/v1/chat/completions", map[string]any{
|
||||
@@ -188,13 +188,14 @@ func TestStreamingUsageIstenmezseGonderilmez(t *testing.T) {
|
||||
continue
|
||||
}
|
||||
if _, ok := ev.data["usage"]; ok {
|
||||
t.Error("istenmediği halde usage chunk'ı gönderildi")
|
||||
t.Error("usage chunk sent although it was not requested")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestIstemciKopmasiEngineeYansir(t *testing.T) {
|
||||
// Faz 1'in çıkış kriterinin temeli: istemci koptuğunda GPU boşa çalışmamalı.
|
||||
func TestClientDisconnectReachesEngine(t *testing.T) {
|
||||
// The basis of phase 1's exit criterion: when a client disconnects the GPU
|
||||
// must stop working on that request.
|
||||
cfg := testConfig()
|
||||
cfg.DecodeTokensPerSec = 20
|
||||
eng := New(cfg)
|
||||
@@ -220,7 +221,7 @@ func TestIstemciKopmasiEngineeYansir(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// İlk token'ı bekle, sonra istemciyi kopar.
|
||||
// Wait for the first token, then drop the client.
|
||||
waitFor(t, 2*time.Second, func() bool { return eng.Stats().Running == 1 })
|
||||
cancel()
|
||||
_ = resp.Body.Close()
|
||||
@@ -231,7 +232,7 @@ func TestIstemciKopmasiEngineeYansir(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestCompletionsUcNoktasi(t *testing.T) {
|
||||
func TestCompletionsEndpoint(t *testing.T) {
|
||||
srv := newTestServer(t, testConfig())
|
||||
|
||||
resp := postJSON(t, srv.URL+"/v1/completions", map[string]any{
|
||||
@@ -252,11 +253,11 @@ func TestCompletionsUcNoktasi(t *testing.T) {
|
||||
t.Errorf("object=%q", out.Object)
|
||||
}
|
||||
if n := len(strings.Fields(out.Choices[0].Text)); n != 3 {
|
||||
t.Errorf("beklenen 3 token, alınan %d", n)
|
||||
t.Errorf("expected 3 tokens, got %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestModelsVeHealth(t *testing.T) {
|
||||
func TestModelsAndHealth(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.Model = "llama-3.3-70b"
|
||||
srv := newTestServer(t, cfg)
|
||||
@@ -267,7 +268,7 @@ func TestModelsVeHealth(t *testing.T) {
|
||||
}
|
||||
resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Errorf("/health durum %d", resp.StatusCode)
|
||||
t.Errorf("/health status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
resp, err = http.Get(srv.URL + "/v1/models") //nolint:noctx // test
|
||||
@@ -285,17 +286,17 @@ func TestModelsVeHealth(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(out.Data) != 1 || out.Data[0].ID != "llama-3.3-70b" {
|
||||
t.Errorf("/v1/models beklenen modeli listelemedi: %+v", out.Data)
|
||||
t.Errorf("/v1/models did not list the expected model: %+v", out.Data)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMetriklerPrometheusIleParseEdilebilir, §7.6'daki engine kontratını doğrular.
|
||||
// Router bu metrikleri gerçek bir Prometheus parser'ıyla okuyacak; format
|
||||
// bozuksa yönlendirme kör kalır.
|
||||
func TestMetriklerPrometheusIleParseEdilebilir(t *testing.T) {
|
||||
// TestMetricsParseWithPrometheus verifies the engine contract from plan §7.6.
|
||||
// The router reads these metrics with a real Prometheus parser; if the format
|
||||
// is wrong, routing goes blind.
|
||||
func TestMetricsParseWithPrometheus(t *testing.T) {
|
||||
srv := newTestServer(t, testConfig())
|
||||
|
||||
// Biraz trafik üret ki sayaçlar dolsun.
|
||||
// Generate some traffic so the counters are non-zero.
|
||||
resp := postJSON(t, srv.URL+"/v1/chat/completions", map[string]any{
|
||||
"messages": []map[string]string{{"role": "user", "content": "merhaba dunya"}},
|
||||
"max_tokens": 4,
|
||||
@@ -308,26 +309,26 @@ func TestMetriklerPrometheusIleParseEdilebilir(t *testing.T) {
|
||||
}
|
||||
defer mresp.Body.Close()
|
||||
|
||||
// TextParser'ın sıfır değeri geçersiz; şema açıkça verilmeli.
|
||||
// Legacy şema daha sıkı: metrik adlarımızın klasik Prometheus kurallarına
|
||||
// uyduğunu da doğrulamış oluyoruz (vLLM adlarındaki ':' legacy'de de geçerli).
|
||||
// TextParser's zero value is invalid; the scheme must be given explicitly.
|
||||
// The legacy scheme is stricter, so this also checks our metric names obey
|
||||
// classic Prometheus rules (the ':' in vLLM names is legal there too).
|
||||
parser := expfmt.NewTextParser(model.LegacyValidation)
|
||||
families, err := parser.TextToMetricFamilies(mresp.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("metrikler parse edilemedi: %v", err)
|
||||
t.Fatalf("metrics failed to parse: %v", err)
|
||||
}
|
||||
|
||||
// §7.6 zorunlu üçlüsü.
|
||||
// The three metrics plan §7.6 makes mandatory.
|
||||
for _, name := range []string{
|
||||
"vllm:num_requests_running",
|
||||
"vllm:num_requests_waiting",
|
||||
"vllm:kv_cache_usage_perc",
|
||||
} {
|
||||
if _, ok := families[name]; !ok {
|
||||
t.Errorf("zorunlu metrik eksik: %s", name)
|
||||
t.Errorf("mandatory metric missing: %s", name)
|
||||
}
|
||||
}
|
||||
// Opsiyoneller ve histogramlar.
|
||||
// Optional metrics and histograms.
|
||||
for _, name := range []string{
|
||||
"vllm:cache_config_info",
|
||||
"vllm:prefix_cache_queries_total",
|
||||
@@ -335,35 +336,35 @@ func TestMetriklerPrometheusIleParseEdilebilir(t *testing.T) {
|
||||
"vllm:time_to_first_token_seconds",
|
||||
} {
|
||||
if _, ok := families[name]; !ok {
|
||||
t.Errorf("metrik eksik: %s", name)
|
||||
t.Errorf("metric missing: %s", name)
|
||||
}
|
||||
}
|
||||
|
||||
if f, ok := families["vllm:generation_tokens_total"]; ok {
|
||||
if got := f.Metric[0].GetCounter().GetValue(); got != 4 {
|
||||
t.Errorf("generation_tokens_total=%v, beklenen 4", got)
|
||||
t.Errorf("generation_tokens_total=%v, expected 4", got)
|
||||
}
|
||||
} else {
|
||||
t.Error("vllm:generation_tokens_total eksik")
|
||||
t.Error("vllm:generation_tokens_total is missing")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageCachedTokensRaporlanir, prefix cache etkisinin istemciye ulaştığını
|
||||
// doğrular. Router ileride bunu kullanmayacak olsa da, cache'in gerçekten
|
||||
// çalıştığını uçtan uca gösteren tek sinyal bu.
|
||||
func TestUsageCachedTokensRaporlanir(t *testing.T) {
|
||||
// TestUsageReportsCachedTokens verifies the prefix cache's effect reaches the
|
||||
// client. The router will not consume this, but it is the only end-to-end
|
||||
// signal that the cache actually works.
|
||||
func TestUsageReportsCachedTokens(t *testing.T) {
|
||||
srv := newTestServer(t, testConfig())
|
||||
|
||||
body := map[string]any{
|
||||
"messages": []map[string]string{{"role": "user", "content": strings.Repeat("ortak onek. ", 100)}},
|
||||
"messages": []map[string]string{{"role": "user", "content": strings.Repeat("shared prefix. ", 100)}},
|
||||
"max_tokens": 3,
|
||||
}
|
||||
|
||||
// İlk istek cache'i doldurur.
|
||||
// The first request populates the cache.
|
||||
resp := postJSON(t, srv.URL+"/v1/chat/completions", body)
|
||||
resp.Body.Close()
|
||||
|
||||
// İkincisi aynı ön eki kullanmalı.
|
||||
// The second one must reuse that prefix.
|
||||
resp = postJSON(t, srv.URL+"/v1/chat/completions", body)
|
||||
defer resp.Body.Close()
|
||||
|
||||
@@ -374,10 +375,10 @@ func TestUsageCachedTokensRaporlanir(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if out.Usage.PromptTokensDetails.CachedTokens == 0 {
|
||||
t.Error("ikinci istekte cached_tokens 0; prefix cache istemciye raporlanmıyor")
|
||||
t.Error("cached_tokens is 0 on the second request; prefix cache is not reported")
|
||||
}
|
||||
if out.Usage.PromptTokensDetails.CachedTokens > out.Usage.PromptTokens {
|
||||
t.Errorf("cached_tokens (%d) prompt_tokens'tan (%d) büyük",
|
||||
t.Errorf("cached_tokens (%d) exceeds prompt_tokens (%d)",
|
||||
out.Usage.PromptTokensDetails.CachedTokens, out.Usage.PromptTokens)
|
||||
}
|
||||
}
|
||||
@@ -2,20 +2,20 @@ package sim
|
||||
|
||||
import "strings"
|
||||
|
||||
// simWords, üretilen sahte çıktının kelime havuzu.
|
||||
// simWords is the word pool the fake output is drawn from.
|
||||
var simWords = []string{
|
||||
"alfa", "beta", "gama", "delta", "epsilon", "zeta", "eta", "teta",
|
||||
"iota", "kappa", "lambda", "mu", "nu", "ksi", "omikron", "pi",
|
||||
"alpha", "beta", "gamma", "delta", "epsilon", "zeta", "eta", "theta",
|
||||
"iota", "kappa", "lambda", "mu", "nu", "xi", "omicron", "pi",
|
||||
}
|
||||
|
||||
// tokenText, n'inci üretilen token'ın metnini verir (n 1'den başlar).
|
||||
// Deterministik olması şart: finish sırasında cache'e yazılan metinle
|
||||
// istemciye akıtılan metin birebir aynı olmalı.
|
||||
// tokenText returns the text of the n-th generated token (n starts at 1).
|
||||
// It must be deterministic: the text written into the cache on completion has
|
||||
// to match exactly what was streamed to the client.
|
||||
func tokenText(n int) string {
|
||||
return " " + simWords[(n-1)%len(simWords)]
|
||||
}
|
||||
|
||||
// generatedText, ilk n token'ın birleşimini verir.
|
||||
// generatedText returns the concatenation of the first n tokens.
|
||||
func generatedText(n int) string {
|
||||
var b strings.Builder
|
||||
b.Grow(n * 7)
|
||||
|
||||
+10
-10
@@ -2,20 +2,20 @@ package sim
|
||||
|
||||
import "hash/fnv"
|
||||
|
||||
// TokensPerChar, sahte tokenizer'ın kaç bayta bir token ürettiği.
|
||||
// Gerçek BPE tokenizer'lara kaba bir yaklaşım; simülasyon için tek gereken
|
||||
// deterministik ve prefix-kararlı olması.
|
||||
// bytesPerToken is how many bytes the fake tokenizer folds into one token.
|
||||
// A crude approximation of real BPE tokenizers; all the simulation needs is
|
||||
// for it to be deterministic and prefix-stable.
|
||||
const bytesPerToken = 4
|
||||
|
||||
// Token, sahte tokenizer'ın ürettiği token kimliği.
|
||||
// Token is a token id produced by the fake tokenizer.
|
||||
type Token uint32
|
||||
|
||||
// Tokenize, metni deterministik ve prefix-kararlı şekilde token'lara böler.
|
||||
// Tokenize splits text into tokens deterministically and prefix-stably.
|
||||
//
|
||||
// Prefix-kararlılık kritik: aynı ön eke sahip iki metnin token dizileri de aynı
|
||||
// ön ekle başlamalı, yoksa prefix cache simülasyonu anlamsız olur. Bu yüzden
|
||||
// bölme, metnin başından itibaren sabit bayt sayısıyla yapılır — içeriğe bağlı
|
||||
// (BPE gibi) bir bölme kullanılmaz.
|
||||
// Prefix stability is essential: two texts sharing a prefix must produce token
|
||||
// sequences sharing that prefix, otherwise the prefix cache simulation is
|
||||
// meaningless. That is why splitting happens at fixed byte offsets from the
|
||||
// start rather than content-dependent (BPE-style) boundaries.
|
||||
func Tokenize(s string) []Token {
|
||||
if s == "" {
|
||||
return nil
|
||||
@@ -31,7 +31,7 @@ func Tokenize(s string) []Token {
|
||||
return out
|
||||
}
|
||||
|
||||
// CountTokens, metnin token sayısını tokenize etmeden verir.
|
||||
// CountTokens returns the token count for a text without tokenizing it.
|
||||
func CountTokens(s string) int {
|
||||
return (len(s) + bytesPerToken - 1) / bytesPerToken
|
||||
}
|
||||
@@ -2,21 +2,21 @@ package sim
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestTokenizeOnEkKararli(t *testing.T) {
|
||||
func TestTokenizeIsPrefixStable(t *testing.T) {
|
||||
a := Tokenize("merhaba")
|
||||
b := Tokenize("merhaba dunya")
|
||||
if len(a) > len(b) {
|
||||
t.Fatal("uzun metin daha az token verdi")
|
||||
t.Fatal("longer text produced fewer tokens")
|
||||
}
|
||||
for i := range a[:len(a)-1] { // son token yarım olabilir
|
||||
for i := range a[:len(a)-1] { // the last token may be partial
|
||||
if a[i] != b[i] {
|
||||
t.Fatalf("token %d ön ek kararlı değil", i)
|
||||
t.Fatalf("token %d is not prefix stable", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCountTokensTokenizeIleTutarli(t *testing.T) {
|
||||
for _, s := range []string{"", "a", "abcd", "abcde", "uzunca bir prompt metni"} {
|
||||
func TestCountTokensMatchesTokenize(t *testing.T) {
|
||||
for _, s := range []string{"", "a", "abcd", "abcde", "a somewhat longer prompt"} {
|
||||
if got, want := CountTokens(s), len(Tokenize(s)); got != want {
|
||||
t.Errorf("%q: CountTokens=%d Tokenize=%d", s, got, want)
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
// Package version, build sırasında ldflags ile doldurulan sürüm bilgisini tutar.
|
||||
// Package version holds build information injected via ldflags.
|
||||
package version
|
||||
|
||||
// Version, `make build` tarafından git describe çıktısıyla doldurulur.
|
||||
// Version is filled in by `make build` from git describe output.
|
||||
var Version = "dev"
|
||||
+564
-531
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user