rpc_util_test.go

Documentation: google.golang.org/grpc

     1  /*
     2   *
     3   * Copyright 2014 gRPC authors.
     4   *
     5   * Licensed under the Apache License, Version 2.0 (the "License");
     6   * you may not use this file except in compliance with the License.
     7   * You may obtain a copy of the License at
     8   *
     9   *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   */
    18  
    19  package grpc
    20  
    21  import (
    22  	"bytes"
    23  	"compress/gzip"
    24  	"io"
    25  	"math"
    26  	"reflect"
    27  	"testing"
    28  
    29  	"google.golang.org/grpc/codes"
    30  	"google.golang.org/grpc/encoding"
    31  	protoenc "google.golang.org/grpc/encoding/proto"
    32  	"google.golang.org/grpc/internal/testutils"
    33  	"google.golang.org/grpc/internal/transport"
    34  	"google.golang.org/grpc/status"
    35  	perfpb "google.golang.org/grpc/test/codec_perf"
    36  	"google.golang.org/protobuf/proto"
    37  )
    38  
    39  type fullReader struct {
    40  	reader io.Reader
    41  }
    42  
    43  func (f fullReader) Read(p []byte) (int, error) {
    44  	return io.ReadFull(f.reader, p)
    45  }
    46  
    47  var _ CallOption = EmptyCallOption{} // ensure EmptyCallOption implements the interface
    48  
    49  func (s) TestSimpleParsing(t *testing.T) {
    50  	bigMsg := bytes.Repeat([]byte{'x'}, 1<<24)
    51  	for _, test := range []struct {
    52  		// input
    53  		p []byte
    54  		// outputs
    55  		err error
    56  		b   []byte
    57  		pt  payloadFormat
    58  	}{
    59  		{nil, io.EOF, nil, compressionNone},
    60  		{[]byte{0, 0, 0, 0, 0}, nil, nil, compressionNone},
    61  		{[]byte{0, 0, 0, 0, 1, 'a'}, nil, []byte{'a'}, compressionNone},
    62  		{[]byte{1, 0}, io.ErrUnexpectedEOF, nil, compressionNone},
    63  		{[]byte{0, 0, 0, 0, 10, 'a'}, io.ErrUnexpectedEOF, nil, compressionNone},
    64  		// Check that messages with length >= 2^24 are parsed.
    65  		{append([]byte{0, 1, 0, 0, 0}, bigMsg...), nil, bigMsg, compressionNone},
    66  	} {
    67  		buf := fullReader{bytes.NewReader(test.p)}
    68  		parser := &parser{r: buf, recvBufferPool: nopBufferPool{}}
    69  		pt, b, err := parser.recvMsg(math.MaxInt32)
    70  		if err != test.err || !bytes.Equal(b, test.b) || pt != test.pt {
    71  			t.Fatalf("parser{%v}.recvMsg(_) = %v, %v, %v\nwant %v, %v, %v", test.p, pt, b, err, test.pt, test.b, test.err)
    72  		}
    73  	}
    74  }
    75  
    76  func (s) TestMultipleParsing(t *testing.T) {
    77  	// Set a byte stream consists of 3 messages with their headers.
    78  	p := []byte{0, 0, 0, 0, 1, 'a', 0, 0, 0, 0, 2, 'b', 'c', 0, 0, 0, 0, 1, 'd'}
    79  	b := fullReader{bytes.NewReader(p)}
    80  	parser := &parser{r: b, recvBufferPool: nopBufferPool{}}
    81  
    82  	wantRecvs := []struct {
    83  		pt   payloadFormat
    84  		data []byte
    85  	}{
    86  		{compressionNone, []byte("a")},
    87  		{compressionNone, []byte("bc")},
    88  		{compressionNone, []byte("d")},
    89  	}
    90  	for i, want := range wantRecvs {
    91  		pt, data, err := parser.recvMsg(math.MaxInt32)
    92  		if err != nil || pt != want.pt || !reflect.DeepEqual(data, want.data) {
    93  			t.Fatalf("after %d calls, parser{%v}.recvMsg(_) = %v, %v, %v\nwant %v, %v, <nil>",
    94  				i, p, pt, data, err, want.pt, want.data)
    95  		}
    96  	}
    97  
    98  	pt, data, err := parser.recvMsg(math.MaxInt32)
    99  	if err != io.EOF {
   100  		t.Fatalf("after %d recvMsgs calls, parser{%v}.recvMsg(_) = %v, %v, %v\nwant _, _, %v",
   101  			len(wantRecvs), p, pt, data, err, io.EOF)
   102  	}
   103  }
   104  
   105  func (s) TestEncode(t *testing.T) {
   106  	for _, test := range []struct {
   107  		// input
   108  		msg proto.Message
   109  		// outputs
   110  		hdr  []byte
   111  		data []byte
   112  		err  error
   113  	}{
   114  		{nil, []byte{0, 0, 0, 0, 0}, []byte{}, nil},
   115  	} {
   116  		data, err := encode(encoding.GetCodec(protoenc.Name), test.msg)
   117  		if err != test.err || !bytes.Equal(data, test.data) {
   118  			t.Errorf("encode(_, %v) = %v, %v; want %v, %v", test.msg, data, err, test.data, test.err)
   119  			continue
   120  		}
   121  		if hdr, _ := msgHeader(data, nil); !bytes.Equal(hdr, test.hdr) {
   122  			t.Errorf("msgHeader(%v, false) = %v; want %v", data, hdr, test.hdr)
   123  		}
   124  	}
   125  }
   126  
   127  func (s) TestCompress(t *testing.T) {
   128  	bestCompressor, err := NewGZIPCompressorWithLevel(gzip.BestCompression)
   129  	if err != nil {
   130  		t.Fatalf("Could not initialize gzip compressor with best compression.")
   131  	}
   132  	bestSpeedCompressor, err := NewGZIPCompressorWithLevel(gzip.BestSpeed)
   133  	if err != nil {
   134  		t.Fatalf("Could not initialize gzip compressor with best speed compression.")
   135  	}
   136  
   137  	defaultCompressor, err := NewGZIPCompressorWithLevel(gzip.BestSpeed)
   138  	if err != nil {
   139  		t.Fatalf("Could not initialize gzip compressor with default compression.")
   140  	}
   141  
   142  	level5, err := NewGZIPCompressorWithLevel(5)
   143  	if err != nil {
   144  		t.Fatalf("Could not initialize gzip compressor with level 5 compression.")
   145  	}
   146  
   147  	for _, test := range []struct {
   148  		// input
   149  		data []byte
   150  		cp   Compressor
   151  		dc   Decompressor
   152  		// outputs
   153  		err error
   154  	}{
   155  		{make([]byte, 1024), NewGZIPCompressor(), NewGZIPDecompressor(), nil},
   156  		{make([]byte, 1024), bestCompressor, NewGZIPDecompressor(), nil},
   157  		{make([]byte, 1024), bestSpeedCompressor, NewGZIPDecompressor(), nil},
   158  		{make([]byte, 1024), defaultCompressor, NewGZIPDecompressor(), nil},
   159  		{make([]byte, 1024), level5, NewGZIPDecompressor(), nil},
   160  	} {
   161  		b := new(bytes.Buffer)
   162  		if err := test.cp.Do(b, test.data); err != test.err {
   163  			t.Fatalf("Compressor.Do(_, %v) = %v, want %v", test.data, err, test.err)
   164  		}
   165  		if b.Len() >= len(test.data) {
   166  			t.Fatalf("The compressor fails to compress data.")
   167  		}
   168  		if p, err := test.dc.Do(b); err != nil || !bytes.Equal(test.data, p) {
   169  			t.Fatalf("Decompressor.Do(%v) = %v, %v, want %v, <nil>", b, p, err, test.data)
   170  		}
   171  	}
   172  }
   173  
   174  func (s) TestToRPCErr(t *testing.T) {
   175  	for _, test := range []struct {
   176  		// input
   177  		errIn error
   178  		// outputs
   179  		errOut error
   180  	}{
   181  		{transport.ErrConnClosing, status.Error(codes.Unavailable, transport.ErrConnClosing.Desc)},
   182  		{io.ErrUnexpectedEOF, status.Error(codes.Internal, io.ErrUnexpectedEOF.Error())},
   183  	} {
   184  		err := toRPCErr(test.errIn)
   185  		if _, ok := status.FromError(err); !ok {
   186  			t.Errorf("toRPCErr{%v} returned type %T, want %T", test.errIn, err, status.Error)
   187  		}
   188  		if !testutils.StatusErrEqual(err, test.errOut) {
   189  			t.Errorf("toRPCErr{%v} = %v \nwant %v", test.errIn, err, test.errOut)
   190  		}
   191  	}
   192  }
   193  
   194  // bmEncode benchmarks encoding a Protocol Buffer message containing mSize
   195  // bytes.
   196  func bmEncode(b *testing.B, mSize int) {
   197  	cdc := encoding.GetCodec(protoenc.Name)
   198  	msg := &perfpb.Buffer{Body: make([]byte, mSize)}
   199  	encodeData, _ := encode(cdc, msg)
   200  	encodedSz := int64(len(encodeData))
   201  	b.ReportAllocs()
   202  	b.ResetTimer()
   203  	for i := 0; i < b.N; i++ {
   204  		encode(cdc, msg)
   205  	}
   206  	b.SetBytes(encodedSz)
   207  }
   208  
   209  func BenchmarkEncode1B(b *testing.B) {
   210  	bmEncode(b, 1)
   211  }
   212  
   213  func BenchmarkEncode1KiB(b *testing.B) {
   214  	bmEncode(b, 1024)
   215  }
   216  
   217  func BenchmarkEncode8KiB(b *testing.B) {
   218  	bmEncode(b, 8*1024)
   219  }
   220  
   221  func BenchmarkEncode64KiB(b *testing.B) {
   222  	bmEncode(b, 64*1024)
   223  }
   224  
   225  func BenchmarkEncode512KiB(b *testing.B) {
   226  	bmEncode(b, 512*1024)
   227  }
   228  
   229  func BenchmarkEncode1MiB(b *testing.B) {
   230  	bmEncode(b, 1024*1024)
   231  }
   232  
   233  // bmCompressor benchmarks a compressor of a Protocol Buffer message containing
   234  // mSize bytes.
   235  func bmCompressor(b *testing.B, mSize int, cp Compressor) {
   236  	payload := make([]byte, mSize)
   237  	cBuf := bytes.NewBuffer(make([]byte, mSize))
   238  	b.ReportAllocs()
   239  	b.ResetTimer()
   240  	for i := 0; i < b.N; i++ {
   241  		cp.Do(cBuf, payload)
   242  		cBuf.Reset()
   243  	}
   244  }
   245  
   246  func BenchmarkGZIPCompressor1B(b *testing.B) {
   247  	bmCompressor(b, 1, NewGZIPCompressor())
   248  }
   249  
   250  func BenchmarkGZIPCompressor1KiB(b *testing.B) {
   251  	bmCompressor(b, 1024, NewGZIPCompressor())
   252  }
   253  
   254  func BenchmarkGZIPCompressor8KiB(b *testing.B) {
   255  	bmCompressor(b, 8*1024, NewGZIPCompressor())
   256  }
   257  
   258  func BenchmarkGZIPCompressor64KiB(b *testing.B) {
   259  	bmCompressor(b, 64*1024, NewGZIPCompressor())
   260  }
   261  
   262  func BenchmarkGZIPCompressor512KiB(b *testing.B) {
   263  	bmCompressor(b, 512*1024, NewGZIPCompressor())
   264  }
   265  
   266  func BenchmarkGZIPCompressor1MiB(b *testing.B) {
   267  	bmCompressor(b, 1024*1024, NewGZIPCompressor())
   268  }
   269
View as plain text