parellized header download in block header checkpointed region

Signed-off-by: Maureen Ononiwu <amaka013@gmail.com>

neutrino.go

@@ -180,21 +180,43 @@ type ServerPeer struct {
 	recvSubscribers  map[spMsgSubscription]struct{}
 	recvSubscribers2 map[msgSubscription]struct{}
 	mtxSubscribers   sync.RWMutex
+
+	//recentReqDuration is the time between sending a request and receiving a response
+	//while querying this peer using the workmanager.
+	recentReqDuration time.Duration
+
+	//recentReqStartTime is the start time of the most recent request sent to this peer while
+	//using the workmanager to orchestrate the query.
+	recentReqStartTime time.Time
 }
 
 // NewServerPeer returns a new ServerPeer instance. The peer needs to be set by
 // the caller.
 func NewServerPeer(s *ChainService, isPersistent bool) *ServerPeer {
 	return &ServerPeer{
-		server:           s,
-		persistent:       isPersistent,
-		knownAddresses:   lru.NewCache[string, *cachedAddr](5000),
-		quit:             make(chan struct{}),
-		recvSubscribers:  make(map[spMsgSubscription]struct{}),
-		recvSubscribers2: make(map[msgSubscription]struct{}),
+		server:            s,
+		persistent:        isPersistent,
+		knownAddresses:    lru.NewCache[string, *cachedAddr](5000),
+		quit:              make(chan struct{}),
+		recvSubscribers:   make(map[spMsgSubscription]struct{}),
+		recvSubscribers2:  make(map[msgSubscription]struct{}),
+		recentReqDuration: 2 * time.Second,
 	}
 }
 
+// LastReqDuration returns the most recent request duaration of the peer.
+func (sp *ServerPeer) LastReqDuration() time.Duration {
+	return sp.recentReqDuration
+}
+
+// UpdateRequestDuration computes the time between the peer's most recent request's start time and now.
+// The duration is assigned to the peer's recentReqDuration field.
+func (sp *ServerPeer) UpdateRequestDuration() {
+	duration := time.Since(sp.recentReqStartTime)
+	sp.recentReqDuration = duration
+	log.Debugf("peer=%v, updated duration to=%v", sp.Addr(), duration.Seconds())
+}
+
 // 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) {
@@ -742,6 +764,8 @@ func NewChainService(cfg Config) (*ChainService, error) {
 		ConnectedPeers: s.ConnectedPeers,
 		NewWorker:      query.NewWorker,
 		Ranking:        query.NewPeerRanking(),
+		OrderPeers:     query.SortPeersByReqDuration,
+		DebugName:      "GeneralWorkManager",
 	})
 
 	var err error
@@ -800,15 +824,24 @@ 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,
+		peerByAddr:              s.PeerByAddr,
+		blkHdrCheckptQueryDispatcher: query.NewWorkManager(&query.Config{
+			ConnectedPeers:       s.ConnectedPeers,
+			NewWorker:            query.NewWorker,
+			Ranking:              query.NewPeerRanking(),
+			OrderPeers:           query.SortPeersByReqDuration,
+			IsEligibleWorkerFunc: query.IsWorkerEligibleForBlkHdrFetch,
+			DebugName:            "BlockHdrWorkManager",
+		}),
 	})
 	if err != nil {
 		return nil, err
@@ -1114,6 +1147,34 @@ func (s *ChainService) AddPeer(sp *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 wire.Message) 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
+}
+
 // AddBytesSent adds the passed number of bytes to the total bytes sent counter
 // for the server.  It is safe for concurrent access.
 func (s *ChainService) AddBytesSent(bytesSent uint64) {
@@ -1609,6 +1670,9 @@ func (s *ChainService) Start() error {
 	// needed by peers.
 	s.addrManager.Start()
 	s.blockManager.Start()
+	if err := s.blockManager.cfg.blkHdrCheckptQueryDispatcher.Start(); err != nil {
+		return fmt.Errorf("unable to start block header work manager: %v", err)
+	}
 	s.blockSubscriptionMgr.Start()
 	if err := s.workManager.Start(); err != nil {
 		return fmt.Errorf("unable to start work manager: %v", err)

query.go

@@ -439,16 +439,33 @@ func (q *cfiltersQuery) request() *query.Request {
 		q.filterType, uint32(q.startHeight), q.stopHash,
 	)
 
+	queryMsg := query.ReqMessage{
+		Message: msg,
+	}
 	return &query.Request{
-		Req:        msg,
+		Req:        &queryMsg,
 		HandleResp: q.handleResponse,
+		SendQuery:  sendQueryMessageWithEncoding,
+		CloneReq: func(req query.ReqMessage) *query.ReqMessage {
+			oldReq, ok := req.Message.(*wire.MsgGetCFilters)
+			if !ok {
+				log.Errorf("request not of type *wire.MsgCFHeaders")
+			}
+			newReq := &query.ReqMessage{
+				Message: wire.NewMsgGetCFilters(
+					oldReq.FilterType, oldReq.StartHeight, &oldReq.StopHash,
+				),
+				PriorityIndex: req.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 {
+	peer query.Peer, _ *error) query.Progress {
 
 	// The request must have been a "getcfilters" msg.
 	request, ok := req.(*wire.MsgGetCFilters)
@@ -462,6 +479,9 @@ func (q *cfiltersQuery) handleResponse(req, resp wire.Message,
 		return noProgress
 	}
 
+	if peer != nil {
+		peer.UpdateRequestDuration()
+	}
 	// If the request filter type doesn't match the type we were expecting,
 	// ignore this message.
 	if q.filterType != request.FilterType {
@@ -691,6 +711,7 @@ func (s *ChainService) prepareCFiltersQuery(blockHash chainhash.Hash,
 func (s *ChainService) GetCFilter(blockHash chainhash.Hash,
 	filterType wire.FilterType, options ...QueryOption) (*gcs.Filter,
 	error) {
+	log.Debugf("Getting CFilter...")
 
 	// The only supported filter atm is the regular filter, so we'll reject
 	// all other filters.
@@ -759,6 +780,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(
@@ -775,6 +797,8 @@ func (s *ChainService) GetCFilter(blockHash chainhash.Hash,
 		return nil, ErrShuttingDown
 	}
 
+	log.Debugf("Gotten CFilter...%v", err)
+
 	// If there are elements left to receive, the query failed.
 	if len(filterQuery.headerIndex) > 0 {
 		numFilters := filterQuery.stopHeight -
@@ -834,12 +858,20 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 	getData := wire.NewMsgGetData()
 	_ = getData.AddInvVect(inv)
 
+	msg := query.ReqMessage{
+		Message: getData,
+	}
 	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(req, resp wire.Message, sp query.Peer, _ *error) query.Progress {
+
+		peer := ""
+		if sp != nil {
+			peer = sp.Addr()
+		}
 		// The request must have been a "getdata" msg.
 		_, ok := req.(*wire.MsgGetData)
 		if !ok {
@@ -852,6 +884,9 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 			return noProgress
 		}
 
+		if sp != nil {
+			sp.UpdateRequestDuration()
+		}
 		// If this isn't the block we asked for, ignore it.
 		if response.BlockHash() != blockHash {
 			return noProgress
@@ -912,15 +947,28 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 
 	// 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)
+
+			newReq := &query.ReqMessage{
+				Message:       newMsg,
+				PriorityIndex: req.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.
@@ -934,6 +982,7 @@ func (s *ChainService) GetBlock(blockHash chainhash.Hash,
 		return nil, ErrShuttingDown
 	}
 
+	log.Debugf("Gotten block, %v", err)
 	if foundBlock == nil {
 		return nil, fmt.Errorf("couldn't retrieve block %s from "+
 			"network", blockHash)

query/interface.go

@@ -43,6 +43,15 @@ type queryOptions struct {
 	// that a query can be retried. If this is set then numRetries has no
 	// effect.
 	noRetryMax bool
+
+	//keepBatch indicates if to delete batch after batch is done. We might need to the keep
+	//tha batch it case of fetching checkpointed block headers where we could be done fetching
+	//headers but we would need to keep the batch as some queries might fail verification and
+	//we need to reschedule the job.
+	keepBatch 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
@@ -75,6 +84,14 @@ func NumRetries(num uint8) QueryOption {
 	}
 }
 
+// KeepBatch is a query option that specifies if to keep batch after
+// query is done.
+func KeepBatch() QueryOption {
+	return func(qo *queryOptions) {
+		qo.keepBatch = true
+	}
+}
+
 // NoRetryMax is a query option that can be used to disable the cap on the
 // number of retries. If this is set then NumRetries has no effect.
 func NoRetryMax() QueryOption {
@@ -83,6 +100,14 @@ func NoRetryMax() QueryOption {
 	}
 }
 
+// ErrChan is a query option that specifies the error channel which the workmanager
+// sends any error to.
+func ErrChan(error chan error) QueryOption {
+	return func(qo *queryOptions) {
+		qo.errChan = error
+	}
+}
+
 // Timeout is a query option that specifies the total time a query is allowed
 // to be tried before it is failed.
 func Timeout(timeout time.Duration) QueryOption {
@@ -107,7 +132,7 @@ func Cancel(cancel chan struct{}) QueryOption {
 	}
 }
 
-// Progress encloses the result of handling a response for a given Request,
+// 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
@@ -125,7 +150,7 @@ type Progress struct {
 // connected peers.
 type Request struct {
 	// Req is the message request to send.
-	Req wire.Message
+	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 +163,22 @@ 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, resp wire.Message, peer Peer, jobErr *error) Progress
+
+	//SendQuery handles sending request to the worker's peer. It returns an error,
+	//if one is encountered while sending the request.
+	SendQuery func(worker Worker, job Task) error
+
+	//CloneReq clones the message.
+	CloneReq func(message ReqMessage) *ReqMessage
+}
+
+// ReqMessage is a struct that contains the wire.Message sent to the peers as well as a
+// priority index, in case the caller wants to manipulate how quickly the message is sent
+// by the workmanager to the worker.
+type ReqMessage struct {
+	wire.Message
+	PriorityIndex float64
 }
 
 // WorkManager defines an API for a manager that dispatches queries to bitcoin
@@ -167,11 +207,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 +219,17 @@ type Peer interface {
 	// OnDisconnect returns a channel that will be closed when this peer is
 	// disconnected.
 	OnDisconnect() <-chan struct{}
+
+	//LastReqDuration returns the last request duration of the peer.
+	LastReqDuration() time.Duration
+
+	//UpdateRequestDuration updates the latest request duration of the peer.
+	UpdateRequestDuration()
+
+	//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 wire.Message) bool
+
+	//IsSyncCandidate returns if the peer is a sync candidate.
+	IsSyncCandidate() bool
 }