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.

Author: davecgh

internal/netsync/manager.go

@@ -2063,7 +2063,7 @@ func (m *SyncManager) SyncPeerID() int32 {
 //
 // 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
@@ -2167,32 +2167,36 @@ 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 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)
+	// Skip mix messages that have already been requested or are already known.
+	m.requestMtx.Lock()
+	_, ok := m.requestedMixMsgs[*mixHash]
+	m.requestMtx.Unlock()
+	alreadyHave := ok || m.cfg.MixPool.HaveMessage(mixHash)
+	if alreadyHave {
+		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)
 	}
@@ -1844,9 +1844,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()) {