conn

Documentation

Index

• Variables
• type Node
  • func NewNode(rc *intern.RaftContext) *Node
  • func (n *Node) AddToCluster(ctx context.Context, pid uint64) error
  • func (n *Node) BatchAndSendMessages()
  • func (n *Node) ConfState() *raftpb.ConfState
  • func (n *Node) Connect(pid uint64, addr string)
  • func (n *Node) DeletePeer(pid uint64)
  • func (n *Node) DoneConfChange(id uint64, err error)
  • func (n *Node) InitFromWal(wal *raftwal.Wal) (idx uint64, restart bool, rerr error)
  • func (n *Node) Peer(pid uint64) (string, bool)
  • func (n *Node) ProposePeerRemoval(ctx context.Context, id uint64) error
  • func (n *Node) Raft() raft.Node
  • func (n *Node) SaveSnapshot(s raftpb.Snapshot)
  • func (n *Node) SaveToStorage(h raftpb.HardState, es []raftpb.Entry)
  • func (n *Node) Send(m raftpb.Message)
  • func (n *Node) SetConfState(cs *raftpb.ConfState)
  • func (n *Node) SetPeer(pid uint64, addr string)
  • func (n *Node) SetRaft(r raft.Node)
  • func (n *Node) WaitForMinProposal(ctx context.Context, read *api.LinRead) error
• type Pool
  • func NewPool(addr string) (*Pool, error)
  • func (p *Pool) Get() *grpc.ClientConn
  • func (p *Pool) IsHealthy() bool
  • func (p *Pool) MonitorHealth()
  • func (p *Pool) UpdateHealthStatus()
• type Pools
  • func Get() *Pools
  • func (p *Pools) Connect(addr string) *Pool
  • func (p *Pools) Get(addr string) (*Pool, error)
  • func (p *Pools) Remove(addr string)
• type RaftServer
  • func (w *RaftServer) Echo(ctx context.Context, in *api.Payload) (*api.Payload, error)
  • func (w *RaftServer) GetNode() *Node
  • func (w *RaftServer) JoinCluster(ctx context.Context, rc *intern.RaftContext) (*api.Payload, error)
  • func (w *RaftServer) RaftMessage(ctx context.Context, query *api.Payload) (*api.Payload, error)

Variables

var (
	ErrNoConnection        = fmt.Errorf("No connection exists")
	ErrUnhealthyConnection = fmt.Errorf("Unhealthy connection")
)

var (
	ErrDuplicateRaftId = x.Errorf("Node is already part of group")
)

Types

type Node

type Node struct {
	x.SafeMutex

	// Fields which are never changed after init.
	Cfg    *raft.Config
	MyAddr string
	Id     uint64

	RaftContext *intern.RaftContext
	Store       *raft.MemoryStorage
	Wal         *raftwal.Wal

	// applied is used to keep track of the applied RAFT proposals.
	// The stages are proposed -> committed (accepted by cluster) ->
	// applied (to PL) -> synced (to BadgerDB).
	Applied x.WaterMark
	// contains filtered or unexported fields
}

func NewNode

func NewNode(rc *intern.RaftContext) *Node

func (*Node) AddToCluster

func (n *Node) AddToCluster(ctx context.Context, pid uint64) error

func (*Node) BatchAndSendMessages

func (n *Node) BatchAndSendMessages()

func (*Node) ConfState

func (n *Node) ConfState() *raftpb.ConfState

ConfState would return the latest ConfState stored in node.

func (*Node) Connect

func (n *Node) Connect(pid uint64, addr string)

Connects the node and makes its peerPool refer to the constructed pool and address (possibly updating ourselves from the old address.) (Unless pid is ourselves, in which case this does nothing.)

func (*Node) DeletePeer

func (n *Node) DeletePeer(pid uint64)

func (*Node) DoneConfChange

func (n *Node) DoneConfChange(id uint64, err error)

func (*Node) InitFromWal

func (n *Node) InitFromWal(wal *raftwal.Wal) (idx uint64, restart bool, rerr error)

func (*Node) Peer

func (n *Node) Peer(pid uint64) (string, bool)

func (*Node) ProposePeerRemoval

func (n *Node) ProposePeerRemoval(ctx context.Context, id uint64) error

func (*Node) Raft

func (n *Node) Raft() raft.Node

Raft would return back the raft.Node stored in the node.

func (*Node) SaveSnapshot

func (n *Node) SaveSnapshot(s raftpb.Snapshot)

func (*Node) SaveToStorage

func (n *Node) SaveToStorage(h raftpb.HardState, es []raftpb.Entry)

func (*Node) Send

func (n *Node) Send(m raftpb.Message)

func (*Node) SetConfState

func (n *Node) SetConfState(cs *raftpb.ConfState)

SetConfState would store the latest ConfState generated by ApplyConfChange.

func (*Node) SetPeer

func (n *Node) SetPeer(pid uint64, addr string)

addr must not be empty.

func (*Node) SetRaft

func (n *Node) SetRaft(r raft.Node)

SetRaft would set the provided raft.Node to this node. It would check fail if the node is already set.

func (*Node) WaitForMinProposal

func (n *Node) WaitForMinProposal(ctx context.Context, read *api.LinRead) error

type Pool

type Pool struct {
	sync.RWMutex

	Addr string
	// contains filtered or unexported fields
}

"Pool" is used to manage the grpc client connection(s) for communicating with other worker instances. Right now it just holds one of them.

func NewPool

func NewPool(addr string) (*Pool, error)

NewPool creates a new "pool" with one gRPC connection, refcount 0.

func (*Pool) Get

func (p *Pool) Get() *grpc.ClientConn

Get returns the connection to use from the pool of connections.

func (*Pool) IsHealthy

func (p *Pool) IsHealthy() bool

func (*Pool) MonitorHealth

func (p *Pool) MonitorHealth()

MonitorHealth monitors the health of the connection via Echo. This function blocks forever.

func (*Pool) UpdateHealthStatus

func (p *Pool) UpdateHealthStatus()

type Pools

type Pools struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func Get

func Get() *Pools

func (*Pools) Connect

func (p *Pools) Connect(addr string) *Pool

func (*Pools) Get

func (p *Pools) Get(addr string) (*Pool, error)

func (*Pools) Remove

func (p *Pools) Remove(addr string)

type RaftServer

type RaftServer struct {
	Node *Node
	// contains filtered or unexported fields
}

func (*RaftServer) Echo

func (w *RaftServer) Echo(ctx context.Context, in *api.Payload) (*api.Payload, error)

Hello rpc call is used to check connection with other workers after worker tcp server for this instance starts.

func (*RaftServer) GetNode

func (w *RaftServer) GetNode() *Node

func (*RaftServer) JoinCluster

func (w *RaftServer) JoinCluster(ctx context.Context,
	rc *intern.RaftContext) (*api.Payload, error)

func (*RaftServer) RaftMessage

func (w *RaftServer) RaftMessage(ctx context.Context,
	query *api.Payload) (*api.Payload, error)