reorg and comment to prepare for refactor of read/write

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
+// eofSignal wraps a reader with eof signaling. the eof channel is
-// of the pipe. They will return ErrPipeClosed. Note that Close does
+// closed when the wrapped reader returns an error or when count bytes
-// not interrupt any reads from a message payload.
+// have been read.
-func (p *MsgPipeRW) Close() error {
+//
-	if atomic.AddInt32(p.closed, 1) != 1 {
+type eofSignal struct {
-		// someone else is already closing
+	wrapped io.Reader
-		atomic.StoreInt32(p.closed, 1) // avoid overflow
+	count   int64
-		return nil
+	eof     chan<- struct{}
-	}
+}
-	close(p.closing)
+
-	return nil
+// 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
 }

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
+}

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
-}