Merge pull request #1325 from Roasbeef/gossip-syncer-fixes

discovery: ensure gossiper syncer has idempotent exit/start, fix deadlock

discovery/gossiper.go

@@ -525,7 +525,7 @@ type msgWithSenders struct {
 // with peers that we have an active gossipSyncer with. We do this to ensure
 // that we don't broadcast messages to any peers that we have active gossip
 // syncers for.
-func (m *msgWithSenders) mergeSyncerMap(syncers map[routing.Vertex]struct{}) {
+func (m *msgWithSenders) mergeSyncerMap(syncers map[routing.Vertex]*gossipSyncer) {
 	for peerPub := range syncers {
 		m.senders[peerPub] = struct{}{}
 	}
@@ -1130,9 +1130,9 @@ func (d *AuthenticatedGossiper) networkHandler() {
 			// syncers, we'll collect their pubkeys so we can avoid
 			// sending them the full message blast below.
 			d.syncerMtx.RLock()
-			syncerPeers := map[routing.Vertex]struct{}{}
-			for peerPub := range d.peerSyncers {
-				syncerPeers[peerPub] = struct{}{}
+			syncerPeers := make(map[routing.Vertex]*gossipSyncer)
+			for peerPub, syncer := range d.peerSyncers {
+				syncerPeers[peerPub] = syncer
 			}
 			d.syncerMtx.RUnlock()
 
@@ -1142,11 +1142,9 @@ func (d *AuthenticatedGossiper) networkHandler() {
 			// We'll first attempt to filter out this new message
 			// for all peers that have active gossip syncers
 			// active.
-			d.syncerMtx.RLock()
-			for _, syncer := range d.peerSyncers {
+			for _, syncer := range syncerPeers {
 				syncer.FilterGossipMsgs(announcementBatch...)
 			}
-			d.syncerMtx.RUnlock()
 
 			// Next, If we have new things to announce then
 			// broadcast them to all our immediately connected
@@ -1234,8 +1232,7 @@ func (d *AuthenticatedGossiper) PruneSyncState(peer *btcec.PublicKey) {
 		peer.SerializeCompressed())
 
 	vertex := routing.NewVertex(peer)
-
-	syncer, ok := d.peerSyncers[routing.NewVertex(peer)]
+	syncer, ok := d.peerSyncers[vertex]
 	if !ok {
 		return
 	}

discovery/syncer.go

@@ -175,6 +175,9 @@ type gossipSyncerCfg struct {
 //
 // TODO(roasbeef): modify to only sync from one peer at a time?
 type gossipSyncer struct {
+	started uint32
+	stopped uint32
+
 	// remoteUpdateHorizon is the update horizon of the remote peer. We'll
 	// use this to properly filter out any messages.
 	remoteUpdateHorizon *lnwire.GossipTimestampRange
@@ -226,6 +229,10 @@ func newGossiperSyncer(cfg gossipSyncerCfg) *gossipSyncer {
 // Start starts the gossipSyncer and any goroutines that it needs to carry out
 // its duties.
 func (g *gossipSyncer) Start() error {
+	if !atomic.CompareAndSwapUint32(&g.started, 0, 1) {
+		return nil
+	}
+
 	log.Debugf("Starting gossipSyncer(%x)", g.peerPub[:])
 
 	g.wg.Add(1)
@@ -237,6 +244,10 @@ func (g *gossipSyncer) Start() error {
 // Stop signals the gossipSyncer for a graceful exit, then waits until it has
 // exited.
 func (g *gossipSyncer) Stop() error {
+	if !atomic.CompareAndSwapUint32(&g.stopped, 0, 1) {
+		return nil
+	}
+
 	close(g.quit)
 
 	g.wg.Wait()
@@ -794,6 +805,12 @@ func (g *gossipSyncer) FilterGossipMsgs(msgs ...msgWithSenders) {
 		return
 	}
 
+	// If we've been signalled to exit, or are exiting, then we'll stop
+	// short.
+	if atomic.LoadUint32(&g.stopped) == 1 {
+		return
+	}
+
 	// TODO(roasbeef): need to ensure that peer still online...send msg to
 	// gossiper on peer termination to signal peer disconnect?