fix(memory): 冷启动放宽改为一次性 + 试算轮区分「没候选」与「向量不可用」
review round1 的 p1/p2 收口（commit 5967272 之后）。
p1：旧实现 `warmed` 只在 2xx 之后才翻 ⇒ 注释里写的「只在第一次成功响应之前」
实际是「服务坏着的全部时间」。reviewer 实测「接受连接但永不回包」时**每次**调用
都等满 30s（fail-fast 变 fail-slow）：审查 20s 预算被一次吃光、索引写入拖成
8×30s、开机探针阻塞 30s。
- `warmupLeft` 改为**一次性许可**（CompareAndSwap 取走即失效，成败都算花掉）
⇒ 挂死时只有第一个请求慢一次，此后一律 DefaultRequestTimeout。
- `warmupTimeout` 30s → 8s：与常态 5s 同一量级（≤2×），且严格小于审查总预算
20s（一次冷启动不许吃光整轮）；两条界都由 TestDefaultRequestTimeout 守住。
- 放宽被花掉且没成功时留一条 WARN（事后可复核「第一个请求」这个判据）。
- 更正两处自述：api.go 的「不会被 30s 拖住」与 budget.go 的不变式说明（批次形状
永远按稳态超时推导；冷启动只放宽第一个请求的截止时刻）。
- p2：删掉零调用者的 `WithRequestTimeout` —— 单次超时只允许改常量，批次上限与它
同源并由编译期/常量断言复核。
p2：试算轮在向量源不可用时 `RecallMode` 与「没有候选可比」同为空串。
- `recallLedger` 增 `unavailable` 计数 + `noteUnavailable()`（strictRecall 失败
那一笔记一笔）；试算汇总行新增 `recall_unavailable_partitions`，**成败都记**。
- 不扩 `recallMode` 取值集合：失败那一路不序列化 result（响应是
500 tidy_recall_unavailable），新取值只会出现在日志里却要动一次契约，不划算。
测试：三种失败形态 × 两次调用的耗时断言（连接被拒 / 500 / 永不回包），
试算轮的「不可用」与「没候选」两态日志断言；两处变异自证（warmupTimeout 调回
30s ⇒ TestDefaultRequestTimeout 红；去掉 noteUnavailable ⇒ 试算汇总行退化成
recall_mode="" + recall_unavailable_partitions=0）。
This commit is contained in:
AskaEth committed
1 parent
59672720bd
commit
6c30ca540a
7 files changed
+273
-72
No files matched your search
@@ -53,7 +53,10 @@ const (
|
||||
// 64 条的实测耗时 7.5~8.5s 已经越过它，客户端放弃、服务端还在算，把整台本机
|
||||
// embedding 服务顶住，连带单条请求（写向量索引）也一起超时。所以修法是**把批次
|
||||
// 收进预算内**，而不是把预算放大到迁就批次（后者会让真挂死的请求也拖住整轮）。
|
||||
// 冷启动那一次另有 warmupTimeout（见 budget.go），它只生效到第一次成功响应为止。
|
||||
//
|
||||
// 冷启动另有 warmupTimeout（见 budget.go），但它**只放宽本进程的第一个请求**、
|
||||
// 且一次性花掉：失败不会让它续期，因此「服务一直挂死」这件事**不会**变成
|
||||
// 「每次调用都慢」—— 第二个请求起就回到本常量。
|
||||
const DefaultRequestTimeout = 5 * time.Second
|
||||
|
||||
// openAICompatibleRequest is the JSON body sent to an OpenAI-compatible
|
||||
@@ -92,12 +95,11 @@ type APIEmbedder struct {
|
||||
cfg config.EmbeddingConfig
|
||||
client *http.Client
|
||||
log *slog.Logger
|
||||
// timeout 是**单次**请求的墙钟上限，与批次上限同源（见 budget.go）。
|
||||
// 默认 DefaultRequestTimeout；WithRequestTimeout 可换，换了批次上限也跟着换。
|
||||
timeout time.Duration
|
||||
// warmed 报告是否已经拿到过成功响应。未热之前用 warmupTimeout（模型加载 /
|
||||
// GPU 缓冲分配），热之后一律用 timeout。
|
||||
warmed atomic.Bool
|
||||
// warmupLeft 是**一次性**冷启动放宽的开关：为真时下一个请求用 warmupTimeout，
|
||||
// 取走即置假（CompareAndSwap ⇒ 并发下也只有一个请求能拿到）。
|
||||
// 它**不是**「未成功就一直放宽」—— 失败也照样花掉这一次，因此服务挂死时
|
||||
// 只会慢一次，之后每个请求都按 DefaultRequestTimeout 快速失败（见 budget.go）。
|
||||
warmupLeft atomic.Bool
|
||||
|
||||
mu sync.RWMutex
|
||||
dim int
|
||||
@@ -135,25 +137,17 @@ func WithLogger(l *slog.Logger) APIOption {
|
||||
}
|
||||
}
|
||||
|
||||
// WithRequestTimeout 换掉单次请求的墙钟上限（非正数忽略，保留默认）。
|
||||
// 它同时决定批次上限 —— 两者是同一条推演（budget.go），因此这里是**唯一**的
|
||||
// 调优点，不允许调用方各自去改批次。
|
||||
func WithRequestTimeout(d time.Duration) APIOption {
|
||||
return func(e *APIEmbedder) {
|
||||
if d > 0 {
|
||||
e.timeout = d
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// NewAPI builds an APIEmbedder from cfg using the default HTTP client.
|
||||
// 单次超时不可由选项更换：它是 budget.go 那条推演的输入，批次上限与它同源，
|
||||
// 因此**只允许改常量**（改了之后批次上限自动跟着改，测试的编译期/常量断言立刻复核）。
|
||||
func NewAPI(cfg config.EmbeddingConfig, opts ...APIOption) *APIEmbedder {
|
||||
e := &APIEmbedder{
|
||||
cfg: cfg,
|
||||
client: defaultHTTPClient(),
|
||||
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
|
||||
timeout: DefaultRequestTimeout,
|
||||
cfg: cfg,
|
||||
client: defaultHTTPClient(),
|
||||
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
|
||||
}
|
||||
// 冷启动放宽是一个**一次性**许可：见 APIEmbedder.warmupLeft。
|
||||
e.warmupLeft.Store(true)
|
||||
for _, opt := range opts {
|
||||
opt(e)
|
||||
}
|
||||
@@ -187,10 +181,10 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro
|
||||
return nil, ErrNoAPIURL
|
||||
}
|
||||
|
||||
// 切片用的上限由**稳态**超时推出（e.timeout），与 requestTimeout() 无关：
|
||||
// 冷启动那一次只是**截止时刻**更宽（warmupTimeout），批次形状不变 —— 否则
|
||||
// 首个请求会按 30s 的预算切出一个巨大的批，正好把冷启动的模型顶得更久。
|
||||
items, chars := batchLimitsFor(e.timeout)
|
||||
// 切片用的上限由**稳态**超时推出（budget.go 的 DefaultRequestTimeout）：
|
||||
// 冷启动那一次只放宽**截止时刻**，批形状不变 —— 否则首个请求会按更宽的预算
|
||||
// 切出一个巨大的批，正好把冷启动的模型顶得更久。
|
||||
items, chars := batchLimitsFor(DefaultRequestTimeout)
|
||||
out := make([]Vector, 0, len(texts))
|
||||
for _, chunk := range chunkTexts(texts, items, chars) {
|
||||
vectors, err := e.embedChunk(ctx, chunk)
|
||||
@@ -202,19 +196,25 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// requestTimeout 返回本次调用该用的单次超时：拿到第一个成功响应之前用
|
||||
// warmupTimeout（冷启动：模型加载 / GPU 缓冲分配），之后一律用配置的 timeout。
|
||||
func (e *APIEmbedder) requestTimeout() time.Duration {
|
||||
if e.warmed.Load() {
|
||||
return e.timeout
|
||||
// callTimeout 返回本次请求该用的单次超时，并**当场花掉**一次性冷启动放宽许可
|
||||
// （成功、失败都算花掉）。因此：
|
||||
// - 本进程的第一个请求最多等 warmupTimeout（冷启动），
|
||||
// - 从第二个请求起一律 DefaultRequestTimeout —— 服务挂死时不会每次都慢。
|
||||
//
|
||||
// 返回值 relaxed 表示这一次是不是拿到了放宽，供失败时留痕。
|
||||
func (e *APIEmbedder) callTimeout() (timeout time.Duration, relaxed bool) {
|
||||
if e.warmupLeft.CompareAndSwap(true, false) {
|
||||
return warmupTimeout, true
|
||||
}
|
||||
return warmupTimeout
|
||||
return DefaultRequestTimeout, false
|
||||
}
|
||||
|
||||
// embedChunk 发**一次** /embeddings 请求。chunk 必须已经落在 batchLimitsFor 推出来
|
||||
// 的边界内（Embed 负责切），因此它的耗时上界不超过超时的一半。
|
||||
// 的边界内（Embed 负责切），因此它的耗时上界不超过单次超时的一半。
|
||||
func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector, error) {
|
||||
callCtx, cancel := context.WithTimeout(ctx, e.requestTimeout())
|
||||
// 先取超时（取走即花掉那次放宽），再建 ctx：放宽许可与这次请求一一对应。
|
||||
timeout, relaxed := e.callTimeout()
|
||||
callCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
req, err := e.newEmbeddingRequest(callCtx, texts)
|
||||
@@ -224,6 +224,16 @@ func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector,
|
||||
|
||||
resp, err := e.client.Do(req)
|
||||
if err != nil {
|
||||
// 冷启动放宽被花掉且没成功时留一条 WARN：它是「本进程第一次请求」这个
|
||||
// 判断唯一能被事后复核的地方（此后一切失败都按常态超时快速失败）。
|
||||
// 只记两个预算与错误（错误文本里已带端点，不另记 URL，避免重复落一条
|
||||
// 可能含 userinfo 的地址）。
|
||||
if relaxed {
|
||||
e.log.Warn("embedding: one-shot cold-start allowance spent without a successful response",
|
||||
"warmup_timeout", warmupTimeout.String(),
|
||||
"steady_timeout", DefaultRequestTimeout.String(),
|
||||
"error", err)
|
||||
}
|
||||
return nil, fmt.Errorf("embedding: request: %w", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
@@ -232,8 +242,6 @@ func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector,
|
||||
msg := readErrorBody(resp.Body)
|
||||
return nil, fmt.Errorf("embedding: status %d: %s", resp.StatusCode, msg)
|
||||
}
|
||||
// 服务端确实答了 ⇒ 冷启动这一关过了，后续调用回到常态超时。
|
||||
e.warmed.Store(true)
|
||||
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
|
||||
if err != nil {
|
||||
|
||||
Reference in new issue
Block a user