assigner

Documentation

Index

• type BlockConsumer
  • func NewBlockConsumer(log zerolog.Logger, processedHeight storage.ConsumerProgress, ...) (*BlockConsumer, int64, error)
  • func (c *BlockConsumer) Done() <-chan struct{}
  • func (c *BlockConsumer) NotifyJobIsDone(jobID module.JobID)
  • func (c *BlockConsumer) OnBlockIncorporated(*model.Block)
  • func (c *BlockConsumer) OnDoubleProposeDetected(*model.Block, *model.Block)
  • func (c *BlockConsumer) OnFinalizedBlock(block *model.Block)
  • func (c *BlockConsumer) Ready() <-chan struct{}
• type BlockJob
  • func (j BlockJob) ID() module.JobID
• type Engine
  • func New(log zerolog.Logger, metrics module.VerificationMetrics, tracer module.Tracer, ...) *Engine
  • func (e *Engine) Done() <-chan struct{}
  • func (e *Engine) ProcessFinalizedBlock(block *flow.Block)
  • func (e *Engine) Ready() <-chan struct{}
• type FinalizedBlockReader
  • func (r *FinalizedBlockReader) AtIndex(index int64) (module.Job, error)
  • func (r *FinalizedBlockReader) Head() (int64, error)
• type Worker
  • func (w *Worker) Notify(blockID flow.Identifier)
  • func (w *Worker) Run(job module.Job)

Types

type BlockConsumer

type BlockConsumer struct {
	// contains filtered or unexported fields
}

BlockConsumer listens to the OnFinalizedBlock event and notify the consumer to Check in the job queue

func NewBlockConsumer

func NewBlockConsumer(
	log zerolog.Logger,
	processedHeight storage.ConsumerProgress,
	state protocol.State,
	engine *Engine,
	maxProcessing int64,
) (*BlockConsumer, int64, error)

NewBlockConsumer creates a new consumer and returns the default processed index for initializing the processed index in storage.

func (*BlockConsumer) Done

func (c *BlockConsumer) Done() <-chan struct{}

func (*BlockConsumer) NotifyJobIsDone

func (c *BlockConsumer) NotifyJobIsDone(jobID module.JobID)

func (*BlockConsumer) OnBlockIncorporated

func (c *BlockConsumer) OnBlockIncorporated(*model.Block)

To implement FinalizationConsumer

func (*BlockConsumer) OnDoubleProposeDetected

func (c *BlockConsumer) OnDoubleProposeDetected(*model.Block, *model.Block)

To implement FinalizationConsumer

func (*BlockConsumer) OnFinalizedBlock

func (c *BlockConsumer) OnFinalizedBlock(block *model.Block)

func (*BlockConsumer) Ready

func (c *BlockConsumer) Ready() <-chan struct{}

type BlockJob

type BlockJob struct {
	Block *flow.Block
}

BlockJob converts a Block into a Job to be used by job queue

func (BlockJob) ID

func (j BlockJob) ID() module.JobID

ID converts block id into job id, which guarantees uniqueness

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

The Assigner engine reads the receipts from each finalized block. For each receipt, it reads its result and find the chunks the assigned to me to verify, and then save it to the chunks job queue for the fetcher engine to process.

func New

func New(
	log zerolog.Logger,
	metrics module.VerificationMetrics,
	tracer module.Tracer,
	me module.Local,
	state protocol.State,
	assigner module.ChunkAssigner,
	chunksQueue storage.ChunksQueue,
	newChunkListener module.NewJobListener,
) *Engine

func (*Engine) Done

func (e *Engine) Done() <-chan struct{}

func (*Engine) ProcessFinalizedBlock

func (e *Engine) ProcessFinalizedBlock(block *flow.Block)

ProcessFinalizedBlock is the entry point of assigner engine. It pushes the block down the pipeline with tracing on it enabled. Through the pipeline the execution receipts included in the block are indexed, and their chunk assignments are done, and the assigned chunks are pushed to the chunks queue, which is the output stream of this engine. Once the assigner engine is done handling all the receipts in the block, it notifies the block consumer.

func (*Engine) Ready

func (e *Engine) Ready() <-chan struct{}

type FinalizedBlockReader

type FinalizedBlockReader struct {
	// contains filtered or unexported fields
}

FinalizedBlockReader provides an abstraction for consumers to read blocks as job

func (*FinalizedBlockReader) AtIndex

func (r *FinalizedBlockReader) AtIndex(index int64) (module.Job, error)

the job index would just be the finalized block height

func (*FinalizedBlockReader) Head

func (r *FinalizedBlockReader) Head() (int64, error)

Head returns the last finalized height as job index

type Worker

type Worker struct {
	// contains filtered or unexported fields
}

Worker receives job from job consumer and converts it back to Block for engine to process Worker is stateless, it acts as a middleman to convert the job into the entity that the engine is expecting, and translating the id of the entity back to JobID notify the consumer a job is done.

func (*Worker) Notify

func (w *Worker) Notify(blockID flow.Identifier)

Notify is a callback for engine to notify a block has been processed by the given blockID the worker will translate the block ID into jobID and notify the consumer that the job is done.

func (*Worker) Run

func (w *Worker) Run(job module.Job)

Run is a block worker that receives a job corresponding to a finalized block. It then converts the job to a block and passes it to the underlying engine for processing.