From 07dcc69dffe60b12f1060bd9bced4a2bae802cde Mon Sep 17 00:00:00 2001 From: slasher Date: Thu, 15 Aug 2024 17:47:44 +0800 Subject: [PATCH] feat(rpc2): add testing of example . #22505814 Signed-off-by: slasher --- blobstore/common/rpc2/body.go | 28 +- blobstore/common/rpc2/body_test.go | 72 +++++ blobstore/common/rpc2/client_test.go | 63 ++++ blobstore/common/rpc2/connector.go | 3 +- blobstore/common/rpc2/connector_test.go | 41 +++ blobstore/common/rpc2/example_test.go | 243 ++++++++++++++++ blobstore/common/rpc2/header.go | 11 +- blobstore/common/rpc2/header_test.go | 125 ++++++++ blobstore/common/rpc2/request.go | 2 +- blobstore/common/rpc2/request_test.go | 102 +++++++ blobstore/common/rpc2/response_test.go | 75 +++++ blobstore/common/rpc2/router_test.go | 85 ++++++ blobstore/common/rpc2/rpc2.go | 3 +- blobstore/common/rpc2/rpc2.pb.go | 234 +++++---------- blobstore/common/rpc2/rpc2.proto | 3 + blobstore/common/rpc2/rpc2_test.go | 364 ++++++++++++++++++++++++ blobstore/common/rpc2/server.go | 26 +- blobstore/common/rpc2/server_test.go | 67 +++++ blobstore/common/rpc2/stream.go | 31 +- blobstore/common/rpc2/stream_test.go | 111 ++++++++ 20 files changed, 1498 insertions(+), 191 deletions(-) create mode 100644 blobstore/common/rpc2/body_test.go create mode 100644 blobstore/common/rpc2/client_test.go create mode 100644 blobstore/common/rpc2/connector_test.go create mode 100644 blobstore/common/rpc2/example_test.go create mode 100644 blobstore/common/rpc2/header_test.go create mode 100644 blobstore/common/rpc2/request_test.go create mode 100644 blobstore/common/rpc2/response_test.go create mode 100644 blobstore/common/rpc2/router_test.go create mode 100644 blobstore/common/rpc2/rpc2_test.go create mode 100644 blobstore/common/rpc2/server_test.go create mode 100644 blobstore/common/rpc2/stream_test.go diff --git a/blobstore/common/rpc2/body.go b/blobstore/common/rpc2/body.go index 183c9754f..9913605de 100644 --- a/blobstore/common/rpc2/body.go +++ b/blobstore/common/rpc2/body.go @@ -28,17 +28,18 @@ type bodyAndTrailer struct { req *Request trailerOnce sync.Once + trailer *FixedHeader closeOnce sync.Once } func (r *bodyAndTrailer) tryReadTrailer() error { if r.remain < 0 { - panic("rpc2: response body read too much") + panic("rpc2: body read too much, remain < 0") } var err error if r.remain == 0 { // try to read trailer r.trailerOnce.Do(func() { - _, err = r.req.Trailer.ReadFrom(r.sr) + _, err = r.trailer.ReadFrom(r.sr) }) } return err @@ -65,15 +66,19 @@ func (r *bodyAndTrailer) WriteTo(w io.Writer) (int64, error) { return 0, io.EOF } - if lw, ok := w.(*LimitedWriter); !ok { + lw, ok := w.(*LimitedWriter) + if !ok { return 0, ErrLimitedWriter - } else if lw.a > int64(r.remain) { + } + if lw.a > int64(r.remain) { return 0, io.ErrShortWrite } - n, err := r.br.WriteTo(w) + + _, err := r.br.WriteTo(lw) if err == errLimitedWrite { err = nil } + n := lw.a - lw.n // actual read bytes r.remain -= int(n) if err == nil { err = r.tryReadTrailer() @@ -100,7 +105,9 @@ func (r *bodyAndTrailer) Close() (err error) { } // makeBodyWithTrailer body and trailer remain. -func makeBodyWithTrailer(sr *transport.SizedReader, req *Request, l int64, decode bool) Body { +func makeBodyWithTrailer(sr *transport.SizedReader, req *Request, + trailer *FixedHeader, l int64, decode bool, +) Body { var br Body if decode { br = newEdBody(*req.checksum, sr, int(l), false) @@ -108,10 +115,11 @@ func makeBodyWithTrailer(sr *transport.SizedReader, req *Request, l int64, decod br = sr } r := &bodyAndTrailer{ - sr: sr, - br: br, - remain: int(l), - req: req, + sr: sr, + br: br, + remain: int(l), + req: req, + trailer: trailer, } r.tryReadTrailer() return r diff --git a/blobstore/common/rpc2/body_test.go b/blobstore/common/rpc2/body_test.go new file mode 100644 index 000000000..b041420f3 --- /dev/null +++ b/blobstore/common/rpc2/body_test.go @@ -0,0 +1,72 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "io" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRpc2Body(t *testing.T) { + var b bodyAndTrailer + b.remain = -1 + require.Panics(t, func() { b.tryReadTrailer() }) + + b.remain = 0 + _, err := b.WriteTo(io.Discard) + require.ErrorIs(t, io.EOF, err) + + b.remain = 1 + _, err = b.WriteTo(io.Discard) + require.ErrorIs(t, ErrLimitedWriter, err) + _, err = b.WriteTo(LimitWriter(io.Discard, 2)) + require.ErrorIs(t, io.ErrShortWrite, err) +} + +func TestRpc2ReadFrame(t *testing.T) { + addr, cli, shutdown := newTcpServer() + defer shutdown() + + cli.connector = defaultConnector(cli.ConnectorConfig) + { + conn, err := cli.connector.Get(testCtx, addr) + require.NoError(t, err) + frame, _ := conn.AllocFrame(1) + frame.Write([]byte{0xee}) + conn.WriteFrame(frame) + _, err = conn.ReadFrame() + require.ErrorIs(t, io.EOF, err) + } + { + conn, err := cli.connector.Get(testCtx, addr) + require.NoError(t, err) + frame, _ := conn.AllocFrame(5) + frame.Write([]byte{0x1, 0x00, 0x00, 0x00}) + conn.WriteFrame(frame) + _, err = conn.ReadFrame() + require.ErrorIs(t, io.EOF, err) + } + { + conn, err := cli.connector.Get(testCtx, addr) + require.NoError(t, err) + frame, _ := conn.AllocFrame(5) + frame.Write([]byte{0x1, 0x00, 0x00, 0x00, 0xee}) + conn.WriteFrame(frame) + _, err = conn.ReadFrame() + require.ErrorIs(t, io.EOF, err) + } +} diff --git a/blobstore/common/rpc2/client_test.go b/blobstore/common/rpc2/client_test.go new file mode 100644 index 000000000..affd33a6b --- /dev/null +++ b/blobstore/common/rpc2/client_test.go @@ -0,0 +1,63 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "bytes" + "fmt" + "io" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +type errNoParameter struct{ Codec } + +func (errNoParameter) Marshal() ([]byte, error) { return nil, fmt.Errorf("codec") } +func (errNoParameter) Unmarshal([]byte) error { return fmt.Errorf("codec") } + +func TestClientRetry(t *testing.T) { + var emptyCli Client + emptyCli.Close() + + addr, cli, shutdown := newTcpServer() + defer shutdown() + + for _, r := range []io.Reader{ + bytes.NewBuffer([]byte("error")), + bytes.NewReader([]byte("error")), + strings.NewReader("error"), + } { + req, err := NewRequest(testCtx, addr, "/error", nil, r) + require.NoError(t, err) + require.Error(t, cli.DoWith(req, nil)) + + req, err = NewRequest(testCtx, addr, "/error", nil, r) + require.NoError(t, err) + req.GetBody = func() (io.ReadCloser, error) { return nil, fmt.Errorf("error") } + require.Error(t, cli.DoWith(req, nil)) + } +} + +func TestClientCodec(t *testing.T) { + addr, cli, shutdown := newTcpServer() + defer shutdown() + _, err := NewRequest(testCtx, addr, "/", errNoParameter{NoParameter}, nil) + require.Error(t, err) + req, err := NewRequest(testCtx, addr, "/", nil, bytes.NewReader(make([]byte, 4))) + require.NoError(t, err) + require.Error(t, cli.DoWith(req, errNoParameter{NoParameter})) +} diff --git a/blobstore/common/rpc2/connector.go b/blobstore/common/rpc2/connector.go index ea3084840..9b1fe39cf 100644 --- a/blobstore/common/rpc2/connector.go +++ b/blobstore/common/rpc2/connector.go @@ -16,6 +16,7 @@ package rpc2 import ( "context" + "errors" "io" "net" "sync" @@ -56,7 +57,7 @@ func (t tcpDialer) Dial(ctx context.Context, addr string) (Conn, error) { type rdmaDialer struct{} func (rdmaDialer) Dial(ctx context.Context, addr string) (Conn, error) { - panic("rpc2: rdma not implements") + return nil, errors.New("rpc2: rdma not implements") } type Connector interface { diff --git a/blobstore/common/rpc2/connector_test.go b/blobstore/common/rpc2/connector_test.go new file mode 100644 index 000000000..cc4a6595a --- /dev/null +++ b/blobstore/common/rpc2/connector_test.go @@ -0,0 +1,41 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestConn(t *testing.T) { + { + var conn tcpConn + require.Nil(t, conn.Allocator()) + require.Panics(t, func() { conn.ReadOnce() }) + } + { + var conf ConnectorConfig + conf.Network = "xxx" + require.Panics(t, func() { defaultConnector(conf) }) + } + { + var conf ConnectorConfig + conf.Network = "rdma" + c := defaultConnector(conf) + _, err := c.Get(testCtx, "") + require.Error(t, err) + } +} diff --git a/blobstore/common/rpc2/example_test.go b/blobstore/common/rpc2/example_test.go new file mode 100644 index 000000000..66e5924a2 --- /dev/null +++ b/blobstore/common/rpc2/example_test.go @@ -0,0 +1,243 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "bytes" + "crypto/md5" + crand "crypto/rand" + "encoding/hex" + "fmt" + "io" + mrand "math/rand" +) + +func ExampleServer_base() { + addr, cli, shutdown := newTcpServer() + defer shutdown() + + // Request & Response + req, _ := NewRequest(testCtx, addr, "/", nil, nil) + if err := cli.DoWith(req, nil); err != nil { + fmt.Println(err) + } + + // Stream + sreq, _ := NewStreamRequest(testCtx, addr, "/stream", nil) + sc := StreamClient[noneCodec, noneCodec]{cli} + scli, _ := sc.Streaming(sreq, nil) + waitc := make(chan struct{}) + go func() { + for { + _, err := scli.Recv() + if err == io.EOF { + close(waitc) + return + } + if err != nil { + fmt.Println(err) + } + } + }() + for range [10]struct{}{} { + if err := scli.Send(&noneCodec{}); err != nil { + fmt.Println(err) + } + } + scli.CloseSend() + <-waitc + + // Output: +} + +type strMessage struct{ str string } + +func (s *strMessage) Size() int { return len(s.str) } +func (s *strMessage) Marshal() ([]byte, error) { return []byte(s.str), nil } +func (s *strMessage) MarshalTo(b []byte) (int, error) { return copy(b, []byte(s.str)), nil } +func (s *strMessage) Unmarshal(b []byte) error { s.str = string(b); return nil } + +func handleMessage(w ResponseWriter, req *Request) error { + var args strMessage + req.ParseParameter(&args) + req.Body.Close() + args.str = "-> " + args.str + return w.WriteOK(&args) +} + +func ExampleServer_request_message() { + handler := &Router{} + handler.Register("/", handleMessage) + server, cli, shutdown := newServer("tcp", handler) + defer shutdown() + + args := &strMessage{str: "request message"} + // message in request & response body + req, _ := NewRequest(testCtx, server.Name, "/", nil, Codec2Reader(args)) + if err := cli.DoWith(req, args); err != nil { + fmt.Println(err) + } + fmt.Println(args.str) + + // Output: + // -> request message +} + +func handleUpload(w ResponseWriter, req *Request) error { + var args strMessage + req.ParseParameter(&args) + hasher := req.checksum.Hasher() + if _, err := req.Body.WriteTo(LimitWriter(hasher, req.ContentLength)); err != nil { + return err + } + args.str = fmt.Sprint(req.checksum.Readable(hasher.Sum(nil))) + req.Body.Close() + return w.WriteOK(&args) +} + +func ExampleServer_request_upload() { + handler := &Router{} + handler.Register("/", handleUpload) + server, cli, shutdown := newServer("tcp", handler) + defer shutdown() + + args := &strMessage{str: "request data"} + buff := make([]byte, mrand.Intn(4<<20)+1<<20) + crand.Read(buff) + // request message in parameter & response message in body + req, _ := NewRequest(testCtx, server.Name, "/", args, bytes.NewReader(buff)) + req.OptionCrcUpload() + req.ContentLength = int64(len(buff)) + if err := cli.DoWith(req, args); err != nil { + fmt.Println(err) + } + hasher := req.checksum.Hasher() + hasher.Write(buff) + fmt.Println(args.str == fmt.Sprint(req.checksum.Readable(hasher.Sum(nil)))) + + // Output: + // true +} + +func handleUpDown(w ResponseWriter, req *Request) error { + var args strMessage + req.ParseParameter(&args) + uhasher := req.checksum.Hasher() + + if _, err := req.Body.WriteTo(LimitWriter(uhasher, req.ContentLength)); err != nil { + return err + } + req.Body.Close() + + buff := make([]byte, mrand.Intn(4<<20)+1<<20) + crand.Read(buff) + dhasher := req.checksum.Hasher() + dhasher.Write(buff) + + rr := req.checksum.Readable + args.str = fmt.Sprintf("%v %v", rr(uhasher.Sum(nil)), rr(dhasher.Sum(nil))) + + w.SetContentLength(int64(len(buff))) + w.WriteHeader(200, &args) + _, err := w.ReadFrom(bytes.NewReader(buff)) + return err +} + +func ExampleServer_request_updown() { + handler := &Router{} + handler.Register("/", handleUpDown) + server, cli, shutdown := newServer("tcp", handler) + defer shutdown() + + args := &strMessage{str: "upload & download"} + buff := make([]byte, mrand.Intn(4<<20)+1<<20) + crand.Read(buff) + // request & response message in parameter + req, _ := NewRequest(testCtx, server.Name, "/", args, bytes.NewReader(buff)) + req.Option(func(r *Request) { _ = req.RequestHeader.ToString() }) + req.OptionCrc() + req.ContentLength = int64(len(buff)) + + uhasher := req.checksum.Hasher() + uhasher.Write(buff) + dhasher := req.checksum.Hasher() + + resp, _ := cli.Do(req, args) + defer resp.Body.Close() + + b := make([]byte, 32<<20) + for { + n, err := resp.Body.Read(b) + dhasher.Write(b[:n]) + if err == io.EOF { + break + } + } + rr := req.checksum.Readable + fmt.Println(args.str == fmt.Sprintf("%v %v", rr(uhasher.Sum(nil)), rr(dhasher.Sum(nil)))) + + // Output: + // true +} + +func handleTrailer(w ResponseWriter, req *Request) error { + hasher := md5.New() + req.Body.WriteTo(LimitWriter(hasher, req.ContentLength)) + req.Body.Close() + + w.Header().Merge(req.Header) + w.Header().Add("add", "header-stable") + w.Trailer().Add("add", "trailer-stable") + w.Trailer().SetLen("md5", 32) + + w.AfterBody(func() error { return nil }) + w.AfterBody(func() error { + w.Trailer().Set("md5", hex.EncodeToString(hasher.Sum(nil))) + return nil + }) + return w.WriteOK(nil) +} + +func ExampleServer_trailer() { + handler := &Router{} + handler.Register("/", handleTrailer) + server, cli, shutdown := newServer("tcp", handler) + defer shutdown() + + hasher := md5.New() + buff := make([]byte, mrand.Intn(4<<20)+1<<20) + crand.Read(buff) + + req, _ := NewRequest(testCtx, server.Name, "/", nil, + io.TeeReader(bytes.NewReader(buff), hasher)) + req.ContentLength = int64(len(buff)) + req.Header.Add("add", "x") + req.Trailer.SetLen("md5", 32) + req.AfterBody = func() error { + req.Trailer.Set("md5", hex.EncodeToString(hasher.Sum(nil))) + return nil + } + + resp, _ := cli.Do(req, nil) + resp.Body.Close() + fmt.Println(resp.Header.Get("add")) + fmt.Println(resp.Trailer.Get("add")) + fmt.Println(hex.EncodeToString(hasher.Sum(nil)) == resp.Trailer.Get("md5")) + + // Output: + // header-stable + // trailer-stable + // true +} diff --git a/blobstore/common/rpc2/header.go b/blobstore/common/rpc2/header.go index 938a6e0df..16d5ca5d4 100644 --- a/blobstore/common/rpc2/header.go +++ b/blobstore/common/rpc2/header.go @@ -86,15 +86,6 @@ func (h *Header) Merge(other Header) { } } -// ToFixedHeader copy to unstabled fixed header -func (h *Header) ToFixedHeader() FixedHeader { - fh := FixedHeader{} - for key, val := range h.M { - fh.Set(key, val) - } - return fh -} - func (fh *FixedHeader) newIfNil() { if fh.M == nil { fh.M = make(map[string]FixedValue) @@ -163,7 +154,7 @@ func (fh *FixedHeader) MergeHeader(h Header) { func (fh *FixedHeader) AllSize() (n int) { for _, v := range fh.M { - n += int(v.GetLen()) + n += int(v.Len) } return } diff --git a/blobstore/common/rpc2/header_test.go b/blobstore/common/rpc2/header_test.go new file mode 100644 index 000000000..b8ae7e03b --- /dev/null +++ b/blobstore/common/rpc2/header_test.go @@ -0,0 +1,125 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "io" + "strconv" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRpc2Header(t *testing.T) { + var header Header + require.False(t, header.Has("a")) + + header.Set("a", "a") + header.Set("b", "b") + header.Set("c", "c") + require.True(t, header.Has("a")) + + maxBuff := make([]byte, MaxHeaderLength+1) + maxBuff[0] = 'a' + header.Set(string(maxBuff), "x") + require.False(t, header.Has(string(maxBuff))) + header.Set("x", string(maxBuff)) + require.False(t, header.Has("x")) + for idx := range [MaxHeaders]struct{}{} { + header.Set("id-"+strconv.Itoa(idx), "") + } + require.Equal(t, MaxHeaders, len(header.M)) + header.Del("a") + require.False(t, header.Has("a")) + header.SetStable() + header.Set("a", "a") + require.False(t, header.Has("a")) + header.Del("b") + require.True(t, header.Has("b")) + + header.Merge(header) + + header = header.Clone() + header.Set("a", "a") + require.True(t, header.Has("a")) + + header.Reset() + require.Nil(t, header.M) +} + +func TestRpc2FixedHeader(t *testing.T) { + var header FixedHeader + require.False(t, header.Has("a")) + + header.Set("a", "a") + header.Set("b", "b") + header.Set("c", "c") + require.True(t, header.Has("a")) + + maxBuff := make([]byte, MaxHeaderLength+1) + maxBuff[0] = 'a' + header.Set(string(maxBuff), "x") + require.False(t, header.Has(string(maxBuff))) + header.Set("x", string(maxBuff)) + require.False(t, header.Has("x")) + for idx := range [MaxHeaders]struct{}{} { + header.Set("id-"+strconv.Itoa(idx), "") + } + require.Equal(t, MaxHeaders, len(header.M)) + header.Del("a") + require.False(t, header.Has("a")) + + o := header.ToHeader() + header.MergeHeader(o) + + header.SetStable() + header.MergeHeader(o) + header.Set("a", "a") + require.False(t, header.Has("a")) + header.Del("b") + require.True(t, header.Has("b")) + + header.Reset() + require.Nil(t, header.M) +} + +func TestRpc2FixedHeaderReader(t *testing.T) { + var header FixedHeader + require.Equal(t, NoBody, header.Reader()) + + header.Add("a", "aaaa") + b, _ := io.ReadAll(header.Reader()) + require.Equal(t, header.Get("a"), string(b)) + + header.Set("c", "cccc") + header.SetLen("b", 8) + header.Set("b", "123456789") + + b, _ = io.ReadAll(header.Reader()) + require.Equal(t, "aaaa12345678cccc", string(b)) + + header.SetStable() + header.Set("b", "87654321") + header.Set("d", "dddd") + b, _ = io.ReadAll(header.Reader()) + require.Equal(t, "aaaa87654321cccc", string(b)) + + r := trailerReader{ + Fn: func() error { return errLimitedWrite }, + Trailer: &header, + } + _, err := r.Read(nil) + require.ErrorIs(t, errLimitedWrite, err) +} diff --git a/blobstore/common/rpc2/request.go b/blobstore/common/rpc2/request.go index 95af243b1..700e629d4 100644 --- a/blobstore/common/rpc2/request.go +++ b/blobstore/common/rpc2/request.go @@ -126,7 +126,7 @@ func (req *Request) request(deadline time.Time) (*Response, error) { payloadSize += int(resp.ContentLength) } resp.Body = makeBodyWithTrailer(req.conn.NewSizedReader(payloadSize, frame), - req, resp.ContentLength, decode) + req, &resp.Trailer, resp.ContentLength, decode) return resp, nil } diff --git a/blobstore/common/rpc2/request_test.go b/blobstore/common/rpc2/request_test.go new file mode 100644 index 000000000..12bb2b905 --- /dev/null +++ b/blobstore/common/rpc2/request_test.go @@ -0,0 +1,102 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "bytes" + "context" + "io" + "testing" + "time" + + "github.com/cubefs/cubefs/blobstore/common/rpc2/transport" + "github.com/stretchr/testify/require" +) + +func handleRequestTimeout(w ResponseWriter, req *Request) error { + time.Sleep(10 * time.Second) + return w.WriteOK(nil) +} + +func TestRequestTimeout(t *testing.T) { + var handler Router + handler.Register("/", handleRequestTimeout) + handler.Register("/none", handleNone) + server, cli, shutdown := newServer("tcp", &handler) + defer shutdown() + + cli.RequestTimeout = 200 * time.Millisecond + req, err := NewRequest(testCtx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.ErrorIs(t, transport.ErrTimeout, err) + cli.RequestTimeout = 0 + + ctx, cancel := context.WithDeadline(testCtx, time.Now().Add(100*time.Millisecond)) + req, err = NewRequest(ctx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.ErrorIs(t, transport.ErrTimeout, err) + cancel() + + cli.Timeout = 200 * time.Millisecond + req, err = NewRequest(testCtx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.ErrorIs(t, transport.ErrTimeout, err) + cli.Timeout = 0 + + buff := make([]byte, 8<<20) + cli.ResponseTimeout = 200 * time.Millisecond + req, err = NewRequest(testCtx, server.Name, "/none", nil, bytes.NewReader(buff)) + require.NoError(t, err) + resp, err := cli.Do(req, nil) + require.NoError(t, err) + time.Sleep(250 * time.Millisecond) + _, err = io.ReadFull(resp.Body, buff) + require.Error(t, err) + cli.ResponseTimeout = 0 + + cli.Timeout = time.Second + ctx, cancel = context.WithDeadline(testCtx, time.Now().Add(100*time.Millisecond)) + req, err = NewRequest(ctx, server.Name, "/none", nil, bytes.NewReader(buff)) + require.NoError(t, err) + resp, err = cli.Do(req, nil) + require.NoError(t, err) + time.Sleep(250 * time.Millisecond) + _, err = io.ReadFull(resp.Body, buff) + require.Error(t, err) + cancel() +} + +func TestRequestErrors(t *testing.T) { + addr, cli, shutdown := newTcpServer() + defer shutdown() + + cli.ConnectorConfig.Transport.MaxFrameSize = 32 + req, err := NewRequest(testCtx, addr, "/", nil, nil) + require.Panics(t, func() { + req.OptionChecksum(ChecksumBlock{ + Algorithm: ChecksumAlgorithm_Alg_None, + Direction: ChecksumDirection_Dir_None, + BlockSize: 1 << 10, + }) + }) + req.OptionCrcDownload() + req.OptionCrcDownload() + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.ErrorIs(t, ErrFrameHeader, err) +} diff --git a/blobstore/common/rpc2/response_test.go b/blobstore/common/rpc2/response_test.go new file mode 100644 index 000000000..5c09e73bf --- /dev/null +++ b/blobstore/common/rpc2/response_test.go @@ -0,0 +1,75 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func handleResponseDoubleOk(w ResponseWriter, req *Request) error { + w.WriteHeader(200, nil) + w.WriteHeader(200, nil) + w.WriteOK(NoParameter) + return w.WriteOK(nil) +} + +func handleResponseDoubleStatus(w ResponseWriter, req *Request) error { + size := int64(10) + w.SetContentLength(size) + w.Flush() + resp := w.(*response) + resp.Write(make([]byte, size-1)) + resp.Write(make([]byte, size)) + resp.Write(make([]byte, size)) + return nil +} + +// response has wrote 200 OK +func handleResponseAfterError(w ResponseWriter, req *Request) error { + w.AfterBody(func() error { return &Error{Status: 511, Detail: "after body"} }) + return w.WriteOK(nil) +} + +func handleResponseClosed(w ResponseWriter, req *Request) error { + resp := w.(*response) + resp.hdr.ToString() + resp.conn.Close() + return w.WriteOK(nil) +} + +func TestResponseError(t *testing.T) { + var handler Router + handler.Register("/ok", handleResponseDoubleOk) + handler.Register("/status", handleResponseDoubleStatus) + handler.Register("/after", handleResponseAfterError) + handler.Register("/closed", handleResponseClosed) + server, cli, shutdown := newServer("tcp", &handler) + defer shutdown() + + req, err := NewRequest(testCtx, server.Name, "/ok", nil, nil) + require.NoError(t, err) + require.NoError(t, cli.DoWith(req, nil)) + req, err = NewRequest(testCtx, server.Name, "/status", nil, nil) + require.NoError(t, err) + require.NoError(t, cli.DoWith(req, nil)) + req, err = NewRequest(testCtx, server.Name, "/after", nil, nil) + require.NoError(t, err) + require.NoError(t, cli.DoWith(req, nil)) + req, err = NewRequest(testCtx, server.Name, "/closed", nil, nil) + require.NoError(t, err) + require.Error(t, cli.DoWith(req, nil)) +} diff --git a/blobstore/common/rpc2/router_test.go b/blobstore/common/rpc2/router_test.go new file mode 100644 index 000000000..25f68ea8a --- /dev/null +++ b/blobstore/common/rpc2/router_test.go @@ -0,0 +1,85 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +func handleMiddlewareNil(w ResponseWriter, req *Request) error { return nil } +func handleMiddlewareError(w ResponseWriter, req *Request) error { return &Error{Status: 555} } +func handleRouteError(w ResponseWriter, req *Request) error { return &Error{Status: 666} } +func handleRoutePanic(w ResponseWriter, req *Request) error { panic("route") } + +func TestRpc2Router(t *testing.T) { + { + var handler Router + handler.Middleware(handleMiddlewareError) + handler.Register("/", handleRouteError) + server, cli, shutdown := newServer("tcp", &handler) + defer shutdown() + req, err := NewRequest(testCtx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.Equal(t, 555, DetectStatusCode(err)) + } + { + var handler Router + handler.Register("/", handleRouteError) + server, cli, shutdown := newServer("tcp", &handler) + defer shutdown() + req, err := NewRequest(testCtx, server.Name, "/404", nil, nil) + require.NoError(t, err, ErrorString(err)) + err = cli.DoWith(req, nil) + require.Equal(t, 404, DetectStatusCode(err), ErrorString(err)) + + req, err = NewRequest(testCtx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.Equal(t, 666, DetectStatusCode(err)) + } + { + var handler Router + handler.Register("/", handleRoutePanic) + server, cli, shutdown := newServer("tcp", &handler) + defer shutdown() + req, err := NewRequest(testCtx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.Equal(t, 597, DetectStatusCode(err)) + + handler.PanicHandler = func(ResponseWriter, *Request, interface{}, []byte) error { + return &Error{Status: 777} + } + req, err = NewRequest(testCtx, server.Name, "/", nil, nil) + require.NoError(t, err) + err = cli.DoWith(req, nil) + require.Equal(t, 777, DetectStatusCode(err), ErrorString(fmt.Errorf("panic 777"))) + } + { + var handler Router + handler.Register("/", handleRoutePanic) + require.Panics(t, func() { handler.Register("/", handleRoutePanic) }) + require.Panics(t, func() { + for range [1 << 10]struct{}{} { + handler.Middleware(handleMiddlewareNil) + } + handler.Middleware(handleMiddlewareNil) + }) + } +} diff --git a/blobstore/common/rpc2/rpc2.go b/blobstore/common/rpc2/rpc2.go index 3dc45f7ea..4bba5366a 100644 --- a/blobstore/common/rpc2/rpc2.go +++ b/blobstore/common/rpc2/rpc2.go @@ -139,11 +139,10 @@ type LimitedWriter struct { w io.Writer a int64 n int64 - e error } func LimitWriter(w io.Writer, limit int64) io.Writer { - return &LimitedWriter{w, limit, limit, nil} + return &LimitedWriter{w: w, a: limit, n: limit} } func (lw *LimitedWriter) Write(p []byte) (int, error) { diff --git a/blobstore/common/rpc2/rpc2.pb.go b/blobstore/common/rpc2/rpc2.pb.go index 4a17dd9bc..279f0fd92 100644 --- a/blobstore/common/rpc2/rpc2.pb.go +++ b/blobstore/common/rpc2/rpc2.pb.go @@ -114,11 +114,8 @@ func (ChecksumDirection) EnumDescriptor() ([]byte, []int) { } type Header struct { - M map[string]string `protobuf:"bytes,1,rep,name=m,proto3" json:"m,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` - stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + M map[string]string `protobuf:"bytes,1,rep,name=m,proto3" json:"m,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"` } func (m *Header) Reset() { *m = Header{} } @@ -169,11 +166,8 @@ func (m *Header) Getstable() bool { } type FixedValue struct { - Len uint32 `protobuf:"varint,1,opt,name=len,proto3" json:"len,omitempty"` - Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + Len uint32 `protobuf:"varint,1,opt,name=len,proto3" json:"len,omitempty"` + Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"` } func (m *FixedValue) Reset() { *m = FixedValue{} } @@ -224,11 +218,8 @@ func (m *FixedValue) GetValue() string { } type FixedHeader struct { - M map[string]FixedValue `protobuf:"bytes,1,rep,name=m,proto3" json:"m" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` - stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + M map[string]FixedValue `protobuf:"bytes,1,rep,name=m,proto3" json:"m" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"` } func (m *FixedHeader) Reset() { *m = FixedHeader{} } @@ -279,18 +270,15 @@ func (m *FixedHeader) Getstable() bool { } type RequestHeader struct { - Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"` - Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"` - StreamCmd StreamCmd `protobuf:"varint,3,opt,name=stream_cmd,json=streamCmd,proto3,enum=cubefs.blobstore.common.rpc2.StreamCmd" json:"stream_cmd,omitempty"` - RemotePath string `protobuf:"bytes,4,opt,name=remote_path,json=remotePath,proto3" json:"remote_path,omitempty"` - TraceID string `protobuf:"bytes,5,opt,name=trace_id,json=traceId,proto3" json:"trace_id,omitempty"` - ContentLength int64 `protobuf:"varint,6,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"` - Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"` - Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"` - Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"` + Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"` + StreamCmd StreamCmd `protobuf:"varint,3,opt,name=stream_cmd,json=streamCmd,proto3,enum=cubefs.blobstore.common.rpc2.StreamCmd" json:"stream_cmd,omitempty"` + RemotePath string `protobuf:"bytes,4,opt,name=remote_path,json=remotePath,proto3" json:"remote_path,omitempty"` + TraceID string `protobuf:"bytes,5,opt,name=trace_id,json=traceId,proto3" json:"trace_id,omitempty"` + ContentLength int64 `protobuf:"varint,6,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"` + Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"` + Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"` + Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"` } func (m *RequestHeader) Reset() { *m = RequestHeader{} } @@ -390,18 +378,15 @@ func (m *RequestHeader) GetParameter() []byte { } type ResponseHeader struct { - Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"` - Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"` - Status int32 `protobuf:"varint,4,opt,name=status,proto3" json:"status,omitempty"` - Reason string `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"` - Error string `protobuf:"bytes,6,opt,name=error,proto3" json:"error,omitempty"` - ContentLength int64 `protobuf:"varint,7,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"` - Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"` - Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"` - Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"` + Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"` + Status int32 `protobuf:"varint,4,opt,name=status,proto3" json:"status,omitempty"` + Reason string `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"` + Error string `protobuf:"bytes,6,opt,name=error,proto3" json:"error,omitempty"` + ContentLength int64 `protobuf:"varint,7,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"` + Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"` + Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"` + Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"` } func (m *ResponseHeader) Reset() { *m = ResponseHeader{} } @@ -501,12 +486,9 @@ func (m *ResponseHeader) GetParameter() []byte { } type Error struct { - Status int32 `protobuf:"varint,1,opt,name=status,proto3" json:"status,omitempty"` - Reason string `protobuf:"bytes,2,opt,name=reason,proto3" json:"reason,omitempty"` - Detail string `protobuf:"bytes,3,opt,name=detail,proto3" json:"detail,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + Status int32 `protobuf:"varint,1,opt,name=status,proto3" json:"status,omitempty"` + Reason string `protobuf:"bytes,2,opt,name=reason,proto3" json:"reason,omitempty"` + Detail string `protobuf:"bytes,3,opt,name=detail,proto3" json:"detail,omitempty"` } func (m *Error) Reset() { *m = Error{} } @@ -564,12 +546,9 @@ func (m *Error) GetDetail() string { } type ChecksumBlock struct { - Algorithm ChecksumAlgorithm `protobuf:"varint,1,opt,name=algorithm,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumAlgorithm" json:"algorithm,omitempty"` - Direction ChecksumDirection `protobuf:"varint,2,opt,name=direction,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumDirection" json:"direction,omitempty"` - BlockSize uint32 `protobuf:"varint,3,opt,name=blockSize,proto3" json:"blockSize,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + Algorithm ChecksumAlgorithm `protobuf:"varint,1,opt,name=algorithm,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumAlgorithm" json:"algorithm,omitempty"` + Direction ChecksumDirection `protobuf:"varint,2,opt,name=direction,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumDirection" json:"direction,omitempty"` + BlockSize uint32 `protobuf:"varint,3,opt,name=blockSize,proto3" json:"blockSize,omitempty"` } func (m *ChecksumBlock) Reset() { *m = ChecksumBlock{} } @@ -644,54 +623,55 @@ func init() { func init() { proto.RegisterFile("rpc2.proto", fileDescriptor_af0916bb5e6806d0) } var fileDescriptor_af0916bb5e6806d0 = []byte{ - // 745 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x55, 0x5f, 0x6f, 0xda, 0x56, - 0x1c, 0xcd, 0xc5, 0xc1, 0xe0, 0x1f, 0x81, 0x79, 0x56, 0x16, 0x59, 0x51, 0x14, 0x10, 0xda, 0x16, - 0x96, 0x69, 0x66, 0x62, 0x79, 0xd8, 0x1f, 0x6d, 0x52, 0x08, 0xb0, 0x20, 0x2d, 0x24, 0xba, 0x64, - 0x93, 0xd6, 0x87, 0x22, 0x63, 0xdf, 0x82, 0x15, 0xdb, 0x97, 0x5e, 0x5f, 0xd2, 0xa4, 0x1f, 0xa7, - 0xea, 0x87, 0x89, 0xd4, 0x97, 0x3e, 0x57, 0x15, 0xaa, 0x78, 0xec, 0xa7, 0xa8, 0xee, 0xb5, 0x1d, - 0x52, 0x35, 0x45, 0xb4, 0x6f, 0x7d, 0xfb, 0x9d, 0xe3, 0x7b, 0x8e, 0x7f, 0xe7, 0xf8, 0x22, 0x00, - 0xd8, 0xc4, 0x69, 0x58, 0x13, 0x46, 0x39, 0x35, 0x76, 0x9c, 0xe9, 0x90, 0x3c, 0x8a, 0xac, 0xa1, - 0x4f, 0x87, 0x11, 0xa7, 0x8c, 0x58, 0x0e, 0x0d, 0x02, 0x1a, 0x5a, 0xe2, 0xcc, 0xf6, 0xe6, 0x88, - 0x8e, 0xa8, 0x3c, 0x58, 0x17, 0x53, 0xac, 0xa9, 0x3e, 0x43, 0xa0, 0x1e, 0x13, 0xdb, 0x25, 0xcc, - 0xf8, 0x0d, 0x50, 0x60, 0xa2, 0x8a, 0x52, 0x2b, 0x34, 0x7e, 0xb4, 0x96, 0x59, 0x59, 0xb1, 0xc0, - 0x3a, 0x69, 0x87, 0x9c, 0x5d, 0x63, 0x14, 0x18, 0x7b, 0xa0, 0x46, 0xdc, 0x1e, 0xfa, 0xc4, 0xcc, - 0x54, 0x50, 0x2d, 0xdf, 0xfc, 0x6a, 0x3e, 0x2b, 0x27, 0xcc, 0xdb, 0x59, 0x19, 0xfd, 0x84, 0x13, - 0xb0, 0x7d, 0x00, 0x6a, 0xac, 0x32, 0x74, 0x50, 0x2e, 0xc8, 0xb5, 0x89, 0x2a, 0xa8, 0xa6, 0x61, - 0x31, 0x1a, 0x9b, 0x90, 0xbd, 0xb4, 0xfd, 0x69, 0xec, 0xa1, 0xe1, 0x18, 0xfc, 0x9e, 0xf9, 0x15, - 0x55, 0x0f, 0x00, 0x3a, 0xde, 0x15, 0x71, 0xff, 0x13, 0x8c, 0x50, 0xfa, 0x24, 0x94, 0xca, 0x22, - 0x16, 0xe3, 0xfd, 0xca, 0xea, 0x6b, 0x04, 0x05, 0x29, 0x4b, 0xf2, 0xb5, 0x16, 0xf9, 0x7e, 0x5e, - 0x9e, 0xef, 0x8e, 0x2a, 0x09, 0xd9, 0x5c, 0xbf, 0x99, 0x95, 0xd7, 0x3e, 0x29, 0xea, 0xc3, 0x25, - 0x51, 0xff, 0xba, 0xbb, 0x70, 0xa1, 0x51, 0x5b, 0x61, 0x1d, 0x99, 0xfd, 0x6e, 0x29, 0xcf, 0x15, - 0x28, 0x62, 0xf2, 0x78, 0x4a, 0x22, 0x9e, 0x04, 0x34, 0x21, 0x77, 0x49, 0x58, 0xe4, 0xd1, 0xb8, - 0x9c, 0x2c, 0x4e, 0xa1, 0x28, 0x28, 0xb0, 0x47, 0x9e, 0x23, 0xdf, 0x97, 0xc5, 0x31, 0x30, 0x3a, - 0x00, 0x11, 0x67, 0xc4, 0x0e, 0x06, 0x4e, 0xe0, 0x9a, 0x4a, 0x05, 0xd5, 0x4a, 0x8d, 0xbd, 0xe5, - 0xab, 0xf4, 0xe5, 0xf9, 0xa3, 0xc0, 0xc5, 0x5a, 0x94, 0x8e, 0x46, 0x19, 0x0a, 0x8c, 0x04, 0x94, - 0x93, 0xc1, 0xc4, 0xe6, 0x63, 0x73, 0x5d, 0xe6, 0x84, 0x98, 0x3a, 0xb3, 0xf9, 0xd8, 0xf8, 0x1e, - 0xf2, 0x9c, 0xd9, 0x0e, 0x19, 0x78, 0xae, 0x99, 0x15, 0x4f, 0x9b, 0x85, 0xf9, 0xac, 0x9c, 0x3b, - 0x17, 0x5c, 0xb7, 0x85, 0x73, 0xf2, 0x61, 0xd7, 0x35, 0xbe, 0x83, 0x92, 0x43, 0x43, 0x4e, 0x42, - 0x3e, 0xf0, 0x49, 0x38, 0xe2, 0x63, 0x53, 0xad, 0xa0, 0x9a, 0x82, 0x8b, 0x09, 0xfb, 0x8f, 0x24, - 0x8d, 0x26, 0xa8, 0x63, 0x99, 0xd8, 0xcc, 0xcb, 0xfa, 0xbe, 0x5d, 0xe5, 0xb6, 0x26, 0x5f, 0x30, - 0x51, 0x1a, 0x5d, 0x10, 0x6f, 0xf5, 0x7c, 0xc2, 0x4c, 0x4d, 0x9a, 0xfc, 0xb0, 0xf2, 0x95, 0x48, - 0x9c, 0x52, 0xbd, 0xb1, 0x03, 0xda, 0xc4, 0x66, 0x76, 0x40, 0x38, 0x61, 0x26, 0x54, 0x50, 0x6d, - 0x03, 0x2f, 0x88, 0xea, 0xab, 0x0c, 0x94, 0x30, 0x89, 0x26, 0x34, 0x8c, 0xc8, 0x67, 0x7e, 0xa7, - 0x2d, 0x79, 0xe5, 0xf8, 0x34, 0x92, 0xd5, 0x66, 0x71, 0x82, 0x04, 0xcf, 0x88, 0x1d, 0xd1, 0x30, - 0x2e, 0x15, 0x27, 0x48, 0xb8, 0x10, 0xc6, 0x28, 0x93, 0xed, 0x69, 0x38, 0x06, 0xf7, 0x94, 0x9b, - 0xfb, 0xe2, 0xcb, 0x3d, 0x85, 0x6c, 0x5b, 0x86, 0x5b, 0x54, 0x84, 0x3e, 0x52, 0x51, 0xe6, 0xbd, - 0x8a, 0xb6, 0x40, 0x75, 0x09, 0xb7, 0x3d, 0x5f, 0x5e, 0x7b, 0x0d, 0x27, 0xa8, 0xfa, 0x02, 0x41, - 0xf1, 0x68, 0x4c, 0x9c, 0x8b, 0x68, 0x1a, 0x34, 0x7d, 0xea, 0x5c, 0x18, 0x27, 0xa0, 0xd9, 0xfe, - 0x88, 0x32, 0x8f, 0x8f, 0x03, 0x69, 0x5e, 0x6a, 0xd4, 0x97, 0xa7, 0x49, 0xf5, 0x87, 0xa9, 0x0c, - 0x2f, 0x1c, 0x84, 0x9d, 0xeb, 0x31, 0xe2, 0x70, 0x2f, 0xd9, 0x69, 0x65, 0xbb, 0x56, 0x2a, 0xc3, - 0x0b, 0x07, 0x51, 0xcf, 0x50, 0xac, 0xd9, 0xf7, 0x9e, 0x12, 0x19, 0xa5, 0x88, 0x17, 0xc4, 0x7e, - 0x1d, 0xb4, 0xdb, 0x1f, 0xac, 0x91, 0x03, 0xa5, 0x77, 0x7a, 0xae, 0xaf, 0x89, 0xa1, 0xff, 0x7f, - 0x4f, 0x47, 0x62, 0x38, 0xeb, 0x1f, 0xeb, 0x19, 0x31, 0x74, 0xba, 0x3d, 0x5d, 0xd9, 0xff, 0x13, - 0xbe, 0xfe, 0x60, 0x7b, 0x63, 0x03, 0xf2, 0x87, 0xfe, 0x68, 0xd0, 0xa3, 0x21, 0xd1, 0xd7, 0x04, - 0x3a, 0x62, 0xce, 0xa0, 0xdb, 0x6e, 0xb7, 0x75, 0x94, 0xa2, 0xce, 0x61, 0xff, 0x5c, 0xcf, 0xec, - 0xff, 0xbd, 0x90, 0xdf, 0x6e, 0x2b, 0x8e, 0xb4, 0x3c, 0x96, 0xca, 0x01, 0xd4, 0xd6, 0x74, 0xe2, - 0x93, 0x2b, 0x1d, 0x89, 0xf9, 0xdf, 0x89, 0x4f, 0x6d, 0x57, 0xcf, 0xc8, 0x53, 0xf4, 0x49, 0x28, - 0x91, 0xd2, 0xfc, 0xe6, 0x66, 0xbe, 0x8b, 0x5e, 0xce, 0x77, 0xd1, 0x9b, 0xf9, 0x2e, 0x7a, 0x90, - 0xb3, 0xea, 0x7f, 0x88, 0x2a, 0x86, 0xaa, 0xfc, 0xcf, 0xfa, 0xe5, 0x5d, 0x00, 0x00, 0x00, 0xff, - 0xff, 0xce, 0x76, 0x64, 0x00, 0xf5, 0x06, 0x00, 0x00, + // 758 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x55, 0x5f, 0x6f, 0xf2, 0x54, + 0x1c, 0xe6, 0xd0, 0x51, 0xe8, 0x8f, 0x81, 0xb5, 0x59, 0x96, 0x66, 0x59, 0x00, 0x89, 0x3a, 0x9c, + 0xb1, 0x18, 0xdc, 0x85, 0x7f, 0xa2, 0xc9, 0x18, 0xe0, 0x48, 0x1c, 0x5b, 0x0e, 0xd3, 0x44, 0x2f, + 0x24, 0xa5, 0x3d, 0x42, 0xb3, 0xb6, 0x07, 0x4f, 0x0f, 0x73, 0xf3, 0x53, 0xf8, 0x19, 0x8c, 0x1f, + 0x66, 0x89, 0x37, 0xbb, 0x34, 0xc6, 0x10, 0xc3, 0xee, 0xfc, 0x14, 0xe6, 0x9c, 0xb6, 0x63, 0xc6, + 0xbd, 0x84, 0xf7, 0xbd, 0x7b, 0xef, 0x7e, 0xcf, 0xd3, 0xf3, 0x3c, 0xfd, 0x3d, 0x4f, 0x0f, 0x01, + 0x80, 0xcd, 0x9c, 0x96, 0x35, 0x63, 0x94, 0x53, 0x63, 0xdf, 0x99, 0x8f, 0xc9, 0x0f, 0x91, 0x35, + 0xf6, 0xe9, 0x38, 0xe2, 0x94, 0x11, 0xcb, 0xa1, 0x41, 0x40, 0x43, 0x4b, 0x9c, 0xd9, 0xdb, 0x99, + 0xd0, 0x09, 0x95, 0x07, 0x9b, 0x62, 0x8a, 0x35, 0xf5, 0x5f, 0x11, 0xa8, 0xa7, 0xc4, 0x76, 0x09, + 0x33, 0x3e, 0x01, 0x14, 0x98, 0xa8, 0xa6, 0x34, 0x8a, 0xad, 0xf7, 0xad, 0x75, 0x56, 0x56, 0x2c, + 0xb0, 0xce, 0xba, 0x21, 0x67, 0xb7, 0x18, 0x05, 0xc6, 0x01, 0xa8, 0x11, 0xb7, 0xc7, 0x3e, 0x31, + 0xb3, 0x35, 0xd4, 0x28, 0xb4, 0xdf, 0x58, 0x2e, 0xaa, 0x09, 0xf3, 0xcf, 0xa2, 0x8a, 0x3e, 0xc0, + 0x09, 0xd8, 0x3b, 0x02, 0x35, 0x56, 0x19, 0x3a, 0x28, 0x57, 0xe4, 0xd6, 0x44, 0x35, 0xd4, 0xd0, + 0xb0, 0x18, 0x8d, 0x1d, 0xc8, 0x5d, 0xdb, 0xfe, 0x3c, 0xf6, 0xd0, 0x70, 0x0c, 0x3e, 0xcd, 0x7e, + 0x8c, 0xea, 0x47, 0x00, 0x3d, 0xef, 0x86, 0xb8, 0xdf, 0x08, 0x46, 0x28, 0x7d, 0x12, 0x4a, 0x65, + 0x09, 0x8b, 0xf1, 0x79, 0x65, 0xfd, 0x2f, 0x04, 0x45, 0x29, 0x4b, 0xf2, 0x75, 0x56, 0xf9, 0x3e, + 0x5c, 0x9f, 0xef, 0x89, 0x2a, 0x09, 0xd9, 0xde, 0xba, 0x5b, 0x54, 0x33, 0x2f, 0x15, 0xf5, 0xfb, + 0x35, 0x51, 0xbf, 0x78, 0xba, 0x70, 0xb1, 0xd5, 0xd8, 0x60, 0x1d, 0x99, 0xfd, 0x69, 0x29, 0xbf, + 0x29, 0x50, 0xc2, 0xe4, 0xc7, 0x39, 0x89, 0x78, 0x12, 0xd0, 0x84, 0xfc, 0x35, 0x61, 0x91, 0x47, + 0xe3, 0x72, 0x72, 0x38, 0x85, 0xa2, 0xa0, 0xc0, 0x9e, 0x78, 0x8e, 0x7c, 0x5f, 0x0e, 0xc7, 0xc0, + 0xe8, 0x01, 0x44, 0x9c, 0x11, 0x3b, 0x18, 0x39, 0x81, 0x6b, 0x2a, 0x35, 0xd4, 0x28, 0xb7, 0x0e, + 0xd6, 0xaf, 0x32, 0x94, 0xe7, 0x4f, 0x02, 0x17, 0x6b, 0x51, 0x3a, 0x1a, 0x55, 0x28, 0x32, 0x12, + 0x50, 0x4e, 0x46, 0x33, 0x9b, 0x4f, 0xcd, 0x2d, 0x99, 0x13, 0x62, 0xea, 0xc2, 0xe6, 0x53, 0xe3, + 0x5d, 0x28, 0x70, 0x66, 0x3b, 0x64, 0xe4, 0xb9, 0x66, 0x4e, 0x3c, 0x6d, 0x17, 0x97, 0x8b, 0x6a, + 0xfe, 0x52, 0x70, 0xfd, 0x0e, 0xce, 0xcb, 0x87, 0x7d, 0xd7, 0x78, 0x07, 0xca, 0x0e, 0x0d, 0x39, + 0x09, 0xf9, 0xc8, 0x27, 0xe1, 0x84, 0x4f, 0x4d, 0xb5, 0x86, 0x1a, 0x0a, 0x2e, 0x25, 0xec, 0x57, + 0x92, 0x34, 0xda, 0xa0, 0x4e, 0x65, 0x62, 0xb3, 0x20, 0xeb, 0x7b, 0x7b, 0x93, 0xdb, 0x9a, 0x7c, + 0xc1, 0x44, 0x69, 0xf4, 0x41, 0xbc, 0xd5, 0xf3, 0x09, 0x33, 0x35, 0x69, 0xf2, 0xde, 0xc6, 0x57, + 0x22, 0x71, 0x4a, 0xf5, 0xc6, 0x3e, 0x68, 0x33, 0x9b, 0xd9, 0x01, 0xe1, 0x84, 0x99, 0x50, 0x43, + 0x8d, 0x6d, 0xbc, 0x22, 0xea, 0x7f, 0x66, 0xa1, 0x8c, 0x49, 0x34, 0xa3, 0x61, 0x44, 0x5e, 0xf1, + 0x3b, 0xed, 0xca, 0x2b, 0xc7, 0xe7, 0x91, 0xac, 0x36, 0x87, 0x13, 0x24, 0x78, 0x46, 0xec, 0x88, + 0x86, 0x71, 0xa9, 0x38, 0x41, 0xc2, 0x85, 0x30, 0x46, 0x99, 0x6c, 0x4f, 0xc3, 0x31, 0x78, 0xa6, + 0xdc, 0xfc, 0x6b, 0x5f, 0xee, 0x39, 0xe4, 0xba, 0x32, 0xdc, 0xaa, 0x22, 0xf4, 0x82, 0x8a, 0xb2, + 0xff, 0xa9, 0x68, 0x17, 0x54, 0x97, 0x70, 0xdb, 0xf3, 0xe5, 0xb5, 0xd7, 0x70, 0x82, 0xea, 0xbf, + 0x23, 0x28, 0x9d, 0x4c, 0x89, 0x73, 0x15, 0xcd, 0x83, 0xb6, 0x4f, 0x9d, 0x2b, 0xe3, 0x0c, 0x34, + 0xdb, 0x9f, 0x50, 0xe6, 0xf1, 0x69, 0x20, 0xcd, 0xcb, 0xad, 0xe6, 0xfa, 0x34, 0xa9, 0xfe, 0x38, + 0x95, 0xe1, 0x95, 0x83, 0xb0, 0x73, 0x3d, 0x46, 0x1c, 0xee, 0x25, 0x3b, 0x6d, 0x6c, 0xd7, 0x49, + 0x65, 0x78, 0xe5, 0x20, 0xea, 0x19, 0x8b, 0x35, 0x87, 0xde, 0xcf, 0x44, 0x46, 0x29, 0xe1, 0x15, + 0x71, 0xd8, 0x04, 0xed, 0xf1, 0x07, 0x6b, 0xe4, 0x41, 0x19, 0x9c, 0x5f, 0xea, 0x19, 0x31, 0x0c, + 0xbf, 0x1d, 0xe8, 0x48, 0x0c, 0x17, 0xc3, 0x53, 0x3d, 0x2b, 0x86, 0x5e, 0x7f, 0xa0, 0x2b, 0x87, + 0x9f, 0xc3, 0x9b, 0xff, 0xdb, 0xde, 0xd8, 0x86, 0xc2, 0xb1, 0x3f, 0x19, 0x0d, 0x68, 0x48, 0xf4, + 0x8c, 0x40, 0x27, 0xcc, 0x19, 0xf5, 0xbb, 0xdd, 0xae, 0x8e, 0x52, 0xd4, 0x3b, 0x1e, 0x5e, 0xea, + 0xd9, 0xc3, 0x2f, 0x57, 0xf2, 0xc7, 0x6d, 0xc5, 0x91, 0x8e, 0xc7, 0x52, 0x39, 0x80, 0xda, 0x99, + 0xcf, 0x7c, 0x72, 0xa3, 0x23, 0x31, 0x7f, 0x3d, 0xf3, 0xa9, 0xed, 0xea, 0x59, 0x79, 0x8a, 0xfe, + 0x14, 0x4a, 0xa4, 0xb4, 0xdf, 0xba, 0x5b, 0x56, 0xd0, 0xfd, 0xb2, 0x82, 0xfe, 0x5e, 0x56, 0xd0, + 0x2f, 0x0f, 0x95, 0xcc, 0xfd, 0x43, 0x25, 0xf3, 0xc7, 0x43, 0x25, 0xf3, 0x5d, 0xde, 0x6a, 0x7e, + 0x26, 0x6a, 0x19, 0xab, 0xf2, 0xff, 0xeb, 0xa3, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0x41, 0x15, + 0x6c, 0xfc, 0x01, 0x07, 0x00, 0x00, } func (m *Header) Marshal() (dAtA []byte, err error) { @@ -714,10 +694,6 @@ func (m *Header) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if m.stable { i-- if m.stable { @@ -770,10 +746,6 @@ func (m *FixedValue) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if len(m.Value) > 0 { i -= len(m.Value) copy(dAtA[i:], m.Value) @@ -809,10 +781,6 @@ func (m *FixedHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if m.stable { i-- if m.stable { @@ -870,10 +838,6 @@ func (m *RequestHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if len(m.Parameter) > 0 { i -= len(m.Parameter) copy(dAtA[i:], m.Parameter) @@ -958,10 +922,6 @@ func (m *ResponseHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if len(m.Parameter) > 0 { i -= len(m.Parameter) copy(dAtA[i:], m.Parameter) @@ -1046,10 +1006,6 @@ func (m *Error) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if len(m.Detail) > 0 { i -= len(m.Detail) copy(dAtA[i:], m.Detail) @@ -1092,10 +1048,6 @@ func (m *ChecksumBlock) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l - if m.XXX_unrecognized != nil { - i -= len(m.XXX_unrecognized) - copy(dAtA[i:], m.XXX_unrecognized) - } if m.BlockSize != 0 { i = encodeVarintRpc2(dAtA, i, uint64(m.BlockSize)) i-- @@ -1142,9 +1094,6 @@ func (m *Header) Size() (n int) { if m.stable { n += 2 } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1161,9 +1110,6 @@ func (m *FixedValue) Size() (n int) { if l > 0 { n += 1 + l + sovRpc2(uint64(l)) } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1185,9 +1131,6 @@ func (m *FixedHeader) Size() (n int) { if m.stable { n += 2 } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1225,9 +1168,6 @@ func (m *RequestHeader) Size() (n int) { if l > 0 { n += 1 + l + sovRpc2(uint64(l)) } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1265,9 +1205,6 @@ func (m *ResponseHeader) Size() (n int) { if l > 0 { n += 1 + l + sovRpc2(uint64(l)) } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1288,9 +1225,6 @@ func (m *Error) Size() (n int) { if l > 0 { n += 1 + l + sovRpc2(uint64(l)) } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1309,9 +1243,6 @@ func (m *ChecksumBlock) Size() (n int) { if m.BlockSize != 0 { n += 1 + sovRpc2(uint64(m.BlockSize)) } - if m.XXX_unrecognized != nil { - n += len(m.XXX_unrecognized) - } return n } @@ -1509,7 +1440,6 @@ func (m *Header) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } @@ -1611,7 +1541,6 @@ func (m *FixedValue) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } @@ -1811,7 +1740,6 @@ func (m *FixedHeader) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } @@ -2102,7 +2030,6 @@ func (m *RequestHeader) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } @@ -2393,7 +2320,6 @@ func (m *ResponseHeader) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } @@ -2527,7 +2453,6 @@ func (m *Error) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } @@ -2635,7 +2560,6 @@ func (m *ChecksumBlock) Unmarshal(dAtA []byte) error { if (iNdEx + skippy) > l { return io.ErrUnexpectedEOF } - m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...) iNdEx += skippy } } diff --git a/blobstore/common/rpc2/rpc2.proto b/blobstore/common/rpc2/rpc2.proto index 611467512..522133c1b 100644 --- a/blobstore/common/rpc2/rpc2.proto +++ b/blobstore/common/rpc2/rpc2.proto @@ -20,6 +20,9 @@ option go_package = "./;rpc2"; option (gogoproto.sizer_all) = true; option (gogoproto.marshaler_all) = true; option (gogoproto.unmarshaler_all) = true; +option (gogoproto.goproto_unkeyed_all) = false; +option (gogoproto.goproto_unrecognized_all) = false; +option (gogoproto.goproto_sizecache_all) = false; import "gogoproto/gogo.proto"; diff --git a/blobstore/common/rpc2/rpc2_test.go b/blobstore/common/rpc2/rpc2_test.go new file mode 100644 index 000000000..2c69c1569 --- /dev/null +++ b/blobstore/common/rpc2/rpc2_test.go @@ -0,0 +1,364 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "context" + "io" + "net" + "testing" + "time" + + "github.com/cubefs/cubefs/blobstore/common/rpc2/transport" + _ "github.com/cubefs/cubefs/blobstore/testing/nolog" + "github.com/cubefs/cubefs/blobstore/util/log" + "github.com/cubefs/cubefs/blobstore/util/retry" + proto "github.com/gogo/protobuf/proto" + "github.com/stretchr/testify/require" +) + +var ( + defHandler = &Router{} + testCtx = context.Background() +) + +func init() { + defHandler.Middleware(handleMiddleware1, handleMiddleware2) + defHandler.Register("/", handleNone) + defHandler.Register("/stream", handleStream) + defHandler.Register("/error", handleError) +} + +func handleMiddleware1(w ResponseWriter, req *Request) error { + log.Info("handle middleware-1") + return nil +} + +func handleMiddleware2(w ResponseWriter, req *Request) error { + log.Info("handle middleware-2") + return nil +} + +type noCopyReadWriter struct{} + +func (noCopyReadWriter) Read(p []byte) (int, error) { return len(p), nil } +func (noCopyReadWriter) Write(p []byte) (int, error) { return len(p), nil } + +func handleNone(w ResponseWriter, req *Request) error { + if req.ContentLength == 0 { + req.Body.Close() + return w.WriteOK(nil) + } + req.Body.WriteTo(LimitWriter(noCopyReadWriter{}, req.ContentLength)) + w.SetContentLength(req.ContentLength) + w.WriteHeader(200, nil) + _, err := w.ReadFrom(noCopyReadWriter{}) + return err +} + +func handleError(w ResponseWriter, req *Request) error { + return w.WriteHeader(500, nil) +} + +func handleStream(_ ResponseWriter, req *Request) error { + stream := GenericServerStream[noneCodec, noneCodec]{ServerStream: req.ServerStream()} + for { + _, err := stream.Recv() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if err = stream.Send(&noneCodec{}); err != nil { + return err + } + } +} + +func newTcpServer() (string, *Client, func()) { + server, client, f := newServer("tcp", defHandler) + return server.Name, client, f +} + +func newServer(network string, handler Handler) (*Server, *Client, func()) { + addr := getAddress(network) + trans := transport.DefaultConfig() + trans.Version = 2 + server := Server{ + Name: addr, + Addresses: []NetworkAddress{{Network: network, Address: addr}}, + Transport: trans, + Handler: handler, + StatDuration: 777 * time.Millisecond, + } + server.RegisterOnShutdown(func() { log.Info("shutdown") }) + go func() { + if err := server.Serve(); err != nil && err != ErrServerClosed { + panic(err) + } + }() + server.waitServe() + client := Client{ + ConnectorConfig: ConnectorConfig{ + Transport: trans, + Network: network, + DialTimeout: 200 * time.Millisecond, + }, + RetryOn: func(err error) bool { + status := DetectStatusCode(err) + return status >= 500 + }, + } + return &server, &client, func() { + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + server.Shutdown(ctx) + client.Close() + } +} + +func getAddress(network string) (addr string) { + if err := retry.Timed(10, 1).On(func() error { + ln, err := net.Listen(network, "127.0.0.1:0") + if err != nil { + return err + } + if err = ln.Close(); err != nil { + return err + } + addr = ln.Addr().String() + return nil + }); err != nil { + panic(err) + } + return +} + +func BenchmarkUploadDownload(b *testing.B) { + handler := &Router{} + handler.Register("/", handleNone) + server, cli, shutdown := newServer("tcp", handler) + defer shutdown() + + l := int64(4 << 20) + b.SetBytes(l) + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + req, _ := NewRequest(testCtx, server.Name, "/", nil, noCopyReadWriter{}) + req.ContentLength = l + resp, _ := cli.Do(req, nil) + resp.Body.WriteTo(LimitWriter(noCopyReadWriter{}, l)) + resp.Body.Close() + } +} + +func TestRpc2None(t *testing.T) { + { + var x noneCodec + x.Size() + x.Marshal() + x.MarshalTo(nil) + x.Unmarshal(nil) + } + { + NoBody.Read(nil) + NoBody.WriteTo(nil) + NoBody.Close() + } + { + NoParameter.Size() + NoParameter.Marshal() + NoParameter.MarshalTo(nil) + NoParameter.Unmarshal(nil) + } + { + require.Panics(t, func() { + var x codecReadWriter + x.Size() + }) + } +} + +func TestRpc2Pb(t *testing.T) { + { + StreamCmd_NOT.EnumDescriptor() + ChecksumAlgorithm_Alg_None.EnumDescriptor() + ChecksumDirection_Dir_None.EnumDescriptor() + } + + run := func(m interface { + Codec + Reset() + String() string + ProtoMessage() + Descriptor() ([]byte, []int) + XXX_Unmarshal(b []byte) error + XXX_Marshal(b []byte, deterministic bool) ([]byte, error) + XXX_Merge(src proto.Message) + XXX_Size() int + XXX_DiscardUnknown() + }, discard bool, + ) { + _ = m.String() + m.ProtoMessage() + m.Descriptor() + b, _ := m.Marshal() + m.XXX_Unmarshal(b) + m.XXX_Marshal(b, false) + m.XXX_Marshal(b, true) + m.XXX_Merge(m) + m.XXX_Size() + if !discard { + m.XXX_DiscardUnknown() + } + m.MarshalTo(b) + m.Reset() + } + + { + var v *FixedValue + v.GetLen() + v.GetValue() + v = &FixedValue{Len: 1, Value: "a"} + run(v, false) + v.GetLen() + v.GetValue() + } + { + var v *Header + v.GetM() + v.Getstable() + v = &Header{} + v.Set("a", "a") + run(v, false) + v.GetM() + v.Getstable() + } + { + var v *FixedHeader + v.GetM() + v.Getstable() + v = &FixedHeader{} + v.Set("a", "a") + run(v, false) + v.GetM() + v.Getstable() + } + { + var v *RequestHeader + v.GetVersion() + v.GetMagic() + v.GetStreamCmd() + v.GetRemotePath() + v.GetTraceID() + v.GetContentLength() + v.GetHeader() + v.GetTrailer() + v.GetParameter() + v.XXX_DiscardUnknown() + v = &RequestHeader{ + Version: 1, + Magic: 1, + StreamCmd: StreamCmd_NOT, + RemotePath: "/", + TraceID: "xxx", + ContentLength: 10, + Header: Header{}, + Trailer: FixedHeader{}, + Parameter: []byte{0xff}, + } + v.Header.Set("a", "a") + v.Trailer.Set("b", "b") + run(v, true) + v.GetVersion() + v.GetMagic() + v.GetStreamCmd() + v.GetRemotePath() + v.GetTraceID() + v.GetContentLength() + v.GetHeader() + v.GetTrailer() + v.GetParameter() + } + { + var v *ResponseHeader + v.GetVersion() + v.GetMagic() + v.GetStatus() + v.GetReason() + v.GetError() + v.GetContentLength() + v.GetHeader() + v.GetTrailer() + v.GetParameter() + v.XXX_DiscardUnknown() + v = &ResponseHeader{ + Version: 1, + Magic: 1, + Status: 200, + Reason: "R", + Error: "E", + ContentLength: 10, + Header: Header{}, + Trailer: FixedHeader{}, + Parameter: []byte{0xff}, + } + v.Header.Set("a", "a") + v.Trailer.Set("b", "b") + run(v, true) + v.GetVersion() + v.GetMagic() + v.GetStatus() + v.GetReason() + v.GetError() + v.GetContentLength() + v.GetHeader() + v.GetTrailer() + v.GetParameter() + } + { + var v *Error + v.GetStatus() + v.GetReason() + v.GetDetail() + v = &Error{ + Status: 100, + Reason: "R", + Detail: "E", + } + run(v, false) + v.GetStatus() + v.GetReason() + v.GetDetail() + } + { + var v *ChecksumBlock + v.GetAlgorithm() + v.GetDirection() + v.GetBlockSize() + v = &ChecksumBlock{ + Algorithm: ChecksumAlgorithm_Crc_IEEE, + Direction: ChecksumDirection_Duplex, + BlockSize: 1 << 10, + } + run(v, false) + v.GetAlgorithm() + v.GetDirection() + v.GetBlockSize() + } +} diff --git a/blobstore/common/rpc2/server.go b/blobstore/common/rpc2/server.go index 983914cb3..e825bcaed 100644 --- a/blobstore/common/rpc2/server.go +++ b/blobstore/common/rpc2/server.go @@ -65,6 +65,7 @@ type Server struct { StatDuration time.Duration statOnce sync.Once + inServe atomic.Value // true when server waiting to accept inShutdown atomic.Value // true when server is in shutdown listenerGroup sync.WaitGroup @@ -96,6 +97,17 @@ func (s *Server) stating() { }) } +func (s *Server) waitServe() { + for { + if val := s.inServe.Load(); val != nil { + if val.(bool) { + return + } + } + time.Sleep(10 * time.Millisecond) + } +} + func (s *Server) shuttingDown() bool { if val := s.inShutdown.Load(); val != nil { return val.(bool) @@ -170,12 +182,11 @@ func (s *Server) Listen(ln net.Listener) error { s.sessions = make(map[*transport.Session]struct{}) } _, has := s.listeners[key] - s.mu.Unlock() if has { + s.mu.Unlock() return nil } - s.mu.Lock() s.listeners[key] = struct{}{} s.listenerGroup.Add(1) s.mu.Unlock() @@ -194,6 +205,7 @@ func (s *Server) Listen(ln net.Listener) error { } for { + s.inServe.Store(true) conn, err := ln.Accept() if err != nil { if s.shuttingDown() { @@ -246,7 +258,11 @@ func (s *Server) handleStream(stream *transport.Stream) { status, reason, detail := DetectError(err) ss.hdr.Status = int32(status) ss.hdr.Reason = reason - ss.hdr.Error = detail.Error() + if detail != nil { + ss.hdr.Error = detail.Error() + } else { + getSpan(ctx).Warn(err) + } } else { ss.hdr.Status = 200 } @@ -266,6 +282,8 @@ func (s *Server) handleStream(stream *transport.Stream) { resp.hdr.Reason = reason if detail != nil { resp.hdr.Error = detail.Error() + } else { + getSpan(ctx).Warn(err) } resp.WriteHeader(status, NoParameter) } @@ -325,7 +343,7 @@ func (s *Server) readRequest(stream *transport.Stream) (*Request, error) { payloadSize += int(req.ContentLength) } req.Body = makeBodyWithTrailer(stream.NewSizedReader(payloadSize, frame), - req, req.ContentLength, decode) + req, &req.Trailer, req.ContentLength, decode) if hdr.StreamCmd == StreamCmd_SYN { req.stream = &serverStream{req: req} diff --git a/blobstore/common/rpc2/server_test.go b/blobstore/common/rpc2/server_test.go new file mode 100644 index 000000000..529c23fa0 --- /dev/null +++ b/blobstore/common/rpc2/server_test.go @@ -0,0 +1,67 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "context" + "testing" + "time" + + "github.com/cubefs/cubefs/blobstore/common/rpc2/transport" + "github.com/stretchr/testify/require" +) + +func TestServerError(t *testing.T) { + { + addr := getAddress("tcp") + server := Server{ + Addresses: []NetworkAddress{{Network: "network", Address: addr}}, + Handler: defHandler, + } + err := server.Serve() + require.Error(t, err) + } + { + addr := getAddress("tcp") + addr1 := getAddress("tcp") + server := Server{ + Addresses: []NetworkAddress{ + {Network: "tcp", Address: addr}, + {Network: "tcp", Address: addr1}, + }, + Handler: defHandler, + } + go func() { server.Serve() }() + server.waitServe() + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + server.Shutdown(ctx) + cancel() + } + { + trans := transport.DefaultConfig() + trans.MaxFrameSize = 1 << 30 + addr := getAddress("tcp") + server := Server{ + Transport: trans, + Addresses: []NetworkAddress{{Network: "tcp", Address: addr}}, + Handler: defHandler, + } + go func() { server.Serve() }() + server.waitServe() + connector := defaultConnector(ConnectorConfig{Network: "tcp"}) + _, err := connector.Get(testCtx, addr) + require.NoError(t, err) + } +} diff --git a/blobstore/common/rpc2/stream.go b/blobstore/common/rpc2/stream.go index f1ea16ad2..a5af9f730 100644 --- a/blobstore/common/rpc2/stream.go +++ b/blobstore/common/rpc2/stream.go @@ -17,6 +17,7 @@ package rpc2 import ( "context" "io" + "sync" ) type ClientStream interface { @@ -181,7 +182,7 @@ func (cs *clientStream) RecvMsg(a any) (err error) { }() if resp.Status > 0 { // end - cs.trailer = resp.Trailer.ToHeader() + cs.trailer.Merge(resp.Trailer.ToHeader()) cs.req.conn.Close() if resp.Status != 200 { return &Error{ @@ -217,6 +218,7 @@ type serverStream struct { req *Request sentHeader bool + supplyOnce sync.Once hdr ResponseHeader } @@ -240,7 +242,9 @@ func (ss *serverStream) SendHeader(obj Marshaler) error { obj = NoParameter } ss.sentHeader = true - ss.hdr.Status = 200 + if ss.hdr.Status == 0 { + ss.hdr.Status = 200 + } ss.hdr.ContentLength = int64(obj.Size()) return ss.writeFrameMsg(&ss.hdr, obj) } @@ -249,16 +253,14 @@ func (ss *serverStream) SetTrailer(h Header) { ss.hdr.Trailer.MergeHeader(h) } -func (ss *serverStream) SendMsg(a any) error { - if !ss.sentHeader { - if err := ss.SendHeader(nil); err != nil { - return err - } - } +func (ss *serverStream) SendMsg(a any) (err error) { msg, is := a.(Codec) if !is { panic("rpc2: stream send message must implement rpc2.Codec") } + if err = ss.supplyHeader(); err != nil { + return err + } hdr := ResponseHeader{Version: Version, Magic: Magic} hdr.ContentLength = int64(msg.Size()) @@ -270,6 +272,10 @@ func (ss *serverStream) RecvMsg(a any) (err error) { if !is { panic("rpc2: stream recv message must implement rpc2.Codec") } + if err = ss.supplyHeader(); err != nil { + return err + } + var req RequestHeader frame, err := readHeaderFrame(ss.req.conn, &req) if err != nil { @@ -292,6 +298,15 @@ func (ss *serverStream) RecvMsg(a any) (err error) { return msg.Unmarshal(frame.Bytes(int(req.ContentLength))) } +func (ss *serverStream) supplyHeader() (err error) { + ss.supplyOnce.Do(func() { + if !ss.sentHeader { + err = ss.SendHeader(nil) + } + }) + return +} + func (ss *serverStream) writeFrameMsg(hdr *ResponseHeader, msg Marshaler) error { size := _headerCell + hdr.Size() + msg.Size() if size > ss.req.conn.MaxPayloadSize() { diff --git a/blobstore/common/rpc2/stream_test.go b/blobstore/common/rpc2/stream_test.go new file mode 100644 index 000000000..c92a42bc0 --- /dev/null +++ b/blobstore/common/rpc2/stream_test.go @@ -0,0 +1,111 @@ +// Copyright 2024 The CubeFS Authors. +// +// Licensed 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 rpc2 + +import ( + "io" + "testing" + + "github.com/cubefs/cubefs/blobstore/cli/common/fmt" + "github.com/stretchr/testify/require" +) + +type ( + streamReq = strMessage + streamResp = strMessage +) + +func handleStreamFull(_ ResponseWriter, req *Request) error { + var para strMessage + req.ParseParameter(¶) + + var header, trailer Header + header.Set("stream-header", "aaa") + trailer.Set("stream-trailer", "") + + stream := GenericServerStream[streamReq, streamResp]{ServerStream: req.ServerStream()} + stream.Context() + stream.SetHeader(header) + stream.SetTrailer(trailer) + stream.SendHeader(¶) + defer func() { + trailer.Set("stream-trailer", "bbb") + stream.SetTrailer(trailer) + }() + for { + msg, err := stream.Recv() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if msg.str == "error" { + return &Error{Status: 500} + } + if err = stream.Send(&streamResp{msg.str}); err != nil { + return err + } + } +} + +func TestStreamBase(t *testing.T) { + handler := &Router{} + handler.Register("/", handleStreamFull) + server, cli, shutdown := newServer("tcp", handler) + defer shutdown() + sc := StreamClient[streamReq, streamResp]{Client: cli} + + { + ecc := &clientStream{} + require.Panics(t, func() { ecc.RecvMsg(struct{}{}) }) + require.Panics(t, func() { ecc.SendMsg(struct{}{}) }) + esc := &serverStream{} + require.Panics(t, func() { esc.RecvMsg(struct{}{}) }) + require.Panics(t, func() { esc.SendMsg(struct{}{}) }) + } + + para := strMessage{"para"} + req, err := NewStreamRequest(testCtx, server.Name, "/", ¶) + require.NoError(t, err) + + cc, err := sc.Streaming(req, ¶) + require.NoError(t, err, cc.Context()) + header, _ := cc.Header() + require.Equal(t, "aaa", header.Get("stream-header")) + trailer := cc.Trailer() + require.Equal(t, "", trailer.Get("stream-trailer")) + + waitc := make(chan struct{}) + go func() { + for { + _, errx := cc.Recv() + if errx != nil { + close(waitc) + return + } + } + }() + for idx := range [10]struct{}{} { + req := streamReq{str: fmt.Sprintf("request-%d", idx)} + if idx == 7 { + req.str = "error" + } + require.NoError(t, cc.Send(&req)) + } + cc.CloseSend() + <-waitc + require.Equal(t, "bbb", trailer.Get("stream-trailer")) +}