fetcher

Documentation

Index

• func IsChunkDataPackValidationError(err error) bool
• func NewChunkDataPackValidationError(originID flow.Identifier, resultID flow.Identifier, chunkIndex uint64, ...) error
• type AssignedChunkProcessor
• type ChunkDataPackHandler
• type ChunkDataPackRequester
• type ChunkDataPackValidationError
  • func (c ChunkDataPackValidationError) Error() string
• type Engine
  • func New(log zerolog.Logger, metrics module.VerificationMetrics, tracer module.Tracer, ...) *Engine
  • func (e *Engine) Done() <-chan struct{}
  • func (e *Engine) HandleChunkDataPack(originID flow.Identifier, response *verification.ChunkDataPackResponse)
  • func (e *Engine) NotifyChunkDataPackSealed(chunkIndex uint64, resultID flow.Identifier)
  • func (e *Engine) ProcessAssignedChunk(locator *chunks.Locator)
  • func (e *Engine) Ready() <-chan struct{}
  • func (e *Engine) WithChunkConsumerNotifier(notifier module.ProcessingNotifier)

Functions

func IsChunkDataPackValidationError

func IsChunkDataPackValidationError(err error) bool

func NewChunkDataPackValidationError

func NewChunkDataPackValidationError(originID flow.Identifier,
	resultID flow.Identifier,
	chunkIndex uint64,
	chunkDataPackID flow.Identifier,
	chunkID flow.Identifier,
	collectionID flow.Identifier,
	err error) error

Types

type AssignedChunkProcessor

type AssignedChunkProcessor interface {
	module.ReadyDoneAware
	// ProcessAssignedChunk receives an assigned chunk locator and processes its corresponding chunk.
	// A chunk processor is expected to shape a verifiable chunk out of the assigned chunk, and pass it to
	// the verifier Engine.
	// Note: it should be implemented in a non-blocking way.
	ProcessAssignedChunk(locator *chunks.Locator)

	// WithChunkConsumerNotifier sets the notifier of this chunk processor.
	// The notifier is called by the internal logic of the processor to let the consumer know that
	// the processor is done by processing a chunk so that the next chunk may be passed to the processor
	// by the consumer through invoking ProcessAssignedChunk of this processor.
	WithChunkConsumerNotifier(notifier module.ProcessingNotifier)
}

AssignedChunkProcessor should be implemented by the verification node engine responsible for processing assigned chunk locators to this node.

In the current version, the fetcher engine is responsible of processing the assigned chunk locators. From the architectural perspective, AssignedChunkProcessor aligns as following on the verification pipeline: ---------------- ------------------ --------------------------- | Chunk Queue | ---> assigned chunk locators --> | Chunk Consumer | ---> chunk workers --> | Assigned Chunk Processor| ---------------- ------------------ ---------------------------

type ChunkDataPackHandler

type ChunkDataPackHandler interface {
	// HandleChunkDataPack is called by the ChunkDataPackRequester anytime a new requested chunk arrives.
	// It contains the logic of handling the chunk data pack.
	HandleChunkDataPack(originID flow.Identifier, response *verification.ChunkDataPackResponse)

	// NotifyChunkDataPackSealed is called by the ChunkDataPackRequester to notify the ChunkDataPackHandler that the specified chunk
	// has been sealed and hence the requester will no longer request it.
	//
	// When the requester calls this callback method, it will never return a chunk data pack for this specified chunk to the handler (i.e.,
	// through HandleChunkDataPack).
	NotifyChunkDataPackSealed(chunkIndex uint64, resultID flow.Identifier)
}

ChunkDataPackHandler encapsulates the logic of handling a requested chunk data pack upon its arrival.

type ChunkDataPackRequester

type ChunkDataPackRequester interface {
	module.ReadyDoneAware
	// Request makes the request of chunk data pack for the specified chunk ID with the specified targets.
	Request(request *verification.ChunkDataPackRequest)
	// WithChunkDataPackHandler registers the handler component of requester engine. The handler implements the logic of handling
	// a requested chunk data pack upon its arrival.
	WithChunkDataPackHandler(handler ChunkDataPackHandler)
}

ChunkDataPackRequester encapsulates the logic of requesting a chunk data pack from an execution node.

type ChunkDataPackValidationError

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

func (ChunkDataPackValidationError) Error

func (c ChunkDataPackValidationError) Error() string

type Engine

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

Engine implements the fetcher engine functionality. It works between a chunk consumer queue, and a verifier engine. Its input is an assigned chunk locator from the chunk consumer it is subscribed to.

Its output is a verifiable chunk that it passes to the verifier engine.

Fetcher engine is an AssignedChunkProcessor implementation: it receives assigned chunks to this verification node from the chunk consumer. The assigned chunks are passed on concurrent executions of its ProcessAssignedChunk method.

On receiving an assigned chunk, the engine requests their chunk data pack through the requester that is attached to it. On receiving a chunk data pack response, the fetcher engine validates it, and shapes a verifiable chunk out of it, and passes it to the verifier engine.

func New

func New(
	log zerolog.Logger,
	metrics module.VerificationMetrics,
	tracer module.Tracer,
	verifier network.Engine,
	state protocol.State,
	pendingChunks mempool.ChunkStatuses,
	headers storage.Headers,
	blocks storage.Blocks,
	results storage.ExecutionResults,
	receipts storage.ExecutionReceipts,
	requester ChunkDataPackRequester,
	stopAtHeight uint64,
) *Engine

func (*Engine) Done

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

Done terminates the engine and returns a channel that is closed when the termination is done

func (*Engine) HandleChunkDataPack

func (e *Engine) HandleChunkDataPack(originID flow.Identifier, response *verification.ChunkDataPackResponse)

HandleChunkDataPack is called by the chunk requester module everytime a new requested chunk data pack arrives. The chunks are supposed to be deduplicated by the requester. So invocation of this method indicates arrival of a distinct requested chunk.

func (*Engine) NotifyChunkDataPackSealed

func (e *Engine) NotifyChunkDataPackSealed(chunkIndex uint64, resultID flow.Identifier)

NotifyChunkDataPackSealed is called by the ChunkDataPackRequester to notify the ChunkDataPackHandler that the specified chunk has been sealed and hence the requester will no longer request it.

When the requester calls this callback method, it will never return a chunk data pack for this specified chunk to the handler (i.e., through HandleChunkDataPack).

func (*Engine) ProcessAssignedChunk

func (e *Engine) ProcessAssignedChunk(locator *chunks.Locator)

ProcessAssignedChunk is the entry point of fetcher engine. It pushes the assigned chunk down the pipeline. Through the pipeline the chunk data pack for this chunk is requested, a verifiable chunk is shaped for it, and is pushed to the verifier engine for verification.

It should not be blocking since multiple chunk consumer workers might be calling it concurrently. It fetches the chunk data pack, once received, verifier engine will be verifying Once a chunk has been processed, it will call the processing notifier callback to notify the chunk consumer in order to process the next chunk.

func (*Engine) Ready

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

Ready initializes the engine and returns a channel that is closed when the initialization is done

func (*Engine) WithChunkConsumerNotifier

func (e *Engine) WithChunkConsumerNotifier(notifier module.ProcessingNotifier)

WithChunkConsumerNotifier sets the processing notifier of fetcher. The fetcher engine uses this notifier to inform the chunk consumer that it is done processing a given chunk, and is ready to receive a new chunk to process.