onflow · yhassanzadeh13 · Jun 11, 2021 · Apr 19, 2021 · Apr 19, 2021 · Apr 19, 2021
diff --git a/engine/testutil/nodes.go b/engine/testutil/nodes.go
@@ -37,7 +37,7 @@ import (
 	"github.com/onflow/flow-go/engine/verification/fetcher/chunkconsumer"
 	"github.com/onflow/flow-go/engine/verification/finder"
 	"github.com/onflow/flow-go/engine/verification/match"
-	verificationrequester "github.com/onflow/flow-go/engine/verification/requester"
+	vereq "github.com/onflow/flow-go/engine/verification/requester"
 	"github.com/onflow/flow-go/engine/verification/verifier"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/ledger/common/pathfinder"
@@ -889,19 +889,22 @@ func NewVerificationNode(t testing.TB,
 	}
 
 	if node.RequesterEngine == nil {
-		node.RequesterEngine, err = verificationrequester.New(node.Log,
+		node.RequesterEngine, err = vereq.New(node.Log,
 			node.State,
 			node.Net,
 			node.Tracer,
 			collector,
 			node.ChunkRequests,
-			100*time.Millisecond,
+			vereq.DefaultRequestInterval,
 			// requests are only qualified if their retryAfter is elapsed.
-			verificationrequester.RetryAfterQualifier,
+			vereq.RetryAfterQualifier,
 			// exponential backoff with multiplier of 2, minimum interval of a second, and
 			// maximum interval of an hour.
-			mempool.ExponentialUpdater(2, time.Hour, time.Second),
-			2)
+			mempool.ExponentialUpdater(
+				vereq.DefaultBackoffMultiplier,
+				vereq.DefaultBackoffMaxInterval,
+				vereq.DefaultBackoffMinInterval),
+			vereq.DefaultRequestTargets)
 
 		require.NoError(t, err)
 	}
@@ -925,7 +928,7 @@ func NewVerificationNode(t testing.TB,
 			node.ProcessedChunkIndex,
 			node.ChunksQueue,
 			node.FetcherEngine,
-			int64(3)) // defaults number of workers to 3.
+			chunkconsumer.DefaultChunkWorkers) // defaults number of workers to 3.
 	}
 
 	if node.AssignerEngine == nil {
@@ -945,7 +948,7 @@ func NewVerificationNode(t testing.TB,
 			node.Blocks,
 			node.State,
 			node.AssignerEngine,
-			int64(3)) // defaults number of workers to 3.
+			blockconsumer.DefaultBlockWorkers)
 		require.NoError(t, err)
 	}
 

@@ -14,6 +14,9 @@ import (
 	"github.com/onflow/flow-go/storage"
 )
 
+// DefaultBlockWorkers is the number of blocks processed in parallel.
+const DefaultBlockWorkers = uint64(5)
+
 // BlockConsumer listens to the OnFinalizedBlock event
 // and notifies the consumer to check in the job queue
 // (i.e., its block reader) for new block jobs.
@@ -42,7 +45,7 @@ func NewBlockConsumer(log zerolog.Logger,
 	blocks storage.Blocks,
 	state protocol.State,
 	blockProcessor assigner.FinalizedBlockProcessor,
-	maxProcessing int64) (*BlockConsumer, uint64, error) {
+	maxProcessing uint64) (*BlockConsumer, uint64, error) {
 
 	lg := log.With().Str("module", "block_consumer").Logger()
 

diff --git a/engine/verification/assigner/blockconsumer/consumer_test.go b/engine/verification/assigner/blockconsumer/consumer_test.go
@@ -116,7 +116,7 @@ func withConsumer(
 	withBlockConsumer func(*blockconsumer.BlockConsumer, []*flow.Block),
 ) {
 	unittest.RunWithBadgerDB(t, func(db *badger.DB) {
-		maxProcessing := int64(workerCount)
+		maxProcessing := uint64(workerCount)
 
 		processedHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressVerificationBlockHeight)
 		collector := &metrics.NoopCollector{}

@@ -12,7 +12,8 @@ import (
 )
 
 const (
-	DefaultJobIndex = uint64(0)
+	DefaultJobIndex     = uint64(0)
+	DefaultChunkWorkers = uint64(20)
 )
 
 // ChunkConsumer consumes the jobs from the job queue, and pass it to the
@@ -28,7 +29,7 @@ func NewChunkConsumer(
 	processedIndex storage.ConsumerProgress, // to persist the processed index
 	chunksQueue storage.ChunksQueue, // to read jobs (chunks) from
 	engine fetcher.AssignedChunkProcessor, // to process jobs (chunks)
-	maxProcessing int64, // max number of jobs to be processed in parallel
+	maxProcessing uint64, // max number of jobs to be processed in parallel
 ) *ChunkConsumer {
 	worker := NewWorker(engine)
 	engine.WithChunkConsumerNotifier(worker)

@@ -141,7 +141,7 @@ func WithConsumer(
 	withConsumer func(*chunkconsumer.ChunkConsumer, *storage.ChunksQueue),
 ) {
 	unittest.RunWithBadgerDB(t, func(db *badger.DB) {
-		maxProcessing := int64(3)
+		maxProcessing := uint64(3)
 
 		processedIndex := storage.NewConsumerProgress(db, module.ConsumeProgressVerificationChunkIndex)
 		chunksQueue := storage.NewChunkQueue(db)

@@ -94,9 +94,6 @@ func (e *Engine) WithChunkConsumerNotifier(notifier module.ProcessingNotifier) {
 
 // Ready initializes the engine and returns a channel that is closed when the initialization is done
 func (e *Engine) Ready() <-chan struct{} {
-	if e.chunkConsumerNotifier == nil {
-		e.log.Fatal().Msg("missing chunk consumer notifier callback in verification fetcher engine")
-	}
 	return e.unit.Ready()
 }
 

diff --git a/engine/verification/requester/requester.go b/engine/verification/requester/requester.go
@@ -22,6 +22,23 @@ import (
 	"github.com/onflow/flow-go/utils/logging"
 )
 
+const (
+	// DefaultRequestInterval is the time interval that requester engine tries requesting chunk data packs.
+	DefaultRequestInterval = 1000 * time.Millisecond
+
+	// DefaultBackoffMultiplier is the base of exponent in exponential backoff multiplier for backing off requests for chunk data packs.
+	DefaultBackoffMultiplier = float64(2)
+
+	// DefaultBackoffMinInterval is the minimum time interval a chunk data pack request waits before dispatching.
+	DefaultBackoffMinInterval = 1 * time.Millisecond
+
+	// DefaultBackoffMaxInterval is the maximum time interval a chunk data pack request waits before dispatching.
+	DefaultBackoffMaxInterval = 1 * time.Minute
+
+	// DefaultRequestTargets is the  maximum number of execution nodes a chunk data pack request is dispatched to.
+	DefaultRequestTargets = 2
+)
+
 // Engine implements a ChunkDataPackRequester that is responsible of receiving chunk data pack requests,
 // dispatching it to the execution nodes, receiving the requested chunk data pack from execution nodes,
 // and passing it to the registered handler.
@@ -41,7 +58,7 @@ type Engine struct {
 
 	// internal logic
 	retryInterval    time.Duration                          // determines time in milliseconds for retrying chunk data requests.
-	requestTargets   uint                                   // maximum number of execution nodes being asked for a chunk data pack.
+	requestTargets   uint64                                 // maximum number of execution nodes being asked for a chunk data pack.
 	pendingRequests  mempool.ChunkRequests                  // used to track requested chunks.
 	reqQualifierFunc RequestQualifierFunc                   // used to decide whether to dispatch a request at a certain cycle.
 	reqUpdaterFunc   mempool.ChunkRequestHistoryUpdaterFunc // used to atomically update chunk request info on mempool.
@@ -56,7 +73,7 @@ func New(log zerolog.Logger,
 	retryInterval time.Duration,
 	reqQualifierFunc RequestQualifierFunc,
 	reqUpdaterFunc mempool.ChunkRequestHistoryUpdaterFunc,
-	requestTargets uint) (*Engine, error) {
+	requestTargets uint64) (*Engine, error) {
 
 	e := &Engine{
 		log:              log.With().Str("engine", "requester").Logger(),
@@ -116,10 +133,6 @@ func (e *Engine) Process(originID flow.Identifier, event interface{}) error {
 
 // Ready initializes the engine and returns a channel that is closed when the initialization is done.
 func (e *Engine) Ready() <-chan struct{} {
-	if e.handler == nil {
-		e.log.Fatal().Msg("could not start requester engine with missing chunk data pack handler")
-	}
-
 	delay := time.Duration(0)
 	// run a periodic check to retry requesting chunk data packs.
 	// if onTimer takes longer than retryInterval, the next call will be blocked until the previous

diff --git a/engine/verification/requester/requester_test.go b/engine/verification/requester/requester_test.go
@@ -41,7 +41,7 @@ type RequesterEngineTestSuite struct {
 	verIdentity *flow.Identity // verification node
 
 	// parameters
-	requestTargets uint
+	requestTargets uint64
 	retryInterval  time.Duration // determines time in milliseconds for retrying chunk data requests.
 }
 

diff --git a/module/jobqueue/consumer.go b/module/jobqueue/consumer.go
@@ -29,7 +29,7 @@ type Consumer struct {
 	worker Worker // to process job and notify consumer when finish processing a job
 
 	// Config
-	maxProcessing int64 // max number of jobs to be processed concurrently
+	maxProcessing uint64 // max number of jobs to be processed concurrently
 
 	// State Variables
 	running bool // a signal to control whether to start processing more jobs. Useful for waiting
@@ -49,7 +49,7 @@ func NewConsumer(
 	jobs module.Jobs,
 	progress storage.ConsumerProgress,
 	worker Worker,
-	maxProcessing int64,
+	maxProcessing uint64,
 ) *Consumer {
 	return &Consumer{
 		log: log.With().Str("sub_module", "job_queue").Logger(),
@@ -259,12 +259,13 @@ func (c *Consumer) processableJobs() ([]*jobAtIndex, uint64, error) {
 // processableJobs check the worker's capacity and if sufficient, read
 // jobs from the storage, return the processable jobs, and the processed
 // index
-func processableJobs(jobs module.Jobs, processings map[uint64]*jobStatus, maxProcessing int64, processedIndex uint64) ([]*jobAtIndex, uint64, error) {
+func processableJobs(jobs module.Jobs, processings map[uint64]*jobStatus, maxProcessing uint64, processedIndex uint64) ([]*jobAtIndex, uint64,
+	error) {
 	processables := make([]*jobAtIndex, 0)
 
 	// count how many jobs are still processing,
 	// in order to decide whether to process a new job
-	processing := int64(0)
+	processing := uint64(0)
 
 	// if still have processing capacity, find the next processable job
 	for i := processedIndex + 1; processing < maxProcessing; i++ {

diff --git a/module/jobqueue/consumer_behavior_test.go b/module/jobqueue/consumer_behavior_test.go
@@ -450,7 +450,7 @@ func assertProcessed(t testing.TB, cp storage.ConsumerProgress, expectProcessed
 
 func newTestConsumer(cp storage.ConsumerProgress, jobs module.Jobs, worker jobqueue.Worker) module.JobConsumer {
 	log := unittest.Logger().With().Str("module", "consumer").Logger()
-	maxProcessing := int64(3)
+	maxProcessing := uint64(3)
 	c := jobqueue.NewConsumer(log, jobs, cp, worker, maxProcessing)
 	return c
 }

diff --git a/module/jobqueue/consumer_test.go b/module/jobqueue/consumer_test.go
@@ -19,7 +19,7 @@ import (
 func TestProcessableJobs(t *testing.T) {
 	t.Parallel()
 
-	maxProcessing := int64(3)
+	maxProcessing := uint64(3)
 
 	t.Run("no job, nothing to process", func(t *testing.T) {
 		jobs := NewMockJobs() // no job in the queue
@@ -136,7 +136,7 @@ func TestProcessedIndexDeletion(t *testing.T) {
 			jobs := NewMockJobs()
 			progress := badger.NewConsumerProgress(db, "consumer")
 			worker := newMockWorker()
-			maxProcessing := int64(3)
+			maxProcessing := uint64(3)
 			c := NewConsumer(log, jobs, progress, worker, maxProcessing)
 			worker.WithConsumer(c)
 

diff --git a/network/mocknetwork/message_processor.go b/network/mocknetwork/message_processor.go