ingestion

Documentation

Index

• Constants
• func NewMachine(logger zerolog.Logger, protocolEvents *events.Distributor, ...) (*Machine, *Core, error)
• type BlockExecutor
• type BlockIDHeight
  • func HeaderToBlockIDHeight(header *flow.Header) BlockIDHeight
• type BlockLoader
• type BlockThrottle
  • func NewBlockThrottle(log zerolog.Logger, state protocol.State, execState state.ExecutionState, ...) (*BlockThrottle, error)
  • func (c *BlockThrottle) Done() error
  • func (c *BlockThrottle) Init(processables chan<- BlockIDHeight, threshold int) error
  • func (c *BlockThrottle) OnBlock(blockID flow.Identifier, height uint64) error
  • func (c *BlockThrottle) OnBlockExecuted(_ flow.Identifier, executed uint64) error
  • func (c *BlockThrottle) OnBlockFinalized(finalizedHeight uint64)
• type CollectionFetcher
• type CollectionRequester
• type Core
  • func NewCore(logger zerolog.Logger, throttle Throttle, execState state.ExecutionState, ...) (*Core, error)
  • func (e *Core) OnCollection(col *flow.Collection)
• type EventConsumer
• type Machine
  • func (e *Machine) BeforeComputationResultSaved(ctx context.Context, result *execution.ComputationResult)
  • func (e *Machine) BlockFinalized(b *flow.Header)
  • func (e *Machine) BlockProcessable(header *flow.Header, qc *flow.QuorumCertificate)
  • func (e *Machine) ExecuteBlock(ctx context.Context, executableBlock *entity.ExecutableBlock) (*execution.ComputationResult, error)
  • func (e *Machine) OnComputationResultSaved(ctx context.Context, result *execution.ComputationResult) string
• type Throttle

Constants

const DefaultCatchUpThreshold = 500

DefaultCatchUpThreshold is the number of blocks that if the execution is far behind the finalization then we will only lazy load the next unexecuted finalized blocks until the execution has caught up

const MaxConcurrentBlockExecutor = 5

MaxConcurrentBlockExecutor is the maximum number of concurrent block executors

const MaxProcessableBlocks = 10000

MaxProcessableBlocks is the maximum number of blocks that is queued to be processed

Functions

func NewMachine

func NewMachine(
	logger zerolog.Logger,
	protocolEvents *events.Distributor,
	collectionRequester CollectionRequester,

	collectionFetcher CollectionFetcher,
	headers storage.Headers,
	blocks storage.Blocks,
	collections storage.Collections,
	execState state.ExecutionState,
	state protocol.State,
	metrics module.ExecutionMetrics,
	computationManager computation.ComputationManager,
	broadcaster provider.ProviderEngine,
	uploader *uploader.Manager,
	stopControl *stop.StopControl,
) (*Machine, *Core, error)

Types

type BlockExecutor

type BlockExecutor interface {
	ExecuteBlock(ctx context.Context, block *entity.ExecutableBlock) (*execution.ComputationResult, error)
}

type BlockIDHeight

type BlockIDHeight struct {
	ID     flow.Identifier
	Height uint64
}

BlockIDHeight is a helper struct that holds the block ID and height

func HeaderToBlockIDHeight

func HeaderToBlockIDHeight(header *flow.Header) BlockIDHeight

type BlockLoader

type BlockLoader interface {
	LoadUnexecuted(ctx context.Context) ([]flow.Identifier, error)
}

type BlockThrottle

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

BlockThrottle is a helper struct that helps throttle the unexecuted blocks to be sent to the block queue for execution. It is useful for case when execution is falling far behind the finalization, in which case we want to throttle the blocks to be sent to the block queue for fetching data to execute them. Without throttle, the block queue will be flooded with blocks, and the network will be flooded with requests fetching collections, and the EN might quickly run out of memory.

func NewBlockThrottle

func NewBlockThrottle(
	log zerolog.Logger,
	state protocol.State,
	execState state.ExecutionState,
	headers storage.Headers,
) (*BlockThrottle, error)

func (*BlockThrottle) Done

func (c *BlockThrottle) Done() error

Done marks the throttle as done, and no more blocks will be processed

func (*BlockThrottle) Init

func (c *BlockThrottle) Init(processables chan<- BlockIDHeight, threshold int) error

func (*BlockThrottle) OnBlock

func (c *BlockThrottle) OnBlock(blockID flow.Identifier, height uint64) error

func (*BlockThrottle) OnBlockExecuted

func (c *BlockThrottle) OnBlockExecuted(_ flow.Identifier, executed uint64) error

func (*BlockThrottle) OnBlockFinalized

func (c *BlockThrottle) OnBlockFinalized(finalizedHeight uint64)

type CollectionFetcher

type CollectionFetcher interface {
	// FetchCollection decides which collection nodes to fetch the collection from
	// No error is expected during normal operation
	FetchCollection(blockID flow.Identifier, height uint64, guarantee *flow.CollectionGuarantee) error

	// Force forces the requests to be sent immediately
	Force()
}

CollectionFetcher abstracts the details of how to fetch collection

type CollectionRequester

type CollectionRequester interface {
	module.ReadyDoneAware
	WithHandle(requester.HandleFunc)
}

type Core

type Core struct {
	*component.ComponentManager
	// contains filtered or unexported fields
}

Core connects the execution components when it receives blocks and collections, it forwards them to the block queue. when the block queue decides to execute blocks, it forwards to the executor for execution when the block queue decides to fetch missing collections, it forwards to the collection fetcher when a block is executed, it notifies the block queue and forwards to execution state to save them.

func NewCore

func NewCore(
	logger zerolog.Logger,
	throttle Throttle,
	execState state.ExecutionState,
	stopControl *stop.StopControl,
	blocks storage.Blocks,
	collections storage.Collections,
	executor BlockExecutor,
	collectionFetcher CollectionFetcher,
	eventConsumer EventConsumer,
	metrics module.ExecutionMetrics,
) (*Core, error)

func (*Core) OnCollection

func (e *Core) OnCollection(col *flow.Collection)

type EventConsumer

type EventConsumer interface {
	BeforeComputationResultSaved(ctx context.Context, result *execution.ComputationResult)
	OnComputationResultSaved(ctx context.Context, result *execution.ComputationResult) string
}

type Machine

type Machine struct {
	events.Noop // satisfy protocol events consumer interface
	// contains filtered or unexported fields
}

Machine forwards blocks and collections to the core for processing.

func (*Machine) BeforeComputationResultSaved

func (e *Machine) BeforeComputationResultSaved(
	ctx context.Context,
	result *execution.ComputationResult,
)

func (*Machine) BlockFinalized

func (e *Machine) BlockFinalized(b *flow.Header)

func (*Machine) BlockProcessable

func (e *Machine) BlockProcessable(header *flow.Header, qc *flow.QuorumCertificate)

Protocol Events implementation

func (*Machine) ExecuteBlock

func (e *Machine) ExecuteBlock(ctx context.Context, executableBlock *entity.ExecutableBlock) (*execution.ComputationResult, error)

func (*Machine) OnComputationResultSaved

func (e *Machine) OnComputationResultSaved(
	ctx context.Context,
	result *execution.ComputationResult,
) string

type Throttle

type Throttle interface {
	// Init initializes the throttle with the processables channel to forward the blocks
	Init(processables chan<- BlockIDHeight, threshold int) error
	// OnBlock is called when a block is received, the throttle will check if the execution
	// is falling far behind the finalization, and add the block to the processables channel
	// if it's not falling far behind.
	OnBlock(blockID flow.Identifier, height uint64) error
	// OnBlockExecuted is called when a block is executed, the throttle will check whether
	// the execution is caught up with the finalization, and allow all the remaining blocks
	// to be added to the processables channel.
	OnBlockExecuted(blockID flow.Identifier, height uint64) error
	// OnBlockFinalized is called when a block is finalized, the throttle will update the
	// finalized height.
	OnBlockFinalized(height uint64)
	// Done stops the throttle, and stop sending new blocks to the processables channel
	Done() error
}

Throttle is used to throttle the blocks to be added to the processables channel