ingestion

Documentation

Index

• Constants
• type BlockExecutor
• type BlockLoader
• type BlockThrottle
  • func NewBlockThrottle(log zerolog.Logger, state protocol.State, execState state.ExecutionState, ...) (*BlockThrottle, error)
  • func (c *BlockThrottle) Init(processables chan<- flow.Identifier) error
  • func (c *BlockThrottle) OnBlock(blockID flow.Identifier) error
  • func (c *BlockThrottle) OnBlockExecuted(_ flow.Identifier, executed uint64) error
  • func (c *BlockThrottle) OnBlockFinalized(lastFinalized *flow.Header)
• type CollectionFetcher
• type Core
  • func NewCore(logger zerolog.Logger, throttle Throttle, execState state.ExecutionState, ...) *Core
  • func (e *Core) Done() <-chan struct{}
  • func (e *Core) OnBlock(header *flow.Header, qc *flow.QuorumCertificate)
  • func (e *Core) OnCollection(col *flow.Collection)
  • func (e *Core) Ready() <-chan struct{}
• type Engine
  • func New(unit *engine.Unit, logger zerolog.Logger, net network.EngineRegistry, ...) (*Engine, error)
  • func (e *Engine) BlockProcessable(b *flow.Header, _ *flow.QuorumCertificate)
  • func (e *Engine) Done() <-chan struct{}
  • func (e *Engine) OnCollection(originID flow.Identifier, entity flow.Entity)
  • func (e *Engine) ProcessLocal(event interface{}) error
  • func (e *Engine) Ready() <-chan struct{}
• type EventConsumer
• type Mempool
  • func (m *Mempool) Run(f func(blockByCollection *stdmap.BlockByCollectionBackdata, ...) error) error
• 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

Types

type BlockExecutor

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

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,
	catchupThreshold int,
) (*BlockThrottle, error)

func (*BlockThrottle) Init

func (c *BlockThrottle) Init(processables chan<- flow.Identifier) error

func (*BlockThrottle) OnBlock

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

func (*BlockThrottle) OnBlockExecuted

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

func (*BlockThrottle) OnBlockFinalized

func (c *BlockThrottle) OnBlockFinalized(lastFinalized *flow.Header)

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 Core

type Core struct {
	// 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,
	headers storage.Headers,
	blocks storage.Blocks,
	collections storage.Collections,
	executor BlockExecutor,
	collectionFetcher CollectionFetcher,
	eventConsumer EventConsumer,
) *Core

func (*Core) Done

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

func (*Core) OnBlock

func (e *Core) OnBlock(header *flow.Header, qc *flow.QuorumCertificate)

func (*Core) OnCollection

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

func (*Core) Ready

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

type Engine

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

An Engine receives and saves incoming blocks.

func New

func New(
	unit *engine.Unit,
	logger zerolog.Logger,
	net network.EngineRegistry,
	collectionFetcher CollectionFetcher,
	headers storage.Headers,
	blocks storage.Blocks,
	collections storage.Collections,
	executionEngine computation.ComputationManager,
	providerEngine provider.ProviderEngine,
	execState state.ExecutionState,
	metrics module.ExecutionMetrics,
	tracer module.Tracer,
	extLog bool,
	pruner *pruner.Pruner,
	uploader *uploader.Manager,
	stopControl *stop.StopControl,
	loader BlockLoader,
) (*Engine, error)

func (*Engine) BlockProcessable

func (e *Engine) BlockProcessable(b *flow.Header, _ *flow.QuorumCertificate)

BlockProcessable handles the new verified blocks (blocks that have passed consensus validation) received from the consensus nodes NOTE: BlockProcessable might be called multiple times for the same block. NOTE: Ready calls reloadUnexecutedBlocks during initialization, which handles dropped protocol events.

func (*Engine) Done

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

Done returns a channel that will close when the engine has successfully stopped.

func (*Engine) OnCollection

func (e *Engine) OnCollection(originID flow.Identifier, entity flow.Entity)

OnCollection is a callback for handling the collections requested by the collection requester.

func (*Engine) ProcessLocal

func (e *Engine) ProcessLocal(event interface{}) error

ProcessLocal processes an event originating on the local node.

func (*Engine) Ready

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

Ready returns a channel that will close when the engine has successfully started.

type EventConsumer

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

type Mempool

type Mempool struct {
	ExecutionQueue    *stdmap.Queues
	BlockByCollection *stdmap.BlockByCollections
}

func (*Mempool) Run

func (m *Mempool) Run(f func(blockByCollection *stdmap.BlockByCollectionBackdata, executionQueue *stdmap.QueuesBackdata) error) error

type Throttle

type Throttle interface {
	Init(processables chan<- flow.Identifier) error
	OnBlock(blockID flow.Identifier) error
	OnBlockExecuted(blockID flow.Identifier, height uint64) error
}