Parallelize header download across peers to fetch headers within a chain's checkpointed region.

blockmanager.go

@@ -5,6 +5,7 @@ package neutrino
 import (
 	"bytes"
 	"container/list"
+	"errors"
 	"fmt"
 	"math"
 	"math/big"
@@ -808,19 +809,39 @@ func (b *blockManager) getUncheckpointedCFHeaders(
 // handle a query for checkpointed filter headers.
 type checkpointedCFHeadersQuery struct {
 	blockMgr    *blockManager
-	msgs        []wire.Message
+	msgs        []*encodedQuery
 	checkpoints []*chainhash.Hash
 	stopHashes  map[chainhash.Hash]uint32
 	headerChan  chan *wire.MsgCFHeaders
 }
 
+// encodedQuery holds all the information needed to query a message that pushes requests
+// using the QueryMessagingWithEncoding method.
+type encodedQuery struct {
+	message       wire.Message
+	encoding      wire.MessageEncoding
+	priorityIndex uint64
+}
+
+// Message returns the wire.Message of encodedQuery's struct.
+func (e *encodedQuery) Message() wire.Message {
+	return e.message
+}
+
+// PriorityIndex returns the specified priority the caller wants
+// the request to take.
+func (e *encodedQuery) PriorityIndex() uint64 {
+	return e.priorityIndex
+}
+
 // requests creates the query.Requests for this CF headers query.
 func (c *checkpointedCFHeadersQuery) requests() []*query.Request {
 	reqs := make([]*query.Request, len(c.msgs))
 	for idx, m := range c.msgs {
 		reqs[idx] = &query.Request{
 			Req:        m,
 			HandleResp: c.handleResponse,
+			SendQuery:  sendQueryMessageWithEncoding,
 		}
 	}
 	return reqs
@@ -924,6 +945,24 @@ func (c *checkpointedCFHeadersQuery) handleResponse(req, resp wire.Message,
 	}
 }
 
+// sendQueryMessageWithEncoding sends a message to the peer with encoding.
+func sendQueryMessageWithEncoding(peer query.Peer, req query.ReqMessage) error {
+	sp, ok := peer.(*ServerPeer)
+	if !ok {
+		err := "peer is not of type ServerPeer"
+		log.Errorf(err)
+		return errors.New(err)
+	}
+	request, ok := req.(*encodedQuery)
+	if !ok {
+		return errors.New("invalid request type")
+	}
+
+	sp.QueueMessageWithEncoding(request.message, nil, request.encoding)
+
+	return nil
+}
+
 // getCheckpointedCFHeaders catches a filter header store up with the
 // checkpoints we got from the network. It assumes that the filter header store
 // matches the checkpoints up to the tip of the store.
@@ -959,7 +998,7 @@ func (b *blockManager) getCheckpointedCFHeaders(checkpoints []*chainhash.Hash,
 	// the remaining checkpoint intervals.
 	numCheckpts := uint32(len(checkpoints)) - startingInterval
 	numQueries := (numCheckpts + maxCFCheckptsPerQuery - 1) / maxCFCheckptsPerQuery
-	queryMsgs := make([]wire.Message, 0, numQueries)
+	queryMsgs := make([]*encodedQuery, 0, numQueries)
 
 	// We'll also create an additional set of maps that we'll use to
 	// re-order the responses as we get them in.
@@ -1004,9 +1043,12 @@ func (b *blockManager) getCheckpointedCFHeaders(checkpoints []*chainhash.Hash,
 
 		// Once we have the stop hash, we can construct the query
 		// message itself.
-		queryMsg := wire.NewMsgGetCFHeaders(
-			fType, startHeightRange, &stopHash,
-		)
+		queryMsg := &encodedQuery{
+			message: wire.NewMsgGetCFHeaders(
+				fType, startHeightRange, &stopHash,
+			),
+			encoding: wire.WitnessEncoding,
+		}
 
 		// We'll mark that the ith interval is queried by this message,
 		// and also map the stop hash back to the index of this message.

blockmanager_test.go

@@ -352,7 +352,7 @@ func TestBlockManagerInitialInterval(t *testing.T) {
 
 			var msgs []wire.Message
 			for _, q := range requests {
-				msgs = append(msgs, q.Req)
+				msgs = append(msgs, q.Req.Message())
 			}
 
 			responses, err := generateResponses(msgs, headers)
@@ -582,7 +582,7 @@ func TestBlockManagerInvalidInterval(t *testing.T) {
 
 			var msgs []wire.Message
 			for _, q := range requests {
-				msgs = append(msgs, q.Req)
+				msgs = append(msgs, q.Req.Message())
 			}
 			responses, err := generateResponses(msgs, headers)
 			require.NoError(t, err)

query.go

@@ -435,13 +435,17 @@ type cfiltersQuery struct {
 // request couples a query message with the handler to be used for the response
 // in a query.Request struct.
 func (q *cfiltersQuery) request() *query.Request {
-	msg := wire.NewMsgGetCFilters(
-		q.filterType, uint32(q.startHeight), q.stopHash,
-	)
+	msg := &encodedQuery{
+		message: wire.NewMsgGetCFilters(
+			q.filterType, uint32(q.startHeight), q.stopHash,
+		),
+		encoding: wire.WitnessEncoding,
+	}
 
 	return &query.Request{
 		Req:        msg,
 		HandleResp: q.handleResponse,
+		SendQuery:  sendQueryMessageWithEncoding,
 	}
 }
 
@@ -833,6 +837,10 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 	// Construct the appropriate getdata message to fetch the target block.
 	getData := wire.NewMsgGetData()
 	_ = getData.AddInvVect(inv)
+	msg := &encodedQuery{
+		message:  getData,
+		encoding: wire.WitnessEncoding,
+	}
 
 	var foundBlock *btcutil.Block
 
@@ -912,8 +920,9 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 
 	// Prepare the query request.
 	request := &query.Request{
-		Req:        getData,
+		Req:        msg,
 		HandleResp: handleResp,
+		SendQuery:  sendQueryMessageWithEncoding,
 	}
 
 	// Prepare the query options.

query/interface.go

@@ -124,8 +124,8 @@ type Progress struct {
 // Request is the main struct that defines a bitcoin network query to be sent to
 // connected peers.
 type Request struct {
-	// Req is the message request to send.
-	Req wire.Message
+	// Req contains the message request to send.
+	Req ReqMessage
 
 	// HandleResp is a response handler that will be called for every
 	// message received from the peer that the request was made to. It
@@ -139,6 +139,22 @@ type Request struct {
 	// The response should be handed off to another goroutine for
 	// processing.
 	HandleResp func(req, resp wire.Message, peer string) Progress
+
+	// SendQuery handles sending request to the worker's peer. It returns an error,
+	// if one is encountered while sending the request.
+	SendQuery func(peer Peer, request ReqMessage) error
+}
+
+// ReqMessage is an interface which all structs containing information
+// required to process a message request must implement.
+type ReqMessage interface {
+
+	// Message returns the message request.
+	Message() wire.Message
+
+	// PriorityIndex returns the priority the caller prefers the request
+	// would take.
+	PriorityIndex() uint64
 }
 
 // WorkManager defines an API for a manager that dispatches queries to bitcoin

query/worker.go

@@ -3,8 +3,6 @@ package query
 import (
 	"errors"
 	"time"
-
-	"github.com/btcsuite/btcd/wire"
 )
 
 var (
@@ -27,7 +25,6 @@ type queryJob struct {
 	tries      uint8
 	index      uint64
 	timeout    time.Duration
-	encoding   wire.MessageEncoding
 	cancelChan <-chan struct{}
 	*Request
 }
@@ -89,6 +86,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 	msgChan, cancel := peer.SubscribeRecvMsg()
 	defer cancel()
 
+nexJobLoop:
 	for {
 		log.Tracef("Worker %v waiting for more work", peer.Addr())
 
@@ -133,7 +131,22 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 			log.Tracef("Worker %v queuing job %T with index %v",
 				peer.Addr(), job.Req, job.Index())
 
-			peer.QueueMessageWithEncoding(job.Req, nil, job.encoding)
+			err := job.SendQuery(peer, job.Req)
+
+			// If any error occurs while sending query, quickly send the result
+			// containing the error to the workmanager.
+			if err != nil {
+				select {
+				case results <- &jobResult{
+					job:  job,
+					peer: peer,
+					err:  err,
+				}:
+				case <-quit:
+					return
+				}
+				goto nexJobLoop
+			}
 		}
 
 		// Wait for the correct response to be received from the peer,
@@ -143,15 +156,15 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 			timeout = time.NewTimer(job.timeout)
 		)
 
-	Loop:
+	feedbackLoop:
 		for {
 			select {
 			// A message was received from the peer, use the
 			// response handler to check whether it was answering
 			// our request.
 			case resp := <-msgChan:
 				progress := job.HandleResp(
-					job.Req, resp, peer.Addr(),
+					job.Req.Message(), resp, peer.Addr(),
 				)
 
 				log.Tracef("Worker %v handled msg %T while "+
@@ -176,12 +189,12 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 							job.timeout,
 						)
 					}
-					continue Loop
+					continue feedbackLoop
 				}
 
 				// We did get a valid response, and can break
 				// the loop.
-				break Loop
+				break feedbackLoop
 
 			// If the timeout is reached before a valid response
 			// has been received, we exit with an error.
@@ -193,7 +206,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 					"with job index %v", peer.Addr(),
 					job.Req, job.Index())
 
-				break Loop
+				break feedbackLoop
 
 			// If the peer disconnects before giving us a valid
 			// answer, we'll also exit with an error.
@@ -203,7 +216,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 					job.Index())
 
 				jobErr = ErrPeerDisconnected
-				break Loop
+				break feedbackLoop
 
 			// If the job was canceled, we report this back to the
 			// work manager.
@@ -212,7 +225,7 @@ func (w *worker) Run(results chan<- *jobResult, quit <-chan struct{}) {
 					peer.Addr(), job.Index())
 
 				jobErr = ErrJobCanceled
-				break Loop
+				break feedbackLoop
 
 			case <-quit:
 				return