Merge pull request #6808 from onflow/tim/6807-verification-engines-remove-Unit

Remove engine.Unit from Assigner engine and Fetcher engine

Author: tim-barry

engine/verification/assigner/engine.go

@@ -8,7 +8,6 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
-	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/model/chunks"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
@@ -24,7 +23,6 @@ import (
 // to me to verify, and then save it to the chunks job queue for the
 // fetcher engine to process.
 type Engine struct {
-	unit                  *engine.Unit
 	log                   zerolog.Logger
 	metrics               module.VerificationMetrics
 	tracer                module.Tracer
@@ -36,8 +34,11 @@ type Engine struct {
 	blockConsumerNotifier module.ProcessingNotifier // to report a block has been processed.
 	stopAtHeight          uint64
 	stopAtBlockID         atomic.Value
+	*module.NoopReadyDoneAware
 }
 
+var _ module.ReadyDoneAware = (*Engine)(nil)
+
 func New(
 	log zerolog.Logger,
 	metrics module.VerificationMetrics,
@@ -50,7 +51,6 @@ func New(
 	stopAtHeight uint64,
 ) *Engine {
 	e := &Engine{
-		unit:             engine.NewUnit(),
 		log:              log.With().Str("engine", "assigner").Logger(),
 		metrics:          metrics,
 		tracer:           tracer,
@@ -69,14 +69,6 @@ func (e *Engine) WithBlockConsumerNotifier(notifier module.ProcessingNotifier) {
 	e.blockConsumerNotifier = notifier
 }
 
-func (e *Engine) Ready() <-chan struct{} {
-	return e.unit.Ready()
-}
-
-func (e *Engine) Done() <-chan struct{} {
-	return e.unit.Done()
-}
-
 // resultChunkAssignment receives an execution result that appears in a finalized incorporating block.
 // In case this verification node is authorized at the reference block of this execution receipt's result,
 // chunk assignment is computed for the result, and the list of assigned chunks returned.
@@ -164,7 +156,8 @@ func (e *Engine) processChunk(chunk *flow.Chunk, resultID flow.Identifier, block
 func (e *Engine) ProcessFinalizedBlock(block *flow.Block) {
 	blockID := block.ID()
 
-	span, ctx := e.tracer.StartBlockSpan(e.unit.Ctx(), blockID, trace.VERProcessFinalizedBlock)
+	// We don't have any existing information and don't need cancellation, so use a background (empty) context
+	span, ctx := e.tracer.StartBlockSpan(context.Background(), blockID, trace.VERProcessFinalizedBlock)
 	defer span.End()
 
 	e.processFinalizedBlock(ctx, block)

engine/verification/fetcher/engine.go

@@ -34,7 +34,6 @@ import (
 // to the verifier engine.
 type Engine struct {
 	// common
-	unit  *engine.Unit
 	state protocol.State // used to verify the origin ID of chunk data response, and sealing status.
 
 	// monitoring
@@ -72,7 +71,6 @@ func New(
 	stopAtHeight uint64,
 ) *Engine {
 	e := &Engine{
-		unit:          engine.NewUnit(),
 		metrics:       metrics,
 		tracer:        tracer,
 		log:           log.With().Str("engine", "fetcher").Logger(),
@@ -104,16 +102,12 @@ 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(func() {
-		<-e.requester.Ready()
-	})
+	return e.requester.Ready()
 }
 
 // Done terminates the engine and returns a channel that is closed when the termination is done
 func (e *Engine) Done() <-chan struct{} {
-	return e.unit.Done(func() {
-		<-e.requester.Done()
-	})
+	return e.requester.Done()
 }
 
 // ProcessAssignedChunk is the entry point of fetcher engine.
@@ -169,7 +163,8 @@ func (e *Engine) ProcessAssignedChunk(locator *chunks.Locator) {
 // processAssignedChunkWithTracing encapsulates the logic of processing assigned chunk with tracing enabled.
 func (e *Engine) processAssignedChunkWithTracing(chunk *flow.Chunk, result *flow.ExecutionResult, chunkLocatorID flow.Identifier) (bool, uint64, error) {
 
-	span, _ := e.tracer.StartBlockSpan(e.unit.Ctx(), result.BlockID, trace.VERProcessAssignedChunk)
+	// We don't have any existing information and don't need cancellation, so use a background (empty) context
+	span, _ := e.tracer.StartBlockSpan(context.Background(), result.BlockID, trace.VERProcessAssignedChunk)
 	span.SetAttributes(attribute.Int("collection_index", int(chunk.CollectionIndex)))
 	defer span.End()