network

Documentation

Index

• type ClaimQueue
• type ConsensusNetwork
• type ConsensusRequestHandler
• type Controller
• type Future
• type HostNetwork
• type InternalTransport
• type NodeKeeper
• type NodeKeeperState
• type Packet
• type PartitionPolicy
• type PulseHandler
• type Request
• type RequestBuilder
• type RequestHandler
• type RequestID
• type Response
• type RoutingTable
• type UnsyncList

Types

type ClaimQueue

type ClaimQueue interface {
	// Pop takes claim from the queue.
	Pop() consensus.ReferendumClaim
	// Front returns claim from the queue without removing it from the queue.
	Front() consensus.ReferendumClaim
	// Length returns the length of the queue
	Length() int
}

ClaimQueue is the queue that contains consensus claims.

type ConsensusNetwork

type ConsensusNetwork interface {
	// Start listening to network requests.
	Start(ctx context.Context)
	// Stop listening to network requests.
	Stop()
	// PublicAddress returns public address that can be published for all nodes.
	PublicAddress() string
	// GetNodeID get current node ID.
	GetNodeID() core.RecordRef

	// SendRequest send request to a remote node.
	SendRequest(request Request, receiver core.RecordRef) error
	// RegisterRequestHandler register a handler function to process incoming requests of a specific type.
	RegisterRequestHandler(t types.PacketType, handler ConsensusRequestHandler)
	// NewRequestBuilder create packet builder for an outgoing request with sender set to current node.
	NewRequestBuilder() RequestBuilder
}

type ConsensusRequestHandler

type ConsensusRequestHandler func(Request)

type Controller

type Controller interface {
	// SendParcel send message to nodeID.
	SendMessage(nodeID core.RecordRef, name string, msg core.Parcel) ([]byte, error)
	// RemoteProcedureRegister register remote procedure that will be executed when message is received.
	RemoteProcedureRegister(name string, method core.RemoteProcedure)
	// SendCascadeMessage sends a message from MessageBus to a cascade of nodes.
	SendCascadeMessage(data core.Cascade, method string, msg core.Parcel) error
	// Bootstrap init complex bootstrap process. Blocks until bootstrap is complete.
	Bootstrap(ctx context.Context) error

	// Inject inject components.
	Inject(cryptographyService core.CryptographyService,
		networkCoordinator core.NetworkCoordinator, nodeKeeper NodeKeeper)

	// SetLastIgnoredPulse set pulse number after which we will begin setting new pulses to PulseManager
	SetLastIgnoredPulse(number core.PulseNumber)
	// GetLastIgnoredPulse get last pulse that will be ignored
	GetLastIgnoredPulse() core.PulseNumber
}

Controller contains network logic.

type Future

type Future interface {
	GetRequest() Request
	Response() <-chan Response
	GetResponse(duration time.Duration) (Response, error)
}

Future allows to handle responses to a previously sent request.

type HostNetwork

type HostNetwork interface {
	// Start listening to network requests.
	Start(ctx context.Context)
	// Stop listening to network requests.
	Stop()
	// PublicAddress returns public address that can be published for all nodes.
	PublicAddress() string
	// GetNodeID get current node ID.
	GetNodeID() core.RecordRef

	// SendRequest send request to a remote node.
	SendRequest(ctx context.Context, request Request, receiver core.RecordRef) (Future, error)
	// RegisterRequestHandler register a handler function to process incoming requests of a specific type.
	RegisterRequestHandler(t types.PacketType, handler RequestHandler)
	// NewRequestBuilder create packet builder for an outgoing request with sender set to current node.
	NewRequestBuilder() RequestBuilder
	// BuildResponse create response to an incoming request with Data set to responseData.
	BuildResponse(ctx context.Context, request Request, responseData interface{}) Response
}

HostNetwork simple interface to send network requests and process network responses.

type InternalTransport

type InternalTransport interface {
	// Start listening to network requests, should be started in goroutine.
	Start(ctx context.Context)
	// Stop listening to network requests.
	Stop()
	// PublicAddress returns public address that can be published for all nodes.
	PublicAddress() string
	// GetNodeID get current node ID.
	GetNodeID() core.RecordRef

	// SendRequestPacket send request packet to a remote node.
	SendRequestPacket(ctx context.Context, request Request, receiver *host.Host) (Future, error)
	// RegisterPacketHandler register a handler function to process incoming requests of a specific type.
	RegisterPacketHandler(t types.PacketType, handler RequestHandler)
	// NewRequestBuilder create packet builder for an outgoing request with sender set to current node.
	NewRequestBuilder() RequestBuilder
	// BuildResponse create response to an incoming request with Data set to responseData.
	BuildResponse(ctx context.Context, request Request, responseData interface{}) Response
}

InternalTransport simple interface to send network requests and process network responses.

type NodeKeeper

type NodeKeeper interface {
	core.NodeNetwork

	// TODO: remove this interface when bootstrap mechanism completed
	core.SwitcherWorkAround

	// SetCloudHash set new cloud hash
	SetCloudHash([]byte)
	// AddActiveNodes add active nodes.
	AddActiveNodes([]core.Node)
	// GetActiveNodeByShortID get active node by short ID. Returns nil if node is not found.
	GetActiveNodeByShortID(shortID core.ShortNodeID) core.Node
	// SetState set state of the NodeKeeper
	SetState(NodeKeeperState)
	// GetState get state of the NodeKeeper
	GetState() NodeKeeperState
	// GetOriginClaim get origin NodeJoinClaim
	GetOriginClaim() (*consensus.NodeJoinClaim, error)
	// NodesJoinedDuringPreviousPulse returns true if the last Sync call contained approved Join claims
	NodesJoinedDuringPreviousPulse() bool
	// AddPendingClaim add pending claim to the internal queue of claims
	AddPendingClaim(consensus.ReferendumClaim) bool
	// GetClaimQueue get the internal queue of claims
	GetClaimQueue() ClaimQueue
	// GetUnsyncList get unsync list for current pulse. Has copy of active node list from nodekeeper as internal state.
	// Should be called when nodekeeper state is Ready.
	GetUnsyncList() UnsyncList
	// GetSparseUnsyncList get sparse unsync list for current pulse with predefined length of active node list.
	// Does not contain active list, should collect active list during its lifetime via AddClaims.
	// Should be called when nodekeeper state is Waiting.
	GetSparseUnsyncList(length int) UnsyncList
	// Sync move unsync -> sync
	Sync(list UnsyncList)
	// MoveSyncToActive merge sync list with active nodes
	MoveSyncToActive()
}

NodeKeeper manages unsync, sync and active lists.

type NodeKeeperState

type NodeKeeperState uint8

const (
	// Undefined is state of NodeKeeper while it is not valid
	Undefined NodeKeeperState = iota + 1
	// Waiting is state of NodeKeeper while it is not part of consensus yet (waits for its join claim to pass)
	Waiting
	// Ready is state of NodeKeeper when it is ready for consensus
	Ready
)

type Packet

type Packet interface {
	GetSender() core.RecordRef
	GetSenderHost() *host.Host
	GetType() types.PacketType
	GetData() interface{}
	GetRequestID() RequestID
}

Packet is a packet that is transported via network by HostNetwork.

type PartitionPolicy

type PartitionPolicy interface {
	ShardsCount() int
}

PartitionPolicy contains all rules how to initiate globule resharding.

type PulseHandler

type PulseHandler interface {
	HandlePulse(ctx context.Context, pulse core.Pulse)
}

PulseHandler interface to process new pulse.

type Request

type Request Packet

Request is a packet that is sent from the current node.

type RequestBuilder

type RequestBuilder interface {
	Type(packetType types.PacketType) RequestBuilder
	Data(data interface{}) RequestBuilder
	Build() Request
}

RequestBuilder allows to build a Request.

type RequestHandler

type RequestHandler func(context.Context, Request) (Response, error)

RequestHandler handler function to process incoming requests from network.

type RequestID

type RequestID uint64

RequestID is 64 bit unsigned int request id.

type Response

type Response Packet

Response is a packet that is received in response to a previously sent Request.

type RoutingTable

type RoutingTable interface {
	// Inject inject dependencies from components
	Inject(nodeKeeper NodeKeeper)
	// Resolve NodeID -> ShortID, Address. Can initiate network requests.
	Resolve(core.RecordRef) (*host.Host, error)
	// ResolveS ShortID -> NodeID, Address for node inside current globe.
	ResolveS(core.ShortNodeID) (*host.Host, error)
	// AddToKnownHosts add host to routing table.
	AddToKnownHosts(*host.Host)
	// Rebalance recreate shards of routing table with known hosts according to new partition policy.
	Rebalance(PartitionPolicy)
	// GetRandomNodes get a specified number of random nodes. Returns less if there are not enough nodes in network.
	GetRandomNodes(count int) []host.Host
}

RoutingTable contains all routing information of the network.

type UnsyncList

type UnsyncList interface {
	consensus.BitSetMapper
	// RemoveClaims
	RemoveClaims(core.RecordRef)
	// AddClaims
	AddClaims(map[core.RecordRef][]consensus.ReferendumClaim, map[core.RecordRef]string)
	// CalculateHash calculate node list hash based on active node list and claims
	CalculateHash() ([]byte, error)
	// GetActiveNode get active node by reference ID for current consensus
	GetActiveNode(ref core.RecordRef) core.Node
	// GetActiveNodes get active nodes for current consensus
	GetActiveNodes() []core.Node
}

UnsyncList is interface to manage unsync list