diff --git a/p2p/messenger.go b/p2p/messenger.go index 57268a6542..56cd0c9da0 100644 --- a/p2p/messenger.go +++ b/p2p/messenger.go @@ -12,6 +12,16 @@ import ( "github.com/ethereum/go-ethereum/rlp" ) +const ( + // maximum amount of time allowed for reading a message + msgReadTimeout = 5 * time.Second + // maximum amount of time allowed for writing a message + msgWriteTimeout = 5 * time.Second + // messages smaller than this many bytes will be read at + // once before passing them to a protocol. + wholePayloadSize = 64 * 1024 +) + var magicToken = []byte{34, 64, 8, 145} /* @@ -37,6 +47,16 @@ func (p *Peer) readLoop(msgc chan<- Msg, errc chan<- error, unblock <-chan bool) close(errc) } +func (p *Peer) writeLoop(msgc <-chan Msg, errc chan<- error) { + p.conn.SetWriteDeadline(time.Now().Add(msgWriteTimeout)) + for msg := range msgc { + if err := writeMsg(p.bufconn, msg); err != nil { + errc <- newPeerError(errWrite, "%v", err) + } + p.bufconn.Flush() + } +} + 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)) @@ -151,9 +171,8 @@ func (p *Peer) writeMsg(msg Msg, timeout time.Duration) error { // 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 + in, out chan Msg maxcode, offset uint64 - peer *Peer } func (rw *proto) WriteMsg(msg Msg) error { @@ -161,7 +180,12 @@ func (rw *proto) WriteMsg(msg Msg) error { return newPeerError(errInvalidMsgCode, "not handled") } msg.Code += rw.offset - return rw.peer.writeMsg(msg, msgWriteTimeout) + select { + case rw.out <- msg: + return nil + case <-time.After(msgWriteTimeout): + return newPeerError(errWrite, "messenger timeout") + } } func (rw *proto) EncodeMsg(code uint64, data ...interface{}) error { diff --git a/p2p/peer.go b/p2p/peer.go index c0cf1dee0c..a9a4cea591 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -64,9 +64,10 @@ type Peer struct { // The mutex protects the connection // so only one protocol can write at a time. - writeMu sync.Mutex - conn net.Conn - bufconn *bufio.ReadWriter + writeMu sync.Mutex + conn net.Conn + bufconn *bufio.ReadWriter + outgoingMsgC chan Msg // These fields maintain the running protocols. protocols []Protocol @@ -122,16 +123,17 @@ func newServerPeer(server *Server, conn net.Conn, dialAddr *peerAddr) *Peer { func newPeer(conn net.Conn, protocols []Protocol, dialAddr *peerAddr) *Peer { p := &Peer{ - Logger: logger.NewLogger("P2P " + conn.RemoteAddr().String()), - conn: conn, - dialAddr: dialAddr, - bufconn: bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)), - protocols: protocols, - running: make(map[string]*proto), - disc: make(chan DiscReason), - protoErr: make(chan error), - closed: make(chan struct{}), - cryptoReady: make(chan struct{}), + Logger: logger.NewLogger("P2P " + conn.RemoteAddr().String()), + conn: conn, + dialAddr: dialAddr, + bufconn: bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)), + protocols: protocols, + running: make(map[string]*proto), + disc: make(chan DiscReason), + protoErr: make(chan error), + closed: make(chan struct{}), + cryptoReady: make(chan struct{}), + outgoingMsgC: make(chan Msg), } return p } @@ -204,16 +206,6 @@ func (p *Peer) String() string { return fmt.Sprintf("Peer(%p %v %s)", p, p.conn.RemoteAddr(), kind) } -const ( - // maximum amount of time allowed for reading a message - msgReadTimeout = 5 * time.Second - // maximum amount of time allowed for writing a message - msgWriteTimeout = 5 * time.Second - // messages smaller than this many bytes will be read at - // once before passing them to a protocol. - wholePayloadSize = 64 * 1024 -) - var ( inactivityTimeout = 2 * time.Second disconnectGracePeriod = 2 * time.Second @@ -237,10 +229,11 @@ func (p *Peer) loop() (reason DiscReason, err error) { // read loop readMsg := make(chan Msg) - readErr := make(chan error) + rwErr := make(chan error) readNext := make(chan bool, 1) protoDone := make(chan struct{}, 1) - go readLoop(readMsg, readErr, readNext) + go readLoop(readMsg, rwErr, readNext) + go p.writeLoop(p.outgoingMsgC, rwErr) readNext <- true close(p.cryptoReady) @@ -269,8 +262,8 @@ loop: // we can continue reading from the socket. readNext <- true - case err := <-readErr: - // read failed. there is no need to run the + case err := <-rwErr: + // read or write failed. there is no need to run the // polite disconnect sequence because the connection // is probably dead anyway. // TODO: handle write errors as well @@ -284,8 +277,9 @@ loop: } // wait for read loop to return. + close(p.outgoingMsgC) close(readNext) - <-readErr + <-rwErr // tell the remote end to disconnect done := make(chan struct{}) go func() { @@ -391,9 +385,9 @@ outer: func (p *Peer) startProto(offset uint64, impl Protocol) *proto { rw := &proto{ in: make(chan Msg), + out: p.outgoingMsgC, offset: offset, maxcode: impl.Length, - peer: p, } p.protoWG.Add(1) go func() { diff --git a/p2p/peer_test.go b/p2p/peer_test.go index 4ee88f112b..814421cb82 100644 --- a/p2p/peer_test.go +++ b/p2p/peer_test.go @@ -33,6 +33,7 @@ func testPeer(protos []Protocol) (net.Conn, *Peer, <-chan error) { peer := newPeer(conn1, protos, nil) peer.ourID = &peerId{} peer.pubkeyHook = func(*peerAddr) error { return nil } + peer.outgoingMsgC = make(chan Msg) errc := make(chan error, 1) go func() { _, err := peer.loop()