lizining1231 commented on code in PR #3746: URL: https://github.com/apache/dubbo-go/pull/3746#discussion_r4064518711
########## protocol/triple/triple_protocol/buffered_writer.go: ########## @@ -0,0 +1,160 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package triple_protocol + +import ( + "bytes" + "io" + "sync" +) + +// defaultStreamWriteBufSize is the aggregation threshold for the streaming +// write buffer. It matches the gRPC-Go default write buffer size: large +// enough to batch small, high-rate messages together, small enough to bound +// how much a burst of tiny messages spends buffered (and therefore delays) +// before hitting the wire. +const defaultStreamWriteBufSize = 32 << 10 // 32 KiB + +// streamBufferWriter is an opt-in aggregation layer over a duplexHTTPCall for +// streaming requests. Without it, every streamed message is flushed down the +// io.Pipe alone, paying a synchronous cross-goroutine handshake per message. +// streamBufferWriter amortizes those handshakes by coalescing small messages +// in a capacity-bounded buffer and flushing to the underlying writer only when +// the buffer fills (or when Flush or Close is called). It is safe for +// concurrent use; StreamingClientConn requires Send and CloseRequest to be +// concurrency-safe. +type streamBufferWriter struct { + mu sync.Mutex + + next io.Writer // wrapped duplexHTTPCall + limit int + buf *bytes.Buffer + + err error + closed bool +} + +// newStreamBufferWriter wraps next with an aggregation buffer of default +// capacity. +func newStreamBufferWriter(next io.Writer) *streamBufferWriter { + return &streamBufferWriter{ + next: next, + limit: defaultStreamWriteBufSize, + buf: makeStreamWriteBuf(), + } +} + +// makeStreamWriteBuf preallocates the aggregation capacity. +func makeStreamWriteBuf() *bytes.Buffer { + return bytes.NewBuffer(make([]byte, 0, defaultStreamWriteBufSize)) +} + +// Write coalesces a small message into the buffer, flushing the batch once the +// buffer reaches its limit. Payloads at or over the limit bypass the buffer +// and go directly to the underlying writer so a single big message is not +// held in memory twice. +func (w *streamBufferWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + + if w.err != nil { + return 0, w.err + } + if w.closed { + // Mirror duplexHTTPCall.Write: a late Send after the write side closed + // reports io.EOF. + return 0, io.EOF + } + if len(p) >= w.limit { + if err := w.flushLocked(); err != nil { + return 0, err + } + n, err := w.next.Write(p) + if err == nil && n < len(p) { + // Same short-write guard as flushLocked: freeze the buffer. + err = io.ErrShortWrite + } + if err != nil { + w.err = err + w.closed = true + } + return n, err + } + w.buf.Write(p) + if w.buf.Len() >= w.limit && w.flushLocked() != nil { + return len(p), w.err + } + return len(p), nil +} Review Comment: 感谢review！情况属实，已复现、修复并追加测试与文档变动 ### 复现 把 `grpcClientConn.Receive` 里的 `flushBeforeWait()` 摘掉（等价于修复前） ```text $ go test -timeout 3s -count=1 -run TestWriteBufferingBidiPingPong ./protocol/triple/triple_protocol/ panic: test timed out after 3s running tests: TestWriteBufferingBidiPingPong (3s) goroutine 8 [chan receive]: dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol.(*duplexHTTPCall).BlockUntilResponseReady(...) .../triple_protocol/duplex_http_call.go:262 dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol.(*grpcClientConn).Receive(...) .../triple_protocol/protocol_grpc.go:416 dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol.(*errorTranslatingClientConn).Receive(...) .../triple_protocol/protocol.go:202 dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol.(*BidiStreamForClient).Receive(...) .../triple_protocol/client_stream.go:246 dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol_test.TestWriteBufferingBidiPingPong.func2() .../triple_protocol/write_buffering_bidi_ext_test.go:98 ``` ### 修复 采纳方案 1：在所有等待响应侧的入口前置 flush。具体改动如下： 代码变更 - `grpcClientConn` 与`tripleUnaryClientConn` 各新增`flushBeforeWait()` ，并在`Receive` 、`ResponseHeader` 、`ResponseTrailer` 、`CloseResponse` 四个入口前置调用，两条 wire 对称。 - 错误处理 ：flush 失败走既有的`SetError` 通道，不引入新的错误形态；flush 必然先触发`ensureRequestMade()` ，所以`responseReady` 一定会 close，防止另一种挂起等待 文档变更 - `WithWriteBuffering()` 注释写清三个 flush 时机（水位 /`CloseRequest` / 开始等待响应），声明首次`Send` 不再立刻把请求头发上线，并补上收益边界：严格乒乓下每次`Receive` 只刷一条、批大小退化为 1，无法省syscall，收益主要体现在`连发多条再读`或`只在`CloseRequest` 收尾`的流路径。 测试变更 - `TestWriteBufferingBidiPingPong` ：gRPC wire 单 goroutine`Send` →`Receive` 交替三次，全部依赖`Receive` 自己触发 flush。 - `TestWriteBufferingBidiWaitEntries` ：表驱动覆盖`Receive` 、`ResponseHeader` 、`ResponseTrailer` 、`CloseResponse` 四个入口，各独占一条 bidi 流。 - `TestWriteBufferingTripleWireUnary` ：Triple wire 的 buffered unary 端到端调用。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
