peer: add async queue for gossiper msgs

This commit refactors the core logic of the
  chanMsgStream to support an additional stream
  that is used to asynchronously queue for in-order
  delivery to the authenticated gossiper. The channel
  streams are slightly adapted to use the more flexible
  primitive. We may look to refactor this using more
  isolated interfaces, but for now this provides a
  minimal change to resolving known flakes.

peer.go

@@ -473,40 +473,38 @@ func (p *peer) readNextMessage() (lnwire.Message, error) {
 	return nextMsg, nil
 }
 
-// chanMsgStream implements a goroutine-safe, in-order stream of messages to be
+// msgStream implements a goroutine-safe, in-order stream of messages to be
-// delivered to an active channel. These messages MUST be in order due to the
+// delivered via closure to a receiver. These messages MUST be in order due to
-// nature of the lightning channel commitment state machine. We utilize
+// the nature of the lightning channel commitment and gossiper state machines.
-// additional synchronization with the fundingManager to ensure we don't
+// TODO(conner): use stream handler interface to abstract out stream
-// attempt to dispatch a message to a channel before it is fully active.
+// state/logging
-type chanMsgStream struct {
+type msgStream struct {
-	fundingMgr *fundingManager
-	htlcSwitch *htlcswitch.Switch
-
-	cid lnwire.ChannelID
-
 	peer *peer
 
+	apply func(lnwire.Message)
+
+	startMsg string
+	stopMsg  string
+
 	msgCond *sync.Cond
 	msgs    []lnwire.Message
 
-	chanLink htlcswitch.ChannelLink
-
 	mtx sync.Mutex
 
 	wg   sync.WaitGroup
 	quit chan struct{}
 }
 
-// newChanMsgStream creates a new instance of a chanMsgStream for a particular
+// newMsgStream creates a new instance of a chanMsgStream for a particular
 // channel identified by its channel ID.
-func newChanMsgStream(f *fundingManager, h *htlcswitch.Switch, p *peer,
+func newMsgStream(p *peer, startMsg, stopMsg string,
-	c lnwire.ChannelID) *chanMsgStream {
+	apply func(lnwire.Message)) *msgStream {
 
-	stream := &chanMsgStream{
+	stream := &msgStream{
-		fundingMgr: f,
-		htlcSwitch: h,
 		peer:     p,
-		cid:        c,
+		apply:    apply,
+		startMsg: startMsg,
+		stopMsg:  stopMsg,
 		quit:     make(chan struct{}),
 	}
 	stream.msgCond = sync.NewCond(&stream.mtx)
@@ -515,45 +513,44 @@ func newChanMsgStream(f *fundingManager, h *htlcswitch.Switch, p *peer,
 }
 
 // Start starts the chanMsgStream.
-func (c *chanMsgStream) Start() {
+func (ms *msgStream) Start() {
-	c.wg.Add(1)
+	ms.wg.Add(1)
-	go c.msgConsumer()
+	go ms.msgConsumer()
 }
 
 // Stop stops the chanMsgStream.
-func (c *chanMsgStream) Stop() {
+func (ms *msgStream) Stop() {
 	// TODO(roasbeef): signal too?
 
-	close(c.quit)
+	close(ms.quit)
 
 	// Wake up the msgConsumer is we've been signalled to exit.
-	c.msgCond.Signal()
+	ms.msgCond.Signal()
 
-	c.wg.Wait()
+	ms.wg.Wait()
 }
 
 // msgConsumer is the main goroutine that streams messages from the peer's
 // readHandler directly to the target channel.
-func (c *chanMsgStream) msgConsumer() {
+func (ms *msgStream) msgConsumer() {
-	defer c.wg.Done()
+	defer ms.wg.Done()
+	defer peerLog.Tracef(ms.stopMsg)
 
-	peerLog.Tracef("Update stream for ChannelID(%x) created", c.cid[:])
+	peerLog.Tracef(ms.startMsg)
 
 	for {
 		// First, we'll check our condition. If the queue of messages
 		// is empty, then we'll wait until a new item is added.
-		c.msgCond.L.Lock()
+		ms.msgCond.L.Lock()
-		for len(c.msgs) == 0 {
+		for len(ms.msgs) == 0 {
-			c.msgCond.Wait()
+			ms.msgCond.Wait()
 
 			// If we were woke up in order to exit, then we'll do
 			// so. Otherwise, we'll check the message queue for any
 			// new items.
 			select {
-			case <-c.quit:
+			case <-ms.quit:
-				peerLog.Tracef("Update stream for "+
+				ms.msgCond.L.Unlock()
-					"ChannelID(%x) exiting", c.cid[:])
-				c.msgCond.L.Unlock()
 				return
 			default:
 			}
@@ -562,48 +559,81 @@ func (c *chanMsgStream) msgConsumer() {
 		// Grab the message off the front of the queue, shifting the
 		// slice's reference down one in order to remove the message
 		// from the queue.
-		msg := c.msgs[0]
+		msg := ms.msgs[0]
-		c.msgs[0] = nil // Set to nil to prevent GC leak.
+		ms.msgs[0] = nil // Set to nil to prevent GC leak.
-		c.msgs = c.msgs[1:]
+		ms.msgs = ms.msgs[1:]
 
-		c.msgCond.L.Unlock()
+		ms.msgCond.L.Unlock()
 
-		// We'll send a message to the funding manager and wait iff an
+		ms.apply(msg)
-		// active funding process for this channel hasn't yet
-		// completed. We do this in order to account for the following
-		// scenario: we send the funding locked message to the other
-		// side, they immediately send a channel update message, but we
-		// haven't yet sent the channel to the channelManager.
-		c.fundingMgr.waitUntilChannelOpen(c.cid)
-
-		// Dispatch the commitment update message to the proper active
-		// goroutine dedicated to this channel.
-		if c.chanLink == nil {
-			link, err := c.htlcSwitch.GetLink(c.cid)
-			if err != nil {
-				peerLog.Errorf("recv'd update for unknown "+
-					"channel %v from %v", c.cid, c.peer)
-				continue
-			}
-			c.chanLink = link
-		}
-
-		c.chanLink.HandleChannelUpdate(msg)
 	}
 }
 
-// AddMsg adds a new message to the chanMsgStream. This function is safe for
+// AddMsg adds a new message to the msgStream. This function is safe for
 // concurrent access.
-func (c *chanMsgStream) AddMsg(msg lnwire.Message) {
+func (ms *msgStream) AddMsg(msg lnwire.Message) {
 	// First, we'll lock the condition, and add the message to the end of
 	// the message queue.
-	c.msgCond.L.Lock()
+	ms.msgCond.L.Lock()
-	c.msgs = append(c.msgs, msg)
+	ms.msgs = append(ms.msgs, msg)
-	c.msgCond.L.Unlock()
+	ms.msgCond.L.Unlock()
 
 	// With the message added, we signal to the msgConsumer that there are
 	// additional messages to consume.
-	c.msgCond.Signal()
+	ms.msgCond.Signal()
+}
+
+// newChanMsgStream is used to create a msgStream between the peer and
+// particular channel link in the htlcswitch.  We utilize additional
+// synchronization with the fundingManager to ensure we don't attempt to
+// dispatch a message to a channel before it is fully active. A reference to the
+// channel this stream forwards to his held in scope to prevent unnecessary
+// lookups.
+func newChanMsgStream(p *peer, cid lnwire.ChannelID) *msgStream {
+
+	var chanLink htlcswitch.ChannelLink
+
+	return newMsgStream(p,
+		fmt.Sprintf("Update stream for ChannelID(%x) created", cid),
+		fmt.Sprintf("Update stream for ChannelID(%x) exiting", cid),
+		func(msg lnwire.Message) {
+			// We'll send a message to the funding manager and wait iff an
+			// active funding process for this channel hasn't yet completed.
+			// We do this in order to account for the following scenario: we
+			// send the funding locked message to the other side, they
+			// immediately send a channel update message, but we haven't yet
+			// sent the channel to the channelManager.
+			p.server.fundingMgr.waitUntilChannelOpen(cid)
+
+			// Dispatch the commitment update message to the proper active
+			// goroutine dedicated to this channel.
+			if chanLink == nil {
+				link, err := p.server.htlcSwitch.GetLink(cid)
+				if err != nil {
+					peerLog.Errorf("recv'd update for unknown "+
+						"channel %v from %v", cid, p)
+					return
+				}
+				chanLink = link
+			}
+
+			chanLink.HandleChannelUpdate(msg)
+		},
+	)
+}
+
+// newDiscMsgStream is used to setup a msgStream between the peer and the
+// authenticated gossiper. This stream should be used to forward all remote
+// channel announcements.
+func newDiscMsgStream(p *peer) *msgStream {
+	return newMsgStream(p,
+		"Update stream for gossiper created",
+		"Update stream for gossiper exited",
+		func(msg lnwire.Message) {
+			p.server.authGossiper.ProcessRemoteAnnouncement(msg,
+				p.addr.IdentityKey)
+		},
+	)
 }
 
 // readHandler is responsible for reading messages off the wire in series, then
@@ -620,7 +650,11 @@ func (p *peer) readHandler() {
 		p.Disconnect(err)
 	})
 
-	chanMsgStreams := make(map[lnwire.ChannelID]*chanMsgStream)
+	discStream := newDiscMsgStream(p)
+	discStream.Start()
+	defer discStream.Stop()
+
+	chanMsgStreams := make(map[lnwire.ChannelID]*msgStream)
 out:
 	for atomic.LoadInt32(&p.disconnect) == 0 {
 		nextMsg, err := p.readNextMessage()
@@ -717,8 +751,8 @@ out:
 			*lnwire.NodeAnnouncement,
 			*lnwire.AnnounceSignatures:
 
-			p.server.authGossiper.ProcessRemoteAnnouncement(msg,
+			discStream.AddMsg(msg)
-				p.addr.IdentityKey)
+
 		default:
 			peerLog.Errorf("unknown message %v received from peer "+
 				"%v", uint16(msg.MsgType()), p)
@@ -732,8 +766,7 @@ out:
 				// If a stream hasn't yet been created, then
 				// we'll do so, add it to the map, and finally
 				// start it.
-				chanStream = newChanMsgStream(p.server.fundingMgr,
+				chanStream = newChanMsgStream(p, targetChan)
-					p.server.htlcSwitch, p, targetChan)
 				chanMsgStreams[targetChan] = chanStream
 				chanStream.Start()
 			}