Merge pull request #1817 from halseth/autopilot-node-update

Reduce goroutine overhead on autopilot updates

autopilot/agent.go

@@ -117,6 +117,29 @@ type Agent struct {
 	// affect the heuristics of the agent will be sent over.
 	stateUpdates chan interface{}
 
+	// balanceUpdates is a channel where notifications about updates to the
+	// wallet's balance will be sent. This channel will be buffered to
+	// ensure we have at most one pending update of this type to handle at
+	// a given time.
+	balanceUpdates chan *balanceUpdate
+
+	// nodeUpdates is a channel that changes to the graph node landscape
+	// will be sent over. This channel will be buffered to ensure we have
+	// at most one pending update of this type to handle at a given time.
+	nodeUpdates chan *nodeUpdates
+
+	// pendingOpenUpdates is a channel where updates about channel pending
+	// opening will be sent. This channel will be buffered to ensure we
+	// have at most one pending update of this type to handle at a given
+	// time.
+	pendingOpenUpdates chan *chanPendingOpenUpdate
+
+	// chanOpenFailures is a channel where updates about channel open
+	// failures will be sent. This channel will be buffered to ensure we
+	// have at most one pending update of this type to handle at a given
+	// time.
+	chanOpenFailures chan *chanOpenFailureUpdate
+
 	// totalBalance is the total number of satoshis the backing wallet is
 	// known to control at any given instance. This value will be updated
 	// when the agent receives external balance update signals.
@@ -136,6 +159,10 @@ func New(cfg Config, initialState []Channel) (*Agent, error) {
 		chanState:          make(map[lnwire.ShortChannelID]Channel),
 		quit:               make(chan struct{}),
 		stateUpdates:       make(chan interface{}),
+		balanceUpdates:     make(chan *balanceUpdate, 1),
+		nodeUpdates:        make(chan *nodeUpdates, 1),
+		chanOpenFailures:   make(chan *chanOpenFailureUpdate, 1),
+		pendingOpenUpdates: make(chan *chanPendingOpenUpdate, 1),
 	}
 
 	for _, c := range initialState {
@@ -206,32 +233,22 @@ type chanCloseUpdate struct {
 	closedChans []lnwire.ShortChannelID
 }
 
-// OnBalanceChange is a callback that should be executed each time the balance of
+// OnBalanceChange is a callback that should be executed each time the balance
-// the backing wallet changes.
+// of the backing wallet changes.
 func (a *Agent) OnBalanceChange() {
-	a.wg.Add(1)
-	go func() {
-		defer a.wg.Done()
-
 	select {
-		case a.stateUpdates <- &balanceUpdate{}:
+	case a.balanceUpdates <- &balanceUpdate{}:
-		case <-a.quit:
+	default:
 	}
-	}()
 }
 
 // OnNodeUpdates is a callback that should be executed each time our channel
 // graph has new nodes or their node announcements are updated.
 func (a *Agent) OnNodeUpdates() {
-	a.wg.Add(1)
-	go func() {
-		defer a.wg.Done()
-
 	select {
-		case a.stateUpdates <- &nodeUpdates{}:
+	case a.nodeUpdates <- &nodeUpdates{}:
-		case <-a.quit:
+	default:
 	}
-	}()
 }
 
 // OnChannelOpen is a callback that should be executed each time a new channel
@@ -252,27 +269,20 @@ func (a *Agent) OnChannelOpen(c Channel) {
 // channel is opened, either by the agent or an external subsystems, but is
 // still pending.
 func (a *Agent) OnChannelPendingOpen() {
-	go func() {
 	select {
-		case a.stateUpdates <- &chanPendingOpenUpdate{}:
+	case a.pendingOpenUpdates <- &chanPendingOpenUpdate{}:
-		case <-a.quit:
+	default:
 	}
-	}()
 }
 
 // OnChannelOpenFailure is a callback that should be executed when the
 // autopilot has attempted to open a channel, but failed. In this case we can
 // retry channel creation with a different node.
 func (a *Agent) OnChannelOpenFailure() {
-	a.wg.Add(1)
-	go func() {
-		defer a.wg.Done()
-
 	select {
-		case a.stateUpdates <- &chanOpenFailureUpdate{}:
+	case a.chanOpenFailures <- &chanOpenFailureUpdate{}:
-		case <-a.quit:
+	default:
 	}
-	}()
 }
 
 // OnChannelClose is a callback that should be executed each time a prior
@@ -377,24 +387,6 @@ func (a *Agent) controller() {
 			log.Infof("Processing new external signal")
 
 			switch update := signal.(type) {
-			// The balance of the backing wallet has changed, if
-			// more funds are now available, we may attempt to open
-			// up an additional channel, or splice in funds to an
-			// existing one.
-			case *balanceUpdate:
-				log.Debug("Applying external balance state " +
-					"update")
-
-				updateBalance()
-
-			// The channel we tried to open previously failed for
-			// whatever reason.
-			case *chanOpenFailureUpdate:
-				log.Debug("Retrying after previous channel " +
-					"open failure.")
-
-				updateBalance()
-
 			// A new channel has been opened successfully. This was
 			// either opened by the Agent, or an external system
 			// that is able to drive the Lightning Node.
@@ -411,13 +403,6 @@ func (a *Agent) controller() {
 				pendingMtx.Unlock()
 
 				updateBalance()
-
-			// A new channel has been opened by the agent or an
-			// external subsystem, but is still pending
-			// confirmation.
-			case *chanPendingOpenUpdate:
-				updateBalance()
-
 			// A channel has been closed, this may free up an
 			// available slot, triggering a new channel update.
 			case *chanCloseUpdate:
@@ -430,32 +415,57 @@ func (a *Agent) controller() {
 				}
 
 				updateBalance()
+			}
+
+		// A new channel has been opened by the agent or an external
+		// subsystem, but is still pending confirmation.
+		case <-a.pendingOpenUpdates:
+			updateBalance()
+
+		// The balance of the backing wallet has changed, if more funds
+		// are now available, we may attempt to open up an additional
+		// channel, or splice in funds to an existing one.
+		case <-a.balanceUpdates:
+			log.Debug("Applying external balance state update")
+
+			updateBalance()
+
+		// The channel we tried to open previously failed for whatever
+		// reason.
+		case <-a.chanOpenFailures:
+			log.Debug("Retrying after previous channel open " +
+				"failure.")
+
+			updateBalance()
 
 		// New nodes have been added to the graph or their node
-			// announcements have been updated. We will consider
+		// announcements have been updated. We will consider opening
-			// opening channels to these nodes if we haven't
+		// channels to these nodes if we haven't stabilized.
-			// stabilized.
+		case <-a.nodeUpdates:
-			case *nodeUpdates:
 			log.Infof("Node updates received, assessing " +
 				"need for more channels")
+
+		// The agent has been signalled to exit, so we'll bail out
+		// immediately.
+		case <-a.quit:
+			return
 		}
 
 		pendingMtx.Lock()
 		log.Debugf("Pending channels: %v", spew.Sdump(pendingOpens))
 		pendingMtx.Unlock()
 
-			// With all the updates applied, we'll obtain a set of
+		// With all the updates applied, we'll obtain a set of the
-			// the current active channels (confirmed channels),
+		// current active channels (confirmed channels), and also
-			// and also factor in our set of unconfirmed channels.
+		// factor in our set of unconfirmed channels.
 		confirmedChans := a.chanState
 		pendingMtx.Lock()
 		totalChans := mergeChanState(pendingOpens, confirmedChans)
 		pendingMtx.Unlock()
 
-			// Now that we've updated our internal state, we'll
+		// Now that we've updated our internal state, we'll consult our
-			// consult our channel attachment heuristic to
+		// channel attachment heuristic to determine if we should open
-			// determine if we should open up any additional
+		// up any additional channels or modify existing channels.
-			// channels or modify existing channels.
 		availableFunds, numChans, needMore := a.cfg.Heuristic.NeedMoreChans(
 			totalChans, a.totalBalance,
 		)
@@ -466,20 +476,19 @@ func (a *Agent) controller() {
 		log.Infof("Triggering attachment directive dispatch, "+
 			"total_funds=%v", a.totalBalance)
 
-			// We're to attempt an attachment so we'll o obtain the
+		// We're to attempt an attachment so we'll o obtain the set of
-			// set of nodes that we currently have channels with so
+		// nodes that we currently have channels with so we avoid
-			// we avoid duplicate edges.
+		// duplicate edges.
 		connectedNodes := a.chanState.ConnectedNodes()
 		pendingMtx.Lock()
 		nodesToSkip := mergeNodeMaps(connectedNodes, failedNodes, pendingOpens)
 		pendingMtx.Unlock()
 
-			// If we reach this point, then according to our
+		// If we reach this point, then according to our heuristic we
-			// heuristic we should modify our channel state to tend
+		// should modify our channel state to tend towards what it
-			// towards what it determines to the optimal state. So
+		// determines to the optimal state. So we'll call Select to get
-			// we'll call Select to get a fresh batch of attachment
+		// a fresh batch of attachment directives, passing in the
-			// directives, passing in the amount of funds available
+		// amount of funds available for us to use.
-			// for us to use.
 		chanCandidates, err := a.cfg.Heuristic.Select(
 			a.cfg.Self, a.cfg.Graph, availableFunds,
 			numChans, nodesToSkip,
@@ -498,18 +507,17 @@ func (a *Agent) controller() {
 		log.Infof("Attempting to execute channel attachment "+
 			"directives: %v", spew.Sdump(chanCandidates))
 
-			// For each recommended attachment directive, we'll
+		// For each recommended attachment directive, we'll launch a
-			// launch a new goroutine to attempt to carry out the
+		// new goroutine to attempt to carry out the directive. If any
-			// directive. If any of these succeed, then we'll
+		// of these succeed, then we'll receive a new state update,
-			// receive a new state update, taking us back to the
+		// taking us back to the top of our controller loop.
-			// top of our controller loop.
 		pendingMtx.Lock()
 		for _, chanCandidate := range chanCandidates {
-				// Before we proceed, we'll check to see if
+			// Before we proceed, we'll check to see if this
-				// this attempt would take us past the total
+			// attempt would take us past the total number of
-				// number of allowed pending opens. If so, then
+			// allowed pending opens. If so, then we'll skip this
-				// we'll skip this round and wait for an
+			// round and wait for an attempt to either fail or
-				// attempt to either fail or succeed.
+			// succeed.
 			if uint16(len(pendingOpens))+1 >
 				a.cfg.MaxPendingOpens {
 
@@ -521,9 +529,9 @@ func (a *Agent) controller() {
 			}
 
 			go func(directive AttachmentDirective) {
-					// We'll start out by attempting to
+				// We'll start out by attempting to connect to
-					// connect to the peer in order to begin
+				// the peer in order to begin the funding
-					// the funding workflow.
+				// workflow.
 				pub := directive.PeerKey
 				alreadyConnected, err := a.cfg.ConnectToPeer(
 					pub, directive.Addrs,
@@ -534,29 +542,26 @@ func (a *Agent) controller() {
 						pub.SerializeCompressed(),
 						err)
 
-						// Since we failed to connect to
+					// Since we failed to connect to them,
-						// them, we'll mark them as
+					// we'll mark them as failed so that we
-						// failed so that we don't
+					// don't attempt to connect to them
-						// attempt to connect to them
 					// again.
 					nodeID := NewNodeID(pub)
 					pendingMtx.Lock()
 					failedNodes[nodeID] = struct{}{}
 					pendingMtx.Unlock()
 
-						// Finally, we'll trigger the
+					// Finally, we'll trigger the agent to
-						// agent to select new peers to
+					// select new peers to connect to.
-						// connect to.
 					a.OnChannelOpenFailure()
 
 					return
 				}
 
-					// If we were succesful, we'll track
+				// If we were succesful, we'll track this peer
-					// this peer in our set of pending
+				// in our set of pending opens. We do this here
-					// opens. We do this here to ensure we
+				// to ensure we don't stall on selecting new
-					// don't stall on selecting new peers if
+				// peers if the connection attempt happens to
-					// the connection attempt happens to
 				// take too long.
 				pendingMtx.Lock()
 				if uint16(len(pendingOpens))+1 >
@@ -564,14 +569,12 @@ func (a *Agent) controller() {
 
 					pendingMtx.Unlock()
 
-						// Since we've reached our max
+					// Since we've reached our max number
-						// number of pending opens,
+					// of pending opens, we'll disconnect
-						// we'll disconnect this peer
+					// this peer and exit. However, if we
-						// and exit. However, if we were
+					// were previously connected to them,
-						// previously connected to them,
+					// then we'll make sure to maintain the
-						// then we'll make sure to
+					// connection alive.
-						// maintain the connection
-						// alive.
 					if alreadyConnected {
 						return
 					}
@@ -596,8 +599,8 @@ func (a *Agent) controller() {
 				}
 				pendingMtx.Unlock()
 
-					// We can then begin the funding
+				// We can then begin the funding workflow with
-					// workflow with this peer.
+				// this peer.
 				err = a.cfg.ChanController.OpenChannel(
 					pub, directive.ChanAmt,
 				)
@@ -607,27 +610,24 @@ func (a *Agent) controller() {
 						pub.SerializeCompressed(),
 						directive.ChanAmt, err)
 
-						// As the attempt failed, we'll
+					// As the attempt failed, we'll clear
-						// clear the peer from the set of
+					// the peer from the set of pending
-						// pending opens and mark them
+					// opens and mark them as failed so we
-						// as failed so we don't attempt
+					// don't attempt to open a channel to
-						// to open a channel to them
+					// them again.
-						// again.
 					pendingMtx.Lock()
 					delete(pendingOpens, nodeID)
 					failedNodes[nodeID] = struct{}{}
 					pendingMtx.Unlock()
 
-						// Trigger the agent to
+					// Trigger the agent to re-evaluate
-						// re-evaluate everything and
+					// everything and possibly retry with a
-						// possibly retry with a
 					// different node.
 					a.OnChannelOpenFailure()
 
-						// Finally, we should also
+					// Finally, we should also disconnect
-						// disconnect the peer if we
+					// the peer if we weren't already
-						// weren't already connected to
+					// connected to them beforehand by an
-						// them beforehand by an
 					// external subsystem.
 					if alreadyConnected {
 						return
@@ -643,19 +643,13 @@ func (a *Agent) controller() {
 					}
 				}
 
-					// Since the channel open was successful
+				// Since the channel open was successful and is
-					// and is currently pending, we'll
+				// currently pending, we'll trigger the
-					// trigger the autopilot agent to query
+				// autopilot agent to query for more peers.
-					// for more peers.
 				a.OnChannelPendingOpen()
 			}(chanCandidate)
 		}
 		pendingMtx.Unlock()
 
-		// The agent has been signalled to exit, so we'll bail out
-		// immediately.
-		case <-a.quit:
-			return
-		}
 	}
 }