From f38052c499c1fee61423efeddb1f52677f1442e9 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 4 Nov 2014 13:21:44 +0100 Subject: p2p: rework protocol API --- p2p/message.go | 201 ++++++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 150 insertions(+), 51 deletions(-) (limited to 'p2p/message.go') diff --git a/p2p/message.go b/p2p/message.go index 446e74dff..366cff5d7 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -1,75 +1,174 @@ package p2p import ( - // "fmt" + "bytes" + "encoding/binary" + "fmt" + "io" + "io/ioutil" + "math/big" + "github.com/ethereum/go-ethereum/ethutil" ) -type MsgCode uint8 +type MsgCode uint64 +// Msg defines the structure of a p2p message. +// +// Note that a Msg can only be sent once since the Payload reader is +// consumed during sending. It is not possible to create a Msg and +// send it any number of times. If you want to reuse an encoded +// structure, encode the payload into a byte array and create a +// separate Msg with a bytes.Reader as Payload for each send. type Msg struct { - code MsgCode // this is the raw code as per adaptive msg code scheme - data *ethutil.Value - encoded []byte + Code MsgCode + Size uint32 // size of the paylod + Payload io.Reader } -func (self *Msg) Code() MsgCode { - return self.code +// NewMsg creates an RLP-encoded message with the given code. +func NewMsg(code MsgCode, params ...interface{}) Msg { + buf := new(bytes.Buffer) + for _, p := range params { + buf.Write(ethutil.Encode(p)) + } + return Msg{Code: code, Size: uint32(buf.Len()), Payload: buf} } -func (self *Msg) Data() *ethutil.Value { - return self.data +func encodePayload(params ...interface{}) []byte { + buf := new(bytes.Buffer) + for _, p := range params { + buf.Write(ethutil.Encode(p)) + } + return buf.Bytes() } -func NewMsg(code MsgCode, params ...interface{}) (msg *Msg, err error) { - - // // data := [][]interface{}{} - // data := []interface{}{} - // for _, value := range params { - // if encodable, ok := value.(ethutil.RlpEncodeDecode); ok { - // data = append(data, encodable.RlpValue()) - // } else if raw, ok := value.([]interface{}); ok { - // data = append(data, raw) - // } else { - // // data = append(data, interface{}(raw)) - // err = fmt.Errorf("Unable to encode object of type %T", value) - // return - // } - // } - return &Msg{ - code: code, - data: ethutil.NewValue(interface{}(params)), - }, nil +// Data returns the decoded RLP payload items in a message. +func (msg Msg) Data() (*ethutil.Value, error) { + // TODO: avoid copying when we have a better RLP decoder + buf := new(bytes.Buffer) + var s []interface{} + if _, err := buf.ReadFrom(msg.Payload); err != nil { + return nil, err + } + for buf.Len() > 0 { + s = append(s, ethutil.DecodeWithReader(buf)) + } + return ethutil.NewValue(s), nil +} + +// Discard reads any remaining payload data into a black hole. +func (msg Msg) Discard() error { + _, err := io.Copy(ioutil.Discard, msg.Payload) + return err +} + +var magicToken = []byte{34, 64, 8, 145} + +func writeMsg(w io.Writer, msg Msg) error { + // TODO: handle case when Size + len(code) + len(listhdr) overflows uint32 + code := ethutil.Encode(uint32(msg.Code)) + listhdr := makeListHeader(msg.Size + uint32(len(code))) + payloadLen := uint32(len(listhdr)) + uint32(len(code)) + msg.Size + + start := make([]byte, 8) + copy(start, magicToken) + binary.BigEndian.PutUint32(start[4:], payloadLen) + + for _, b := range [][]byte{start, listhdr, code} { + if _, err := w.Write(b); err != nil { + return err + } + } + _, err := io.CopyN(w, msg.Payload, int64(msg.Size)) + return err } -func NewMsgFromBytes(encoded []byte) (msg *Msg, err error) { - value := ethutil.NewValueFromBytes(encoded) - // Type of message - code := value.Get(0).Uint() - // Actual data - data := value.SliceFrom(1) - - msg = &Msg{ - code: MsgCode(code), - data: data, - // data: ethutil.NewValue(data), - encoded: encoded, +func makeListHeader(length uint32) []byte { + if length < 56 { + return []byte{byte(length + 0xc0)} } - return + enc := big.NewInt(int64(length)).Bytes() + lenb := byte(len(enc)) + 0xf7 + return append([]byte{lenb}, enc...) } -func (self *Msg) Decode(offset MsgCode) { - self.code = self.code - offset +type byteReader interface { + io.Reader + io.ByteReader } -// encode takes an offset argument to implement adaptive message coding -// the encoded message is memoized to make msgs relayed to several peers more efficient -func (self *Msg) Encode(offset MsgCode) (res []byte) { - if len(self.encoded) == 0 { - res = ethutil.NewValue(append([]interface{}{byte(self.code + offset)}, self.data.Slice()...)).Encode() - self.encoded = res +// readMsg reads a message header. +func readMsg(r byteReader) (msg Msg, err error) { + // read magic and payload size + start := make([]byte, 8) + if _, err = io.ReadFull(r, start); err != nil { + return msg, NewPeerError(ReadError, "%v", err) + } + if !bytes.HasPrefix(start, magicToken) { + return msg, NewPeerError(MagicTokenMismatch, "got %x, want %x", start[:4], magicToken) + } + size := binary.BigEndian.Uint32(start[4:]) + + // decode start of RLP message to get the message code + _, hdrlen, err := readListHeader(r) + if err != nil { + return msg, err + } + code, codelen, err := readMsgCode(r) + if err != nil { + return msg, err + } + + rlpsize := size - hdrlen - codelen + return Msg{ + Code: code, + Size: rlpsize, + Payload: io.LimitReader(r, int64(rlpsize)), + }, nil +} + +// readListHeader reads an RLP list header from r. +func readListHeader(r byteReader) (len uint64, hdrlen uint32, err error) { + b, err := r.ReadByte() + if err != nil { + return 0, 0, err + } + if b < 0xC0 { + return 0, 0, fmt.Errorf("expected list start byte >= 0xC0, got %x", b) + } else if b < 0xF7 { + len = uint64(b - 0xc0) + hdrlen = 1 } else { - res = self.encoded + lenlen := b - 0xF7 + lenbuf := make([]byte, 8) + if _, err := io.ReadFull(r, lenbuf[8-lenlen:]); err != nil { + return 0, 0, err + } + len = binary.BigEndian.Uint64(lenbuf) + hdrlen = 1 + uint32(lenlen) + } + return len, hdrlen, nil +} + +// readUint reads an RLP-encoded unsigned integer from r. +func readMsgCode(r byteReader) (code MsgCode, codelen uint32, err error) { + b, err := r.ReadByte() + if err != nil { + return 0, 0, err + } + if b < 0x80 { + return MsgCode(b), 1, nil + } else if b < 0x89 { // max length for uint64 is 8 bytes + codelen = uint32(b - 0x80) + if codelen == 0 { + return 0, 1, nil + } + buf := make([]byte, 8) + if _, err := io.ReadFull(r, buf[8-codelen:]); err != nil { + return 0, 0, err + } + return MsgCode(binary.BigEndian.Uint64(buf)), codelen, nil } - return + return 0, 0, fmt.Errorf("bad RLP type for message code: %x", b) } -- cgit v1.2.3 From 7149191dd999f4d192398e4b0821b656e62f3345 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Wed, 5 Nov 2014 01:28:46 +0100 Subject: p2p: fix issues found during review --- p2p/message.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'p2p/message.go') diff --git a/p2p/message.go b/p2p/message.go index 366cff5d7..97d440a27 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -98,7 +98,7 @@ type byteReader interface { io.ByteReader } -// readMsg reads a message header. +// readMsg reads a message header from r. func readMsg(r byteReader) (msg Msg, err error) { // read magic and payload size start := make([]byte, 8) -- cgit v1.2.3 From 59b63caf5e4de64ceb7dcdf01551a080f53b1672 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 21 Nov 2014 21:48:49 +0100 Subject: p2p: API cleanup and PoC 7 compatibility Whoa, one more big commit. I didn't manage to untangle the changes while working towards compatibility. --- p2p/message.go | 62 +++++++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 53 insertions(+), 9 deletions(-) (limited to 'p2p/message.go') diff --git a/p2p/message.go b/p2p/message.go index 97d440a27..89ad189d7 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -11,8 +11,6 @@ import ( "github.com/ethereum/go-ethereum/ethutil" ) -type MsgCode uint64 - // Msg defines the structure of a p2p message. // // Note that a Msg can only be sent once since the Payload reader is @@ -21,13 +19,13 @@ type MsgCode uint64 // structure, encode the payload into a byte array and create a // separate Msg with a bytes.Reader as Payload for each send. type Msg struct { - Code MsgCode + Code uint64 Size uint32 // size of the paylod Payload io.Reader } // NewMsg creates an RLP-encoded message with the given code. -func NewMsg(code MsgCode, params ...interface{}) Msg { +func NewMsg(code uint64, params ...interface{}) Msg { buf := new(bytes.Buffer) for _, p := range params { buf.Write(ethutil.Encode(p)) @@ -63,6 +61,52 @@ func (msg Msg) Discard() error { return err } +type MsgReader interface { + ReadMsg() (Msg, error) +} + +type MsgWriter interface { + // WriteMsg sends an existing message. + // The Payload reader of the message is consumed. + // Note that messages can be sent only once. + WriteMsg(Msg) error + + // EncodeMsg writes an RLP-encoded message with the given + // code and data elements. + EncodeMsg(code uint64, data ...interface{}) error +} + +// MsgReadWriter provides reading and writing of encoded messages. +type MsgReadWriter interface { + MsgReader + MsgWriter +} + +// MsgLoop reads messages off the given reader and +// calls the handler function for each decoded message until +// it returns an error or the peer connection is closed. +// +// If a message is larger than the given maximum size, +// MsgLoop returns an appropriate error. +func MsgLoop(r MsgReader, maxsize uint32, f func(code uint64, data *ethutil.Value) error) error { + for { + msg, err := r.ReadMsg() + if err != nil { + return err + } + if msg.Size > maxsize { + return newPeerError(errInvalidMsg, "size %d exceeds maximum size of %d", msg.Size, maxsize) + } + value, err := msg.Data() + if err != nil { + return err + } + if err := f(msg.Code, value); err != nil { + return err + } + } +} + var magicToken = []byte{34, 64, 8, 145} func writeMsg(w io.Writer, msg Msg) error { @@ -103,10 +147,10 @@ func readMsg(r byteReader) (msg Msg, err error) { // read magic and payload size start := make([]byte, 8) if _, err = io.ReadFull(r, start); err != nil { - return msg, NewPeerError(ReadError, "%v", err) + return msg, newPeerError(errRead, "%v", err) } if !bytes.HasPrefix(start, magicToken) { - return msg, NewPeerError(MagicTokenMismatch, "got %x, want %x", start[:4], magicToken) + return msg, newPeerError(errMagicTokenMismatch, "got %x, want %x", start[:4], magicToken) } size := binary.BigEndian.Uint32(start[4:]) @@ -152,13 +196,13 @@ func readListHeader(r byteReader) (len uint64, hdrlen uint32, err error) { } // readUint reads an RLP-encoded unsigned integer from r. -func readMsgCode(r byteReader) (code MsgCode, codelen uint32, err error) { +func readMsgCode(r byteReader) (code uint64, codelen uint32, err error) { b, err := r.ReadByte() if err != nil { return 0, 0, err } if b < 0x80 { - return MsgCode(b), 1, nil + return uint64(b), 1, nil } else if b < 0x89 { // max length for uint64 is 8 bytes codelen = uint32(b - 0x80) if codelen == 0 { @@ -168,7 +212,7 @@ func readMsgCode(r byteReader) (code MsgCode, codelen uint32, err error) { if _, err := io.ReadFull(r, buf[8-codelen:]); err != nil { return 0, 0, err } - return MsgCode(binary.BigEndian.Uint64(buf)), codelen, nil + return binary.BigEndian.Uint64(buf), codelen, nil } return 0, 0, fmt.Errorf("bad RLP type for message code: %x", b) } -- cgit v1.2.3 From c1fca72552386868d28ce7541691e53e55673549 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Mon, 24 Nov 2014 19:02:48 +0100 Subject: p2p: use package rlp --- p2p/message.go | 93 +++++++++++++++++----------------------------------------- 1 file changed, 27 insertions(+), 66 deletions(-) (limited to 'p2p/message.go') diff --git a/p2p/message.go b/p2p/message.go index 89ad189d7..ade39d25a 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -3,12 +3,12 @@ package p2p import ( "bytes" "encoding/binary" - "fmt" "io" "io/ioutil" "math/big" "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/rlp" ) // Msg defines the structure of a p2p message. @@ -43,16 +43,10 @@ func encodePayload(params ...interface{}) []byte { // Data returns the decoded RLP payload items in a message. func (msg Msg) Data() (*ethutil.Value, error) { - // TODO: avoid copying when we have a better RLP decoder - buf := new(bytes.Buffer) - var s []interface{} - if _, err := buf.ReadFrom(msg.Payload); err != nil { - return nil, err - } - for buf.Len() > 0 { - s = append(s, ethutil.DecodeWithReader(buf)) - } - return ethutil.NewValue(s), nil + s := rlp.NewListStream(msg.Payload, uint64(msg.Size)) + var v []interface{} + err := s.Decode(&v) + return ethutil.NewValue(v), err } // Discard reads any remaining payload data into a black hole. @@ -137,13 +131,9 @@ func makeListHeader(length uint32) []byte { return append([]byte{lenb}, enc...) } -type byteReader interface { - io.Reader - io.ByteReader -} - // readMsg reads a message header from r. -func readMsg(r byteReader) (msg Msg, err error) { +// It takes an rlp.ByteReader to ensure that the decoding doesn't buffer. +func readMsg(r rlp.ByteReader) (msg Msg, err error) { // read magic and payload size start := make([]byte, 8) if _, err = io.ReadFull(r, start); err != nil { @@ -155,64 +145,35 @@ func readMsg(r byteReader) (msg Msg, err error) { size := binary.BigEndian.Uint32(start[4:]) // decode start of RLP message to get the message code - _, hdrlen, err := readListHeader(r) - if err != nil { + posr := &postrack{r, 0} + s := rlp.NewStream(posr) + if _, err := s.List(); err != nil { return msg, err } - code, codelen, err := readMsgCode(r) + code, err := s.Uint() if err != nil { return msg, err } + payloadsize := size - posr.p + return Msg{code, payloadsize, io.LimitReader(r, int64(payloadsize))}, nil +} - rlpsize := size - hdrlen - codelen - return Msg{ - Code: code, - Size: rlpsize, - Payload: io.LimitReader(r, int64(rlpsize)), - }, nil +// postrack wraps an rlp.ByteReader with a position counter. +type postrack struct { + r rlp.ByteReader + p uint32 } -// readListHeader reads an RLP list header from r. -func readListHeader(r byteReader) (len uint64, hdrlen uint32, err error) { - b, err := r.ReadByte() - if err != nil { - return 0, 0, err - } - if b < 0xC0 { - return 0, 0, fmt.Errorf("expected list start byte >= 0xC0, got %x", b) - } else if b < 0xF7 { - len = uint64(b - 0xc0) - hdrlen = 1 - } else { - lenlen := b - 0xF7 - lenbuf := make([]byte, 8) - if _, err := io.ReadFull(r, lenbuf[8-lenlen:]); err != nil { - return 0, 0, err - } - len = binary.BigEndian.Uint64(lenbuf) - hdrlen = 1 + uint32(lenlen) - } - return len, hdrlen, nil +func (r *postrack) Read(buf []byte) (int, error) { + n, err := r.r.Read(buf) + r.p += uint32(n) + return n, err } -// readUint reads an RLP-encoded unsigned integer from r. -func readMsgCode(r byteReader) (code uint64, codelen uint32, err error) { - b, err := r.ReadByte() - if err != nil { - return 0, 0, err - } - if b < 0x80 { - return uint64(b), 1, nil - } else if b < 0x89 { // max length for uint64 is 8 bytes - codelen = uint32(b - 0x80) - if codelen == 0 { - return 0, 1, nil - } - buf := make([]byte, 8) - if _, err := io.ReadFull(r, buf[8-codelen:]); err != nil { - return 0, 0, err - } - return binary.BigEndian.Uint64(buf), codelen, nil +func (r *postrack) ReadByte() (byte, error) { + b, err := r.r.ReadByte() + if err == nil { + r.p++ } - return 0, 0, fmt.Errorf("bad RLP type for message code: %x", b) + return b, err } -- cgit v1.2.3 From 6049fcd52ab10362721a352cfd7a93a01c3ffa97 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 25 Nov 2014 12:25:31 +0100 Subject: p2p: use package rlp for baseProtocol --- p2p/message.go | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) (limited to 'p2p/message.go') diff --git a/p2p/message.go b/p2p/message.go index ade39d25a..845c832f0 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -41,14 +41,22 @@ func encodePayload(params ...interface{}) []byte { return buf.Bytes() } -// Data returns the decoded RLP payload items in a message. -func (msg Msg) Data() (*ethutil.Value, error) { - s := rlp.NewListStream(msg.Payload, uint64(msg.Size)) +// Value returns the decoded RLP payload items in a message. +func (msg Msg) Value() (*ethutil.Value, error) { var v []interface{} - err := s.Decode(&v) + err := msg.Decode(&v) return ethutil.NewValue(v), err } +// Decode parse the RLP content of a message into +// the given value, which must be a pointer. +// +// For the decoding rules, please see package rlp. +func (msg Msg) Decode(val interface{}) error { + s := rlp.NewListStream(msg.Payload, uint64(msg.Size)) + return s.Decode(val) +} + // Discard reads any remaining payload data into a black hole. func (msg Msg) Discard() error { _, err := io.Copy(ioutil.Discard, msg.Payload) @@ -91,7 +99,7 @@ func MsgLoop(r MsgReader, maxsize uint32, f func(code uint64, data *ethutil.Valu if msg.Size > maxsize { return newPeerError(errInvalidMsg, "size %d exceeds maximum size of %d", msg.Size, maxsize) } - value, err := msg.Data() + value, err := msg.Value() if err != nil { return err } -- cgit v1.2.3 From 9b85002b700500d421ba7e13ac2062a6b8090a83 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 25 Nov 2014 16:01:39 +0100 Subject: p2p: remove Msg.Value and MsgLoop --- p2p/message.go | 32 -------------------------------- 1 file changed, 32 deletions(-) (limited to 'p2p/message.go') diff --git a/p2p/message.go b/p2p/message.go index 845c832f0..d3b8b74d4 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -41,13 +41,6 @@ func encodePayload(params ...interface{}) []byte { return buf.Bytes() } -// Value returns the decoded RLP payload items in a message. -func (msg Msg) Value() (*ethutil.Value, error) { - var v []interface{} - err := msg.Decode(&v) - return ethutil.NewValue(v), err -} - // Decode parse the RLP content of a message into // the given value, which must be a pointer. // @@ -84,31 +77,6 @@ type MsgReadWriter interface { MsgWriter } -// MsgLoop reads messages off the given reader and -// calls the handler function for each decoded message until -// it returns an error or the peer connection is closed. -// -// If a message is larger than the given maximum size, -// MsgLoop returns an appropriate error. -func MsgLoop(r MsgReader, maxsize uint32, f func(code uint64, data *ethutil.Value) error) error { - for { - msg, err := r.ReadMsg() - if err != nil { - return err - } - if msg.Size > maxsize { - return newPeerError(errInvalidMsg, "size %d exceeds maximum size of %d", msg.Size, maxsize) - } - value, err := msg.Value() - if err != nil { - return err - } - if err := f(msg.Code, value); err != nil { - return err - } - } -} - var magicToken = []byte{34, 64, 8, 145} func writeMsg(w io.Writer, msg Msg) error { -- cgit v1.2.3