etcdraft

Documentation

Index

• Constants
• Variables
• func ListSnapshots(logger raft.Logger, snapDir string) []uint64
• func NewNode(opts ...order.Option) (order.Order, error)
• type BatchTimer
  • func NewTimer(d time.Duration, logger logrus.FieldLogger) *BatchTimer
  • func (timer *BatchTimer) BatchTimeoutEvent() chan bool
  • func (timer *BatchTimer) IsBatchTimerActive() bool
  • func (timer *BatchTimer) StartBatchTimer()
  • func (timer *BatchTimer) StopBatchTimer()
• type GetTxReq
• type MemoryStorage
• type MempoolConfig
• type Node
  • func (n *Node) Commit() chan *pb.CommitEvent
  • func (n *Node) DelNode(delID uint64) error
  • func (n *Node) GetPendingNonceByAccount(account string) uint64
  • func (n *Node) GetPendingTxByHash(hash *types.Hash) pb.Transaction
  • func (n *Node) Prepare(tx pb.Transaction) error
  • func (n *Node) Quorum() uint64
  • func (n *Node) Ready() error
  • func (n *Node) ReportState(height uint64, blockHash *types.Hash, txHashList []*types.Hash)
  • func (n *Node) Start() error
  • func (n *Node) Step(msg []byte) error
  • func (n *Node) Stop()
  • func (n *Node) SubscribeTxEvent(events chan<- pb.Transactions) event.Subscription
• type RAFT
• type RAFTConfig
• type RaftStorage
  • func CreateStorage(lg raft.Logger, walDir string, snapDir string, dbDir string, ram MemoryStorage) (*RaftStorage, storage.Storage, error)
  • func (rs *RaftStorage) Close() error
  • func (rs *RaftStorage) Store(entries []raftpb.Entry, hardstate raftpb.HardState, snapshot raftpb.Snapshot) error
  • func (rs *RaftStorage) TakeSnapshot(i uint64, cs raftpb.ConfState, data []byte) error
• type SyncerConfig

Constants

const (
	DefaultBatchTick     = 500 * time.Millisecond
	DefaultSnapshotCount = 1000
	DefaultCheckInterval = 3 * time.Minute
)

Variables

var MaxSnapshotFiles = 5

MaxSnapshotFiles defines max number of etcd/raft snapshot files to retain on filesystem. Snapshot files are read from newest to oldest, until first intact file is found. The more snapshot files we keep around, the more we mitigate the impact of a corrupted snapshots. This is exported for testing purpose. This MUST be greater equal than 1.

Functions

func ListSnapshots

func ListSnapshots(logger raft.Logger, snapDir string) []uint64

ListSnapshots returns a list of RaftIndex of snapshots stored on disk. If a file is corrupted, rename the file.

func NewNode

func NewNode(opts ...order.Option) (order.Order, error)

NewNode new raft node

Types

type BatchTimer

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

func NewTimer

func NewTimer(d time.Duration, logger logrus.FieldLogger) *BatchTimer

NewTimer news a timer with default timeout.

func (*BatchTimer) BatchTimeoutEvent

func (timer *BatchTimer) BatchTimeoutEvent() chan bool

func (*BatchTimer) IsBatchTimerActive

func (timer *BatchTimer) IsBatchTimerActive() bool

func (*BatchTimer) StartBatchTimer

func (timer *BatchTimer) StartBatchTimer()

TODO (YH): add restartTimer??? StartBatchTimer starts the batch timer and reset the batchTimerActive to true.

func (*BatchTimer) StopBatchTimer

func (timer *BatchTimer) StopBatchTimer()

StopBatchTimer stops the batch timer and reset the batchTimerActive to false.

type GetTxReq

type GetTxReq struct {
	Hash *types.Hash
	Tx   chan pb.Transaction
}

type MemoryStorage

type MemoryStorage interface {
	raft.Storage
	Append(entries []raftpb.Entry) error
	SetHardState(st raftpb.HardState) error
	CreateSnapshot(i uint64, cs *raftpb.ConfState, data []byte) (raftpb.Snapshot, error)
	Compact(compactIndex uint64) error
	ApplySnapshot(snap raftpb.Snapshot) error
}

MemoryStorage is currently backed by etcd/raft.MemoryStorage. This interface is defined to expose dependencies of fsm so that it may be swapped in the future. TODO(jay) Add other necessary methods to this interface once we need them in implementation, e.g. ApplySnapshot.

type MempoolConfig

type MempoolConfig struct {
	BatchSize      uint64        `mapstructure:"batch_size"`
	PoolSize       uint64        `mapstructure:"pool_size"`
	TxSliceSize    uint64        `mapstructure:"tx_slice_size"`
	TxSliceTimeout time.Duration `mapstructure:"tx_slice_timeout"`
}

type Node

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

func (*Node) Commit

func (n *Node) Commit() chan *pb.CommitEvent

func (*Node) DelNode

func (n *Node) DelNode(delID uint64) error

DelNode sends a delete vp request by given id.

func (*Node) GetPendingNonceByAccount

func (n *Node) GetPendingNonceByAccount(account string) uint64

func (*Node) GetPendingTxByHash

func (n *Node) GetPendingTxByHash(hash *types.Hash) pb.Transaction

func (*Node) Prepare

func (n *Node) Prepare(tx pb.Transaction) error

Add the transaction into txpool and broadcast it to other nodes

func (*Node) Quorum

func (n *Node) Quorum() uint64

func (*Node) Ready

func (n *Node) Ready() error

func (*Node) ReportState

func (n *Node) ReportState(height uint64, blockHash *types.Hash, txHashList []*types.Hash)

func (*Node) Start

func (n *Node) Start() error

Start or restart raft node

func (*Node) Step

func (n *Node) Step(msg []byte) error

func (*Node) Stop

func (n *Node) Stop()

Stop the raft node

func (*Node) SubscribeTxEvent

func (n *Node) SubscribeTxEvent(events chan<- pb.Transactions) event.Subscription

SubscribeTxEvent subscribes tx event

type RAFT

type RAFT struct {
	BatchTimeout              time.Duration `mapstructure:"batch_timeout"`
	CheckInterval             time.Duration `mapstructure:"check_interval"`
	TickTimeout               time.Duration `mapstructure:"tick_timeout"`
	ElectionTick              int           `mapstructure:"election_tick"`
	HeartbeatTick             int           `mapstructure:"heartbeat_tick"`
	MaxSizePerMsg             uint64        `mapstructure:"max_size_per_msg"`
	MaxInflightMsgs           int           `mapstructure:"max_inflight_msgs"`
	CheckQuorum               bool          `mapstructure:"check_quorum"`
	PreVote                   bool          `mapstructure:"pre_vote"`
	DisableProposalForwarding bool          `mapstructure:"disable_proposal_forwarding"`
	MempoolConfig             MempoolConfig `mapstructure:"mempool"`
	SyncerConfig              SyncerConfig  `mapstructure:"syncer"`
}

type RAFTConfig

type RAFTConfig struct {
	RAFT RAFT
}

type RaftStorage

type RaftStorage struct {
	SnapshotCatchUpEntries uint64
	// contains filtered or unexported fields
}

RaftStorage encapsulates storages needed for etcd/raft data, i.e. memory, wal

func CreateStorage

func CreateStorage(
	lg raft.Logger,
	walDir string,
	snapDir string,
	dbDir string,
	ram MemoryStorage,
) (*RaftStorage, storage.Storage, error)

CreateStorage attempts to create a storage to persist etcd/raft data. If data presents in specified disk, they are loaded to reconstruct storage state.

func (*RaftStorage) Close

func (rs *RaftStorage) Close() error

Close closes storage

func (*RaftStorage) Store

func (rs *RaftStorage) Store(entries []raftpb.Entry, hardstate raftpb.HardState, snapshot raftpb.Snapshot) error

Store persists etcd/raft data

func (*RaftStorage) TakeSnapshot

func (rs *RaftStorage) TakeSnapshot(i uint64, cs raftpb.ConfState, data []byte) error

TakeSnapshot takes a snapshot at index i from MemoryStorage, and persists it to wal and disk.

type SyncerConfig

type SyncerConfig struct {
	SyncBlocks    uint64 `mapstructure:"sync_blocks"`
	SnapshotCount uint64 `mapstructure:"snapshot_count"`
}