From c5732f93740d1d1869997fe1580cc6d1bbd62a49 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Sat, 19 Jul 2014 16:40:49 -0700 Subject: [PATCH] Moved PEX logic to PeerManager where it belongs. --- common/throttler.go | 6 ++++-- main.go | 52 ++++++--------------------------------------- p2p/addrbook.go | 14 +++++++++--- p2p/peer_manager.go | 45 ++++++++++++++++++++++++++++++++------- 4 files changed, 59 insertions(+), 58 deletions(-) diff --git a/common/throttler.go b/common/throttler.go index 5a378ec4a..2076ddf1e 100644 --- a/common/throttler.go +++ b/common/throttler.go @@ -6,8 +6,10 @@ import ( ) /* -Throttler sends a struct{}{} to .Ch "dur" after the last .Set(). -It's good for ensuring that something happens last after a burst of events. +Throttler fires an event at most "dur" after each .Set() call. +If a short burst of .Set() calls happens, Throttler fires once. +If a long continuous burst of .Set() calls happens, Throttler fires +at most once every "dur". */ type Throttler struct { Ch chan struct{} diff --git a/main.go b/main.go index fa7a3ba2c..4f29d10a0 100644 --- a/main.go +++ b/main.go @@ -9,11 +9,10 @@ import ( ) type Node struct { - lz []p2p.Listener - sw *p2p.Switch - swEvents chan interface{} - book *p2p.AddrBook - pmgr *p2p.PeerManager + lz []p2p.Listener + sw *p2p.Switch + book *p2p.AddrBook + pmgr *p2p.PeerManager } func NewNode() *Node { @@ -41,16 +40,13 @@ func NewNode() *Node { }, } sw := p2p.NewSwitch(chDescs) - swEvents := make(chan interface{}) - sw.AddEventListener("Node.swEvents", swEvents) book := p2p.NewAddrBook(config.RootDir + "/addrbook.json") pmgr := p2p.NewPeerManager(sw, book) return &Node{ - sw: sw, - swEvents: swEvents, - book: book, - pmgr: pmgr, + sw: sw, + book: book, + pmgr: pmgr, } } @@ -59,7 +55,6 @@ func (n *Node) Start() { for _, l := range n.lz { go n.inboundConnectionHandler(l) } - go n.switchEventsHandler() n.sw.Start() n.book.Start() n.pmgr.Start() @@ -69,7 +64,6 @@ func (n *Node) Stop() { log.Info("Stopping node") // TODO: gracefully disconnect from peers. n.sw.Stop() - close(n.swEvents) n.book.Stop() n.pmgr.Stop() } @@ -102,38 +96,6 @@ func (n *Node) inboundConnectionHandler(l p2p.Listener) { // cleanup } -func (n *Node) switchEventsHandler() { - for { - swEvent, ok := <-n.swEvents - if !ok { - break - } - switch swEvent.(type) { - case p2p.SwitchEventNewPeer: - event := swEvent.(p2p.SwitchEventNewPeer) - if event.Peer.IsOutbound() { - n.sendOurExternalAddrs(event.Peer) - if n.book.NeedMoreAddrs() { - n.pmgr.RequestPEX(event.Peer) - } - } - case p2p.SwitchEventDonePeer: - // TODO - } - } -} - -func (n *Node) sendOurExternalAddrs(peer *p2p.Peer) { - // Send listener our external address(es) - addrs := []*p2p.NetAddress{} - for _, l := range n.lz { - addrs = append(addrs, l.ExternalAddress()) - } - n.pmgr.SendAddrs(peer, addrs) - // On the remote end, the pexHandler may choose - // to add these to its book. -} - //----------------------------------------------------------------------------- func main() { diff --git a/p2p/addrbook.go b/p2p/addrbook.go index 70eab54ed..592e77592 100644 --- a/p2p/addrbook.go +++ b/p2p/addrbook.go @@ -80,7 +80,7 @@ type AddrBook struct { mtx sync.Mutex rand *rand.Rand key string - ourAddrs map[string]struct{} + ourAddrs map[string]*NetAddress addrLookup map[string]*knownAddress // new & old addrNew []map[string]*knownAddress addrOld []map[string]*knownAddress @@ -101,7 +101,7 @@ const ( func NewAddrBook(filePath string) *AddrBook { am := AddrBook{ rand: rand.New(rand.NewSource(time.Now().UnixNano())), - ourAddrs: make(map[string]struct{}), + ourAddrs: make(map[string]*NetAddress), addrLookup: make(map[string]*knownAddress), quit: make(chan struct{}), filePath: filePath, @@ -145,7 +145,15 @@ func (a *AddrBook) Stop() { func (a *AddrBook) AddOurAddress(addr *NetAddress) { a.mtx.Lock() defer a.mtx.Unlock() - a.ourAddrs[addr.String()] = struct{}{} + a.ourAddrs[addr.String()] = addr +} + +func (a *AddrBook) OurAddresses() []*NetAddress { + addrs := []*NetAddress{} + for _, addr := range a.ourAddrs { + addrs = append(addrs, addr) + } + return addrs } func (a *AddrBook) AddAddress(addr *NetAddress, src *NetAddress) { diff --git a/p2p/peer_manager.go b/p2p/peer_manager.go index b8bcbc758..d331a230c 100644 --- a/p2p/peer_manager.go +++ b/p2p/peer_manager.go @@ -26,18 +26,22 @@ PeerManager handles PEX (peer exchange) and ensures that an adequate number of peers are connected to the switch. */ type PeerManager struct { - sw *Switch - book *AddrBook - quit chan struct{} - started uint32 - stopped uint32 + sw *Switch + swEvents chan interface{} + book *AddrBook + quit chan struct{} + started uint32 + stopped uint32 } func NewPeerManager(sw *Switch, book *AddrBook) *PeerManager { + swEvents := make(chan interface{}) + sw.AddEventListener("PeerManager.swEvents", swEvents) pm := &PeerManager{ - sw: sw, - book: book, - quit: make(chan struct{}), + sw: sw, + swEvents: swEvents, + book: book, + quit: make(chan struct{}), } return pm } @@ -45,6 +49,7 @@ func NewPeerManager(sw *Switch, book *AddrBook) *PeerManager { func (pm *PeerManager) Start() { if atomic.CompareAndSwapUint32(&pm.started, 0, 1) { log.Info("Starting PeerManager") + go pm.switchEventsHandler() go pm.ensurePeersHandler() go pm.pexHandler() } @@ -54,6 +59,7 @@ func (pm *PeerManager) Stop() { if atomic.CompareAndSwapUint32(&pm.stopped, 0, 1) { log.Info("Stopping PeerManager") close(pm.quit) + close(pm.swEvents) } } @@ -70,6 +76,29 @@ func (pm *PeerManager) SendAddrs(peer *Peer, addrs []*NetAddress) { peer.Send(NewPacket(pexCh, tm)) } +// For new outbound peers, announce our listener addresses if any, +// and if .book needs more addresses, ask for them. +func (pm *PeerManager) switchEventsHandler() { + for { + swEvent, ok := <-pm.swEvents + if !ok { + break + } + switch swEvent.(type) { + case SwitchEventNewPeer: + event := swEvent.(SwitchEventNewPeer) + if event.Peer.IsOutbound() { + pm.SendAddrs(event.Peer, pm.book.OurAddresses()) + if pm.book.NeedMoreAddrs() { + pm.RequestPEX(event.Peer) + } + } + case SwitchEventDonePeer: + // TODO + } + } +} + // Ensures that sufficient peers are connected. (continuous) func (pm *PeerManager) ensurePeersHandler() { // fire once immediately.