netsync: Separate mix message request.

The current logic for requesting mix messages from peers was added to
the generic request from peer paths because the old async model required
a lot of additional plumbing.

Now that the sync manager has been converted to a synchronous model, no
additional plumbing is needed and therefore it is much simpler and more
efficient to implement the request logic for mix messages independently.

Also, only a single mix message for a missing pair request is ever
needed at a time, so the additional overhead of taking a slice is not
needed.

Consequently, this separates the logic for requesting a mix message from
a peer from the generic request from peer path into its own specialized
method that is much more efficient.

Finally, it also changes the logic for determining if the mix message
should be requested to make use of the same method as other paths that
handle requesting mix messages.  In particular, this means it will no
longer attempt to request recently rejected or removed messages.

Author: davecgh

internal/netsync/manager.go

@@ -2057,14 +2057,13 @@ func (m *SyncManager) SyncPeerID() int32 {
 	return 0
 }
 
-// RequestFromPeer requests any combination of blocks, votes, treasury spends,
-// and mix messages from the given peer.  It ensures all of the requests are
-// tracked so the peer is not banned for sending unrequested data when it
-// responds.
+// RequestFromPeer requests any combination of blocks, votes, and treasury
+// spends from the given peer.  It ensures all of the requests are tracked so
+// the peer is not banned for sending unrequested data when it responds.
 //
 // This function is safe for concurrent access.
 func (m *SyncManager) RequestFromPeer(peer *Peer, blocks, voteHashes,
-	tSpendHashes, mixHashes []chainhash.Hash) error {
+	tSpendHashes []chainhash.Hash) error {
 
 	if m.shutdownRequested() {
 		return nil
@@ -2168,32 +2167,37 @@ func (m *SyncManager) RequestFromPeer(peer *Peer, blocks, voteHashes,
 		return err
 	}
 
-	for i := range mixHashes {
-		// Skip mix messages that have already been requested.
-		mh := &mixHashes[i]
-		if _, ok := m.requestedMixMsgs[*mh]; ok {
-			continue
-		}
-
-		// Skip the message when it is already known.
-		if m.cfg.MixPool.HaveMessage(mh) {
-			continue
-		}
+	if len(gdMsg.InvList) > 0 {
+		peer.QueueMessage(gdMsg, nil)
+	}
 
-		err := gdMsg.AddInvVect(wire.NewInvVect(wire.InvTypeMix, mh))
-		if err != nil {
-			return fmt.Errorf("unexpected error encountered building request "+
-				"for inv vect mix hash %v: %w", mh, err)
-		}
+	return nil
+}
 
-		m.requestedMixMsgs[*mh] = peer
+// RequestMixMsgFromPeer requests the specified mix message from the given peer.
+// It ensures all of the requests are tracked so the peer is not banned for
+// sending unrequested data when it responds.
+//
+// This function is safe for concurrent access.
+func (m *SyncManager) RequestMixMsgFromPeer(peer *Peer, mixHash *chainhash.Hash) {
+	if m.shutdownRequested() {
+		return
 	}
 
-	if len(gdMsg.InvList) > 0 {
-		peer.QueueMessage(gdMsg, nil)
+	defer m.requestMtx.Unlock()
+	m.requestMtx.Lock()
+
+	// Skip mix messages that have already been requested or are otherwise not
+	// needed.
+	_, alreadyRequested := m.requestedMixMsgs[*mixHash]
+	if alreadyRequested || !m.needMixMsg(mixHash) {
+		return
 	}
 
-	return nil
+	gdMsg := wire.NewMsgGetDataSizeHint(1)
+	gdMsg.AddInvVect(wire.NewInvVect(wire.InvTypeMix, mixHash))
+	m.requestedMixMsgs[*mixHash] = peer
+	peer.QueueMessage(gdMsg, nil)
 }
 
 // ProcessBlock processes the provided block using the chain instance associated

server.go

@@ -1344,7 +1344,7 @@ func (sp *serverPeer) OnMiningState(_ *peer.Peer, msg *wire.MsgMiningState) {
 	}
 
 	err := sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer, blockHashes,
-		voteHashes, nil, nil)
+		voteHashes, nil)
 	if err != nil {
 		peerLog.Warnf("couldn't handle mining state message: %v", err)
 	}
@@ -1430,7 +1430,7 @@ func (sp *serverPeer) OnGetInitState(_ *peer.Peer, msg *wire.MsgGetInitState) {
 // requests the data advertised in the message from the peer.
 func (sp *serverPeer) OnInitState(_ *peer.Peer, msg *wire.MsgInitState) {
 	err := sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer,
-		msg.BlockHashes, msg.VoteHashes, msg.TSpendHashes, nil)
+		msg.BlockHashes, msg.VoteHashes, msg.TSpendHashes)
 	if err != nil {
 		peerLog.Warnf("couldn't handle init state message: %v", err)
 	}
@@ -1852,9 +1852,8 @@ func (sp *serverPeer) onMixMessage(msg mixing.Message) {
 	}
 	var missingOwnPRErr *mixpool.MissingOwnPRError
 	if errors.As(err, &missingOwnPRErr) {
-		mixHashes := []chainhash.Hash{missingOwnPRErr.MissingPR}
-		sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer, nil, nil, nil,
-			mixHashes)
+		mixHash := &missingOwnPRErr.MissingPR
+		sp.server.syncManager.RequestMixMsgFromPeer(sp.syncMgrPeer, mixHash)
 		return
 	}
 	if mixpool.IsBannable(err, sp.Services()) {