From b0b2475660a1561f8930847646dd60bf457a6b2d Mon Sep 17 00:00:00 2001 From: Conner Fromknecht Date: Thu, 1 Nov 2018 17:08:24 -0700 Subject: [PATCH 1/3] peer: remove quit chan from AddMsg signature --- peer.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/peer.go b/peer.go index e95f30ee1..63e526210 100644 --- a/peer.go +++ b/peer.go @@ -793,7 +793,7 @@ func (ms *msgStream) msgConsumer() { // AddMsg adds a new message to the msgStream. This function is safe for // concurrent access. -func (ms *msgStream) AddMsg(msg lnwire.Message, quit chan struct{}) { +func (ms *msgStream) AddMsg(msg lnwire.Message) { // First, we'll attempt to receive from the producerSema struct. This // acts as a sempahore to prevent us from indefinitely buffering // incoming items from the wire. Either the msg queue isn't full, and @@ -801,7 +801,7 @@ func (ms *msgStream) AddMsg(msg lnwire.Message, quit chan struct{}) { // we're signalled to quit, or a slot is freed up. select { case <-ms.producerSema: - case <-quit: + case <-ms.peer.quit: return case <-ms.quit: return @@ -1019,7 +1019,7 @@ out: // forward the error to all channels with this peer. case msg.ChanID == lnwire.ConnectionWideID: for chanID, chanStream := range chanMsgStreams { - chanStream.AddMsg(nextMsg, p.quit) + chanStream.AddMsg(nextMsg) // Also marked this channel as failed, // so we won't try to restart it on @@ -1081,7 +1081,7 @@ out: *lnwire.ReplyChannelRange, *lnwire.ReplyShortChanIDsEnd: - discStream.AddMsg(msg, p.quit) + discStream.AddMsg(msg) default: peerLog.Errorf("unknown message %v received from peer "+ @@ -1104,7 +1104,7 @@ out: // With the stream obtained, add the message to the // stream so we can continue processing message. - chanStream.AddMsg(nextMsg, p.quit) + chanStream.AddMsg(nextMsg) } idleTimer.Reset(idleTimeout) From 96c47f7de4835ffc1fbb74d3dc207fef93658c7d Mon Sep 17 00:00:00 2001 From: Conner Fromknecht Date: Thu, 1 Nov 2018 17:13:13 -0700 Subject: [PATCH 2/3] discovery/gossiper: bypass main event loop for queries This commit restructures the delivery of gossip query related messages, such that they are delivered directly to the gossip syncers. Gossip query rate limiting was introduced in #1824 on a per-peer basis. However, since all gossip query messages were being delivered in the main event loop, the end result is that one rate-limited peer could stall all other peers. In addition, since no other peers would be able to submit gossip-related messages through the blocked event loop, the back pressure would eventually rate limit the read handlers of all peers as well. The end result would be lengthy delays in reading messages related to htlc forwarding. The fix is to lift the delivery of gossip query messages outside of the main event loop. With this change, the rate limiting backpressure is delivered only to the intended peer. --- discovery/gossiper.go | 104 +++++++++++++++++++++++------------------- 1 file changed, 57 insertions(+), 47 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 88fd401aa..8257c2fc5 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -456,12 +456,67 @@ func (d *AuthenticatedGossiper) Stop() { func (d *AuthenticatedGossiper) ProcessRemoteAnnouncement(msg lnwire.Message, peer lnpeer.Peer) chan error { + errChan := make(chan error, 1) + + // For messages in the known set of channel series queries, we'll + // dispatch the message directly to the gossipSyncer, and skip the main + // processing loop. + switch m := msg.(type) { + case *lnwire.QueryShortChanIDs, + *lnwire.QueryChannelRange, + *lnwire.ReplyChannelRange, + *lnwire.ReplyShortChanIDsEnd: + + syncer, err := d.findGossipSyncer(peer.IdentityKey()) + if err != nil { + log.Warnf("Unable to find gossip syncer for "+ + "peer=%x: %v", peer.PubKey(), err) + + errChan <- err + return errChan + } + + // If we've found the message target, then we'll dispatch the + // message directly to it. + syncer.ProcessQueryMsg(m) + + errChan <- nil + return errChan + + // If a peer is updating its current update horizon, then we'll dispatch + // that directly to the proper gossipSyncer. + case *lnwire.GossipTimestampRange: + syncer, err := d.findGossipSyncer(peer.IdentityKey()) + if err != nil { + log.Warnf("Unable to find gossip syncer for "+ + "peer=%x: %v", peer.PubKey(), err) + + errChan <- err + return errChan + } + + // If we've found the message target, then we'll dispatch the + // message directly to it. + err = syncer.ApplyGossipFilter(m) + if err != nil { + log.Warnf("unable to apply gossip "+ + "filter for peer=%x: %v", + peer.PubKey(), err) + + errChan <- err + return errChan + } + + errChan <- nil + return errChan + } + nMsg := &networkMsg{ msg: msg, isRemote: true, peer: peer, source: peer.IdentityKey(), - err: make(chan error, 1), + err: errChan, } select { @@ -965,7 +1020,7 @@ func (d *AuthenticatedGossiper) networkHandler() { policyUpdate.errResp <- nil case announcement := <-d.networkMsgs: - switch msg := announcement.msg.(type) { + switch announcement.msg.(type) { // Channel announcement signatures are amongst the only // messages that we'll process serially. case *lnwire.AnnounceSignatures: @@ -978,51 +1033,6 @@ func (d *AuthenticatedGossiper) networkHandler() { ) } continue - - // If a peer is updating its current update horizon, - // then we'll dispatch that directly to the proper - // gossipSyncer. - case *lnwire.GossipTimestampRange: - syncer, err := d.findGossipSyncer( - announcement.source, - ) - if err != nil { - log.Warnf("Unable to find gossip "+ - "syncer for peer=%x: %v", - announcement.peer.PubKey(), err) - continue - } - - // If we've found the message target, then - // we'll dispatch the message directly to it. - err = syncer.ApplyGossipFilter(msg) - if err != nil { - log.Warnf("unable to apply gossip "+ - "filter for peer=%x: %v", - announcement.peer.PubKey(), err) - } - continue - - // For messages in the known set of channel series - // queries, we'll dispatch the message directly to the - // peer, and skip the main processing loop. - case *lnwire.QueryShortChanIDs, - *lnwire.QueryChannelRange, - *lnwire.ReplyChannelRange, - *lnwire.ReplyShortChanIDsEnd: - - syncer, err := d.findGossipSyncer( - announcement.source, - ) - if err != nil { - log.Warnf("Unable to find gossip "+ - "syncer for peer=%x: %v", - announcement.source, err) - continue - } - - syncer.ProcessQueryMsg(announcement.msg) - continue } // If this message was recently rejected, then we won't From ef309f43d2b5bc981b6c6bc3c07021c04dbefdf0 Mon Sep 17 00:00:00 2001 From: Conner Fromknecht Date: Thu, 1 Nov 2018 17:35:55 -0700 Subject: [PATCH 3/3] discovery: pass peer quit signal to ProcessQueryMsg This commit passes the peer's quit signal to the gossipSyncer when attempt to hand off gossip query messages. This allows a rate-limited peer's read handler to break out immediately, which would otherwise remain stuck until the rate-limited gossip syncer pulled the message. --- discovery/gossiper.go | 2 +- discovery/syncer.go | 5 ++--- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 8257c2fc5..66aba91b7 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -478,7 +478,7 @@ func (d *AuthenticatedGossiper) ProcessRemoteAnnouncement(msg lnwire.Message, // If we've found the message target, then we'll dispatch the // message directly to it. - syncer.ProcessQueryMsg(m) + syncer.ProcessQueryMsg(m, peer.QuitSignal()) errChan <- nil return errChan diff --git a/discovery/syncer.go b/discovery/syncer.go index 5568b3707..91793b72d 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -982,12 +982,11 @@ func (g *gossipSyncer) FilterGossipMsgs(msgs ...msgWithSenders) { // ProcessQueryMsg is used by outside callers to pass new channel time series // queries to the internal processing goroutine. -func (g *gossipSyncer) ProcessQueryMsg(msg lnwire.Message) { +func (g *gossipSyncer) ProcessQueryMsg(msg lnwire.Message, peerQuit <-chan struct{}) { select { case g.gossipMsgs <- msg: - return + case <-peerQuit: case <-g.quit: - return } }