From de2686ed1033dc0da7860e5850dc7d6f31dda2b0 Mon Sep 17 00:00:00 2001 From: Guillaume Ballet Date: Mon, 16 Apr 2018 12:58:46 +0200 Subject: [PATCH] whisper: activate write deadline This imposes to cleanup peers from the server as the deadline occurs. As a result, I decided to make the `update` function dependent on the type of peer; the libp2p version will try to remove the peers as itreceives the shutdown signal from a peer. Note that the read deadlines triggers no matter what, and so wnode stops working. --- whisper/whisperv6/devp2p_glue.go | 76 ++++++++++++++++++++++++++++++++ whisper/whisperv6/libp2p_glue.go | 58 +++++++++++++++++++++--- whisper/whisperv6/peer.go | 75 ------------------------------- 3 files changed, 129 insertions(+), 80 deletions(-) diff --git a/whisper/whisperv6/devp2p_glue.go b/whisper/whisperv6/devp2p_glue.go index 8625942ffa..56106481fe 100644 --- a/whisper/whisperv6/devp2p_glue.go +++ b/whisper/whisperv6/devp2p_glue.go @@ -17,9 +17,85 @@ package whisperv6 import ( + "fmt" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" + set "gopkg.in/fatih/set.v0" + "time" ) +// DevP2PPeer is the DevP2P implementation of the Peer interface +type DevP2PPeer struct { + *PeerBase + + peer *p2p.Peer +} + +// newPeer creates a new whisper peer object, but does not run the handshake itself. +func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer { + return &DevP2PPeer{ + &PeerBase{ + host: host, + ws: rw, + trusted: false, + powRequirement: 0.0, + known: set.New(), + quit: make(chan struct{}), + bloomFilter: makeFullNodeBloom(), + fullNode: true, + }, + remote, + } +} + +// handshake sends the protocol initiation status message to the remote peer and +// verifies the remote status too. +func (peer *DevP2PPeer) handshake() error { + err := peer.handshakeBase() + if err != nil { + return fmt.Errorf("peer [%x] %s", peer.ID(), err.Error()) + } + return nil +} + +// start initiates the peer updater, periodically broadcasting the whisper packets +// into the network. +func (peer *DevP2PPeer) start() { + go peer.update() + log.Trace("start", "peer", peer.ID()) +} + +// stop terminates the peer updater, stopping message forwarding to it. +func (peer *DevP2PPeer) stop() { + close(peer.quit) + log.Trace("stop", "peer", peer.ID()) +} + +// update executes periodic operations on the peer, including message transmission +// and expiration. +func (peer *DevP2PPeer) update() { + // Start the tickers for the updates + expire := time.NewTicker(expirationCycle) + transmit := time.NewTicker(transmissionCycle) + + // Loop and transmit until termination is requested + for { + select { + case <-expire.C: + peer.expire() + + case <-transmit.C: + if err := peer.broadcast(); err != nil { + log.Trace("broadcast failed", "reason", err) + return + } + + case <-peer.quit: + return + } + } +} + // DevP2PWhisperServer implements WhisperServer with a DevP2P backend type DevP2PWhisperServer struct { Server *p2p.Server diff --git a/whisper/whisperv6/libp2p_glue.go b/whisper/whisperv6/libp2p_glue.go index 5f7fac9b5b..450708fb29 100644 --- a/whisper/whisperv6/libp2p_glue.go +++ b/whisper/whisperv6/libp2p_glue.go @@ -23,6 +23,7 @@ import ( "fmt" "github.com/ethereum/go-ethereum/log" "io" + "time" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" @@ -77,8 +78,6 @@ func (stream *LibP2PStream) ReadMsg() (p2p.Msg, error) { // WriteMsg implements the MsgReadWriter interface to write messages // to lilbp2p streams. func (stream *LibP2PStream) WriteMsg(msg p2p.Msg) error { - - stream.lp2pStream.SetWriteDeadline(time.Now().Add(transmissionCycle)) if err := rlp.Encode(stream.lp2pStream, msg.Code); err != nil { @@ -94,10 +93,12 @@ type LibP2PPeer struct { id peer.ID + server *LibP2PWhisperServer + connectionStream *LibP2PStream } -func newLibP2PPeer(w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer { +func newLibP2PPeer(s *LibP2PWhisperServer, w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer { return &LibP2PPeer{ &PeerBase{ host: w, @@ -110,6 +111,7 @@ func newLibP2PPeer(w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer { fullNode: true, }, pid, + s, nil, } } @@ -127,6 +129,52 @@ func (p *LibP2PPeer) handshake() error { return nil } +// start initiates the peer updater, periodically broadcasting the whisper packets +// into the network. +func (p *LibP2PPeer) start() { + go p.update() + log.Trace("start", "peer", p.ID()) +} + +// stop terminates the peer updater, stopping message forwarding to it. +func (p *LibP2PPeer) stop() { + close(p.quit) + fmt.Println("stop", "peer", p.ID()) +} + +// update executes periodic operations on the peer, including message transmission +// and expiration. +func (p *LibP2PPeer) update() { + // Start the tickers for the updates + expire := time.NewTicker(expirationCycle) + transmit := time.NewTicker(transmissionCycle) + + // Loop and transmit until termination is requested +updateLoop: + for { + select { + case <-expire.C: + p.expire() + + case <-transmit.C: + if err := p.broadcast(); err != nil { + break updateLoop + } + + case <-p.quit: + break updateLoop + } + } + + // Cleanup and remove the peer from the list + for i, it := range p.server.Peers { + if it.id == p.id { + p.server.Peers = append(p.server.Peers[:i], p.server.Peers[i+1:]...) + break + } + } +} + // LibP2PWhisperServer implements WhisperServer for libp2p. type LibP2PWhisperServer struct { Host host.Host @@ -177,7 +225,7 @@ func (server *LibP2PWhisperServer) Start() error { // Unknown peer if peer == nil { - peer = newLibP2PPeer(server.whisper, pid, lps) + peer = newLibP2PPeer(server, server.whisper, pid, lps) // TODO check critical section server.Peers = append(server.Peers, peer.(*LibP2PPeer)) } @@ -236,7 +284,7 @@ func (server *LibP2PWhisperServer) AddPeer(addr ma.Multiaddr) *LibP2PPeer { ipfsaddrpart, _ := ma.NewMultiaddr(fmt.Sprintf("/ipfs/%s", pid)) ipaddr := addr.Decapsulate(ipfsaddrpart) server.Host.Peerstore().AddAddr(peerid, ipaddr, pstore.PermanentAddrTTL) - newPeer := newLibP2PPeer(server.whisper, peerid, nil).(*LibP2PPeer) + newPeer := newLibP2PPeer(server, server.whisper, peerid, nil).(*LibP2PPeer) server.Peers = append(server.Peers, newPeer) return newPeer diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go index 52ed6e7d71..57abbe4afa 100644 --- a/whisper/whisperv6/peer.go +++ b/whisper/whisperv6/peer.go @@ -52,7 +52,6 @@ type Peer interface { start() stop() handshake() error - update() mark(*Envelope) marked(*Envelope) bool expire() @@ -65,55 +64,6 @@ type Peer interface { setTrusted(bool) setPoWRequirement(float64) stream() p2p.MsgReadWriter - - // newPeer(*Whisper, p2p.MsgReadWriter) Peer -} - -// DevP2PPeer is the DevP2P implementation of the Peer interface -type DevP2PPeer struct { - *PeerBase - - peer *p2p.Peer -} - -// newPeer creates a new whisper peer object, but does not run the handshake itself. -func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer { - return &DevP2PPeer{ - &PeerBase{ - host: host, - ws: rw, - trusted: false, - powRequirement: 0.0, - known: set.New(), - quit: make(chan struct{}), - bloomFilter: MakeFullNodeBloom(), - fullNode: true, - }, - remote, - } -} - -// start initiates the peer updater, periodically broadcasting the whisper packets -// into the network. -func (peer *PeerBase) start() { - go peer.update() - // log.Trace("start", "peer", peer.ID()) -} - -// stop terminates the peer updater, stopping message forwarding to it. -func (peer *PeerBase) stop() { - close(peer.quit) - // log.Trace("stop", "peer", peer.ID()) -} - -// handshake sends the protocol initiation status message to the remote peer and -// verifies the remote status too. -func (peer *DevP2PPeer) handshake() error { - err := peer.handshakeBase() - if err != nil { - return fmt.Errorf("peer [%x] %s", peer.ID(), err.Error()) - } - return nil } func (peer *PeerBase) handshakeBase() error { @@ -173,31 +123,6 @@ func (peer *PeerBase) handshakeBase() error { return nil } -// update executes periodic operations on the peer, including message transmission -// and expiration. -func (peer *DevP2PPeer) update() { - // Start the tickers for the updates - expire := time.NewTicker(expirationCycle) - transmit := time.NewTicker(transmissionCycle) - - // Loop and transmit until termination is requested - for { - select { - case <-expire.C: - peer.expire() - - case <-transmit.C: - if err := peer.broadcast(); err != nil { - log.Trace("broadcast failed", "reason", err, "peer", peer.ID()) - return - } - - case <-peer.quit: - return - } - } -} - // mark marks an envelope known to the peer so that it won't be sent back. func (peer *PeerBase) mark(envelope *Envelope) { peer.known.Add(envelope.Hash())