diff --git a/p2p/message.go b/p2p/message.go index daf2bf05c1..09eefb7658 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -2,12 +2,10 @@ package p2p import ( "bytes" - "encoding/binary" "errors" "fmt" "io" "io/ioutil" - "math/big" "sync/atomic" "github.com/ethereum/go-ethereum/ethutil" @@ -23,7 +21,7 @@ import ( // separate Msg with a bytes.Reader as Payload for each send. type Msg struct { Code uint64 - Size uint32 // size of the paylod + Size uint32 // size of the payload Payload io.Reader } @@ -36,6 +34,7 @@ func NewMsg(code uint64, params ...interface{}) Msg { return Msg{Code: code, Size: uint32(buf.Len()), Payload: buf} } +// this should probably use the new rlp encoding func encodePayload(params ...interface{}) []byte { buf := new(bytes.Buffer) for _, p := range params { @@ -66,6 +65,11 @@ func (msg Msg) Discard() error { return err } +/* +MsgReadWriter is an interface for reading and writing messages +It is aware of message structure and knows how to encode/decode +*/ + type MsgReader interface { ReadMsg() (Msg, error) } @@ -84,89 +88,15 @@ type MsgReadWriter interface { MsgWriter } +// EncodeMsg should be explicit in the interface and should have the MsgWriter +// as receiver + // EncodeMsg writes an RLP-encoded message with the given code and // data elements. func EncodeMsg(w MsgWriter, code uint64, data ...interface{}) error { return w.WriteMsg(NewMsg(code, data...)) } -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 makeListHeader(length uint32) []byte { - if length < 56 { - return []byte{byte(length + 0xc0)} - } - enc := big.NewInt(int64(length)).Bytes() - lenb := byte(len(enc)) + 0xf7 - return append([]byte{lenb}, enc...) -} - -// readMsg reads a message header from r. -// 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 { - return msg, newPeerError(errRead, "%v", err) - } - if !bytes.HasPrefix(start, magicToken) { - return msg, newPeerError(errMagicTokenMismatch, "got %x, want %x", start[:4], magicToken) - } - size := binary.BigEndian.Uint32(start[4:]) - - // decode start of RLP message to get the message code - posr := &postrack{r, 0} - s := rlp.NewStream(posr) - if _, err := s.List(); err != nil { - return msg, err - } - code, err := s.Uint() - if err != nil { - return msg, err - } - payloadsize := size - posr.p - return Msg{code, payloadsize, io.LimitReader(r, int64(payloadsize))}, nil -} - -// postrack wraps an rlp.ByteReader with a position counter. -type postrack struct { - r rlp.ByteReader - p uint32 -} - -func (r *postrack) Read(buf []byte) (int, error) { - n, err := r.r.Read(buf) - r.p += uint32(n) - return n, err -} - -func (r *postrack) ReadByte() (byte, error) { - b, err := r.r.ReadByte() - if err == nil { - r.p++ - } - return b, err -} - // MsgPipe creates a message pipe. Reads on one end are matched // with writes on the other. The pipe is full-duplex, both ends // implement MsgReadWriter. @@ -185,6 +115,19 @@ func MsgPipe() (*MsgPipeRW, *MsgPipeRW) { // pipe has been closed. var ErrPipeClosed = errors.New("p2p: read or write on closed message pipe") +// Close unblocks any pending ReadMsg and WriteMsg calls on both ends +// of the pipe. They will return ErrPipeClosed. Note that Close does +// not interrupt any reads from a message payload. +func (p *MsgPipeRW) Close() error { + if atomic.AddInt32(p.closed, 1) != 1 { + // someone else is already closing + atomic.StoreInt32(p.closed, 1) // avoid overflow + return nil + } + close(p.closing) + return nil +} + // MsgPipeRW is an endpoint of a MsgReadWriter pipe. type MsgPipeRW struct { w chan<- Msg @@ -224,15 +167,25 @@ func (p *MsgPipeRW) ReadMsg() (Msg, error) { return Msg{}, ErrPipeClosed } -// Close unblocks any pending ReadMsg and WriteMsg calls on both ends -// of the pipe. They will return ErrPipeClosed. Note that Close does -// not interrupt any reads from a message payload. -func (p *MsgPipeRW) Close() error { - if atomic.AddInt32(p.closed, 1) != 1 { - // someone else is already closing - atomic.StoreInt32(p.closed, 1) // avoid overflow - return nil - } - close(p.closing) - return nil +// eofSignal wraps a reader with eof signaling. the eof channel is +// closed when the wrapped reader returns an error or when count bytes +// have been read. +// +type eofSignal struct { + wrapped io.Reader + count int64 + eof chan<- struct{} +} + +// note: when using eofSignal to detect whether a message payload +// has been read, Read might not be called for zero sized messages. + +func (r *eofSignal) Read(buf []byte) (int, error) { + n, err := r.wrapped.Read(buf) + r.count -= int64(n) + if (err != nil || r.count <= 0) && r.eof != nil { + r.eof <- struct{}{} // tell Peer that msg has been consumed + r.eof = nil + } + return n, err } diff --git a/p2p/messenger.go b/p2p/messenger.go new file mode 100644 index 0000000000..57268a6542 --- /dev/null +++ b/p2p/messenger.go @@ -0,0 +1,178 @@ +package p2p + +import ( + "bytes" + "encoding/binary" + "fmt" + "io" + "math/big" + "time" + + "github.com/ethereum/go-ethereum/ethutil" + "github.com/ethereum/go-ethereum/rlp" +) + +var magicToken = []byte{34, 64, 8, 145} + +/* + writeMsg, readMsg (makeListHeader) should all be internal to the default legacy + MsgReadWriter implementation + + A MsgReadWriter implementation will typically sit on a multiplexed peer connection and runs a single read and a write loop without need to use locking. + + It passes on incoming messages to its channel picked up by the interface methods Read + the peer runs a dispatch loop that figures out which protocol to forward the message to. + + Because framing and header structure will change there will be hardly any overlap with the new code so I do not abstract readers any further. +*/ +func (p *Peer) readLoop(msgc chan<- Msg, errc chan<- error, unblock <-chan bool) { + for _ = range unblock { + p.conn.SetReadDeadline(time.Now().Add(msgReadTimeout)) + if msg, err := readMsg(p.bufconn); err != nil { + errc <- err + } else { + msgc <- msg + } + } + close(errc) +} + +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 makeListHeader(length uint32) []byte { + if length < 56 { + return []byte{byte(length + 0xc0)} + } + enc := big.NewInt(int64(length)).Bytes() + lenb := byte(len(enc)) + 0xf7 + return append([]byte{lenb}, enc...) +} + +// readMsg reads a message header from r. +// 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 { + return msg, newPeerError(errRead, "%v", err) + } + if !bytes.HasPrefix(start, magicToken) { + return msg, newPeerError(errMagicTokenMismatch, "got %x, want %x", start[:4], magicToken) + } + size := binary.BigEndian.Uint32(start[4:]) + + // decode start of RLP message to get the message code + posr := &postrack{r, 0} + s := rlp.NewStream(posr) + if _, err := s.List(); err != nil { + return msg, err + } + code, err := s.Uint() + if err != nil { + return msg, err + } + payloadsize := size - posr.p + return Msg{code, payloadsize, io.LimitReader(r, int64(payloadsize))}, nil +} + +// postrack wraps an rlp.ByteReader with a position counter. +type postrack struct { + r rlp.ByteReader + p uint32 +} + +func (r *postrack) Read(buf []byte) (int, error) { + n, err := r.r.Read(buf) + r.p += uint32(n) + return n, err +} + +func (r *postrack) ReadByte() (byte, error) { + b, err := r.r.ReadByte() + if err == nil { + r.p++ + } + return b, err +} + +// this duplicates functionality of proto.WriteMsg +// if we need this for broadcasting via a server interface then +// simply call the appropriate Write function of the protocol RW +// writeProtoMsg sends the given message on behalf of the given named protocol. +func (p *Peer) writeProtoMsg(protoName string, msg Msg) error { + p.runlock.RLock() + proto, ok := p.running[protoName] + p.runlock.RUnlock() + if !ok { + return fmt.Errorf("protocol %s not handled by peer", protoName) + } + if msg.Code >= proto.maxcode { + return newPeerError(errInvalidMsgCode, "code %x is out of range for protocol %q", msg.Code, protoName) + } + msg.Code += proto.offset + return p.writeMsg(msg, msgWriteTimeout) +} + +/* +this function is not needed write will be done directly by the msgReadWriter +with the connection attached +if the interface is channel, then no write locking is needed for synchronous write +*/ +// writeMsg writes a message to the connection. +func (p *Peer) writeMsg(msg Msg, timeout time.Duration) error { + p.writeMu.Lock() + defer p.writeMu.Unlock() + p.conn.SetWriteDeadline(time.Now().Add(timeout)) + if err := writeMsg(p.bufconn, msg); err != nil { + return newPeerError(errWrite, "%v", err) + } + return p.bufconn.Flush() +} + +// proto will embed the same writer channel as given to the readwriter +// in the legacy code it knows about the code offset +// no need to go through peer for writing , so do not need to embed peer as field +type proto struct { + name string + in chan Msg + maxcode, offset uint64 + peer *Peer +} + +func (rw *proto) WriteMsg(msg Msg) error { + if msg.Code >= rw.maxcode { + return newPeerError(errInvalidMsgCode, "not handled") + } + msg.Code += rw.offset + return rw.peer.writeMsg(msg, msgWriteTimeout) +} + +func (rw *proto) EncodeMsg(code uint64, data ...interface{}) error { + return rw.WriteMsg(NewMsg(code, data...)) +} + +func (rw *proto) ReadMsg() (Msg, error) { + msg, ok := <-rw.in + if !ok { + return msg, io.EOF + } + msg.Code -= rw.offset + return msg, nil +} diff --git a/p2p/peer.go b/p2p/peer.go index 99f1a61d38..c0cf1dee0c 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -301,18 +301,6 @@ loop: return reason, err } -func (p *Peer) readLoop(msgc chan<- Msg, errc chan<- error, unblock <-chan bool) { - for _ = range unblock { - p.conn.SetReadDeadline(time.Now().Add(msgReadTimeout)) - if msg, err := readMsg(p.bufconn); err != nil { - errc <- err - } else { - msgc <- msg - } - } - close(errc) -} - func (p *Peer) dispatch(msg Msg, protoDone chan struct{}) (wait bool, err error) { proto, err := p.getProto(msg.Code) if err != nil { @@ -446,80 +434,3 @@ func (p *Peer) closeProtocols() { p.runlock.RUnlock() p.protoWG.Wait() } - -// writeProtoMsg sends the given message on behalf of the given named protocol. -func (p *Peer) writeProtoMsg(protoName string, msg Msg) error { - p.runlock.RLock() - proto, ok := p.running[protoName] - p.runlock.RUnlock() - if !ok { - return fmt.Errorf("protocol %s not handled by peer", protoName) - } - if msg.Code >= proto.maxcode { - return newPeerError(errInvalidMsgCode, "code %x is out of range for protocol %q", msg.Code, protoName) - } - msg.Code += proto.offset - return p.writeMsg(msg, msgWriteTimeout) -} - -// writeMsg writes a message to the connection. -func (p *Peer) writeMsg(msg Msg, timeout time.Duration) error { - p.writeMu.Lock() - defer p.writeMu.Unlock() - p.conn.SetWriteDeadline(time.Now().Add(timeout)) - if err := writeMsg(p.bufconn, msg); err != nil { - return newPeerError(errWrite, "%v", err) - } - return p.bufconn.Flush() -} - -type proto struct { - name string - in chan Msg - maxcode, offset uint64 - peer *Peer -} - -func (rw *proto) WriteMsg(msg Msg) error { - if msg.Code >= rw.maxcode { - return newPeerError(errInvalidMsgCode, "not handled") - } - msg.Code += rw.offset - return rw.peer.writeMsg(msg, msgWriteTimeout) -} - -func (rw *proto) EncodeMsg(code uint64, data ...interface{}) error { - return rw.WriteMsg(NewMsg(code, data...)) -} - -func (rw *proto) ReadMsg() (Msg, error) { - msg, ok := <-rw.in - if !ok { - return msg, io.EOF - } - msg.Code -= rw.offset - return msg, nil -} - -// eofSignal wraps a reader with eof signaling. the eof channel is -// closed when the wrapped reader returns an error or when count bytes -// have been read. -// -type eofSignal struct { - wrapped io.Reader - count int64 - eof chan<- struct{} -} - -// note: when using eofSignal to detect whether a message payload -// has been read, Read might not be called for zero sized messages. - -func (r *eofSignal) Read(buf []byte) (int, error) { - n, err := r.wrapped.Read(buf) - r.count -= int64(n) - if (err != nil || r.count <= 0) && r.eof != nil { - r.eof <- struct{}{} // tell Peer that msg has been consumed - r.eof = nil - } - return n, err -}