diff --git a/p2p/discover/node.go b/p2p/discover/node.go index 99cb549a5947532d9e4de39e2f37ad33a234f0c6..6662a6cb7b85fb0e505777d5dc4de38ac755b14b 100644 --- a/p2p/discover/node.go +++ b/p2p/discover/node.go @@ -14,8 +14,6 @@ import ( "strconv" "strings" "sync" - "sync/atomic" - "time" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto/secp256k1" @@ -31,9 +29,6 @@ type Node struct { DiscPort int // UDP listening port for discovery protocol TCPPort int // TCP listening port for RLPx - - // this must be set/read using atomic load and store. - activeStamp int64 } func newNode(id NodeID, addr *net.UDPAddr) *Node { @@ -50,16 +45,6 @@ func (n *Node) isValid() bool { return !n.IP.IsMulticast() && !n.IP.IsUnspecified() && n.TCPPort != 0 && n.DiscPort != 0 } -func (n *Node) bumpActive() { - stamp := time.Now().Unix() - atomic.StoreInt64(&n.activeStamp, stamp) -} - -func (n *Node) active() time.Time { - stamp := atomic.LoadInt64(&n.activeStamp) - return time.Unix(stamp, 0) -} - func (n *Node) addr() *net.UDPAddr { return &net.UDPAddr{IP: n.IP, Port: n.DiscPort} } diff --git a/p2p/discover/table.go b/p2p/discover/table.go index dbf86c0840aa6e2e66a32efd5b5bbe0e4e3c6d25..e2e8464569534ab99f326e652ad0b1cae80d2514 100644 --- a/p2p/discover/table.go +++ b/p2p/discover/table.go @@ -326,7 +326,6 @@ outer: func (b *bucket) bump(n *Node) bool { for i := range b.entries { if b.entries[i].ID == n.ID { - n.bumpActive() // move it to the front copy(b.entries[1:], b.entries[:i]) b.entries[0] = n diff --git a/p2p/discover/udp.go b/p2p/discover/udp.go index d37260e7d769ade08268b379bedb5b346ed575f0..61a0abed995904e0dd92b9ab42b73b944d80dc34 100644 --- a/p2p/discover/udp.go +++ b/p2p/discover/udp.go @@ -267,11 +267,12 @@ func (t *udp) loop() { defer timeout.Stop() rearmTimeout := func() { - if len(pending) == 0 || nextDeadline == pending[0].deadline { + now := time.Now() + if len(pending) == 0 || now.Before(nextDeadline) { return } nextDeadline = pending[0].deadline - timeout.Reset(nextDeadline.Sub(time.Now())) + timeout.Reset(nextDeadline.Sub(now)) } for { diff --git a/p2p/handshake.go b/p2p/handshake.go index 43361364fa6b820766e3235001beda4cfbd23921..79395f23ffe4dddbaba578abb45797dafb43cbd7 100644 --- a/p2p/handshake.go +++ b/p2p/handshake.go @@ -115,7 +115,7 @@ func setupOutboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, // returning the handshake read error. If the remote side // disconnects us early with a valid reason, we should return it // as the error so it can be tracked elsewhere. - werr := make(chan error) + werr := make(chan error, 1) go func() { werr <- Send(rw, handshakeMsg, our) }() rhs, err := readProtocolHandshake(rw, secrets.RemoteID, our) if err != nil { diff --git a/p2p/peer.go b/p2p/peer.go index 7bc4f9cf6bb3b4ec86e5dac1d03304786fa02245..1262ba64a7cbf3e26e2dcbe5fa796d9fe217e2b6 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -4,7 +4,6 @@ import ( "errors" "fmt" "io" - "io/ioutil" "net" "sort" "sync" @@ -20,8 +19,7 @@ const ( baseProtocolLength = uint64(16) baseProtocolMaxMsgSize = 10 * 1024 * 1024 - pingInterval = 15 * time.Second - disconnectGracePeriod = 2 * time.Second + pingInterval = 15 * time.Second ) const ( @@ -129,39 +127,27 @@ func (p *Peer) run() DiscReason { case err := <-readErr: if r, ok := err.(DiscReason); ok { reason = r - break + } else { + // Note: We rely on protocols to abort if there is a write + // error. It might be more robust to handle them here as well. + p.DebugDetailf("Read error: %v\n", err) + reason = DiscNetworkError } - // Note: We rely on protocols to abort if there is a write - // error. It might be more robust to handle them here as well. - p.DebugDetailf("Read error: %v\n", err) - p.conn.Close() - reason = DiscNetworkError case err := <-p.protoErr: reason = discReasonForError(err) case reason = <-p.disc: } close(p.closed) + p.politeDisconnect(reason) p.wg.Wait() - if reason != DiscNetworkError { - p.politeDisconnect(reason) - } p.Debugf("Disconnected: %v\n", reason) return reason } func (p *Peer) politeDisconnect(reason DiscReason) { - done := make(chan struct{}) - go func() { + if reason != DiscNetworkError { SendItems(p.rw, discMsg, uint(reason)) - // Wait for the other side to close the connection. - // Discard any data that they send until then. - io.Copy(ioutil.Discard, p.conn) - close(done) - }() - select { - case <-done: - case <-time.After(disconnectGracePeriod): } p.conn.Close() } diff --git a/p2p/server.go b/p2p/server.go index 5cd3dc2adf60a5072084332de26393866e7f9970..61e0d71e93ff93eda94d6731a69e3e00b6237ba5 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -260,9 +260,11 @@ func (srv *Server) Stop() { // No new peers can be added at this point because dialLoop and // listenLoop are down. It is safe to call peerWG.Wait because // peerWG.Add is not called outside of those loops. + srv.lock.Lock() for _, peer := range srv.peers { peer.Disconnect(DiscQuitting) } + srv.lock.Unlock() srv.peerWG.Wait() }