consensus

Documentation

Overview

Package consensus provides BFT-like algorithm to distribute list of records between participants

Index

• func CalculateHash(list []*core.ActiveNode) (result []byte, err error)
• type Communicator
• type CommunicatorReceiver
• type Consensus
  • func NewConsensus(communicator Communicator) Consensus
• type NodeKeeper
• type NodeUnsyncHash
  • func CalculateNodeUnsyncHash(nodeID core.RecordRef, list []*core.ActiveNode) (*NodeUnsyncHash, error)
• type Participant
• type Processor
• type UnsyncHolder
• type UnsyncList

Functions

func CalculateHash

func CalculateHash(list []*core.ActiveNode) (result []byte, err error)

CalculateHash calculates hash of active node list

Types

type Communicator

type Communicator interface {
	// ExchangeData used in first consensus step to exchange data between participants
	ExchangeData(ctx context.Context, pulse core.PulseNumber, p Participant, data []*core.ActiveNode) ([]*core.ActiveNode, error)

	// ExchangeHash used in second consensus step to exchange only hashes of merged data vectors
	ExchangeHash(ctx context.Context, pulse core.PulseNumber, p Participant, data []*NodeUnsyncHash) ([]*NodeUnsyncHash, error)
}

Communicator interface is used to exchange messages between participants

type CommunicatorReceiver

type CommunicatorReceiver interface {
	// ExchangeData used in first consensus step to exchange data between participants
	ExchangeData(ctx context.Context, pulse core.PulseNumber, from core.RecordRef, data []*core.ActiveNode) ([]*core.ActiveNode, error)
	// ExchangeHash used in second consensus step to exchange only hashes of merged data vectors
	ExchangeHash(ctx context.Context, pulse core.PulseNumber, from core.RecordRef, data []*NodeUnsyncHash) ([]*NodeUnsyncHash, error)
}

CommunicatorReceiver

type Consensus

type Consensus interface {
	// DoConsensus is sync method, it performs all consensus steps and returns list of synced nodes
	// method should be executed in goroutine
	DoConsensus(ctx context.Context, holder UnsyncHolder, self Participant, allParticipants []Participant) ([]*core.ActiveNode, error)
}

Consensus interface provides method to make consensus between participants

func NewConsensus

func NewConsensus(communicator Communicator) Consensus

NewConsensus creates consensus

type NodeKeeper

type NodeKeeper interface {
	// GetID get current node ID
	GetID() core.RecordRef
	// GetSelf get active node for the current insolard. Returns nil if the current insolard is not an active node.
	GetSelf() *core.ActiveNode
	// GetActiveNode get active node by its reference. Returns nil if node is not found.
	GetActiveNode(ref core.RecordRef) *core.ActiveNode
	// GetActiveNodes get active nodes.
	GetActiveNodes() []*core.ActiveNode
	// GetActiveNodesByRole get active nodes by role
	GetActiveNodesByRole(role core.JetRole) []core.RecordRef
	// AddActiveNodes add active nodes.
	AddActiveNodes([]*core.ActiveNode)
	// SetPulse sets internal PulseNumber to number. Returns true if set was successful, false if number is less
	// or equal to internal PulseNumber. If set is successful, returns collected unsync list and starts collecting new unsync list
	SetPulse(number core.PulseNumber) (bool, UnsyncList)
	// Sync initiates transferring syncCandidates -> sync, sync -> active.
	// If number is less than internal PulseNumber then ignore Sync.
	Sync(syncCandidates []*core.ActiveNode, number core.PulseNumber)
	// AddUnsync add unsync node to the unsync list. Returns channel that receives active node on successful sync.
	// Channel will return nil node if added node has not passed the consensus.
	// Returns error if current node is not active and cannot participate in consensus.
	AddUnsync(nodeID core.RecordRef, roles []core.NodeRole, address string,
		version string) (chan *core.ActiveNode, error)
	// GetUnsyncHolder get unsync list executed in consensus for specific pulse.
	// 1. If pulse is less than internal NodeKeeper pulse, returns error.
	// 2. If pulse is equal to internal NodeKeeper pulse, returns unsync list holder for currently executed consensus.
	// 3. If pulse is more than internal NodeKeeper pulse, blocks till next SetPulse or duration timeout and then acts like in par. 2
	GetUnsyncHolder(pulse core.PulseNumber, duration time.Duration) (UnsyncList, error)
}

NodeKeeper manages unsync, sync and active lists

type NodeUnsyncHash

type NodeUnsyncHash struct {
	NodeID core.RecordRef
	Hash   []byte
}

func CalculateNodeUnsyncHash

func CalculateNodeUnsyncHash(nodeID core.RecordRef, list []*core.ActiveNode) (*NodeUnsyncHash, error)

CalculateNodeUnsyncHash calculates hash for a NodeUnsyncHash

type Participant

type Participant interface {
	GetActiveNode() *core.ActiveNode
}

Participant describes one consensus participant

type Processor

type Processor interface {
	// ProcessPulse is called when we get new pulse from pulsar. Should be called in goroutine
	ProcessPulse(ctx context.Context, pulse core.Pulse)
	// IsPartOfConsensus returns whether we should perform all consensus interactions or not
	IsPartOfConsensus() bool
	// ReceiverHandler return handler that is responsible to handle consensus network requests
	ReceiverHandler() CommunicatorReceiver
	// SetNodeKeeper set NodeKeeper for the processor to integrate Processor with unsync -> sync -> active pipeline
	SetNodeKeeper(keeper NodeKeeper)
}

Processor is an interface to bind all functionality related to consensus with the network layer

type UnsyncHolder

type UnsyncHolder interface {
	// GetUnsync returns list of local unsync nodes. This list is created
	GetUnsync() []*core.ActiveNode
	// GetPulse returns actual pulse for current consensus process.
	GetPulse() core.PulseNumber
	// SetHash sets hash of unsync lists for each node of consensus.
	SetHash([]*NodeUnsyncHash)
	// GetHash get hash of unsync lists for each node of consensus. If hash is not calculated yet, then this call blocks
	// until the hash is calculated with SetHash() call
	GetHash(blockTimeout time.Duration) ([]*NodeUnsyncHash, error)
}

UnsyncHolder

type UnsyncList

type UnsyncList interface {
	// GetUnsync returns list of local unsync nodes. This list is created
	GetUnsync() []*core.ActiveNode
	// GetPulse returns actual pulse for current consensus process.
	GetPulse() core.PulseNumber
	// SetHash sets hash of unsync lists for each node of consensus.
	SetHash([]*NodeUnsyncHash)
	// GetHash get hash of unsync lists for each node of consensus. If hash is not calculated yet, then this call blocks
	// until the hash is calculated with SetHash() call
	GetHash(blockTimeout time.Duration) ([]*NodeUnsyncHash, error)
	// AddUnsyncList add unsync list for remote ref
	AddUnsyncList(ref core.RecordRef, unsync []*core.ActiveNode)
	// AddUnsyncHash add unsync hash for remote ref
	AddUnsyncHash(ref core.RecordRef, hash []*NodeUnsyncHash)
	// GetUnsyncList get unsync list for remote ref
	GetUnsyncList(ref core.RecordRef) ([]*core.ActiveNode, bool)
	// GetUnsyncHash get unsync hash for remote ref
	GetUnsyncHash(ref core.RecordRef) ([]*NodeUnsyncHash, bool)
}