-
Notifications
You must be signed in to change notification settings - Fork 3.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
138368: all: host a DRPC server and use it for node-node BatchRequests r=cthumuluru-crdb a=cthumuluru-crdb drpc (github.com/storj/drpc) is a lightweight, drop-in replacement for gRPC. Initial benchmark results from the Perf Efficiency team, using a simple ping service (github.com/cockroachlabs/perf-efficiency-team/tree/main/rpctoy), demonstrate that switching from gRPC to drpc delivers significant performance improvements. ``` $ benchdiff --old lastmerge ./pkg/sql/tests -b -r 'Sysbench/SQL/3node/oltp_read_write' -d 1000x -c 10 name old time/op new time/op delta Sysbench/SQL/3node/oltp_read_write-24 14.0ms ± 2% 14.1ms ± 1% ~ (p=0.063 n=10+10) name old alloc/op new alloc/op delta Sysbench/SQL/3node/oltp_read_write-24 2.55MB ± 1% 2.26MB ± 2% -11.39% (p=0.000 n=9+9) name old allocs/op new allocs/op delta Sysbench/SQL/3node/oltp_read_write-24 17.2k ± 1% 11.7k ± 0% -32.30% (p=0.000 n=10+8) ``` This pull request introduces experimental support for using drpc to serve BatchRequests. The feature is disabled by default but can be enabled explicitly by setting the `COCKROACH_EXPERIMENTAL_DRPC_ENABLED` environment variable or automatically in CRDB test builds. The prototype lacks key features, such as authorization checks, interceptors, etc, and is not production-ready. When DRPC is disabled, sever, muxer and listener mock implementations are injected. If it is enabled real implementations are used. DRPC is only used for BatchRequests. This PR also added support for pooled streaming unary operations similar to #136648. Added unit tests to ensure: - CRDB nodes can server BatchRequests. - Simple queries when DRPC is enabled or disabled. Note: drpc protocol generation is not yet integrated into the build process. Protobuf files were generated manually and committed directly to the repository. This PR has a few parts: - it adds a copy of drpc generated code for a service that supports unary BatchRequest. - in `rpc.NewServerEx`, we now also set up a drpc server and return it up the stack. - Registers BatchRequest service with both DRPC and gRPC. If DRPC is enabled, it is used to make BatchRequests, if not gRPC is used. - in our listener setup, we configure cmux to match on the `DRPC!!!1` header and serve the resulting listener on the drpc server. The drpc example uses `drpcmigrate.ListenMux` instead of cmux; we keep cmux but must make sure the header is read off the connection before delegating (which the drpxmigrate mux would have done for us). - if using TLS, wrap the drpc listener with TLS config and use it to servr DRPC requests. - add support to reuse drpc streams across unary BatchRequests. However, the DRPC implementation is not on par with the gRPC counterpart in terms of allocation optimizations. Closes #136794. Epic: CRDB-42584 Release note: None Co-authored-by: Chandra Thumuluru <[email protected]> Co-authored-by: Tobias Grieger <[email protected]>
- Loading branch information
Showing
26 changed files
with
900 additions
and
53 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9302,6 +9302,16 @@ def go_deps(): | |
| "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/zeebo/assert/com_github_zeebo_assert-v1.3.0.zip", | ||
| ], | ||
| ) | ||
| go_repository( | ||
| name = "com_github_zeebo_errs", | ||
| build_file_proto_mode = "disable_global", | ||
| importpath = "github.com/zeebo/errs", | ||
| sha256 = "d2fa293e275c21bfb413e2968d79036931a55f503d8b62381563ed189b523cd2", | ||
| strip_prefix = "github.com/zeebo/[email protected]", | ||
| urls = [ | ||
| "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/zeebo/errs/com_github_zeebo_errs-v1.2.2.zip", | ||
| ], | ||
| ) | ||
| go_repository( | ||
| name = "com_github_zeebo_xxh3", | ||
| build_file_proto_mode = "disable_global", | ||
|
|
@@ -11435,6 +11445,16 @@ def go_deps(): | |
| "https://storage.googleapis.com/cockroach-godeps/gomod/rsc.io/sampler/io_rsc_sampler-v1.3.0.zip", | ||
| ], | ||
| ) | ||
| go_repository( | ||
| name = "io_storj_drpc", | ||
| build_file_proto_mode = "disable_global", | ||
| importpath = "storj.io/drpc", | ||
| sha256 = "e297ccead2763d354959a3c04b0c9c27c9c84c99d129f216ec07da663ee0091a", | ||
| strip_prefix = "storj.io/[email protected]", | ||
| urls = [ | ||
| "https://storage.googleapis.com/cockroach-godeps/gomod/storj.io/drpc/io_storj_drpc-v0.0.34.zip", | ||
| ], | ||
| ) | ||
| go_repository( | ||
| name = "io_vitess_vitess", | ||
| build_file_proto_mode = "disable_global", | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,207 @@ | ||
| // Copyright 2025 The Cockroach Authors. | ||
| // | ||
| // Use of this software is governed by the CockroachDB Software License | ||
| // included in the /LICENSE file. | ||
|
|
||
| // This file was manually generated with the DRPC protogen plugin using a dummy | ||
| // `api.proto` that includes a subset of relevant service methods. | ||
| // | ||
| // For instance, to generate this file, following proto file was used: | ||
| // | ||
| // -- api.proto -- begin -- | ||
| // syntax = "proto3"; | ||
| // package cockroach.kv.kvpb; | ||
| // option go_package = "github.com/cockroachdb/cockroach/pkg/kv/kvpb"; | ||
| // service Batch { | ||
| // rpc Batch (BatchRequest) returns (BatchResponse) {} | ||
| // rpc BatchStream (stream BatchRequest) returns (stream BatchResponse) {} | ||
| // } | ||
| // message BatchRequest{} | ||
| // message BatchResponse{} | ||
| // -- api.proto -- end -- | ||
| // | ||
| // NB: The use of empty BatchRequest and BatchResponse messages is a deliberate | ||
| // decision to avoid dependencies. | ||
| // | ||
| // | ||
| // To generate this file using DRPC protogen plugin from the dummy `api.proto` | ||
| // defined above, use the following command: | ||
| // | ||
| // ``` | ||
| // protoc --gogo_out=paths=source_relative:. \ | ||
| // --go-drpc_out=paths=source_relative,protolib=github.com/gogo/protobuf:. \ | ||
| // api.proto | ||
| // ``` | ||
| // | ||
| // NB: Make sure you have `protoc` installed and `protoc-gen-gogoroach` is | ||
| // built from $COCKROACH_SRC/pkg/cmd/protoc-gen-gogoroach. | ||
| // | ||
| // This code-gen should be automated as part of productionizing drpc. | ||
|
|
||
| package kvpb | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/util/protoutil" | ||
| "github.com/cockroachdb/errors" | ||
| "storj.io/drpc" | ||
| "storj.io/drpc/drpcerr" | ||
| ) | ||
|
|
||
| type drpcEncoding_File_api_proto struct{} | ||
|
|
||
| func (drpcEncoding_File_api_proto) Marshal(msg drpc.Message) ([]byte, error) { | ||
| return protoutil.Marshal(msg.(protoutil.Message)) | ||
| } | ||
|
|
||
| func (drpcEncoding_File_api_proto) Unmarshal(buf []byte, msg drpc.Message) error { | ||
| return protoutil.Unmarshal(buf, msg.(protoutil.Message)) | ||
| } | ||
|
|
||
| type DRPCBatchClient interface { | ||
| DRPCConn() drpc.Conn | ||
|
|
||
| Batch(ctx context.Context, in *BatchRequest) (*BatchResponse, error) | ||
| BatchStream(ctx context.Context) (DRPCBatch_BatchStreamClient, error) | ||
| } | ||
|
|
||
| type drpcBatchClient struct { | ||
| cc drpc.Conn | ||
| } | ||
|
|
||
| func NewDRPCBatchClient(cc drpc.Conn) DRPCBatchClient { | ||
| return &drpcBatchClient{cc} | ||
| } | ||
|
|
||
| func (c *drpcBatchClient) DRPCConn() drpc.Conn { return c.cc } | ||
|
|
||
| func (c *drpcBatchClient) Batch(ctx context.Context, in *BatchRequest) (*BatchResponse, error) { | ||
| out := new(BatchResponse) | ||
| err := c.cc.Invoke(ctx, "/cockroach.kv.kvpb.Batch/Batch", drpcEncoding_File_api_proto{}, in, out) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return out, nil | ||
| } | ||
|
|
||
| func (c *drpcBatchClient) BatchStream(ctx context.Context) (DRPCBatch_BatchStreamClient, error) { | ||
| stream, err := c.cc.NewStream(ctx, "/cockroach.kv.kvpb.Batch/BatchStream", drpcEncoding_File_api_proto{}) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| x := &drpcBatch_BatchStreamClient{stream} | ||
| return x, nil | ||
| } | ||
|
|
||
| type DRPCBatch_BatchStreamClient interface { | ||
| drpc.Stream | ||
| Send(*BatchRequest) error | ||
| Recv() (*BatchResponse, error) | ||
| } | ||
|
|
||
| type drpcBatch_BatchStreamClient struct { | ||
| drpc.Stream | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamClient) GetStream() drpc.Stream { | ||
| return x.Stream | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamClient) Send(m *BatchRequest) error { | ||
| return x.MsgSend(m, drpcEncoding_File_api_proto{}) | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamClient) Recv() (*BatchResponse, error) { | ||
| m := new(BatchResponse) | ||
| if err := x.MsgRecv(m, drpcEncoding_File_api_proto{}); err != nil { | ||
| return nil, err | ||
| } | ||
| return m, nil | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamClient) RecvMsg(m *BatchResponse) error { | ||
| return x.MsgRecv(m, drpcEncoding_File_api_proto{}) | ||
| } | ||
|
|
||
| type DRPCBatchServer interface { | ||
| Batch(context.Context, *BatchRequest) (*BatchResponse, error) | ||
| BatchStream(DRPCBatch_BatchStreamStream) error | ||
| } | ||
|
|
||
| type DRPCBatchUnimplementedServer struct{} | ||
|
|
||
| func (s *DRPCBatchUnimplementedServer) Batch( | ||
| context.Context, *BatchRequest, | ||
| ) (*BatchResponse, error) { | ||
| return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented) | ||
| } | ||
|
|
||
| func (s *DRPCBatchUnimplementedServer) BatchStream(DRPCBatch_BatchStreamStream) error { | ||
| return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented) | ||
| } | ||
|
|
||
| type DRPCBatchDescription struct{} | ||
|
|
||
| func (DRPCBatchDescription) NumMethods() int { return 2 } | ||
|
|
||
| func (DRPCBatchDescription) Method( | ||
| n int, | ||
| ) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) { | ||
| switch n { | ||
| case 0: | ||
| return "/cockroach.kv.kvpb.Batch/Batch", drpcEncoding_File_api_proto{}, | ||
| func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) { | ||
| return srv.(DRPCBatchServer). | ||
| Batch( | ||
| ctx, | ||
| in1.(*BatchRequest), | ||
| ) | ||
| }, DRPCBatchServer.Batch, true | ||
| case 1: | ||
| return "/cockroach.kv.kvpb.Batch/BatchStream", drpcEncoding_File_api_proto{}, | ||
| func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) { | ||
| return nil, srv.(DRPCBatchServer). | ||
| BatchStream( | ||
| &drpcBatch_BatchStreamStream{in1.(drpc.Stream)}, | ||
| ) | ||
| }, DRPCBatchServer.BatchStream, true | ||
| default: | ||
| return "", nil, nil, nil, false | ||
| } | ||
| } | ||
|
|
||
| func DRPCRegisterBatch(mux drpc.Mux, impl DRPCBatchServer) error { | ||
| return mux.Register(impl, DRPCBatchDescription{}) | ||
| } | ||
|
|
||
| type DRPCBatch_BatchStream interface { | ||
| drpc.Stream | ||
| SendAndClose(*BatchResponse) error | ||
| } | ||
|
|
||
| type DRPCBatch_BatchStreamStream interface { | ||
| drpc.Stream | ||
| Send(*BatchResponse) error | ||
| Recv() (*BatchRequest, error) | ||
| } | ||
|
|
||
| type drpcBatch_BatchStreamStream struct { | ||
| drpc.Stream | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamStream) Send(m *BatchResponse) error { | ||
| return x.MsgSend(m, drpcEncoding_File_api_proto{}) | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamStream) Recv() (*BatchRequest, error) { | ||
| m := new(BatchRequest) | ||
| if err := x.MsgRecv(m, drpcEncoding_File_api_proto{}); err != nil { | ||
| return nil, err | ||
| } | ||
| return m, nil | ||
| } | ||
|
|
||
| func (x *drpcBatch_BatchStreamStream) RecvMsg(m *BatchRequest) error { | ||
| return x.MsgRecv(m, drpcEncoding_File_api_proto{}) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.