queue

Documentation

Index

• Variables
• func FilterAny(*DecodedSSVMessage) bool
• type DecodedSSVMessage
  • func DecodeSSVMessage(m *spectypes.SSVMessage) (*DecodedSSVMessage, error)
• type Filter
• type MessagePrioritizer
  • func NewMessagePrioritizer(state *State) MessagePrioritizer
• type Metrics
• type Queue
  • func New(capacity int) Queue
  • func NewDefault() Queue
  • func WithMetrics(q Queue, metrics Metrics) Queue
• type State

Variables

var (
	ErrUnknownMessageType = fmt.Errorf("unknown message type")
)

Functions

func FilterAny

func FilterAny(*DecodedSSVMessage) bool

FilterAny returns a Filter that returns true for any message.

Types

type DecodedSSVMessage

type DecodedSSVMessage struct {
	*spectypes.SSVMessage

	// Body is the decoded Data.
	Body interface{} // *SignedMessage | *SignedPartialSignatureMessage | *EventMsg
}

DecodedSSVMessage is a bundle of SSVMessage and it's decoding. TODO: try to make it generic

func DecodeSSVMessage

func DecodeSSVMessage(m *spectypes.SSVMessage) (*DecodedSSVMessage, error)

DecodeSSVMessage decodes an SSVMessage and returns a DecodedSSVMessage.

type Filter

type Filter func(*DecodedSSVMessage) bool

Filter is a function that returns true if the message should be popped.

type MessagePrioritizer

type MessagePrioritizer interface {
	// Prior returns true if message A should be prioritized over B.
	Prior(a, b *DecodedSSVMessage) bool
}

MessagePrioritizer is an interface for prioritizing messages.

func NewMessagePrioritizer

func NewMessagePrioritizer(state *State) MessagePrioritizer

NewMessagePrioritizer returns a standard implementation for MessagePrioritizer which prioritizes messages according to the given State.

type Metrics

type Metrics interface {
	DroppedQueueMessage(messageID spectypes.MessageID)
}

Metrics records metrics about the Queue.

type Queue

type Queue interface {
	// Push blocks until the message is pushed to the queue.
	Push(*DecodedSSVMessage)

	// TryPush returns immediately with true if the message was pushed to the queue,
	// or false if the queue is full.
	TryPush(*DecodedSSVMessage) bool

	// Pop returns and removes the next message in the queue, or blocks until a message is available.
	// When the context is canceled, Pop returns immediately with any leftover message or nil.
	Pop(context.Context, MessagePrioritizer, Filter) *DecodedSSVMessage

	// TryPop returns immediately with the next message in the queue, or nil if there is none.
	TryPop(MessagePrioritizer, Filter) *DecodedSSVMessage

	// Empty returns true if the queue is empty.
	Empty() bool

	// Len returns the number of messages in the queue.
	Len() int
}

Queue is a queue of DecodedSSVMessage with dynamic (per-pop) prioritization.

func New

func New(capacity int) Queue

New returns an implementation of Queue optimized for concurrent push and sequential pop. Pops aren't thread-safe, so don't call Pop from multiple goroutines.

func NewDefault

func NewDefault() Queue

NewDefault returns an implementation of Queue optimized for concurrent push and sequential pop, with a capacity of 32 and a PusherDropping.

func WithMetrics

func WithMetrics(q Queue, metrics Metrics) Queue

WithMetrics returns a wrapping of the given Queue that records metrics using the given Metrics.

type State

type State struct {
	HasRunningInstance bool
	Height             qbft.Height
	Round              qbft.Round
	Slot               phase0.Slot
	Quorum             uint64
}

State represents a portion of the the current state that is relevant to the prioritization of messages.