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

neutrino.go

@@ -195,6 +195,31 @@ func NewServerPeer(s *ChainService, isPersistent bool) *ServerPeer {
 	}
 }
 
+// IsSyncCandidate returns whether or not the peer is a candidate to consider
+// syncing from.
+func (sp *ServerPeer) IsSyncCandidate() bool {
+	// The peer is not a candidate for sync if it's not a full node.
+	return sp.Services()&wire.SFNodeNetwork == wire.SFNodeNetwork
+}
+
+// IsPeerBehindStartHeight returns a boolean indicating if the peer's last block height
+// is behind the start height of the request. If the peer is not behind the request start
+// height false is returned, otherwise, true is.
+func (sp *ServerPeer) IsPeerBehindStartHeight(req query.ReqMessage) bool {
+	queryGetHeaders, ok := req.(*headerQuery)
+
+	if !ok {
+		log.Debugf("request is not type headerQuery")
+
+		return true
+	}
+
+	if sp.LastBlock() < queryGetHeaders.startHeight {
+		return true
+	}
+	return false
+}
+
 // newestBlock returns the current best block hash and height using the format
 // required by the configuration for the peer package.
 func (sp *ServerPeer) newestBlock() (*chainhash.Hash, int32, error) {
@@ -800,15 +825,21 @@ func NewChainService(cfg Config) (*ChainService, error) {
 	}
 
 	bm, err := newBlockManager(&blockManagerCfg{
-		ChainParams:      s.chainParams,
-		BlockHeaders:     s.BlockHeaders,
-		RegFilterHeaders: s.RegFilterHeaders,
-		TimeSource:       s.timeSource,
-		QueryDispatcher:  s.workManager,
-		BanPeer:          s.BanPeer,
-		GetBlock:         s.GetBlock,
-		firstPeerSignal:  s.firstPeerConnect,
-		queryAllPeers:    s.queryAllPeers,
+		ChainParams:             s.chainParams,
+		BlockHeaders:            s.BlockHeaders,
+		RegFilterHeaders:        s.RegFilterHeaders,
+		TimeSource:              s.timeSource,
+		cfHeaderQueryDispatcher: s.workManager,
+		BanPeer:                 s.BanPeer,
+		GetBlock:                s.GetBlock,
+		firstPeerSignal:         s.firstPeerConnect,
+		queryAllPeers:           s.queryAllPeers,
+		blkHdrCheckptQueryDispatcher: query.NewWorkManager(&query.Config{
+			ConnectedPeers:       s.ConnectedPeers,
+			NewWorker:            query.NewWorker,
+			Ranking:              query.NewPeerRanking(),
+			IsEligibleWorkerFunc: query.IsWorkerEligibleForBlkHdrFetch,
+		}),
 	})
 	if err != nil {
 		return nil, err
@@ -1610,6 +1641,9 @@ func (s *ChainService) Start() error {
 	s.addrManager.Start()
 	s.blockManager.Start()
 	s.blockSubscriptionMgr.Start()
+	if err := s.blockManager.cfg.blkHdrCheckptQueryDispatcher.Start(); err != nil {
+		return fmt.Errorf("unable to start block header work manager: %v", err)
+	}
 	if err := s.workManager.Start(); err != nil {
 		return fmt.Errorf("unable to start work manager: %v", err)
 	}

query.go

@@ -66,10 +66,7 @@ var (
 
 	// noProgress will be used to indicate to a query.WorkManager that a
 	// response makes no progress towards the completion of the query.
-	noProgress = query.Progress{
-		Finished:   false,
-		Progressed: false,
-	}
+	noProgress = query.NoResponse
 )
 
 // queries are a set of options that can be modified per-query, unlike global
@@ -430,26 +427,50 @@ type cfiltersQuery struct {
 	headerIndex   map[chainhash.Hash]int
 	targetHash    chainhash.Hash
 	targetFilter  *gcs.Filter
+	mtx           sync.Mutex
 }
 
 // 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,
+		CloneReq: func(req query.ReqMessage) query.ReqMessage {
+			oldReq, ok := req.(*encodedQuery)
+			if !ok {
+				log.Errorf("request not of type *encodedQuery")
+			}
+			oldReqMessage, ok := oldReq.message.(*wire.MsgGetCFilters)
+			if !ok {
+				log.Errorf("request not of type *wire.MsgGetCFilters")
+			}
+			newReq := &encodedQuery{
+				message: wire.NewMsgGetCFilters(
+					oldReqMessage.FilterType, oldReqMessage.StartHeight, &oldReqMessage.StopHash,
+				),
+				encoding:      oldReq.encoding,
+				priorityIndex: oldReq.priorityIndex,
+			}
+			return newReq
+		},
 	}
 }
 
 // handleResponse validates that the cfilter response we get from a peer is
 // sane given the getcfilter query that we made.
-func (q *cfiltersQuery) handleResponse(req, resp wire.Message,
-	_ string) query.Progress {
+func (q *cfiltersQuery) handleResponse(r query.ReqMessage, resp wire.Message,
+	peer query.Peer) query.Progress {
 
+	req := r.Message()
 	// The request must have been a "getcfilters" msg.
 	request, ok := req.(*wire.MsgGetCFilters)
 	if !ok {
@@ -476,6 +497,8 @@ func (q *cfiltersQuery) handleResponse(req, resp wire.Message,
 
 	// If this filter is for a block not in our index, we can ignore it, as
 	// we either already got it, or it is out of our queried range.
+	q.mtx.Lock()
+	defer q.mtx.Unlock()
 	i, ok := q.headerIndex[response.BlockHash]
 	if !ok {
 		return noProgress
@@ -548,17 +571,11 @@ func (q *cfiltersQuery) handleResponse(req, resp wire.Message,
 	// If there are still entries left in the headerIndex then the query
 	// has made progress but has not yet completed.
 	if len(q.headerIndex) != 0 {
-		return query.Progress{
-			Finished:   false,
-			Progressed: true,
-		}
+		return query.Progressed
 	}
 
 	// The headerIndex is empty and so this query is complete.
-	return query.Progress{
-		Finished:   true,
-		Progressed: true,
-	}
+	return query.Finished
 }
 
 // prepareCFiltersQuery creates a cfiltersQuery that can be used to fetch a
@@ -759,6 +776,7 @@ func (s *ChainService) GetCFilter(blockHash chainhash.Hash,
 		query.Cancel(s.quit),
 		query.Encoding(qo.encoding),
 		query.NumRetries(qo.numRetries),
+		query.ErrChan(make(chan error, 1)),
 	}
 
 	errChan := s.workManager.Query(
@@ -833,13 +851,22 @@ 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
 
 	// handleResp will be called for each message received from a peer. It
 	// will be used to signal to the work manager whether progress has been
 	// made or not.
-	handleResp := func(req, resp wire.Message, peer string) query.Progress {
+	handleResp := func(request query.ReqMessage, resp wire.Message, sp query.Peer) query.Progress {
+		req := request.Message()
+		peer := ""
+		if sp != nil {
+			peer = sp.Addr()
+		}
 		// The request must have been a "getdata" msg.
 		_, ok := req.(*wire.MsgGetData)
 		if !ok {
@@ -904,23 +931,39 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 		// we declare it sane. We can kill the query and pass the
 		// response back to the caller.
 		foundBlock = block
-		return query.Progress{
-			Finished:   true,
-			Progressed: true,
-		}
+		return query.Finished
 	}
 
 	// Prepare the query request.
 	request := &query.Request{
-		Req:        getData,
+		Req:        msg,
 		HandleResp: handleResp,
+		SendQuery:  sendQueryMessageWithEncoding,
+		CloneReq: func(req query.ReqMessage) query.ReqMessage {
+			newMsg := wire.NewMsgGetData()
+			_ = newMsg.AddInvVect(inv)
+
+			oldReq, ok := req.(*encodedQuery)
+			if !ok {
+				log.Errorf("request not of type *encodedQuery")
+			}
+
+			newReq := &encodedQuery{
+				message:       newMsg,
+				encoding:      oldReq.encoding,
+				priorityIndex: oldReq.priorityIndex,
+			}
+
+			return newReq
+		},
 	}
 
 	// Prepare the query options.
 	queryOpts := []query.QueryOption{
 		query.Encoding(qo.encoding),
 		query.NumRetries(qo.numRetries),
 		query.Cancel(s.quit),
+		query.ErrChan(make(chan error, 1)),
 	}
 
 	// Send the request to the work manager and await a response.

query/interface.go

@@ -43,6 +43,9 @@ type queryOptions struct {
 	// that a query can be retried. If this is set then numRetries has no
 	// effect.
 	noRetryMax bool
+
+	// errChan error channel with which the workmananger sends error.
+	errChan chan error
 }
 
 // QueryOption is a functional option argument to any of the network query
@@ -67,6 +70,14 @@ func (qo *queryOptions) applyQueryOptions(options ...QueryOption) {
 	}
 }
 
+// ErrChan is a query option that specifies the error channel which the workmanager
+// sends any error to.
+func ErrChan(err chan error) QueryOption {
+	return func(qo *queryOptions) {
+		qo.errChan = err
+	}
+}
+
 // NumRetries is a query option that specifies the number of times a query
 // should be retried.
 func NumRetries(num uint8) QueryOption {
@@ -107,25 +118,40 @@ func Cancel(cancel chan struct{}) QueryOption {
 	}
 }
 
-// Progress encloses the result of handling a response for a given Request,
-// determining whether the response did progress the query.
-type Progress struct {
-	// Finished is true if the query was finished as a result of the
-	// received response.
-	Finished bool
-
-	// Progressed is true if the query made progress towards fully
-	// answering the request as a result of the received response. This is
-	// used for the requests types where more than one response is
-	// expected.
-	Progressed bool
-}
+// Progress encloses the result of handling a response for a given Request.
+type Progress string
+
+var (
+
+	// Finished indicates we have received the complete, valid response for this request,
+	// and so we are done with it.
+	Finished Progress = "Received complete  and valid response for request."
+
+	// Progressed indicates that we have received a valid response, but we are expecting more.
+	Progressed Progress = "Received valid response, expecting more response for query."
+
+	// UnFinishedRequest indicates that we have received some response, but we need to rescheule the job
+	// to completely fetch all the response required for this request.
+	UnFinishedRequest Progress = "Received valid response, reschedule to complete request"
+
+	// ResponseErr indicates we obtained a valid response but response fails checks and needs to
+	// be rescheduled.
+	ResponseErr Progress = "Received valid response but fails checks "
+
+	// IgnoreRequest indicates that we have received a valid response but the workmanager need take
+	// no action on the result of this job.
+	IgnoreRequest Progress = "Received response but ignoring"
+
+	// NoResponse indicates that we have received an invalid response for this request, and we need
+	// to wait for a valid one.
+	NoResponse Progress = "Received invalid response"
+)
 
 // 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
@@ -138,7 +164,26 @@ type Request struct {
 	// should validate the response and immediately return the progress.
 	// The response should be handed off to another goroutine for
 	// processing.
-	HandleResp func(req, resp wire.Message, peer string) Progress
+	HandleResp func(req ReqMessage, resp wire.Message, peer Peer) 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
+
+	// CloneReq clones the message.
+	CloneReq func(message ReqMessage) ReqMessage
+}
+
+// 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() float64
 }
 
 // WorkManager defines an API for a manager that dispatches queries to bitcoin
@@ -167,11 +212,6 @@ type Dispatcher interface {
 // Peer is the interface that defines the methods needed by the query package
 // to be able to make requests and receive responses from a network peer.
 type Peer interface {
-	// QueueMessageWithEncoding adds the passed bitcoin message to the peer
-	// send queue.
-	QueueMessageWithEncoding(msg wire.Message, doneChan chan<- struct{},
-		encoding wire.MessageEncoding)
-
 	// SubscribeRecvMsg adds a OnRead subscription to the peer. All bitcoin
 	// messages received from this peer will be sent on the returned
 	// channel. A closure is also returned, that should be called to cancel
@@ -184,4 +224,11 @@ type Peer interface {
 	// OnDisconnect returns a channel that will be closed when this peer is
 	// disconnected.
 	OnDisconnect() <-chan struct{}
+
+	// IsPeerBehindStartHeight returns a boolean indicating if the peer's known last height is behind
+	// the request's start Height which it receives as an argument.
+	IsPeerBehindStartHeight(req ReqMessage) bool
+
+	// IsSyncCandidate returns true if the peer is a sync candidate.
+	IsSyncCandidate() bool
 }