Alanxtl commented on code in PR #3706: URL: https://github.com/apache/dubbo-go/pull/3706#discussion_r3868040015
########## 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: 在 `Write` 中，即使 `data` 长度为 0，也会从 `bufferPool` 获取 buffer。随后 `makeRequest` 发现 buffer 长度为 0 后直接使用 `http.NoBody`，但没有把原 buffer 放回 pool。 因此 `Send(nil)`、空 protobuf 消息或显式 `Write([]byte{})` 每次都会造成一次池化失效和额外分配。建议在零长度写入时直接返回，或在切换到 `http.NoBody` 前归还 buffer，并增加对应回归测试。 -- 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]
