fix: WithDefaultTimeout 不再套用到流式输出/双向流
流式输出(Invoke[chan T])返回的 channel 由调用方通过 range 自行决定 消费多久,典型场景是流式聊天回复,可能正常持续几十秒甚至更久。之前 WithDefaultTimeout 会不分场景地把这个全局默认超时套到整个调用生命周期 上(包括流还在正常输出的过程中),导致长时间运行的正常流式响应被腰斩。 改成流式输出/双向流完全不受 WithDefaultTimeout 影响,只认调用方显式传入 的 ctx deadline;需要超时保护的话必须自己 context.WithTimeout。普通调用 和流式输入不受影响,继续吃池子的默认超时。 新增 TestStreamOutIgnoresDefaultTimeout 验证;example/main.go 的 demoTimeout 补充示例4 演示这一行为;README 同步更新。
This commit is contained in:
@@ -209,16 +209,25 @@ defer cancel()
|
|||||||
result, err := gobridge.Invoke[string](ctx, pool, "sleep_seconds", 1.0) // 正常返回
|
result, err := gobridge.Invoke[string](ctx, pool, "sleep_seconds", 1.0) // 正常返回
|
||||||
```
|
```
|
||||||
|
|
||||||
|
> **`WithDefaultTimeout` 只对普通调用和流式输入生效,对流式输出/双向流完全不生效**(即使不设置也不会被套用)。
|
||||||
|
> 原因:`Invoke[chan T]` 返回的 channel 由调用方通过 `range` 自行决定消费多久——典型场景是流式聊天回复,
|
||||||
|
> 可能正常持续几十秒甚至更久,如果套一个全局默认超时,会在输出还在正常进行时把它腰斩。
|
||||||
|
> 流式输出/双向流的调用**只认调用方显式传入的 deadline**;不想让它无限期阻塞(包括等待连接、等待数据),
|
||||||
|
> 必须自己 `context.WithTimeout` 包一层传进去,`WithDefaultTimeout` 在这里不提供任何兜底。
|
||||||
|
|
||||||
**超时是否会通过 `error` 返回,取决于调用模式:**
|
**超时是否会通过 `error` 返回,取决于调用模式:**
|
||||||
|
|
||||||
| 调用模式 | 超时表现 |
|
| 调用模式 | `WithDefaultTimeout` 是否生效 | 超时表现 |
|
||||||
|---|---|
|
|---|---|---|
|
||||||
| 普通调用 / 流式输入 | `Invoke` 返回非 nil 的 `error`(`context.DeadlineExceeded` / `context.Canceled`) |
|
| 普通调用 / 流式输入 | 生效 | `Invoke` 返回非 nil 的 `error`(`context.DeadlineExceeded` / `context.Canceled`) |
|
||||||
| 流式输出 / 双向流 | `Invoke` 建立阶段失败会返回 `error`;**建立成功后**若中途超时,只会静默关闭已返回的 channel,不会有第二次 `error` |
|
| 流式输出 / 双向流 | **不生效**,只认显式 deadline | `Invoke` 建立阶段失败会返回 `error`;**建立成功后**若显式 ctx 超时,只会静默关闭已返回的 channel,不会有第二次 `error` |
|
||||||
|
|
||||||
流式模式下,`for v := range ch` 结束后无法区分"正常读完"还是"被超时打断",需要调用方自行检查传入的 `ctx.Err()`:
|
流式模式下,`for v := range ch` 结束后无法区分"正常读完"还是"被超时打断",需要调用方自行检查传入的 `ctx.Err()`:
|
||||||
|
|
||||||
```go
|
```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)
|
ch, err := gobridge.Invoke[chan int](ctx, pool, "slow_range_gen", 1, 10, 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// 建立阶段失败/超时
|
// 建立阶段失败/超时
|
||||||
|
|||||||
@@ -180,20 +180,19 @@ func invokeRegular[R any](ctx context.Context, pool Pool, method string, args ..
|
|||||||
return result, nil
|
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) {
|
func invokeStreamOut[R any](ctx context.Context, pool Pool, method string, rt reflect.Type, args ...any) (R, error) {
|
||||||
var zero R
|
var zero R
|
||||||
|
|
||||||
ctx, cancel := applyDefaultTimeout(ctx, pool)
|
|
||||||
|
|
||||||
argsJSON, err := json.Marshal(args)
|
argsJSON, err := json.Marshal(args)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cancel()
|
|
||||||
return zero, fmt.Errorf("marshal args: %w", err)
|
return zero, fmt.Errorf("marshal args: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
conn, w, err := pool.acquire(ctx)
|
conn, w, err := pool.acquire(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cancel()
|
|
||||||
return zero, err
|
return zero, err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -205,7 +204,6 @@ func invokeStreamOut[R any](ctx context.Context, pool Pool, method string, rt re
|
|||||||
Args: argsJSON,
|
Args: argsJSON,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
w.release(conn, false)
|
w.release(conn, false)
|
||||||
cancel()
|
|
||||||
return zero, contextErr(ctx, fmt.Errorf("write call: %w", err))
|
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()
|
stop()
|
||||||
ch.Close()
|
ch.Close()
|
||||||
w.release(conn, ctx.Err() == nil)
|
w.release(conn, ctx.Err() == nil)
|
||||||
cancel()
|
|
||||||
}()
|
}()
|
||||||
for {
|
for {
|
||||||
msg, err := readResult(ctx, conn, pool, write)
|
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
|
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) {
|
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
|
var zero R
|
||||||
|
|
||||||
ctx, cancel := applyDefaultTimeout(ctx, pool)
|
|
||||||
|
|
||||||
jsonArgs := make([]any, len(args))
|
jsonArgs := make([]any, len(args))
|
||||||
copy(jsonArgs, args)
|
copy(jsonArgs, args)
|
||||||
jsonArgs[streamArgIdx] = nil
|
jsonArgs[streamArgIdx] = nil
|
||||||
|
|
||||||
argsJSON, err := json.Marshal(jsonArgs)
|
argsJSON, err := json.Marshal(jsonArgs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cancel()
|
|
||||||
return zero, fmt.Errorf("marshal args: %w", err)
|
return zero, fmt.Errorf("marshal args: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
conn, w, err := pool.acquire(ctx)
|
conn, w, err := pool.acquire(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cancel()
|
|
||||||
return zero, err
|
return zero, err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -359,7 +355,6 @@ func invokeStreamBoth[R any](ctx context.Context, pool Pool, method string, stre
|
|||||||
StreamArgIdx: streamArgIdx,
|
StreamArgIdx: streamArgIdx,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
w.release(conn, false)
|
w.release(conn, false)
|
||||||
cancel()
|
|
||||||
return zero, contextErr(ctx, fmt.Errorf("write call: %w", err))
|
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
|
var mu sync.Mutex
|
||||||
write := func(msg Message) { mu.Lock(); writeMsg(conn, msg); mu.Unlock() } //nolint
|
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
|
// 写入 goroutine:输入 channel → Python chunks
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
|
||||||
for {
|
for {
|
||||||
val, ok, cancelled := chanRecv(ctx, streamCh)
|
val, ok, cancelled := chanRecv(ctx, streamCh)
|
||||||
if cancelled || !ok {
|
if cancelled || !ok {
|
||||||
@@ -398,7 +386,6 @@ func invokeStreamBoth[R any](ctx context.Context, pool Pool, method string, stre
|
|||||||
|
|
||||||
// 读取 goroutine:Python chunks → 输出 channel,内联处理 callback
|
// 读取 goroutine:Python chunks → 输出 channel,内联处理 callback
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
|
||||||
stop := watchCtx(ctx, conn, id, write)
|
stop := watchCtx(ctx, conn, id, write)
|
||||||
defer func() {
|
defer func() {
|
||||||
stop()
|
stop()
|
||||||
|
|||||||
@@ -323,6 +323,19 @@ func demoTimeout(script string) {
|
|||||||
if ctx3.Err() != nil {
|
if ctx3.Err() != nil {
|
||||||
fmt.Println("channel 提前关闭,ctx.Err() =", ctx3.Err())
|
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) {
|
func demoBlocking(script string) {
|
||||||
|
|||||||
@@ -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 线程阻塞在
|
// TestStreamInputCancelDoesNotLeakThread 验证流式输入模式下,handler 线程阻塞在
|
||||||
// _ChunkIter.__next__ → chunk_q.get() 等待下一个输入块时,如果 ctx 取消导致连接被
|
// _ChunkIter.__next__ → chunk_q.get() 等待下一个输入块时,如果 ctx 取消导致连接被
|
||||||
// Go 关闭,Python 侧能否真正把这个线程唤醒退出,而不是永久卡住。
|
// Go 关闭,Python 侧能否真正把这个线程唤醒退出,而不是永久卡住。
|
||||||
|
|||||||
Reference in New Issue
Block a user