lizining1231 commented on code in PR #3706: URL: https://github.com/apache/dubbo-go/pull/3706#discussion_r3870754195
########## protocol/triple/triple_protocol/unary_fastpath.go: ########## @@ -0,0 +1,286 @@ +/* + * 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" + "context" + "fmt" + "io" + "net/http" + "net/url" + "sync" +) + +// unaryClientCall is the subset of *duplexHTTPCall that tripleUnaryClientConn +// depends on. unary calls use unaryFastPathCall; streaming calls keep using +// duplexHTTPCall. +type unaryClientCall interface { + io.Writer + io.Reader + Header() http.Header + CloseWrite() error + CloseRead() error + BlockUntilResponseReady() + SetValidateResponse(func(*http.Response) *Error) + SetError(error) + ResponseTrailer() http.Header +} + +// Both the original and the fast path must satisfy the interface. +var ( + _ unaryClientCall = (*duplexHTTPCall)(nil) + _ unaryClientCall = (*unaryFastPathCall)(nil) +) + +// unaryFastPathCall is a synchronous replacement for duplexHTTPCall on the +// unary hot path. Write accumulates the marshaled payload into a pooled +// buffer; CloseWrite issues the request synchronously (no io.Pipe, no +// per-request goroutine). +type unaryFastPathCall struct { + ctx context.Context + httpClient HTTPClient + request *http.Request + bufferPool *bufferPool + validateResponse func(*http.Response) *Error + + // writeMu serializes Write against CloseWrite. StreamingClientConn's + // contract requires Send and CloseRequest to be safe to call concurrently; + // CloseWrite runs makeRequest synchronously under this lock. + writeMu sync.Mutex + + // body accumulates the marshaled payload between Write and CloseWrite. + body *bytes.Buffer + + // bodySent is set once makeRequest hands the body to the transport; a + // later Write then returns io.EOF instead of racing with the transport. + bodySent bool + + errMu sync.Mutex + err error + + response *http.Response + responseReady chan struct{} + + sendOnce sync.Once +} + +func newUnaryFastPathCall( + ctx context.Context, + httpClient HTTPClient, + url *url.URL, + spec Spec, + header http.Header, + bufferPool *bufferPool, +) *unaryFastPathCall { + // Clone the URL so a transport we don't control can't mutate the caller's, + // then bind the concrete RPC path. + url = cloneURL(url) + url.Path = spec.Procedure + url.RawPath = "" + request := (&http.Request{ + Method: http.MethodPost, + URL: url, + Header: header, + Proto: "HTTP/1.1", + ProtoMajor: 1, + ProtoMinor: 1, + Host: url.Host, + }).WithContext(ctx) + return &unaryFastPathCall{ + ctx: ctx, + httpClient: httpClient, + request: request, + bufferPool: bufferPool, + responseReady: make(chan struct{}), + } +} + +// Write accumulates the request payload into the pooled body buffer. Unlike +// duplexHTTPCall.Write it never touches the network. +func (c *unaryFastPathCall) Write(data []byte) (int, error) { + c.writeMu.Lock() + defer c.writeMu.Unlock() + if c.bodySent { + // A racing CloseWrite already handed the body to the transport; + // writing now would race with its background read. Mirror + // duplexHTTPCall.Write, which returns io.EOF once the pipe is closed. + return 0, io.EOF + } + if err := c.getError(); err != nil { + return 0, err + } + if err := c.ctx.Err(); err != nil { + c.SetError(err) + return 0, wrapIfContextError(err) + } + if c.body == nil { + c.body = c.bufferPool.Get() + } + return c.body.Write(data) +} Review Comment: 感谢review，代码改动如下 - `Write` 现在对零长度写入直接短路：`len(data) == 0` 直接返回 `(0, nil)`，完全不碰池，`c.body` 保持 nil。 - `makeRequest` 保留兜底防护：如果仍有零长度 buffer 进入发送路径，会在回退到 `http.NoBody` 之前将其归还池。 测试改动如下（`unary_fastpath_body_test.go`）： - `TestUnaryFastPathEmptyWriteSkipsPool`：单元测试，`Write(nil)` / `Write([]byte{})` 后 `c.body` 保持 nil，请求使用 `http.NoBody`。 - `TestUnaryFastPathMakeRequestRecyclesEmptyBuffer`：单元测试，在 `CloseWrite` 前预置的零长度 buffer 会被归还到池。 - `TestUnaryFastPathEmptyMessageCall`：端到端测试，空 protobuf 消息通过 HTTP/2 完整完成一次 unary 调用。 当前行为：空 unary 载荷永远不会从请求 buffer 池中分配内存。请求仍会正常发送（`Content-Length: 0`，body 为 `http.NoBody`），且不会在池中遗留任何东西。 CI错误仍为已知问题，修复中 -- 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]
