mirror of
https://github.com/tendermint/tendermint.git
synced 2026-01-07 13:55:17 +00:00
p2p/conn: migrate to Protobuf (#4990)
Migrates the p2p connections to Protobuf. Supersedes #4800. gogoproto's `NewDelimitedReader()` uses an internal buffer, which makes it unsuitable for reading individual messages from a shared reader (since any remaining data in the buffer will be discarded). We therefore add a new `protoio` package with an unbuffered `NewDelimitedReader()`. Additionally, the `NewDelimitedWriter()` returns the number of bytes written, and we've added `MarshalDelimited()` and `UnmarshalDelimited()`, to ease migration of existing code.
This commit is contained in:
Erik Grinaker
GitHub
96
libs/protoio/io.go
Normal file
96
libs/protoio/io.go
Normal file
@@ -0,0 +1,96 @@
|
||||
// Protocol Buffers for Go with Gadgets
|
||||
//
|
||||
// Copyright (c) 2013, The GoGo Authors. All rights reserved.
|
||||
// http://github.com/gogo/protobuf
|
||||
//
|
||||
// Redistribution and use in source and binary forms, with or without
|
||||
// modification, are permitted provided that the following conditions are
|
||||
// met:
|
||||
//
|
||||
// * Redistributions of source code must retain the above copyright
|
||||
// notice, this list of conditions and the following disclaimer.
|
||||
// * Redistributions in binary form must reproduce the above
|
||||
// copyright notice, this list of conditions and the following disclaimer
|
||||
// in the documentation and/or other materials provided with the
|
||||
// distribution.
|
||||
//
|
||||
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
//
|
||||
// Modified to return number of bytes written by Writer.WriteMsg(), and added byteReader.
|
||||
|
||||
package protoio
|
||||
|
||||
import (
|
||||
"io"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
)
|
||||
|
||||
type Writer interface {
|
||||
WriteMsg(proto.Message) (int, error)
|
||||
}
|
||||
|
||||
type WriteCloser interface {
|
||||
Writer
|
||||
io.Closer
|
||||
}
|
||||
|
||||
type Reader interface {
|
||||
ReadMsg(msg proto.Message) error
|
||||
}
|
||||
|
||||
type ReadCloser interface {
|
||||
Reader
|
||||
io.Closer
|
||||
}
|
||||
|
||||
type marshaler interface {
|
||||
MarshalTo(data []byte) (n int, err error)
|
||||
}
|
||||
|
||||
func getSize(v interface{}) (int, bool) {
|
||||
if sz, ok := v.(interface {
|
||||
Size() (n int)
|
||||
}); ok {
|
||||
return sz.Size(), true
|
||||
} else if sz, ok := v.(interface {
|
||||
ProtoSize() (n int)
|
||||
}); ok {
|
||||
return sz.ProtoSize(), true
|
||||
} else {
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
|
||||
// byteReader wraps an io.Reader and implements io.ByteReader. Reading one byte at a
|
||||
// time is extremely slow, but this is what Amino did already, and the caller can
|
||||
// wrap the reader in bufio.Reader if appropriate.
|
||||
type byteReader struct {
|
||||
io.Reader
|
||||
bytes []byte
|
||||
}
|
||||
|
||||
func newByteReader(r io.Reader) *byteReader {
|
||||
return &byteReader{
|
||||
Reader: r,
|
||||
bytes: make([]byte, 1),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *byteReader) ReadByte() (byte, error) {
|
||||
_, err := r.Read(r.bytes)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return r.bytes[0], nil
|
||||
}
|
||||
157
libs/protoio/io_test.go
Normal file
157
libs/protoio/io_test.go
Normal file
@@ -0,0 +1,157 @@
|
||||
// Protocol Buffers for Go with Gadgets
|
||||
//
|
||||
// Copyright (c) 2013, The GoGo Authors. All rights reserved.
|
||||
// http://github.com/gogo/protobuf
|
||||
//
|
||||
// Redistribution and use in source and binary forms, with or without
|
||||
// modification, are permitted provided that the following conditions are
|
||||
// met:
|
||||
//
|
||||
// * Redistributions of source code must retain the above copyright
|
||||
// notice, this list of conditions and the following disclaimer.
|
||||
// * Redistributions in binary form must reproduce the above
|
||||
// copyright notice, this list of conditions and the following disclaimer
|
||||
// in the documentation and/or other materials provided with the
|
||||
// distribution.
|
||||
//
|
||||
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
|
||||
package protoio_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
"github.com/gogo/protobuf/test"
|
||||
|
||||
"github.com/tendermint/tendermint/libs/protoio"
|
||||
)
|
||||
|
||||
func iotest(writer protoio.WriteCloser, reader protoio.ReadCloser) error {
|
||||
varint := make([]byte, binary.MaxVarintLen64)
|
||||
size := 1000
|
||||
msgs := make([]*test.NinOptNative, size)
|
||||
r := rand.New(rand.NewSource(time.Now().UnixNano()))
|
||||
for i := range msgs {
|
||||
msgs[i] = test.NewPopulatedNinOptNative(r, true)
|
||||
//issue 31
|
||||
if i == 5 {
|
||||
msgs[i] = &test.NinOptNative{}
|
||||
}
|
||||
//issue 31
|
||||
if i == 999 {
|
||||
msgs[i] = &test.NinOptNative{}
|
||||
}
|
||||
// FIXME Check size
|
||||
bz, err := proto.Marshal(msgs[i])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
visize := binary.PutUvarint(varint, uint64(len(bz)))
|
||||
n, err := writer.WriteMsg(msgs[i])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n != len(bz)+visize {
|
||||
return fmt.Errorf("WriteMsg() wrote %v bytes, expected %v", n, len(bz)+visize) // nolint
|
||||
}
|
||||
}
|
||||
if err := writer.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
i := 0
|
||||
for {
|
||||
msg := &test.NinOptNative{}
|
||||
if err := reader.ReadMsg(msg); err != nil {
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
return err
|
||||
}
|
||||
if err := msg.VerboseEqual(msgs[i]); err != nil {
|
||||
return err
|
||||
}
|
||||
i++
|
||||
}
|
||||
if i != size {
|
||||
panic("not enough messages read")
|
||||
}
|
||||
if err := reader.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type buffer struct {
|
||||
*bytes.Buffer
|
||||
closed bool
|
||||
}
|
||||
|
||||
func (b *buffer) Close() error {
|
||||
b.closed = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func newBuffer() *buffer {
|
||||
return &buffer{bytes.NewBuffer(nil), false}
|
||||
}
|
||||
|
||||
func TestVarintNormal(t *testing.T) {
|
||||
buf := newBuffer()
|
||||
writer := protoio.NewDelimitedWriter(buf)
|
||||
reader := protoio.NewDelimitedReader(buf, 1024*1024)
|
||||
if err := iotest(writer, reader); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if !buf.closed {
|
||||
t.Fatalf("did not close buffer")
|
||||
}
|
||||
}
|
||||
|
||||
func TestVarintNoClose(t *testing.T) {
|
||||
buf := bytes.NewBuffer(nil)
|
||||
writer := protoio.NewDelimitedWriter(buf)
|
||||
reader := protoio.NewDelimitedReader(buf, 1024*1024)
|
||||
if err := iotest(writer, reader); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
//issue 32
|
||||
func TestVarintMaxSize(t *testing.T) {
|
||||
buf := newBuffer()
|
||||
writer := protoio.NewDelimitedWriter(buf)
|
||||
reader := protoio.NewDelimitedReader(buf, 20)
|
||||
if err := iotest(writer, reader); err == nil {
|
||||
t.Error(err)
|
||||
} else {
|
||||
t.Logf("%s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestVarintError(t *testing.T) {
|
||||
buf := newBuffer()
|
||||
buf.Write([]byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x7f})
|
||||
reader := protoio.NewDelimitedReader(buf, 1024*1024)
|
||||
msg := &test.NinOptNative{}
|
||||
err := reader.ReadMsg(msg)
|
||||
if err == nil {
|
||||
t.Fatalf("Expected error")
|
||||
}
|
||||
}
|
||||
88
libs/protoio/reader.go
Normal file
88
libs/protoio/reader.go
Normal file
@@ -0,0 +1,88 @@
|
||||
// Protocol Buffers for Go with Gadgets
|
||||
//
|
||||
// Copyright (c) 2013, The GoGo Authors. All rights reserved.
|
||||
// http://github.com/gogo/protobuf
|
||||
//
|
||||
// Redistribution and use in source and binary forms, with or without
|
||||
// modification, are permitted provided that the following conditions are
|
||||
// met:
|
||||
//
|
||||
// * Redistributions of source code must retain the above copyright
|
||||
// notice, this list of conditions and the following disclaimer.
|
||||
// * Redistributions in binary form must reproduce the above
|
||||
// copyright notice, this list of conditions and the following disclaimer
|
||||
// in the documentation and/or other materials provided with the
|
||||
// distribution.
|
||||
//
|
||||
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
//
|
||||
// Modified from original GoGo Protobuf to not buffer the reader.
|
||||
|
||||
package protoio
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
)
|
||||
|
||||
// NewDelimitedReader reads varint-delimited Protobuf messages from a reader. Unlike the gogoproto
|
||||
// NewDelimitedReader, this does not buffer the reader, which may cause poor performance but is
|
||||
// necessary when only reading single messages (e.g. in the p2p package).
|
||||
func NewDelimitedReader(r io.Reader, maxSize int) ReadCloser {
|
||||
var closer io.Closer
|
||||
if c, ok := r.(io.Closer); ok {
|
||||
closer = c
|
||||
}
|
||||
return &varintReader{newByteReader(r), nil, maxSize, closer}
|
||||
}
|
||||
|
||||
type varintReader struct {
|
||||
r *byteReader
|
||||
buf []byte
|
||||
maxSize int
|
||||
closer io.Closer
|
||||
}
|
||||
|
||||
func (r *varintReader) ReadMsg(msg proto.Message) error {
|
||||
length64, err := binary.ReadUvarint(newByteReader(r.r))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
length := int(length64)
|
||||
if length < 0 || length > r.maxSize {
|
||||
return fmt.Errorf("message exceeds max size (%v > %v)", length, r.maxSize)
|
||||
}
|
||||
if len(r.buf) < length {
|
||||
r.buf = make([]byte, length)
|
||||
}
|
||||
buf := r.buf[:length]
|
||||
if _, err := io.ReadFull(r.r, buf); err != nil {
|
||||
return err
|
||||
}
|
||||
return proto.Unmarshal(buf, msg)
|
||||
}
|
||||
|
||||
func (r *varintReader) Close() error {
|
||||
if r.closer != nil {
|
||||
return r.closer.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func UnmarshalDelimited(data []byte, msg proto.Message) error {
|
||||
return NewDelimitedReader(bytes.NewReader(data), len(data)).ReadMsg(msg)
|
||||
}
|
||||
100
libs/protoio/writer.go
Normal file
100
libs/protoio/writer.go
Normal file
@@ -0,0 +1,100 @@
|
||||
// Protocol Buffers for Go with Gadgets
|
||||
//
|
||||
// Copyright (c) 2013, The GoGo Authors. All rights reserved.
|
||||
// http://github.com/gogo/protobuf
|
||||
//
|
||||
// Redistribution and use in source and binary forms, with or without
|
||||
// modification, are permitted provided that the following conditions are
|
||||
// met:
|
||||
//
|
||||
// * Redistributions of source code must retain the above copyright
|
||||
// notice, this list of conditions and the following disclaimer.
|
||||
// * Redistributions in binary form must reproduce the above
|
||||
// copyright notice, this list of conditions and the following disclaimer
|
||||
// in the documentation and/or other materials provided with the
|
||||
// distribution.
|
||||
//
|
||||
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
//
|
||||
// Modified from original GoGo Protobuf to return number of bytes written.
|
||||
|
||||
package protoio
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"io"
|
||||
|
||||
"github.com/gogo/protobuf/proto"
|
||||
)
|
||||
|
||||
// NewDelimitedWriter writes a varint-delimited Protobuf message to a writer. It is
|
||||
// equivalent to the gogoproto NewDelimitedWriter, except WriteMsg() also returns the
|
||||
// number of bytes written, which is necessary in the p2p package.
|
||||
func NewDelimitedWriter(w io.Writer) WriteCloser {
|
||||
return &varintWriter{w, make([]byte, binary.MaxVarintLen64), nil}
|
||||
}
|
||||
|
||||
type varintWriter struct {
|
||||
w io.Writer
|
||||
lenBuf []byte
|
||||
buffer []byte
|
||||
}
|
||||
|
||||
func (w *varintWriter) WriteMsg(msg proto.Message) (int, error) {
|
||||
if m, ok := msg.(marshaler); ok {
|
||||
n, ok := getSize(m)
|
||||
if ok {
|
||||
if n+binary.MaxVarintLen64 >= len(w.buffer) {
|
||||
w.buffer = make([]byte, n+binary.MaxVarintLen64)
|
||||
}
|
||||
lenOff := binary.PutUvarint(w.buffer, uint64(n))
|
||||
_, err := m.MarshalTo(w.buffer[lenOff:])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
_, err = w.w.Write(w.buffer[:lenOff+n])
|
||||
return lenOff + n, err
|
||||
}
|
||||
}
|
||||
|
||||
// fallback
|
||||
data, err := proto.Marshal(msg)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
length := uint64(len(data))
|
||||
n := binary.PutUvarint(w.lenBuf, length)
|
||||
_, err = w.w.Write(w.lenBuf[:n])
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
_, err = w.w.Write(data)
|
||||
return len(data) + n, err
|
||||
}
|
||||
|
||||
func (w *varintWriter) Close() error {
|
||||
if closer, ok := w.w.(io.Closer); ok {
|
||||
return closer.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func MarshalDelimited(msg proto.Message) ([]byte, error) {
|
||||
var buf bytes.Buffer
|
||||
_, err := NewDelimitedWriter(&buf).WriteMsg(msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return buf.Bytes(), nil
|
||||
}
|
||||
Reference in New Issue
Block a user