synchronization

Documentation

Index

• type Config
  • func DefaultConfig() *Config
• type Engine
  • func New(log zerolog.Logger, metrics module.EngineMetrics, net network.Network, ...) (*Engine, error)
  • func (e *Engine) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error
• type OptionFunc
  • func WithPollInterval(interval time.Duration) OptionFunc
  • func WithScanInterval(interval time.Duration) OptionFunc
• type RequestHandler
  • func NewRequestHandler(log zerolog.Logger, metrics module.EngineMetrics, ...) *RequestHandler
  • func (r *RequestHandler) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error
• type RequestHandlerEngine
  • func NewRequestHandlerEngine(logger zerolog.Logger, metrics module.EngineMetrics, net network.Network, ...) (*RequestHandlerEngine, error)
  • func (r *RequestHandlerEngine) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error
• type RequestHeap
  • func NewRequestHeap(limit uint) *RequestHeap
  • func (q *RequestHeap) Get() (*engine.Message, bool)
  • func (q *RequestHeap) Put(message *engine.Message) bool
• type ResponseSender
• type ResponseSenderImpl
  • func NewResponseSender(con network.Conduit) *ResponseSenderImpl
  • func (r *ResponseSenderImpl) SendResponse(res interface{}, target flow.Identifier) error

Types

type Config

type Config struct {
	PollInterval time.Duration
	ScanInterval time.Duration
}

func DefaultConfig

func DefaultConfig() *Config

type Engine

type Engine struct {
	component.Component
	hotstuff.FinalizationConsumer
	// contains filtered or unexported fields
}

Engine is the synchronization engine, responsible for synchronizing chain state.

func New

func New(
	log zerolog.Logger,
	metrics module.EngineMetrics,
	net network.Network,
	me module.Local,
	state protocol.State,
	blocks storage.Blocks,
	comp consensus.Compliance,
	core module.SyncCore,
	participantsProvider module.IdentifierProvider,
	opts ...OptionFunc,
) (*Engine, error)

New creates a new main chain synchronization engine.

func (*Engine) Process

func (e *Engine) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error

Process processes the given event from the node with the given origin ID in a blocking manner. It returns the potential processing error when done.

type OptionFunc

type OptionFunc func(*Config)

func WithPollInterval

func WithPollInterval(interval time.Duration) OptionFunc

WithPollInterval sets a custom interval at which we scan for poll items

func WithScanInterval

func WithScanInterval(interval time.Duration) OptionFunc

WithScanInterval sets a custom interval at which we scan for pending items and batch them for requesting.

type RequestHandler

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

RequestHandler encapsulates message queues and processing logic for the sync engine. It logically separates request processing from active participation (sending requests), primarily to simplify nodes which bridge the public and private networks.

The RequestHandlerEngine embeds RequestHandler to create an engine which only responds to requests on the public network (does not send requests over this network). The Engine embeds RequestHandler and additionally includes logic for sending sync requests.

Although the RequestHandler defines a notifier, message queue, and processing worker logic, it is not itself a component.Component and does not manage any worker threads. The containing engine is responsible for starting the worker threads for processing requests.

func NewRequestHandler

func NewRequestHandler(
	log zerolog.Logger,
	metrics module.EngineMetrics,
	responseSender ResponseSender,
	me module.Local,
	finalizedHeaderCache *events.FinalizedHeaderCache,
	blocks storage.Blocks,
	core module.SyncCore,
	queueMissingHeights bool,
) *RequestHandler

func (*RequestHandler) Process

func (r *RequestHandler) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error

Process processes the given event from the node with the given origin ID in a blocking manner. No errors are expected during normal operation.

type RequestHandlerEngine

type RequestHandlerEngine struct {
	component.Component
	hotstuff.FinalizationConsumer
	// contains filtered or unexported fields
}

RequestHandlerEngine is an engine which operates only the request-handling portion of the block sync protocol. It is used by Access/Observer nodes attached to the public network, enabling them to provide block synchronization data to nodes on the public network, but not requesting any data from these nodes. (Requests are sent only on the private network.)

func NewRequestHandlerEngine

func NewRequestHandlerEngine(
	logger zerolog.Logger,
	metrics module.EngineMetrics,
	net network.Network,
	me module.Local,
	state protocol.State,
	blocks storage.Blocks,
	core module.SyncCore,
) (*RequestHandlerEngine, error)

func (*RequestHandlerEngine) Process

func (r *RequestHandlerEngine) Process(channel channels.Channel, originID flow.Identifier, event interface{}) error

type RequestHeap

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

RequestHeap is a special structure that implements engine.MessageStore interface and indexes requests by originator. If request will be sent by same originator then it will replace the old one. Comparing to default FIFO queue this one can contain MAX one request for origin ID. Getting value from queue as well as ejecting is pseudo-random.

func NewRequestHeap

func NewRequestHeap(limit uint) *RequestHeap

func (*RequestHeap) Get

func (q *RequestHeap) Get() (*engine.Message, bool)

Get returns pseudo-random element from request storage using go map properties.

func (*RequestHeap) Put

func (q *RequestHeap) Put(message *engine.Message) bool

Put stores message into requests map using OriginID as key. Returns always true

type ResponseSender

type ResponseSender interface {
	SendResponse(interface{}, flow.Identifier) error
}

type ResponseSenderImpl

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

func NewResponseSender

func NewResponseSender(con network.Conduit) *ResponseSenderImpl

func (*ResponseSenderImpl) SendResponse

func (r *ResponseSenderImpl) SendResponse(res interface{}, target flow.Identifier) error