This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new b7086bd1e feat(triple): performance optimization for Triple unary hot
path (#3706)
b7086bd1e is described below
commit b7086bd1e5a3254f3220e8ca06d25f337302de21
Author: Li Zining <[email protected]>
AuthorDate: Wed Sep 2 13:23:56 2026 +0800
feat(triple): performance optimization for Triple unary hot path (#3706)
* feat(protocol): add unary fast path config switch
Add the UnaryFastPath toggle so the fast path stays off by default:
- global: add UnaryFastPath field to TripleConfig (yaml/json/property)
and copy it in Clone
- option: add WithUnaryFastPath client option
- config chain: propagate through clientConfig and
protocolClientParams into NewConn
Signed-off-by: lizining <[email protected]>
* feat(protocol): implement unary fast path for Triple unary RPCs
Replace duplexHTTPCall with a synchronous unaryFastPathCall: no
io.Pipe, no per-request makeRequest goroutine, pooled buffers,
declared Content-Length, and direct response reading. Select it in
tripleClient.NewConn when the stream type is unary and the switch is
on.
Also hoist tri.WithTriple() so HTTP/1.1, HTTP/2 and HTTP/3 all
explicitly select the Triple protocol: HTTP/2 (the actual default
transport) previously fell back to the gRPC protocol and the fast
path living on tripleClient.NewConn was unreachable.
Signed-off-by: lizining <[email protected]>
* test(triple): add NewConn routing A/B test for the unary fast path
Signed-off-by: lizining <[email protected]>
* test(triple): guard the unary fast path against concurrent
Send/RequestHeader/CloseRequest races
Signed-off-by: lizining <[email protected]>
* test(triple): cover unary fast path call behavior including wire format
and error handling
Signed-off-by: lizining <[email protected]>
* test(triple): verify unary fast path body round-trips, Content-Length and
buffer reuse
Signed-off-by: lizining <[email protected]>
* test(triple): add A/B benchmark for unary fast path vs duplex
Signed-off-by: lizining <[email protected]>
* style(triple): apply make fmt to the unary fast path tests
Signed-off-by: lizining <[email protected]>
* feat(triple): enable the unary fast path by default
Flip the UnaryFastPath default from off to on so every Triple unary
call runs through unaryFastPathCall (no io.Pipe, no per-request
goroutine) unless explicitly opted out:
- triple_protocol: newClientConfig defaults UnaryFastPath to true, and
a new WithoutUnaryFastPath client option lets callers roll back to
the duplex (io.Pipe) path
- global: DefaultTripleConfig sets UnaryFastPath to true, so configs
that never mention unary-fast-path inherit the on-by-default behavior
- protocol/triple: newClientManager explicitly maps the config flag to
With/WithoutUnaryFastPath, so an explicit unary-fast-path: false in
the config opts back out of the default
- bench_test: BenchmarkUnaryDuplex pins the control group to the
explicit opt-out so it keeps measuring duplexHTTPCall
Streaming calls and the gRPC protocol are unaffected: the fast path
branch is still gated on StreamTypeUnary in tripleClient.NewConn, and
grpcClient.NewConn never reads the toggle.
Signed-off-by: lizining <[email protected]>
* fix(triple): recycle pooled buffer on zero-length fast-path writes
Empty payloads (Send(nil), an empty protobuf message, or an explicit
Write of an empty slice) pulled a buffer out of the pool in Write, but
makeRequest switched to http.NoBody for the zero-length body without
returning it, leaking one pooled slot plus an allocation per call.
Write now short-circuits zero-length writes and makeRequest recycles a
leftover empty buffer before falling back to http.NoBody.
Signed-off-by: lizining <[email protected]>
* test(triple): add regression tests for zero-length fast-path writes
Cover the empty-payload pool behavior: zero-length writes leave the
body buffer untouched, makeRequest recycles a leftover empty buffer
before falling back to http.NoBody, and an empty protobuf message
completes a unary call end to end.
Signed-off-by: lizining <[email protected]>
* fix(triple): restore gRPC as the default wire format for Triple clients
Unconditionally applying WithTriple forced every Triple client onto the
connect wire format and broke interop with dubbo-java (415 Unsupported
Media Type). Remove the unconditional option so HTTP/2/3 clients go back
to the default gRPC wire format, and keep WithTriple only on the
HTTP/1.1 branch, where the connect format is required because HTTP/1.1
has no trailer support. The unary fast path stays on by default and
remains reachable over the connect wire format.
Signed-off-by: lizining <[email protected]>
* feat(triple): route gRPC protocol unary calls through the fast path
The unary fast path was only reachable on the connect (Triple) wire
format, so the default gRPC format never took it and the switch had no
effect on the RPC-layer hot path. Route unary calls through
unaryFastPathCall in grpcClient.NewConn as well, keeping streaming calls
on duplexHTTPCall. grpcClientConn now stores the unaryClientCall
interface instead of a concrete *duplexHTTPCall, and CloseResponse reads
trailers through ResponseTrailer(). Update the config and protocol
comments that claimed the gRPC wire never reaches the fast path, and
tidy stale or informal comments in unary_fastpath.go.
Signed-off-by: lizining <[email protected]>
* test(triple): verify fast path routing and public-entry wire signature
Flip the gRPC routing assertion from "always duplex" to "unary routes to
the fast path when enabled and to duplex otherwise; streaming calls
always use duplex". Add a regression test that issues a call through the
public client entry (clientManager.callUnary, the path reached by
TripleInvoker.Invoke) and checks the server sees a pre-declared
Content-Length with the fast path on and -1 when it is explicitly
disabled, plus error-path equivalence between the fast path and duplex.
Signed-off-by: lizining <[email protected]>
* test(triple): drive public-entry tests through Refer and
TripleInvoker.Invoke
Replace the direct newClientManager/callUnary setup in the unary fast path
public-entry tests with the production entry: TripleProtocol.Refer followed
by TripleInvoker.Invoke with a real invocation, so the tests guard the full
Refer/Invoker/Invoke dispatch instead of the client manager in isolation.
The server-side Content-Length capture now excludes the background gRPC
health-check stream, whose streaming body previously clobbered the
fast-path signature.
Signed-off-by: lizining <[email protected]>
* docs(triple): clarify unary fast path scope and caveats
Signed-off-by: lizining <[email protected]>
* fix(triple): keep unary fast path enabled when config omits the toggle
UnaryFastPath was a plain bool, so any TripleConfig built or decoded
without the field silently disabled the fast path even though the
default is on. Change it to *bool: nil means "not explicitly set" and
keeps the default (enabled); explicit true/false still override. Align
Clone with the *bool deep-copy style used across the global config
structs.
Signed-off-by: lizining <[email protected]>
* test(triple): cover unary fast path config toggle states
Exercise the *bool three-state semantics end to end through the
production Refer/Invoke entry: nil and explicit true keep the fast path
on, explicit false falls back to duplex, and a config built without the
toggle (e.g. only KeepAliveInterval) keeps the default-on behavior.
Signed-off-by: lizining <[email protected]>
---------
Signed-off-by: lizining <[email protected]>
---
global/triple_config.go | 21 +-
protocol/triple/client.go | 17 +-
protocol/triple/triple_protocol/client.go | 6 +-
protocol/triple/triple_protocol/option.go | 33 ++
protocol/triple/triple_protocol/protocol.go | 3 +
protocol/triple/triple_protocol/protocol_grpc.go | 58 +-
protocol/triple/triple_protocol/protocol_triple.go | 33 +-
protocol/triple/triple_protocol/unary_fastpath.go | 304 +++++++++++
.../unary_fastpath_behavior_test.go | 606 +++++++++++++++++++++
.../triple_protocol/unary_fastpath_bench_test.go | 131 +++++
.../triple_protocol/unary_fastpath_body_test.go | 308 +++++++++++
.../unary_fastpath_concurrency_test.go | 158 ++++++
.../unary_fastpath_conn_type_test.go | 78 +++
.../triple/unary_fastpath_public_entry_test.go | 260 +++++++++
14 files changed, 1973 insertions(+), 43 deletions(-)
diff --git a/global/triple_config.go b/global/triple_config.go
index 25bdf6314..f395b3a63 100644
--- a/global/triple_config.go
+++ b/global/triple_config.go
@@ -49,14 +49,22 @@ type TripleConfig struct {
// KeepAliveTimeout defines the keep-alive timeout for client.
KeepAliveTimeout string `yaml:"keep-alive-timeout"
json:"keep-alive-timeout,omitempty" property:"keep-alive-timeout"`
+ // UnaryFastPath enables the unary fast path for client, on by default.
+ // It applies to unary calls on both the gRPC and the Triple (connect)
+ // wire formats; streaming calls always use duplexHTTPCall.
+ // A nil value means the field is not explicitly set, and the default
+ // (enabled) applies.
+ UnaryFastPath *bool `default:"true" yaml:"unary-fast-path"
json:"unary-fast-path,omitempty" property:"unary-fast-path"`
}
// DefaultTripleConfig returns a default TripleConfig instance.
func DefaultTripleConfig() *TripleConfig {
+ unaryFastPath := true
return &TripleConfig{
- Http3: DefaultHttp3Config(),
- Cors: DefaultCorsConfig(),
- OpenAPI: DefaultOpenAPIConfig(),
+ Http3: DefaultHttp3Config(),
+ Cors: DefaultCorsConfig(),
+ OpenAPI: DefaultOpenAPIConfig(),
+ UnaryFastPath: &unaryFastPath,
}
}
@@ -66,6 +74,12 @@ func (t *TripleConfig) Clone() *TripleConfig {
return nil
}
+ var newUnaryFastPath *bool
+ if t.UnaryFastPath != nil {
+ newUnaryFastPath = new(bool)
+ *newUnaryFastPath = *t.UnaryFastPath
+ }
+
return &TripleConfig{
MaxServerSendMsgSize: t.MaxServerSendMsgSize,
MaxServerRecvMsgSize: t.MaxServerRecvMsgSize,
@@ -75,5 +89,6 @@ func (t *TripleConfig) Clone() *TripleConfig {
KeepAliveInterval: t.KeepAliveInterval,
KeepAliveTimeout: t.KeepAliveTimeout,
+ UnaryFastPath: newUnaryFastPath,
}
}
diff --git a/protocol/triple/client.go b/protocol/triple/client.go
index 6aa949609..d7ef521aa 100644
--- a/protocol/triple/client.go
+++ b/protocol/triple/client.go
@@ -181,6 +181,19 @@ func newClientManager(url *common.URL) (*clientManager,
error) {
}
cliOpts = append(cliOpts, clientKeepAliveOpts...)
+ // The unary fast path is on by default. It routes unary calls through
+ // unaryFastPathCall on both the gRPC and the Triple (connect) wire
+ // formats; streaming calls always use duplexHTTPCall.
+ // A nil UnaryFastPath means the field was not explicitly set, so the
+ // default (enabled) applies.
+ if tripleConf != nil {
+ if tripleConf.UnaryFastPath == nil || *tripleConf.UnaryFastPath
{
+ cliOpts = append(cliOpts, tri.WithUnaryFastPath())
+ } else {
+ cliOpts = append(cliOpts, tri.WithoutUnaryFastPath())
+ }
+ }
+
// Build the HTTP transport used by the Triple client.
var transport http.RoundTripper
@@ -195,7 +208,9 @@ func newClientManager(url *common.URL) (*clientManager,
error) {
switch callProtocol {
case constant.CallHTTP:
// Backward compatibility path for callers that still request
HTTP/1.1.
- // Triple itself requires HTTP/2 or HTTP/3 trailer support.
+ // Triple itself requires HTTP/2 or HTTP/3 trailer support, so
HTTP/1.1
+ // callers use the Triple (connect) protocol, which does not
rely on
+ // trailers.
transport = &http.Transport{
TLSClientConfig: cfg,
}
diff --git a/protocol/triple/triple_protocol/client.go
b/protocol/triple/triple_protocol/client.go
index 9b079b211..156ad3a90 100644
--- a/protocol/triple/triple_protocol/client.go
+++ b/protocol/triple/triple_protocol/client.go
@@ -69,6 +69,7 @@ func NewClient(httpClient HTTPClient, url string, options
...ClientOption) *Clie
ReadMaxBytes: config.ReadMaxBytes,
SendMaxBytes: config.SendMaxBytes,
GetURLMaxBytes: config.GetURLMaxBytes,
+ UnaryFastPath: config.UnaryFastPath,
},
)
if protocolErr != nil {
@@ -230,6 +231,7 @@ type clientConfig struct {
Timeout time.Duration
Group string
Version string
+ UnaryFastPath bool
}
func newClientConfig(rawURL string, options []ClientOption) (*clientConfig,
*Error) {
@@ -241,7 +243,9 @@ func newClientConfig(rawURL string, options []ClientOption)
(*clientConfig, *Err
config := clientConfig{
URL: url,
// use gRPC by default
- Protocol: &protocolGRPC{},
+ Protocol: &protocolGRPC{},
+ // the unary fast path is on by default
+ UnaryFastPath: true,
Procedure: protoPath,
CompressionPools: make(map[string]*compressionPool),
BufferPool: newBufferPool(),
diff --git a/protocol/triple/triple_protocol/option.go
b/protocol/triple/triple_protocol/option.go
index 61fb29d76..a3301ab6e 100644
--- a/protocol/triple/triple_protocol/option.go
+++ b/protocol/triple/triple_protocol/option.go
@@ -70,6 +70,39 @@ func WithTriple() ClientOption {
return &tripleOption{}
}
+// WithUnaryFastPath explicitly enables the unary fast path (unaryFastPathCall)
+// for unary calls. The fast path is on by default; this option is provided for
+// callers that want to state the intent explicitly.
+//
+// This optimization primarily targets small, CPU-dominated unary requests,
+// roughly in the 128 B to 32 KiB range: the complete request body is
+// buffered before it is submitted to the transport. For large or highly
+// concurrent requests, memory residency and buffer growth may increase, and
+// the performance benefit is not guaranteed. The implementation does not
+// automatically fall back based on payload size.
+func WithUnaryFastPath() ClientOption {
+ return &unaryFastPathOption{}
+}
+
+type unaryFastPathOption struct{}
+
+func (o *unaryFastPathOption) applyToClient(config *clientConfig) {
+ config.UnaryFastPath = true
+}
+
+// WithoutUnaryFastPath disables the unary fast path (unaryFastPathCall) for
+// unary calls, falling back to the duplex (io.Pipe) call path. Use it as a
+// rollback switch when the fast path misbehaves under a specific workload.
+func WithoutUnaryFastPath() ClientOption {
+ return &withoutUnaryFastPathOption{}
+}
+
+type withoutUnaryFastPathOption struct{}
+
+func (o *withoutUnaryFastPathOption) applyToClient(config *clientConfig) {
+ config.UnaryFastPath = false
+}
+
// WithProtoJSON configures a client to send JSON-encoded data instead of
// binary Protobuf. It uses the standard Protobuf JSON mapping as implemented
// by [google.golang.org/protobuf/encoding/protojson]: fields are named using
diff --git a/protocol/triple/triple_protocol/protocol.go
b/protocol/triple/triple_protocol/protocol.go
index 517910218..cc0451ef0 100644
--- a/protocol/triple/triple_protocol/protocol.go
+++ b/protocol/triple/triple_protocol/protocol.go
@@ -129,6 +129,9 @@ type protocolClientParams struct {
EnableGet bool
GetURLMaxBytes int
GetUseFallback bool
+ // UnaryFastPath enables the unary fast path (unaryFastPathCall) for
unary
+ // calls. It is on by default; disable it to fall back to the duplex
path.
+ UnaryFastPath bool
// The gRPC family of protocols always needs access to a Protobuf codec
to
// marshal and unmarshal errors.
Protobuf Codec
diff --git a/protocol/triple/triple_protocol/protocol_grpc.go
b/protocol/triple/triple_protocol/protocol_grpc.go
index 21f0fbc5b..e21f7a3d7 100644
--- a/protocol/triple/triple_protocol/protocol_grpc.go
+++ b/protocol/triple/triple_protocol/protocol_grpc.go
@@ -280,23 +280,31 @@ func (g *grpcClient) NewConn(
header[grpcHeaderTimeout] = []string{encodedDeadline}
}
}
- duplexCall := newDuplexHTTPCall(
- ctx,
- g.HTTPClient,
- g.URL,
- spec,
- header,
- )
+ var call unaryClientCall
+ if spec.StreamType == StreamTypeUnary && g.UnaryFastPath {
+ // Unary fast path: no io.Pipe, no per-request goroutine.
Streaming
+ // calls always keep using duplexHTTPCall.
+ call = newUnaryFastPathCall(
+ ctx,
+ g.HTTPClient,
+ g.URL,
+ spec,
+ header,
+ g.BufferPool,
+ )
+ } else {
+ call = newDuplexHTTPCall(ctx, g.HTTPClient, g.URL, spec, header)
+ }
conn := &grpcClientConn{
spec: spec,
peer: g.Peer(),
- duplexCall: duplexCall,
+ call: call,
compressionPools: g.CompressionPools,
bufferPool: g.BufferPool,
protobuf: g.Protobuf,
marshaler: grpcMarshaler{
envelopeWriter: envelopeWriter{
- writer: duplexCall,
+ writer: call,
compressionPool:
g.CompressionPools.Get(g.CompressionName),
codec: g.Codec,
compressMinBytes: g.CompressMinBytes,
@@ -306,7 +314,7 @@ func (g *grpcClient) NewConn(
},
unmarshaler: grpcUnmarshaler{
envelopeReader: envelopeReader{
- reader: duplexCall,
+ reader: call,
codec: g.Codec,
bufferPool: g.BufferPool,
readMaxBytes: g.ReadMaxBytes,
@@ -315,8 +323,8 @@ func (g *grpcClient) NewConn(
responseHeader: make(http.Header),
responseTrailer: make(http.Header),
}
- duplexCall.SetValidateResponse(conn.validateResponse)
- conn.readTrailers = func(_ *grpcUnmarshaler, call *duplexHTTPCall)
http.Header {
+ call.SetValidateResponse(conn.validateResponse)
+ conn.readTrailers = func(_ *grpcUnmarshaler, call unaryClientCall)
http.Header {
// To access HTTP trailers, we need to read the body to EOF.
_ = discard(call)
return call.ResponseTrailer()
@@ -328,7 +336,7 @@ func (g *grpcClient) NewConn(
type grpcClientConn struct {
spec Spec
peer Peer
- duplexCall *duplexHTTPCall
+ call unaryClientCall
compressionPools readOnlyCompressionPools
bufferPool *bufferPool
protobuf Codec // for errors
@@ -336,7 +344,7 @@ type grpcClientConn struct {
unmarshaler grpcUnmarshaler
responseHeader http.Header
responseTrailer http.Header
- readTrailers func(*grpcUnmarshaler, *duplexHTTPCall) http.Header
+ readTrailers func(*grpcUnmarshaler, unaryClientCall) http.Header
}
func (cc *grpcClientConn) Spec() Spec {
@@ -355,15 +363,15 @@ func (cc *grpcClientConn) Send(msg any) error {
}
func (cc *grpcClientConn) RequestHeader() http.Header {
- return cc.duplexCall.Header()
+ return cc.call.Header()
}
func (cc *grpcClientConn) CloseRequest() error {
- return cc.duplexCall.CloseWrite()
+ return cc.call.CloseWrite()
}
func (cc *grpcClientConn) Receive(msg any) error {
- cc.duplexCall.BlockUntilResponseReady()
+ cc.call.BlockUntilResponseReady()
err := cc.unmarshaler.Unmarshal(msg)
if err == nil {
return nil
@@ -377,7 +385,7 @@ func (cc *grpcClientConn) Receive(msg any) error {
// See if the server sent an explicit error in the HTTP or gRPC-Web
trailers.
mergeHeaders(
cc.responseTrailer,
- cc.readTrailers(&cc.unmarshaler, cc.duplexCall),
+ cc.readTrailers(&cc.unmarshaler, cc.call),
)
serverErr := grpcErrorFromTrailer(cc.bufferPool, cc.protobuf,
cc.responseTrailer)
if serverErr != nil && (errors.Is(err, io.EOF) || !errors.Is(serverErr,
errTrailersWithoutGRPCStatus)) {
@@ -391,33 +399,33 @@ func (cc *grpcClientConn) Receive(msg any) error {
// the stream has ended, Receive must return an error.
serverErr.meta = cc.responseHeader.Clone()
mergeHeaders(serverErr.meta, cc.responseTrailer)
- cc.duplexCall.SetError(serverErr)
+ cc.call.SetError(serverErr)
return serverErr
}
// This was probably an error converting the bytes to a message or an
error
// reading from the network. We're going to return it to the
// user, but we also want to setResponseError so Send errors out.
- cc.duplexCall.SetError(err)
+ cc.call.SetError(err)
return err
}
func (cc *grpcClientConn) ResponseHeader() http.Header {
- cc.duplexCall.BlockUntilResponseReady()
+ cc.call.BlockUntilResponseReady()
return cc.responseHeader
}
func (cc *grpcClientConn) ResponseTrailer() http.Header {
- cc.duplexCall.BlockUntilResponseReady()
+ cc.call.BlockUntilResponseReady()
return cc.responseTrailer
}
func (cc *grpcClientConn) CloseResponse() error {
- err := cc.duplexCall.CloseRead()
+ err := cc.call.CloseRead()
if err != nil {
return err
}
- if cc.duplexCall.response != nil && cc.duplexCall.response.Trailer !=
nil {
- cc.responseTrailer = cc.duplexCall.response.Trailer.Clone()
+ if trailer := cc.call.ResponseTrailer(); len(trailer) > 0 {
+ cc.responseTrailer = trailer.Clone()
}
return nil
}
diff --git a/protocol/triple/triple_protocol/protocol_triple.go
b/protocol/triple/triple_protocol/protocol_triple.go
index 04de63618..d5c093863 100644
--- a/protocol/triple/triple_protocol/protocol_triple.go
+++ b/protocol/triple/triple_protocol/protocol_triple.go
@@ -264,27 +264,34 @@ func (c *tripleClient) NewConn(
} // else effectively unbounded
}
}
- duplexCall := newDuplexHTTPCall(ctx, c.HTTPClient, c.URL, spec, header)
+ var call unaryClientCall
+ if spec.StreamType == StreamTypeUnary && c.UnaryFastPath {
+ // Unary fast path: no io.Pipe, no per-request goroutine.
Streaming
+ // calls always keep using duplexHTTPCall.
+ call = newUnaryFastPathCall(ctx, c.HTTPClient, c.URL, spec,
header, c.BufferPool)
+ } else {
+ call = newDuplexHTTPCall(ctx, c.HTTPClient, c.URL, spec, header)
+ }
unaryConn := &tripleUnaryClientConn{
spec: spec,
peer: c.Peer(),
- duplexCall: duplexCall,
+ call: call,
compressionPools: c.CompressionPools,
bufferPool: c.BufferPool,
marshaler: tripleUnaryRequestMarshaler{
tripleUnaryMarshaler: tripleUnaryMarshaler{
- writer: duplexCall,
+ writer: call,
codec: c.Codec,
compressMinBytes: c.CompressMinBytes,
compressionName: c.CompressionName,
compressionPool:
c.CompressionPools.Get(c.CompressionName),
bufferPool: c.BufferPool,
- header: duplexCall.Header(),
+ header: call.Header(),
sendMaxBytes: c.SendMaxBytes,
},
},
unmarshaler: tripleUnaryUnmarshaler{
- reader: duplexCall,
+ reader: call,
codec: c.Codec,
bufferPool: c.BufferPool,
readMaxBytes: c.ReadMaxBytes,
@@ -292,14 +299,14 @@ func (c *tripleClient) NewConn(
responseHeader: make(http.Header),
responseTrailer: make(http.Header),
}
- duplexCall.SetValidateResponse(unaryConn.validateResponse)
+ call.SetValidateResponse(unaryConn.validateResponse)
return wrapClientConnWithCodedErrors(unaryConn)
}
type tripleUnaryClientConn struct {
spec Spec
peer Peer
- duplexCall *duplexHTTPCall
+ call unaryClientCall
compressionPools readOnlyCompressionPools
bufferPool *bufferPool
marshaler tripleUnaryRequestMarshaler
@@ -324,15 +331,15 @@ func (cc *tripleUnaryClientConn) Send(msg any) error {
}
func (cc *tripleUnaryClientConn) RequestHeader() http.Header {
- return cc.duplexCall.Header()
+ return cc.call.Header()
}
func (cc *tripleUnaryClientConn) CloseRequest() error {
- return cc.duplexCall.CloseWrite()
+ return cc.call.CloseWrite()
}
func (cc *tripleUnaryClientConn) Receive(msg any) error {
- cc.duplexCall.BlockUntilResponseReady()
+ cc.call.BlockUntilResponseReady()
if err := cc.unmarshaler.Unmarshal(msg); err != nil {
return err
}
@@ -340,17 +347,17 @@ func (cc *tripleUnaryClientConn) Receive(msg any) error {
}
func (cc *tripleUnaryClientConn) ResponseHeader() http.Header {
- cc.duplexCall.BlockUntilResponseReady()
+ cc.call.BlockUntilResponseReady()
return cc.responseHeader
}
func (cc *tripleUnaryClientConn) ResponseTrailer() http.Header {
- cc.duplexCall.BlockUntilResponseReady()
+ cc.call.BlockUntilResponseReady()
return cc.responseTrailer
}
func (cc *tripleUnaryClientConn) CloseResponse() error {
- return cc.duplexCall.CloseRead()
+ return cc.call.CloseRead()
}
func (cc *tripleUnaryClientConn) validateResponse(response *http.Response)
*Error {
diff --git a/protocol/triple/triple_protocol/unary_fastpath.go
b/protocol/triple/triple_protocol/unary_fastpath.go
new file mode 100644
index 000000000..63f8d6a5e
--- /dev/null
+++ b/protocol/triple/triple_protocol/unary_fastpath.go
@@ -0,0 +1,304 @@
+/*
+ * 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 transport surface shared by grpcClientConn and
+// tripleUnaryClientConn. 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 duplexHTTPCall and unaryFastPathCall 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).
+//
+// This optimization primarily targets small, CPU-dominated unary requests,
+// roughly in the 128 B to 32 KiB range: the complete request body is
+// buffered before it is submitted to the transport. For large or highly
+// concurrent requests, memory residency and buffer growth may increase, and
+// the performance benefit is not guaranteed. The implementation does not
+// automatically fall back based on payload size.
+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; a
+ // write now could race with the transport reading it. 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 len(data) == 0 {
+ // Empty writes must not pull a buffer from the pool: a
zero-length
+ // body is sent as http.NoBody, so the buffer would never be
+ // returned via unaryRequestBody.Close.
+ return 0, nil
+ }
+ if c.body == nil {
+ c.body = c.bufferPool.Get()
+ }
+ return c.body.Write(data)
+}
+
+// CloseWrite issues the request once the body is complete. Failures are
+// stored via SetError and surface from Read. Runs under writeMu so it never
+// sees a body being appended concurrently.
+func (c *unaryFastPathCall) CloseWrite() error {
+ c.writeMu.Lock()
+ defer c.writeMu.Unlock()
+ c.sendOnce.Do(func() {
+ c.makeRequest()
+ })
+ return nil
+}
+
+// unaryRequestBody is an io.ReadCloser over the pooled request buffer with no
+// extra allocation. The buffer is returned to the pool from Close, which
+// x/net/http2 invokes exactly once after it stops reading the body. Read and
+// Close are guarded by a mutex so an aborted-write Close can never race with
+// an in-flight Read.
+type unaryRequestBody struct {
+ mu sync.Mutex
+ buf *bytes.Buffer
+ pool *bufferPool
+}
+
+func (b *unaryRequestBody) Read(p []byte) (int, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ if b.buf == nil {
+ return 0, io.EOF
+ }
+ return b.buf.Read(p)
+}
+
+func (b *unaryRequestBody) Close() error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ if b.buf != nil {
+ b.pool.Put(b.buf)
+ b.buf = nil
+ }
+ return nil
+}
+
+// makeRequest issues the HTTP request with the same error wrapping chain as
+// duplexHTTPCall. The request body buffer is returned to the pool from
+// unaryRequestBody.Close, which x/net/http2 calls after it stops reading.
+func (c *unaryFastPathCall) makeRequest() {
+ defer close(c.responseReady)
+ // Advertise Content-Length so the server can pre-size its HTTP/2 data
+ // buffer; empty payloads use http.NoBody.
+ var bodyLen int
+ if c.body != nil {
+ bodyLen = c.body.Len()
+ if bodyLen > 0 {
+ c.request.Body = &unaryRequestBody{buf: c.body, pool:
c.bufferPool}
+ } else {
+ // Zero-length body: no transport read will ever
happen, so
+ // unaryRequestBody.Close can't return the buffer to
the pool.
+ // Recycle it here instead.
+ c.bufferPool.Put(c.body)
+ c.body = nil
+ c.request.Body = http.NoBody
+ }
+ } else {
+ c.request.Body = http.NoBody
+ }
+ c.bodySent = true
+ c.request.ContentLength = int64(bodyLen)
+ response, err := c.httpClient.Do(c.request) //nolint:bodyclose
+ if err != nil {
+ err = wrapIfContextError(err)
+ err = wrapIfLikelyH2CNotConfiguredError(c.request, err)
+ err = wrapIfLikelyWithGRPCNotUsedError(err)
+ err = wrapIfRSTError(err)
+ if _, ok := asError(err); !ok {
+ err = NewError(CodeUnavailable, err)
+ }
+ c.SetError(err)
+ return
+ }
+ c.response = response
+ if err := c.validateResponse(response); err != nil {
+ // Leave the response body open for CloseRead: callers may
still read
+ // the error body, and CloseResponse is the single close point.
+ c.SetError(err)
+ }
+}
+
+// Read reads the response body. BlockUntilResponseReady is already resolved
+// by the time Read is called.
+func (c *unaryFastPathCall) Read(data []byte) (int, error) {
+ c.BlockUntilResponseReady()
+ 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.response == nil {
+ return 0, fmt.Errorf("nil response from %v", c.request.URL)
+ }
+ n, err := c.response.Body.Read(data)
+ return n, wrapIfRSTError(err)
+}
+
+// CloseRead closes the response body. The request body buffer is recycled by
+// unaryRequestBody.Close, not here.
+func (c *unaryFastPathCall) CloseRead() error {
+ c.BlockUntilResponseReady()
+ if c.response == nil {
+ return nil
+ }
+ return wrapIfRSTError(c.response.Body.Close())
+}
+
+// Header returns the HTTP request headers.
+func (c *unaryFastPathCall) Header() http.Header {
+ return c.request.Header
+}
+
+// SetValidateResponse sets the response validation function.
+func (c *unaryFastPathCall) SetValidateResponse(validate func(*http.Response)
*Error) {
+ c.validateResponse = validate
+}
+
+// BlockUntilResponseReady blocks until the response is available. The fast
+// path resolves it synchronously inside CloseWrite.
+func (c *unaryFastPathCall) BlockUntilResponseReady() {
+ <-c.responseReady
+}
+
+// SetError stores the first error encountered; safe for concurrent use.
+func (c *unaryFastPathCall) SetError(err error) {
+ c.errMu.Lock()
+ defer c.errMu.Unlock()
+ if c.err == nil {
+ c.err = wrapIfContextError(err)
+ }
+}
+
+// ResponseTrailer returns the response HTTP trailers.
+func (c *unaryFastPathCall) ResponseTrailer() http.Header {
+ c.BlockUntilResponseReady()
+ if c.response != nil {
+ return c.response.Trailer
+ }
+ return make(http.Header)
+}
+
+func (c *unaryFastPathCall) getError() error {
+ c.errMu.Lock()
+ defer c.errMu.Unlock()
+ return c.err
+}
diff --git a/protocol/triple/triple_protocol/unary_fastpath_behavior_test.go
b/protocol/triple/triple_protocol/unary_fastpath_behavior_test.go
new file mode 100644
index 000000000..fe5865661
--- /dev/null
+++ b/protocol/triple/triple_protocol/unary_fastpath_behavior_test.go
@@ -0,0 +1,606 @@
+/*
+ * 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 (
+ "compress/gzip"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+)
+
+import (
+ pingv1
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1"
+)
+
+// behaviorHTTPClient adapts a closure to HTTPClient so tests can capture or
+// fake the wire behavior of the fast path.
+type behaviorHTTPClient struct {
+ do func(*http.Request) (*http.Response, error)
+}
+
+func (c *behaviorHTTPClient) Do(req *http.Request) (*http.Response, error) {
+ return c.do(req)
+}
+
+// newPingHandler builds a Triple unary handler in-package, avoiding the
+// generated pingv1connect package (which would create an import cycle).
+func newPingHandler() *Handler {
+ return NewUnaryHandler(
+ "/connect.ping.v1.PingService/Ping",
+ func() any { return &pingv1.PingRequest{} },
+ func(ctx context.Context, req *Request) (*Response, error) {
+ pingReq, ok := req.Any().(*pingv1.PingRequest)
+ if !ok {
+ return nil, fmt.Errorf("unexpected request type
%T", req.Any())
+ }
+ return NewResponse(&pingv1.PingResponse{Text:
pingReq.Text}), nil
+ },
+ WithCompression(
+ compressionGzip,
+ func() Decompressor { return &gzip.Reader{} },
+ func() Compressor { return gzip.NewWriter(io.Discard) },
+ ),
+ )
+}
+
+// pingCompressionPools registers gzip so response compression is negotiated,
+// and is shared by the behavior tests to keep the pools warm.
+var pingCompressionPools = newReadOnlyCompressionPools(
+ map[string]*compressionPool{
+ compressionGzip: newCompressionPool(
+ func() Decompressor { return &gzip.Reader{} },
+ func() Compressor { return gzip.NewWriter(io.Discard) },
+ ),
+ },
+ []string{compressionGzip},
+)
+
+// newBehaviorCall constructs an unaryFastPathCall wired to the given client,
+// mirroring tripleClient.NewConn's construction.
+func newBehaviorCall(ctx context.Context, httpClient HTTPClient, serverURL
*url.URL, pool *bufferPool) *unaryFastPathCall {
+ call := newUnaryFastPathCall(
+ ctx,
+ httpClient,
+ serverURL,
+ Spec{
+ StreamType: StreamTypeUnary,
+ Procedure: "/connect.ping.v1.PingService/Ping",
+ },
+ make(http.Header),
+ pool,
+ )
+ call.SetValidateResponse(func(*http.Response) *Error { return nil })
+ return call
+}
+
+// newTripleClientConn constructs a tripleClient and routes the unary call
+// through NewConn, exercising the real production entry point.
+func newTripleClientConn(useFastPath bool, codec Codec, httpClient HTTPClient,
serverURL *url.URL, header http.Header, pool *bufferPool) StreamingClientConn {
+ client := &tripleClient{
+ protocolClientParams: protocolClientParams{
+ HTTPClient: httpClient,
+ URL: serverURL,
+ BufferPool: pool,
+ Codec: codec,
+ CompressionPools: pingCompressionPools,
+ UnaryFastPath: useFastPath,
+ },
+ peer: Peer{Addr: serverURL.String(), Protocol: ProtocolTriple},
+ }
+ return client.NewConn(context.Background(), Spec{
+ StreamType: StreamTypeUnary,
+ Procedure: "/connect.ping.v1.PingService/Ping",
+ }, header)
+}
+
+// TestUnaryFastPathWriteAccumulates verifies that Write appends into the
+// pooled body without touching the network, and that CloseWrite hands the
+// whole payload to the transport once, with an exact Content-Length.
+func TestUnaryFastPathWriteAccumulates(t *testing.T) {
+ pool := newBufferPool()
+ serverURL, err := url.Parse("https://example.com";)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var (
+ calls atomic.Int32
+ wireBody []byte
+ contentLen int64
+ )
+ httpClient := &behaviorHTTPClient{do: func(req *http.Request)
(*http.Response, error) {
+ calls.Add(1)
+ data, err := io.ReadAll(req.Body)
+ if err != nil {
+ return nil, err
+ }
+ if err := req.Body.Close(); err != nil {
+ return nil, err
+ }
+ wireBody = data
+ contentLen = req.ContentLength
+ return &http.Response{
+ StatusCode: http.StatusOK,
+ Header: make(http.Header),
+ Body: io.NopCloser(strings.NewReader("")),
+ }, nil
+ }}
+
+ call := newBehaviorCall(context.Background(), httpClient, serverURL,
pool)
+ if n, err := call.Write([]byte("hello")); n != 5 || err != nil {
+ t.Fatalf("first write = (n=%d, err=%v), want (5, nil)", n, err)
+ }
+ if n, err := call.Write([]byte("-world")); n != 6 || err != nil {
+ t.Fatalf("second write = (n=%d, err=%v), want (6, nil)", n, err)
+ }
+ if err := call.CloseWrite(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ // A second CloseWrite must not dispatch the request again.
+ if err := call.CloseWrite(); err != nil {
+ t.Fatalf("second close request: %v", err)
+ }
+ if err := call.CloseRead(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("transport saw %d requests, want 1", got)
+ }
+ if got := string(wireBody); got != "hello-world" {
+ t.Fatalf("wire body = %q, want %q", got, "hello-world")
+ }
+ if contentLen != int64(len("hello-world")) {
+ t.Fatalf("content length = %d, want %d", contentLen,
len("hello-world"))
+ }
+}
+
+// TestUnaryFastPathEmptyBodyUsesNoBody verifies that a call with no Write uses
+// http.NoBody and a zero Content-Length, skipping the pooled buffer.
+func TestUnaryFastPathEmptyBodyUsesNoBody(t *testing.T) {
+ pool := newBufferPool()
+ serverURL, err := url.Parse("https://example.com";)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var (
+ calls atomic.Int32
+ sawNoBody bool
+ contentLen int64
+ )
+ httpClient := &behaviorHTTPClient{do: func(req *http.Request)
(*http.Response, error) {
+ calls.Add(1)
+ sawNoBody = req.Body == http.NoBody
+ contentLen = req.ContentLength
+ return &http.Response{
+ StatusCode: http.StatusOK,
+ Header: make(http.Header),
+ Body: io.NopCloser(strings.NewReader("")),
+ }, nil
+ }}
+
+ call := newBehaviorCall(context.Background(), httpClient, serverURL,
pool)
+ if err := call.CloseWrite(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ if err := call.CloseRead(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("transport saw %d requests, want 1", got)
+ }
+ if !sawNoBody {
+ t.Fatal("empty call did not use http.NoBody")
+ }
+ if contentLen != 0 {
+ t.Fatalf("content length = %d, want 0", contentLen)
+ }
+}
+
+// TestUnaryFastPathConcurrentReadClose verifies that a Read racing a Close
+// never panics or reads recycled bytes: both paths are guarded by the body
+// mutex and surface a clean EOF.
+func TestUnaryFastPathConcurrentReadClose(t *testing.T) {
+ pool := newBufferPool()
+ buf := pool.Get()
+ buf.WriteString("payload")
+ body := &unaryRequestBody{buf: buf, pool: pool}
+
+ const (
+ readers = 8
+ closers = 8
+ )
+ var wg sync.WaitGroup
+ wg.Add(readers + closers)
+ for range closers {
+ go func() {
+ defer wg.Done()
+ _ = body.Close()
+ }()
+ }
+ for range readers {
+ go func() {
+ defer wg.Done()
+ for {
+ if _, err := body.Read(make([]byte, 4)); err !=
nil {
+ return
+ }
+ }
+ }()
+ }
+ wg.Wait()
+}
+
+// TestUnaryFastPathRequestHeaderConcurrent verifies that RequestHeader is
+// safe to race with Write and CloseWrite, per the streaming client conn
+// contract's write-side group.
+func TestUnaryFastPathRequestHeaderConcurrent(t *testing.T) {
+ httpClient, serverURL, _, _ := concurrentEchoServer(t)
+ pool := newBufferPool()
+ call := newBehaviorCall(context.Background(), httpClient, serverURL,
pool)
+
+ var wg sync.WaitGroup
+ wg.Add(3)
+ go func() {
+ defer wg.Done()
+ for range 50 {
+ _ = call.Header()
+ }
+ }()
+ go func() {
+ defer wg.Done()
+ for range 50 {
+ _, _ = call.Write([]byte("payload"))
+ }
+ }()
+ go func() {
+ defer wg.Done()
+ _ = call.CloseWrite()
+ }()
+ wg.Wait()
+
+ if call.Header() == nil {
+ t.Fatal("header must remain readable after CloseWrite")
+ }
+ if _, err := io.Copy(io.Discard, call); err != nil {
+ t.Fatalf("read response: %v", err)
+ }
+ if err := call.CloseRead(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+}
+
+// TestUnaryFastPathWriteAfterTransportError verifies that a failed request
+// rejects later writes with io.EOF and Read surfaces the wrapped
+// CodeUnavailable error.
+func TestUnaryFastPathWriteAfterTransportError(t *testing.T) {
+ pool := newBufferPool()
+ serverURL, err := url.Parse("https://example.com";)
+ if err != nil {
+ t.Fatal(err)
+ }
+ transportErr := errors.New("transport failed")
+ httpClient := &behaviorHTTPClient{do: func(*http.Request)
(*http.Response, error) {
+ return nil, transportErr
+ }}
+
+ call := newBehaviorCall(context.Background(), httpClient, serverURL,
pool)
+ if _, writeErr := call.Write([]byte("payload")); writeErr != nil {
+ t.Fatalf("write: %v", writeErr)
+ }
+ if closeErr := call.CloseWrite(); closeErr != nil {
+ t.Fatalf("close request: %v", closeErr)
+ }
+ n, err := call.Write([]byte("late"))
+ if n != 0 || !errors.Is(err, io.EOF) {
+ t.Fatalf("write after transport error = (n=%d, err=%v), want
(0, io.EOF)", n, err)
+ }
+ _, err = call.Read(make([]byte, 16))
+ var connErr *Error
+ if !errors.As(err, &connErr) || connErr.Code() != CodeUnavailable {
+ t.Fatalf("read after transport error = %v, want
CodeUnavailable", err)
+ }
+}
+
+// TestUnaryFastPathSetErrorRejectsWrite verifies that a stored error rejects
+// writes before the body is sent. Unlike duplexHTTPCall's io.EOF, the fast
+// path surfaces the stored error so the underlying failure stays visible.
+func TestUnaryFastPathSetErrorRejectsWrite(t *testing.T) {
+ pool := newBufferPool()
+ serverURL, err := url.Parse("https://example.com";)
+ if err != nil {
+ t.Fatal(err)
+ }
+ httpClient := &behaviorHTTPClient{do: func(*http.Request)
(*http.Response, error) {
+ return &http.Response{
+ StatusCode: http.StatusOK,
+ Header: make(http.Header),
+ Body: io.NopCloser(strings.NewReader("")),
+ }, nil
+ }}
+
+ call := newBehaviorCall(context.Background(), httpClient, serverURL,
pool)
+ storedErr := errors.New("stored")
+ call.SetError(storedErr)
+ n, err := call.Write([]byte("payload"))
+ if n != 0 || !errors.Is(err, storedErr) {
+ t.Fatalf("write after SetError = (n=%d, err=%v), want (0, %v)",
n, err, storedErr)
+ }
+}
+
+// TestUnaryFastPathWireConsistent verifies that the fast path sends the same
+// request body bytes as the duplex path for the same message.
+func TestUnaryFastPathWireConsistent(t *testing.T) {
+ httpClient, serverURL, bodies, bodiesMu := concurrentEchoServer(t)
+ header := make(http.Header)
+ header.Set(headerContentType, "application/proto")
+ pool := newBufferPool()
+
+ for _, tc := range []struct {
+ name string
+ fast bool
+ }{
+ {"duplex", false},
+ {"fastpath", true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ conn := newTripleClientConn(tc.fast,
&protoBinaryCodec{}, httpClient, serverURL, header, pool)
+ if err := conn.Send(&pingv1.PingRequest{Text:
"wire-consistency"}); err != nil {
+ t.Fatalf("send: %v", err)
+ }
+ if err := conn.CloseRequest(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ // The echo server returns an empty 200 body, which
unmarshals
+ // into an empty message; only the request side matters
here.
+ if err := conn.Receive(&pingv1.PingResponse{}); err !=
nil {
+ t.Fatalf("receive: %v", err)
+ }
+ if err := conn.CloseResponse(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+ })
+ }
+
+ bodiesMu.Lock()
+ defer bodiesMu.Unlock()
+ if len(*bodies) != 2 {
+ t.Fatalf("server received %d requests, want 2", len(*bodies))
+ }
+ if (*bodies)[0] != (*bodies)[1] {
+ t.Fatalf("wire mismatch: duplex %q vs fastpath %q",
(*bodies)[0], (*bodies)[1])
+ }
+}
+
+// TestUnaryFastPathContextCancel verifies that canceling the call context
+// aborts the in-flight request and Read surfaces a CodeCanceled error.
+func TestUnaryFastPathContextCancel(t *testing.T) {
+ pool := newBufferPool()
+ serverURL, err := url.Parse("https://example.com";)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ httpClient := &behaviorHTTPClient{do: func(req *http.Request)
(*http.Response, error) {
+ <-req.Context().Done()
+ return nil, req.Context().Err()
+ }}
+
+ call := newBehaviorCall(ctx, httpClient, serverURL, pool)
+ if _, writeErr := call.Write([]byte("payload")); writeErr != nil {
+ t.Fatalf("write: %v", writeErr)
+ }
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ _ = call.CloseWrite()
+ }()
+ cancel()
+ <-done
+ _, err = call.Read(make([]byte, 16))
+ var connErr *Error
+ if !errors.As(err, &connErr) || connErr.Code() != CodeCanceled {
+ t.Fatalf("read after cancel = %v, want CodeCanceled", err)
+ }
+}
+
+// TestUnaryFastPathEndToEnd verifies that the fast path is wire- and
+// response-compatible with the duplex path: same message, headers, and
+// trailers on a real HTTP/2 server.
+func TestUnaryFastPathEndToEnd(t *testing.T) {
+ mux := http.NewServeMux()
+ mux.Handle("/connect.ping.v1.PingService/Ping", NewUnaryHandler(
+ "/connect.ping.v1.PingService/Ping",
+ func() any { return &pingv1.PingRequest{} },
+ func(ctx context.Context, req *Request) (*Response, error) {
+ pingReq, ok := req.Any().(*pingv1.PingRequest)
+ if !ok {
+ return nil, fmt.Errorf("unexpected request type
%T", req.Any())
+ }
+ resp := NewResponse(&pingv1.PingResponse{Text:
pingReq.Text})
+ resp.Header().Set("X-Triple-Echo", "header")
+ resp.Trailer().Set("X-Triple-Echo-Trailer", "trailer")
+ return resp, nil
+ },
+ ))
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ t.Cleanup(server.Close)
+ httpClient := server.Client()
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ header := make(http.Header)
+ header.Set(headerContentType, "application/proto")
+ header.Set(tripleUnaryHeaderAcceptCompression, "gzip")
+ pool := newBufferPool()
+
+ for _, tc := range []struct {
+ name string
+ fast bool
+ }{
+ {"duplex", false},
+ {"fastpath", true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ conn := newTripleClientConn(tc.fast,
&protoBinaryCodec{}, httpClient, serverURL, header, pool)
+ if err := conn.Send(&pingv1.PingRequest{Text:
"hello"}); err != nil {
+ t.Fatalf("send: %v", err)
+ }
+ if err := conn.CloseRequest(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ resp := NewResponse(&pingv1.PingResponse{})
+ if err := receiveUnaryResponse(conn, resp); err != nil {
+ t.Fatalf("receive: %v", err)
+ }
+ pingResp, ok := resp.Any().(*pingv1.PingResponse)
+ if !ok {
+ t.Fatalf("unexpected response type %T",
resp.Any())
+ }
+ if pingResp.Text != "hello" {
+ t.Fatalf("response text = %q, want %q",
pingResp.Text, "hello")
+ }
+ if got := resp.Header().Get("X-Triple-Echo"); got !=
"header" {
+ t.Fatalf("response header X-Triple-Echo = %q,
want %q", got, "header")
+ }
+ if got := resp.Trailer().Get("X-Triple-Echo-Trailer");
got != "trailer" {
+ t.Fatalf("response trailer
X-Triple-Echo-Trailer = %q, want %q", got, "trailer")
+ }
+ if err := conn.CloseResponse(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+ })
+ }
+}
+
+// TestUnaryFastPathProtoJSONCodec verifies that the fast path carries
+// JSON-encoded payloads end to end, matching the duplex path's codec support.
+func TestUnaryFastPathProtoJSONCodec(t *testing.T) {
+ mux := http.NewServeMux()
+ mux.Handle("/connect.ping.v1.PingService/Ping", newPingHandler())
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ t.Cleanup(server.Close)
+ httpClient := server.Client()
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ header := make(http.Header)
+ header.Set(headerContentType, tripleUnaryContentTypeJSON)
+ header.Set(tripleUnaryHeaderAcceptCompression, "gzip")
+ pool := newBufferPool()
+
+ for _, tc := range []struct {
+ name string
+ fast bool
+ }{
+ {"duplex", false},
+ {"fastpath", true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ conn := newTripleClientConn(tc.fast,
&protoJSONCodec{name: codecNameJSON}, httpClient, serverURL, header, pool)
+ if err := conn.Send(&pingv1.PingRequest{Text:
"hello"}); err != nil {
+ t.Fatalf("send: %v", err)
+ }
+ if err := conn.CloseRequest(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ resp := NewResponse(&pingv1.PingResponse{})
+ if err := receiveUnaryResponse(conn, resp); err != nil {
+ t.Fatalf("receive: %v", err)
+ }
+ pingResp, ok := resp.Any().(*pingv1.PingResponse)
+ if !ok {
+ t.Fatalf("unexpected response type %T",
resp.Any())
+ }
+ if pingResp.Text != "hello" {
+ t.Fatalf("response text = %q, want %q",
pingResp.Text, "hello")
+ }
+ if err := conn.CloseResponse(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+ })
+ }
+}
+
+// TestUnaryFastPathGRPCRoutesUnary verifies that grpcClient.NewConn routes
+// unary calls through unaryFastPathCall when the switch is enabled and keeps
+// duplexHTTPCall otherwise; streaming calls always use duplexHTTPCall.
+func TestUnaryFastPathGRPCRoutesUnary(t *testing.T) {
+ newClient := func(fast bool) *grpcClient {
+ return &grpcClient{
+ protocolClientParams: protocolClientParams{
+ HTTPClient: &http.Client{},
+ URL: &url.URL{Scheme: "http",
Host: "example.com"},
+ BufferPool: newBufferPool(),
+ Codec: &protoBinaryCodec{},
+ CompressionPools:
newReadOnlyCompressionPools(map[string]*compressionPool{}, nil),
+ UnaryFastPath: fast,
+ },
+ peer: Peer{},
+ }
+ }
+ assertCallType := func(t *testing.T, conn StreamingClientConn, want
string) {
+ t.Helper()
+ translated, ok := conn.(*errorTranslatingClientConn)
+ if !ok {
+ t.Fatalf("unexpected conn wrapper %T", conn)
+ }
+ grpcConn, ok := translated.StreamingClientConn.(*grpcClientConn)
+ if !ok {
+ t.Fatalf("unexpected grpc conn %T",
translated.StreamingClientConn)
+ }
+ switch want {
+ case "fastpath":
+ if _, ok := grpcConn.call.(*unaryFastPathCall); !ok {
+ t.Fatalf("grpc call type = %T, want
*unaryFastPathCall", grpcConn.call)
+ }
+ case "duplex":
+ if _, ok := grpcConn.call.(*duplexHTTPCall); !ok {
+ t.Fatalf("grpc call type = %T, want
*duplexHTTPCall", grpcConn.call)
+ }
+ default:
+ // Guard against a misspelled want string silently
passing.
+ t.Fatalf("assertCallType: unknown want %q", want)
+ }
+ }
+
+ unarySpec := Spec{StreamType: StreamTypeUnary, Procedure:
"/connect.ping.v1.PingService/Ping"}
+ // Enabled -> unary calls take the fast path on the gRPC protocol too.
+ assertCallType(t, newClient(true).NewConn(context.Background(),
unarySpec, make(http.Header)), "fastpath")
+ // Disabled -> unary calls keep using duplexHTTPCall.
+ assertCallType(t, newClient(false).NewConn(context.Background(),
unarySpec, make(http.Header)), "duplex")
+ // Streaming calls always use duplexHTTPCall, regardless of the switch.
+ streamSpec := Spec{StreamType: StreamTypeBidi, Procedure:
"/connect.ping.v1.PingService/Ping"}
+ assertCallType(t, newClient(true).NewConn(context.Background(),
streamSpec, make(http.Header)), "duplex")
+}
diff --git a/protocol/triple/triple_protocol/unary_fastpath_bench_test.go
b/protocol/triple/triple_protocol/unary_fastpath_bench_test.go
new file mode 100644
index 000000000..1acc2f762
--- /dev/null
+++ b/protocol/triple/triple_protocol/unary_fastpath_bench_test.go
@@ -0,0 +1,131 @@
+/*
+ * 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_test
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol"
+
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/assert"
+ pingv1
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1"
+
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1/pingv1connect"
+)
+
+// newPingBenchmarkServer starts an HTTP/2 test server serving the PingService
+// handler, and returns the server and its HTTP client.
+func newPingBenchmarkServer(b *testing.B) (*httptest.Server, *http.Client) {
+ b.Helper()
+ mux := http.NewServeMux()
+ mux.Handle(
+ pingv1connect.NewPingServiceHandler(
+ &ExamplePingServer{},
+ ),
+ )
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ b.Cleanup(server.Close)
+
+ httpClient := server.Client()
+ httpTransport, ok := httpClient.Transport.(*http.Transport)
+ assert.True(b, ok)
+ httpTransport.DisableCompression = true
+ return server, httpClient
+}
+
+// unaryBenchPayloadSizes covers small (fixed-overhead dominated) through large
+// (bandwidth dominated) message sizes.
+var unaryBenchPayloadSizes = []int{
+ 128, 1024, 16 * 1024, 1024 * 1024,
+}
+
+// BenchmarkUnaryDuplex measures the duplex unary path with the fast path
+// explicitly disabled. The generated client defaults to the gRPC protocol,
+// so WithTriple selects the Triple protocol; WithoutUnaryFastPath opts back
+// out of the on-by-default fast path, keeping duplexHTTPCall for unary RPCs.
+// It is the control group for BenchmarkUnaryFastPathProduction.
+func BenchmarkUnaryDuplex(b *testing.B) {
+ server, httpClient := newPingBenchmarkServer(b)
+ client := pingv1connect.NewPingServiceClient(
+ httpClient,
+ server.URL,
+ triple_protocol.WithTriple(),
+ triple_protocol.WithoutUnaryFastPath(),
+ )
+ benchmarkUnaryPing(b, client)
+}
+
+// BenchmarkUnaryFastPathProduction measures the production unary fast path via
+// the generated client. The fast path is on by default; the WithUnaryFastPath
+// option is repeated here to state the intent explicitly.
+func BenchmarkUnaryFastPathProduction(b *testing.B) {
+ server, httpClient := newPingBenchmarkServer(b)
+ client := pingv1connect.NewPingServiceClient(
+ httpClient,
+ server.URL,
+ triple_protocol.WithTriple(),
+ triple_protocol.WithUnaryFastPath(),
+ )
+ benchmarkUnaryPing(b, client)
+}
+
+// benchmarkUnaryPing runs one parallel sub-benchmark per payload size through
+// the generated client.
+func benchmarkUnaryPing(b *testing.B, client pingv1connect.PingServiceClient) {
+ b.Helper()
+ for _, size := range unaryBenchPayloadSizes {
+ text := strings.Repeat("a", size)
+ b.Run(sizeLabel(size), func(b *testing.B) {
+ b.RunParallel(func(pb *testing.PB) {
+ for pb.Next() {
+ unaryPingIteration(b, client, text)
+ }
+ })
+ })
+ }
+}
+
+// unaryPingIteration performs one unary Ping call through the generated
client.
+func unaryPingIteration(b *testing.B, client pingv1connect.PingServiceClient,
text string) {
+ b.Helper()
+ req := pingv1.PingRequest{Text: text}
+ res := pingv1.PingResponse{}
+ if err := client.Ping(
+ context.Background(),
+ triple_protocol.NewRequest(&req),
+ triple_protocol.NewResponse(&res),
+ ); err != nil {
+ b.Fatalf("ping: %v", err)
+ }
+}
+
+func sizeLabel(size int) string {
+ switch size {
+ case 1024 * 1024:
+ return "1MiB"
+ default:
+ return fmt.Sprintf("%dB", size)
+ }
+}
diff --git a/protocol/triple/triple_protocol/unary_fastpath_body_test.go
b/protocol/triple/triple_protocol/unary_fastpath_body_test.go
new file mode 100644
index 000000000..9ba832a2a
--- /dev/null
+++ b/protocol/triple/triple_protocol/unary_fastpath_body_test.go
@@ -0,0 +1,308 @@
+/*
+ * 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 (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "sync/atomic"
+ "testing"
+)
+
+import (
+ pingv1
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1"
+)
+
+// TestUnaryFastPathBodyReturnOnAsyncClose verifies that the pooled request
+// body is returned to the pool exactly once and left clean when the transport
+// closes it after http.Client.Do has returned: Do may return while the
+// transport's background writer still reads the body, so the buffer must be
+// recycled from the Close callback rather than the Do call site. Repeated
+// Close (redirect / retry paths) and a concurrent Read must stay safe.
+func TestUnaryFastPathBodyReturnOnAsyncClose(t *testing.T) {
+ pool := newBufferPool()
+ buf := pool.Get()
+ buf.WriteString("payload")
+ body := &unaryRequestBody{buf: buf, pool: pool}
+
+ // The transport reads the whole body, then closes it after
http.Client.Do
+ // has already returned (abort path).
+ if _, err := io.Copy(io.Discard, body); err != nil {
+ t.Fatalf("read body: %v", err)
+ }
+ if err := body.Close(); err != nil {
+ t.Fatalf("close body: %v", err)
+ }
+
+ // The buffer must have been returned to the pool.
+ body.mu.Lock()
+ returned := body.buf == nil
+ body.mu.Unlock()
+ if !returned {
+ t.Fatal("request body buffer was not returned to the pool")
+ }
+ // Idempotent Close: a second or third Close from redirect / retry /
+ // context-cancel paths must not double-return the buffer.
+ if err := body.Close(); err != nil {
+ t.Fatalf("second close: %v", err)
+ }
+ if err := body.Close(); err != nil {
+ t.Fatalf("third close: %v", err)
+ }
+ // After return, Read must surface EOF instead of the pooled buffer's
+ // stale bytes (use-after-return guard).
+ if _, err := body.Read(make([]byte, 4)); err != io.EOF {
+ t.Fatalf("read after return = %v, want io.EOF", err)
+ }
+ // Reuse must be clean: whatever buffer the pool hands out next has been
+ // Reset, so appending new payload must not carry this request's data.
+ reuse := pool.Get()
+ reuse.WriteString("new-payload")
+ if got := reuse.String(); got != "new-payload" {
+ t.Fatalf("reused buffer not clean after return: got %q", got)
+ }
+ reuse.Reset()
+ pool.Put(reuse)
+}
+
+// TestUnaryFastPathBodyAbortServer verifies end to end that an early non-2xx
+// response aborts the body write, returns the pooled buffer, and leaves the
+// pool clean for a later request reusing it.
+func TestUnaryFastPathBodyAbortServer(t *testing.T) {
+ var first atomic.Int32
+ mux := http.NewServeMux()
+ mux.Handle("/connect.ping.v1.PingService/Ping", NewUnaryHandler(
+ "/connect.ping.v1.PingService/Ping",
+ func() any { return &pingv1.PingRequest{} },
+ func(ctx context.Context, req *Request) (*Response, error) {
+ // First request aborts early with a non-2xx error
without reading
+ // the request body, so the transport aborts the body
write. Later
+ // requests decode and echo the payload back, so a
poisoned pool is
+ // detectable via the echoed text.
+ if first.Add(1) == 1 {
+ err := NewError(CodePermissionDenied,
errors.New("early response"))
+ err.meta = make(http.Header)
+ err.meta.Set("X-Triple-Error", "meta")
+ return nil, err
+ }
+ pingReq, ok := req.Any().(*pingv1.PingRequest)
+ if !ok {
+ return nil, fmt.Errorf("unexpected request type
%T", req.Any())
+ }
+ return NewResponse(&pingv1.PingResponse{Text:
pingReq.Text}), nil
+ },
+ ))
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ t.Cleanup(server.Close)
+ httpClient := server.Client()
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ header := make(http.Header)
+ header.Set(headerContentType, "application/proto")
+ header.Set(tripleUnaryHeaderAcceptCompression, "gzip")
+ // One shared pool across requests, matching production where the conn
+ // reuses the protocol-level BufferPool.
+ pool := newBufferPool()
+
+ // First call hits the early-response abort path.
+ conn1 := newTripleClientConn(true, &protoBinaryCodec{}, httpClient,
serverURL, header, pool)
+ if err := conn1.Send(&pingv1.PingRequest{Text: "first-payload"}); err
!= nil {
+ t.Fatalf("send: %v", err)
+ }
+ if err := conn1.CloseRequest(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ resp1 := NewResponse(&pingv1.PingResponse{})
+ if err := receiveUnaryResponse(conn1, resp1); err == nil {
+ t.Fatal("expected error from early-response abort, got nil")
+ } else {
+ var connErr *Error
+ if !errors.As(err, &connErr) || connErr.Code() !=
CodePermissionDenied {
+ t.Fatalf("abort error = %v, want CodePermissionDenied",
err)
+ }
+ if got := connErr.meta.Get("X-Triple-Error"); got != "meta" {
+ t.Fatalf("abort error meta = %q, want %q", got, "meta")
+ }
+ }
+ if err := conn1.CloseResponse(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+
+ // Second call reuses the same pool: the body must be clean.
+ conn2 := newTripleClientConn(true, &protoBinaryCodec{}, httpClient,
serverURL, header, pool)
+ if err := conn2.Send(&pingv1.PingRequest{Text: "second-payload"}); err
!= nil {
+ t.Fatalf("send: %v", err)
+ }
+ if err := conn2.CloseRequest(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ resp2 := NewResponse(&pingv1.PingResponse{})
+ if err := receiveUnaryResponse(conn2, resp2); err != nil {
+ t.Fatalf("second call after abort: %v", err)
+ }
+ pingResp, ok := resp2.Any().(*pingv1.PingResponse)
+ if !ok {
+ t.Fatalf("unexpected response type %T", resp2.Any())
+ }
+ if pingResp.Text != "second-payload" {
+ t.Fatalf("second call body contaminated: got text %q",
pingResp.Text)
+ }
+ if err := conn2.CloseResponse(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+}
+
+// TestUnaryFastPathEmptyWriteSkipsPool verifies that zero-length writes do
+// not pull a buffer out of the pool: Send(nil), an empty protobuf message,
+// or an explicit Write of an empty slice must leave c.body nil so that
+// makeRequest hands http.NoBody to the transport and the pool stays intact.
+func TestUnaryFastPathEmptyWriteSkipsPool(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ if r.ContentLength != 0 {
+ t.Errorf("request ContentLength = %d, want 0 for empty
payload", r.ContentLength)
+ }
+ w.WriteHeader(http.StatusOK)
+ }))
+ t.Cleanup(server.Close)
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ pool := newBufferPool()
+ call := newUnaryFastPathCall(
+ context.Background(),
+ server.Client(),
+ serverURL,
+ Spec{Procedure: "/connect.ping.v1.PingService/Ping"},
+ make(http.Header),
+ pool,
+ )
+ call.SetValidateResponse(func(*http.Response) *Error { return nil })
+
+ for _, empty := range [][]byte{nil, []byte{}} {
+ if n, err := call.Write(empty); n != 0 || err != nil {
+ t.Fatalf("Write(%v) = (%d, %v), want (0, nil)", empty,
n, err)
+ }
+ }
+ if call.body != nil {
+ t.Fatal("zero-length write pulled a buffer from the pool")
+ }
+ if err := call.CloseWrite(); err != nil {
+ t.Fatalf("close write: %v", err)
+ }
+ if call.request.Body != http.NoBody {
+ t.Fatalf("request body = %T, want http.NoBody",
call.request.Body)
+ }
+}
+
+// TestUnaryFastPathMakeRequestRecyclesEmptyBuffer verifies the fallback guard
+// in makeRequest: if a zero-length buffer reaches the send path despite the
+// Write short-circuit, it must be returned to the pool instead of being
+// replaced by http.NoBody and leaked.
+func TestUnaryFastPathMakeRequestRecyclesEmptyBuffer(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(http.StatusOK)
+ }))
+ t.Cleanup(server.Close)
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ pool := newBufferPool()
+ call := newUnaryFastPathCall(
+ context.Background(),
+ server.Client(),
+ serverURL,
+ Spec{Procedure: "/connect.ping.v1.PingService/Ping"},
+ make(http.Header),
+ pool,
+ )
+ call.SetValidateResponse(func(*http.Response) *Error { return nil })
+ // Simulate a zero-length buffer reaching the send path without going
+ // through Write (the short-circuit normally prevents this).
+ call.body = pool.Get()
+ if err := call.CloseWrite(); err != nil {
+ t.Fatalf("close write: %v", err)
+ }
+ if call.body != nil {
+ t.Fatal("empty buffer was not returned to the pool")
+ }
+ if call.request.Body != http.NoBody {
+ t.Fatalf("request body = %T, want http.NoBody",
call.request.Body)
+ }
+}
+
+// TestUnaryFastPathEmptyMessageCall verifies end to end that an empty
+// protobuf request (marshaled to zero bytes) still completes a unary call:
+// the fast path must hand http.NoBody to the transport without leaking a
+// pooled buffer.
+func TestUnaryFastPathEmptyMessageCall(t *testing.T) {
+ mux := http.NewServeMux()
+ mux.Handle("/connect.ping.v1.PingService/Ping", NewUnaryHandler(
+ "/connect.ping.v1.PingService/Ping",
+ func() any { return &pingv1.PingRequest{} },
+ func(ctx context.Context, req *Request) (*Response, error) {
+ return NewResponse(&pingv1.PingResponse{Text:
"empty-ok"}), nil
+ },
+ ))
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ t.Cleanup(server.Close)
+ httpClient := server.Client()
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ header := make(http.Header)
+ header.Set(headerContentType, "application/proto")
+ pool := newBufferPool()
+
+ conn := newTripleClientConn(true, &protoBinaryCodec{}, httpClient,
serverURL, header, pool)
+ if err := conn.Send(&pingv1.PingRequest{}); err != nil {
+ t.Fatalf("send empty message: %v", err)
+ }
+ if err := conn.CloseRequest(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ resp := NewResponse(&pingv1.PingResponse{})
+ if err := receiveUnaryResponse(conn, resp); err != nil {
+ t.Fatalf("receive: %v", err)
+ }
+ pingResp, ok := resp.Any().(*pingv1.PingResponse)
+ if !ok {
+ t.Fatalf("unexpected response type %T", resp.Any())
+ }
+ if pingResp.Text != "empty-ok" {
+ t.Fatalf("echo text = %q, want %q", pingResp.Text, "empty-ok")
+ }
+ if err := conn.CloseResponse(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+}
diff --git a/protocol/triple/triple_protocol/unary_fastpath_concurrency_test.go
b/protocol/triple/triple_protocol/unary_fastpath_concurrency_test.go
new file mode 100644
index 000000000..3955059cd
--- /dev/null
+++ b/protocol/triple/triple_protocol/unary_fastpath_concurrency_test.go
@@ -0,0 +1,158 @@
+/*
+ * 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 (
+ "context"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "sync"
+ "testing"
+)
+
+// concurrentEchoServer starts an HTTP/2 server that records every request body
+// it receives, so tests can assert on what actually reached the wire.
+func concurrentEchoServer(t *testing.T) (HTTPClient, *url.URL, *[]string,
*sync.Mutex) {
+ t.Helper()
+ var bodies []string
+ var mu sync.Mutex
+ mux := http.NewServeMux()
+ mux.HandleFunc("/connect.ping.v1.PingService/Ping", func(w
http.ResponseWriter, r *http.Request) {
+ b, _ := io.ReadAll(r.Body)
+ mu.Lock()
+ bodies = append(bodies, string(b))
+ mu.Unlock()
+ w.WriteHeader(http.StatusOK)
+ })
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ t.Cleanup(server.Close)
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return server.Client(), serverURL, &bodies, &mu
+}
+
+func newConcurrentCall(httpClient HTTPClient, serverURL *url.URL, pool
*bufferPool) *unaryFastPathCall {
+ call := newUnaryFastPathCall(
+ context.Background(),
+ httpClient,
+ serverURL,
+ Spec{
+ StreamType: StreamTypeUnary,
+ Procedure: "/connect.ping.v1.PingService/Ping",
+ },
+ make(http.Header),
+ pool,
+ )
+ // The conn layer injects validateResponse via SetValidateResponse;
mirror
+ // that here.
+ call.SetValidateResponse(func(*http.Response) *Error { return nil })
+ return call
+}
+
+// TestUnaryFastPathWriteCloseConcurrent verifies the write-side concurrency
+// contract: a Write racing a CloseWrite must fully succeed or fail with
+// io.EOF, and the server must never see a torn body.
+func TestUnaryFastPathWriteCloseConcurrent(t *testing.T) {
+ httpClient, serverURL, bodies, bodiesMu := concurrentEchoServer(t)
+ pool := newBufferPool()
+ payload := []byte("payload-concurrent")
+
+ const iterations = 30
+ for i := range iterations {
+ call := newConcurrentCall(httpClient, serverURL, pool)
+ var (
+ writeN int
+ writeErr error
+ wg sync.WaitGroup
+ )
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ writeN, writeErr = call.Write(payload)
+ }()
+ go func() {
+ defer wg.Done()
+ // CloseWrite always returns nil; transport failures
surface from
+ // Read below, as on the duplex path.
+ _ = call.CloseWrite()
+ }()
+ wg.Wait()
+
+ // Drain the response so the request body Close callback runs
and the
+ // pooled buffer returns to the pool.
+ if _, err := io.Copy(io.Discard, call); err != nil {
+ t.Fatalf("iter %d: read response: %v", i, err)
+ }
+ if err := call.CloseRead(); err != nil {
+ t.Fatalf("iter %d: close response: %v", i, err)
+ }
+
+ // Write must have fully succeeded, or been rejected with io.EOF
+ // because the racing CloseWrite already sent the body.
+ if writeErr != nil && !errors.Is(writeErr, io.EOF) {
+ t.Fatalf("iter %d: write error = %v, want nil or
io.EOF", i, writeErr)
+ }
+ if writeErr == nil && writeN != len(payload) {
+ t.Fatalf("iter %d: write n = %d, want %d", i, writeN,
len(payload))
+ }
+ }
+
+ bodiesMu.Lock()
+ defer bodiesMu.Unlock()
+ if len(*bodies) != iterations {
+ t.Fatalf("server received %d requests, want %d", len(*bodies),
iterations)
+ }
+ for _, body := range *bodies {
+ if body != "" && body != string(payload) {
+ t.Fatalf("server received torn body %q, want empty or
the full payload", body)
+ }
+ }
+}
+
+// TestUnaryFastPathWriteAfterClose verifies that Write after CloseWrite has
+// dispatched the body fails with io.EOF instead of racing the transport.
+func TestUnaryFastPathWriteAfterClose(t *testing.T) {
+ httpClient, serverURL, _, _ := concurrentEchoServer(t)
+ pool := newBufferPool()
+
+ call := newConcurrentCall(httpClient, serverURL, pool)
+ if err := call.CloseWrite(); err != nil {
+ t.Fatalf("close request: %v", err)
+ }
+ n, err := call.Write([]byte("late-write"))
+ if !errors.Is(err, io.EOF) {
+ t.Fatalf("write after close = (n=%d, err=%v), want (0,
io.EOF)", n, err)
+ }
+ if n != 0 {
+ t.Fatalf("write after close n = %d, want 0", n)
+ }
+
+ if _, err := io.Copy(io.Discard, call); err != nil {
+ t.Fatalf("read response: %v", err)
+ }
+ if err := call.CloseRead(); err != nil {
+ t.Fatalf("close response: %v", err)
+ }
+}
diff --git a/protocol/triple/triple_protocol/unary_fastpath_conn_type_test.go
b/protocol/triple/triple_protocol/unary_fastpath_conn_type_test.go
new file mode 100644
index 000000000..e3497f7da
--- /dev/null
+++ b/protocol/triple/triple_protocol/unary_fastpath_conn_type_test.go
@@ -0,0 +1,78 @@
+/*
+ * 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 (
+ "context"
+ "net/http"
+ "net/url"
+ "testing"
+)
+
+import (
+
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/assert"
+)
+
+// TestUnaryFastPathNewConnType verifies that tripleClient.NewConn routes unary
+// calls through unaryFastPathCall when enabled and duplexHTTPCall otherwise;
+// streaming calls always use duplexHTTPCall.
+func TestUnaryFastPathNewConnType(t *testing.T) {
+ newClient := func(fast bool) *tripleClient {
+ // CompressionPools must be non-nil; NewConn consults it when
+ // building the marshaler even if compression is disabled.
+ return &tripleClient{
+ protocolClientParams: protocolClientParams{
+ HTTPClient: &http.Client{},
+ URL: &url.URL{Scheme: "http",
Host: "example.com"},
+ BufferPool: newBufferPool(),
+ Codec: &protoBinaryCodec{},
+ CompressionPools:
newReadOnlyCompressionPools(map[string]*compressionPool{}, nil),
+ UnaryFastPath: fast,
+ },
+ peer: Peer{},
+ }
+ }
+ assertCallType := func(t *testing.T, conn StreamingClientConn, want
string) {
+ t.Helper()
+ translated, ok := conn.(*errorTranslatingClientConn)
+ assert.True(t, ok, assert.Sprintf("unexpected conn wrapper %T",
conn))
+ unaryConn, ok :=
translated.StreamingClientConn.(*tripleUnaryClientConn)
+ assert.True(t, ok, assert.Sprintf("unexpected unary conn %T",
translated.StreamingClientConn))
+ switch want {
+ case "fastpath":
+ _, ok := unaryConn.call.(*unaryFastPathCall)
+ assert.True(t, ok, assert.Sprintf("unary call type =
%T, want *unaryFastPathCall", unaryConn.call))
+ case "duplex":
+ _, ok := unaryConn.call.(*duplexHTTPCall)
+ assert.True(t, ok, assert.Sprintf("unary call type =
%T, want *duplexHTTPCall", unaryConn.call))
+ default:
+ // Guard against a misspelled want string silently
passing:
+ // the switch would otherwise skip every case and
report success.
+ t.Fatalf("assertCallType: unknown want %q", want)
+ }
+ }
+
+ unarySpec := Spec{StreamType: StreamTypeUnary, Procedure:
"/connect.ping.v1.PingService/Ping"}
+ // WithUnaryFastPath enabled -> unary calls take the fast path.
+ assertCallType(t, newClient(true).NewConn(context.Background(),
unarySpec, make(http.Header)), "fastpath")
+ // Default (option disabled) -> unary calls keep using duplexHTTPCall.
+ assertCallType(t, newClient(false).NewConn(context.Background(),
unarySpec, make(http.Header)), "duplex")
+ // Streaming calls always use duplexHTTPCall, regardless of the switch.
+ streamSpec := Spec{StreamType: StreamTypeBidi, Procedure:
"/connect.ping.v1.PingService/Ping"}
+ assertCallType(t, newClient(true).NewConn(context.Background(),
streamSpec, make(http.Header)), "duplex")
+}
diff --git a/protocol/triple/unary_fastpath_public_entry_test.go
b/protocol/triple/unary_fastpath_public_entry_test.go
new file mode 100644
index 000000000..f49a6c958
--- /dev/null
+++ b/protocol/triple/unary_fastpath_public_entry_test.go
@@ -0,0 +1,260 @@
+/*
+ * 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
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/http"
+ "strings"
+ "sync"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "golang.org/x/net/http2"
+ "golang.org/x/net/http2/h2c"
+
+ "google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/global"
+ "dubbo.apache.org/dubbo-go/v3/protocol/invocation"
+ tri "dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol"
+)
+
+// TestCallUnaryFastPathWireSignature verifies that a unary call issued through
+// the production entry — TripleProtocol.Refer, then TripleInvoker.Invoke with
+// a real invocation (method, parameters, call type, metadata and attachments
+// converted exactly as production does) — reaches the server with a
+// pre-declared Content-Length when the fast path is on, and streams without
+// one when it is explicitly disabled. The fast path buffers the whole request
+// body and sets Content-Length; duplexHTTPCall streams through an io.Pipe and
+// cannot, so the server sees Content-Length == -1. This guards the dispatch
+// wiring from the RPC layer down to the gRPC protocol client: if Refer ever
+// routed to a different Invoker, or Invoke stopped reaching this manager, the
+// wire signature assertion fails even though the manager-level path is intact.
+func TestCallUnaryFastPathWireSignature(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ tripleConf *global.TripleConfig
+ wantPositive bool
+ }{
+ {"default-on", nil, true},
+ {"explicitly-on", &global.TripleConfig{UnaryFastPath:
boolPtr(true)}, true},
+ {"explicitly-off", &global.TripleConfig{UnaryFastPath:
boolPtr(false)}, false},
+ {"unset-keeps-default", &global.TripleConfig{KeepAliveInterval:
"10s"}, true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ var (
+ mu sync.Mutex
+ contentLen int64
+ )
+ pingHandler := tri.NewUnaryHandler(
+ "/connect.ping.v1.PingService/Ping",
+ func() any { return &wrapperspb.StringValue{} },
+ func(_ context.Context, req *tri.Request)
(*tri.Response, error) {
+ sv :=
req.Any().(*wrapperspb.StringValue)
+ return
tri.NewResponse(&wrapperspb.StringValue{Value: sv.Value}), nil
+ },
+ )
+ // Refer starts a background gRPC health-check stream;
it hits a
+ // different path (grpc.health.v1.Health/...) with a
streaming body
+ // (Content-Length == -1) that would clobber the Ping
signature.
+ // Capture every non-health request instead, so the
assertion is
+ // immune to both the health stream and the exact Ping
path.
+ wrapped := http.HandlerFunc(func(w http.ResponseWriter,
r *http.Request) {
+ if !strings.HasPrefix(r.URL.Path,
"/grpc.health.v1.Health/") {
+ mu.Lock()
+ contentLen = r.ContentLength
+ mu.Unlock()
+ }
+ pingHandler.ServeHTTP(w, r)
+ })
+ server := &http.Server{Handler: h2c.NewHandler(wrapped,
&http2.Server{})}
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ defer server.Close()
+ go func() {
+ _ = server.Serve(ln)
+ }()
+
+ url, err := common.NewURL(
+
"tri://"+ln.Addr().String()+"/connect.ping.v1.PingService",
+ common.WithMethods([]string{"Ping"}),
+ common.WithProtocol(TRIPLE),
+ common.WithParamsValue(constant.IDLMode,
constant.NONIDL),
+ )
+ require.NoError(t, err)
+ if tc.tripleConf != nil {
+ url.SetAttribute(constant.TripleConfigKey,
tc.tripleConf)
+ }
+
+ invoker := GetProtocol().Refer(url)
+ require.NotNil(t, invoker)
+ defer invoker.Destroy()
+
+ resp := &wrapperspb.StringValue{}
+ inv := invocation.NewRPCInvocationWithOptions(
+ invocation.WithMethodName("Ping"),
+ invocation.WithParameterRawValues([]any{
+ &wrapperspb.StringValue{Value: "hello"},
+ resp,
+ }),
+ )
+ inv.SetAttribute(constant.CallTypeKey,
constant.CallUnary)
+
+ res := invoker.Invoke(context.Background(), inv)
+ require.NoError(t, res.Error())
+ assert.Equal(t, "hello", resp.Value)
+
+ mu.Lock()
+ defer mu.Unlock()
+ if tc.wantPositive {
+ assert.Positive(t, contentLen,
+ "server saw Content-Length %d, want > 0
(fast path signature)", contentLen)
+ } else {
+ assert.Equal(t, int64(-1), contentLen,
+ "server saw Content-Length %d, want -1
(duplex streams the body)", contentLen)
+ }
+ })
+ }
+}
+
+// TestCallUnaryFastPathErrorPaths verifies that error propagation through the
+// production entry (TripleProtocol.Refer + TripleInvoker.Invoke) behaves
+// identically whether the fast path is on (default) or disabled (duplex): a
+// canceled context fails fast, a handler error surfaces with its code and
+// message, and a gRPC trailers-only response (grpc-status with no message
+// body) is decoded correctly. This guards the gRPC wire error path of the
+// fast path through the same Refer/Invoker dispatch the RPC layer uses.
+func TestCallUnaryFastPathErrorPaths(t *testing.T) {
+ scenarios := []struct {
+ name string
+ serve func(context.Context, *tri.Request) (*tri.Response,
error)
+ wantErr func(*testing.T, error)
+ }{
+ {
+ name: "handler-error",
+ serve: func(_ context.Context, _ *tri.Request)
(*tri.Response, error) {
+ return nil, tri.NewError(tri.CodeUnavailable,
errors.New("boom"))
+ },
+ wantErr: func(t *testing.T, err error) {
+ require.Error(t, err)
+ assert.Equal(t, tri.CodeUnavailable,
tri.CodeOf(err))
+ assert.Contains(t, err.Error(), "boom")
+ },
+ },
+ {
+ name: "trailers-only",
+ serve: func(_ context.Context, _ *tri.Request)
(*tri.Response, error) {
+ // A gRPC error response carries
grpc-status/grpc-message in
+ // trailers only, with no message body.
+ return nil, tri.NewError(tri.CodeCanceled, nil)
+ },
+ wantErr: func(t *testing.T, err error) {
+ require.Error(t, err)
+ assert.Equal(t, tri.CodeCanceled,
tri.CodeOf(err))
+ },
+ },
+ {
+ name: "context-canceled",
+ serve: func(_ context.Context, _ *tri.Request)
(*tri.Response, error) {
+ return
tri.NewResponse(&wrapperspb.StringValue{Value: "echo"}), nil
+ },
+ wantErr: func(t *testing.T, err error) {
+ require.Error(t, err)
+ assert.ErrorIs(t, err, context.Canceled)
+ },
+ },
+ }
+
+ for _, sc := range scenarios {
+ t.Run(sc.name, func(t *testing.T) {
+ for _, cfg := range []struct {
+ name string
+ tripleConf *global.TripleConfig
+ }{
+ {"fastpath", nil},
+ {"duplex", &global.TripleConfig{UnaryFastPath:
boolPtr(false)}},
+ } {
+ t.Run(cfg.name, func(t *testing.T) {
+ pingHandler := tri.NewUnaryHandler(
+
"/connect.ping.v1.PingService/Ping",
+ func() any { return
&wrapperspb.StringValue{} },
+ sc.serve,
+ )
+ server := &http.Server{Handler:
h2c.NewHandler(pingHandler, &http2.Server{})}
+ ln, err := net.Listen("tcp",
"127.0.0.1:0")
+ require.NoError(t, err)
+ defer server.Close()
+ go func() {
+ _ = server.Serve(ln)
+ }()
+
+ url, err := common.NewURL(
+
"tri://"+ln.Addr().String()+"/connect.ping.v1.PingService",
+
common.WithMethods([]string{"Ping"}),
+ common.WithProtocol(TRIPLE),
+
common.WithParamsValue(constant.IDLMode, constant.NONIDL),
+ )
+ require.NoError(t, err)
+ if cfg.tripleConf != nil {
+
url.SetAttribute(constant.TripleConfigKey, cfg.tripleConf)
+ }
+
+ invoker := GetProtocol().Refer(url)
+ require.NotNil(t, invoker)
+ defer invoker.Destroy()
+
+ ctx := context.Background()
+ if sc.name == "context-canceled" {
+ var cancel context.CancelFunc
+ ctx, cancel =
context.WithCancel(ctx)
+ cancel()
+ }
+ resp := &wrapperspb.StringValue{}
+ inv :=
invocation.NewRPCInvocationWithOptions(
+
invocation.WithMethodName("Ping"),
+
invocation.WithParameterRawValues([]any{
+
&wrapperspb.StringValue{Value: "hello"},
+ resp,
+ }),
+ )
+ inv.SetAttribute(constant.CallTypeKey,
constant.CallUnary)
+
+ res := invoker.Invoke(ctx, inv)
+ sc.wantErr(t, res.Error())
+ })
+ }
+ })
+ }
+}
+
+// boolPtr returns a pointer to v, mirroring the *bool config field style used
+// across the global config structs.
+func boolPtr(v bool) *bool {
+ return &v
+}
