From 6ccec66a1ea4430cdb845cf371529038687dc607 Mon Sep 17 00:00:00 2001 From: what Date: Thu, 23 Jul 2026 14:24:10 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20WithDefaultTimeout=20=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E5=A5=97=E7=94=A8=E5=88=B0=E6=B5=81=E5=BC=8F=E8=BE=93=E5=87=BA?= =?UTF-8?q?/=E5=8F=8C=E5=90=91=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 流式输出（Invoke[chan T]）返回的 channel 由调用方通过 range 自行决定 消费多久，典型场景是流式聊天回复，可能正常持续几十秒甚至更久。之前 WithDefaultTimeout 会不分场景地把这个全局默认超时套到整个调用生命周期 上（包括流还在正常输出的过程中），导致长时间运行的正常流式响应被腰斩。 改成流式输出/双向流完全不受 WithDefaultTimeout 影响，只认调用方显式传入 的 ctx deadline；需要超时保护的话必须自己 context.WithTimeout。普通调用 和流式输入不受影响，继续吃池子的默认超时。 新增 TestStreamOutIgnoresDefaultTimeout 验证；example/main.go 的 demoTimeout 补充示例4 演示这一行为；README 同步更新。 --- README.md | 17 +++++++++++++---- client.go | 25 ++++++------------------- example/main.go | 13 +++++++++++++ pool_test.go | 25 +++++++++++++++++++++++++ 4 files changed, 57 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 8a8ae87..25cb7d4 100644 --- a/README.md +++ b/README.md @@ -209,16 +209,25 @@ defer cancel() result, err := gobridge.Invoke[string](ctx, pool, "sleep_seconds", 1.0) // 正常返回 ``` +> **`WithDefaultTimeout` 只对普通调用和流式输入生效，对流式输出/双向流完全不生效**（即使不设置也不会被套用）。 +> 原因：`Invoke[chan T]` 返回的 channel 由调用方通过 `range` 自行决定消费多久——典型场景是流式聊天回复， +> 可能正常持续几十秒甚至更久，如果套一个全局默认超时，会在输出还在正常进行时把它腰斩。 +> 流式输出/双向流的调用**只认调用方显式传入的 deadline**；不想让它无限期阻塞（包括等待连接、等待数据）， +> 必须自己 `context.WithTimeout` 包一层传进去，`WithDefaultTimeout` 在这里不提供任何兜底。 + **超时是否会通过 `error` 返回，取决于调用模式：** -| 调用模式 | 超时表现 | -|---|---| -| 普通调用 / 流式输入 | `Invoke` 返回非 nil 的 `error`（`context.DeadlineExceeded` / `context.Canceled`） | -| 流式输出 / 双向流 | `Invoke` 建立阶段失败会返回 `error`；**建立成功后**若中途超时，只会静默关闭已返回的 channel，不会有第二次 `error` | +| 调用模式 | `WithDefaultTimeout` 是否生效 | 超时表现 | +|---|---|---| +| 普通调用 / 流式输入 | 生效 | `Invoke` 返回非 nil 的 `error`（`context.DeadlineExceeded` / `context.Canceled`） | +| 流式输出 / 双向流 | **不生效**，只认显式 deadline | `Invoke` 建立阶段失败会返回 `error`；**建立成功后**若显式 ctx 超时，只会静默关闭已返回的 channel，不会有第二次 `error` | 流式模式下，`for v := range ch` 结束后无法区分"正常读完"还是"被超时打断"，需要调用方自行检查传入的 `ctx.Err()`： ```go +// 流式输出必须自己设置超时，WithDefaultTimeout 不会兜底 +ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) +defer cancel() ch, err := gobridge.Invoke[chan int](ctx, pool, "slow_range_gen", 1, 10, 100) if err != nil { // 建立阶段失败/超时 diff --git a/client.go b/client.go index 930207f..9fa549c 100644 --- a/client.go +++ b/client.go @@ -180,20 +180,19 @@ func invokeRegular[R any](ctx context.Context, pool Pool, method string, args .. return result, nil } +// invokeStreamOut 不套用 WithDefaultTimeout：返回的 channel 生命周期由调用方通过 +// range 消费决定，可能持续很久（比如流式聊天回复），套一个全局默认超时会在正常 +// 流式输出过程中把它腰斩。想要超时保护的话，调用方必须显式传入带 deadline 的 ctx。 func invokeStreamOut[R any](ctx context.Context, pool Pool, method string, rt reflect.Type, args ...any) (R, error) { var zero R - ctx, cancel := applyDefaultTimeout(ctx, pool) - argsJSON, err := json.Marshal(args) if err != nil { - cancel() return zero, fmt.Errorf("marshal args: %w", err) } conn, w, err := pool.acquire(ctx) if err != nil { - cancel() return zero, err } @@ -205,7 +204,6 @@ func invokeStreamOut[R any](ctx context.Context, pool Pool, method string, rt re Args: argsJSON, }); err != nil { w.release(conn, false) - cancel() return zero, contextErr(ctx, fmt.Errorf("write call: %w", err)) } @@ -219,7 +217,6 @@ func invokeStreamOut[R any](ctx context.Context, pool Pool, method string, rt re stop() ch.Close() w.release(conn, ctx.Err() == nil) - cancel() }() for { msg, err := readResult(ctx, conn, pool, write) @@ -328,24 +325,23 @@ func invokeStreamIn[R any](ctx context.Context, pool Pool, method string, stream return result, nil } +// invokeStreamBoth 同样不套用 WithDefaultTimeout，理由同 invokeStreamOut—— +// 双向流的生命周期由输入/输出两端共同决定，可能持续很久，需要超时保护时调用方 +// 必须显式传入带 deadline 的 ctx。 func invokeStreamBoth[R any](ctx context.Context, pool Pool, method string, streamArgIdx int, streamCh reflect.Value, rt reflect.Type, args ...any) (R, error) { var zero R - ctx, cancel := applyDefaultTimeout(ctx, pool) - jsonArgs := make([]any, len(args)) copy(jsonArgs, args) jsonArgs[streamArgIdx] = nil argsJSON, err := json.Marshal(jsonArgs) if err != nil { - cancel() return zero, fmt.Errorf("marshal args: %w", err) } conn, w, err := pool.acquire(ctx) if err != nil { - cancel() return zero, err } @@ -359,7 +355,6 @@ func invokeStreamBoth[R any](ctx context.Context, pool Pool, method string, stre StreamArgIdx: streamArgIdx, }); err != nil { w.release(conn, false) - cancel() return zero, contextErr(ctx, fmt.Errorf("write call: %w", err)) } @@ -368,15 +363,8 @@ func invokeStreamBoth[R any](ctx context.Context, pool Pool, method string, stre var mu sync.Mutex write := func(msg Message) { mu.Lock(); writeMsg(conn, msg); mu.Unlock() } //nolint - // 两个 goroutine 共用同一个 ctx，defaultTimeout 产生的 cancel 必须等两者都结束才能调用， - // 否则先完成的一方会提前取消掉另一方仍在进行的操作。 - var wg sync.WaitGroup - wg.Add(2) - go func() { wg.Wait(); cancel() }() - // 写入 goroutine：输入 channel → Python chunks go func() { - defer wg.Done() for { val, ok, cancelled := chanRecv(ctx, streamCh) if cancelled || !ok { @@ -398,7 +386,6 @@ func invokeStreamBoth[R any](ctx context.Context, pool Pool, method string, stre // 读取 goroutine：Python chunks → 输出 channel，内联处理 callback go func() { - defer wg.Done() stop := watchCtx(ctx, conn, id, write) defer func() { stop() diff --git a/example/main.go b/example/main.go index e20108f..0c6b5fb 100644 --- a/example/main.go +++ b/example/main.go @@ -323,6 +323,19 @@ func demoTimeout(script string) { if ctx3.Err() != nil { fmt.Println("channel 提前关闭，ctx.Err() =", ctx3.Err()) } + + // ── 示例4：WithDefaultTimeout 对流式输出完全不生效——即使不设 ctx deadline， + // 池子的默认超时（这里是 500ms）也不会把一个耗时 900ms 的正常流式输出腰斩。 + // 想要超时保护，流式调用必须自己显式传 deadline（见示例3）。 + ch2, err := gobridge.Invoke[chan int](context.Background(), pool, "slow_range_gen", 1, 10, 100) + if err != nil { + log.Fatal(err) + } + fmt.Print("slow_range_gen(1, 10, 100ms)（无 ctx deadline，池默认超时 500ms 但流式输出不受影响）=") + for v := range ch2 { + fmt.Print(" ", v) + } + fmt.Println() // 应该完整输出 1~9，不会被 500ms 默认超时打断 } func demoBlocking(script string) { diff --git a/pool_test.go b/pool_test.go index 3a97192..8a5a640 100644 --- a/pool_test.go +++ b/pool_test.go @@ -177,6 +177,31 @@ func TestStreamErrorMidwayCorruptsNextCall(t *testing.T) { } } +// TestStreamOutIgnoresDefaultTimeout 验证 WithDefaultTimeout 不会套用到流式输出/ +// 双向流调用上：流式聊天这类响应可能持续很久，如果套用全局默认超时会在正常输出 +// 过程中把它腰斩。没有显式传入 deadline 时，流式调用应该完全不受池子默认超时影响， +// 只受调用方显式设置的 ctx 约束。 +func TestStreamOutIgnoresDefaultTimeout(t *testing.T) { + pool := newTestPool(t, WithWorkers(1), WithMaxConns(1), WithDefaultTimeout(200*time.Millisecond)) + + // slow_range_gen(1, 10, 100) 总耗时 ~900ms，远超 200ms 的默认超时。 + // 用 context.Background()（无 deadline）发起调用：如果默认超时被错误地套用到 + // 流式输出上，channel 会在 ~200ms 提前关闭，读不满 9 个元素。 + ch, err := Invoke[chan int](context.Background(), pool, "slow_range_gen", 1, 10, 100) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + + var got []int + for v := range ch { + got = append(got, v) + } + + if len(got) != 9 { + t.Fatalf("want 9 items (WithDefaultTimeout 不应影响流式输出), got %v", got) + } +} + // TestStreamInputCancelDoesNotLeakThread 验证流式输入模式下，handler 线程阻塞在 // _ChunkIter.__next__ → chunk_q.get() 等待下一个输入块时，如果 ctx 取消导致连接被 // Go 关闭，Python 侧能否真正把这个线程唤醒退出，而不是永久卡住。