raftlog

Documentation

Index

• Constants
• Variables
• func IsValidSnapshot(snapshot raftpb.Snapshot) bool
• type DataType
• type DataWrapper
  • func Unmarshal(dst []byte) (*DataWrapper, error)
  • func (d *DataWrapper) GetData() []byte
  • func (d *DataWrapper) GetDataType() DataType
  • func (d *DataWrapper) Marshal() []byte
• type FileWrap
  • func (fw *FileWrap) Close() error
  • func (fw *FileWrap) Delete() error
  • func (fw *FileWrap) GetEntryData(start, end int) []byte
  • func (fw *FileWrap) Name() string
  • func (fw *FileWrap) ReadSlice(offset int64) []byte
  • func (fw *FileWrap) Size() int
  • func (fw *FileWrap) SliceSize(offset int) int
  • func (fw *FileWrap) Truncate(size int64) error
  • func (fw *FileWrap) TrySync() error
  • func (fw *FileWrap) Write(dat []byte) (int, error)
  • func (fw *FileWrap) WriteAt(offset int64, dat []byte) (int, error)
  • func (fw *FileWrap) WriteSlice(offset int64, dat []byte) error
• type FileWrapper
  • func OpenFile(fpath string, flag int, maxSz int) (FileWrapper, error)
• type MetaInfo
• type RaftDiskStorage
  • func Init(dir string) (*RaftDiskStorage, error)
  • func (rds *RaftDiskStorage) Close() error
  • func (rds *RaftDiskStorage) CreateSnapshot(i uint64, cs *raftpb.ConfState, data []byte) error
  • func (rds *RaftDiskStorage) DeleteBefore(index uint64)
  • func (rds *RaftDiskStorage) Entries(lo, hi, maxSize uint64) ([]raftpb.Entry, error)
  • func (rds *RaftDiskStorage) Exist() bool
  • func (rds *RaftDiskStorage) FirstIndex() (uint64, error)
  • func (rds *RaftDiskStorage) FirstIndexWithSnap() (uint64, error)
  • func (rds *RaftDiskStorage) GetFirstLast() (uint64, uint64)
  • func (rds *RaftDiskStorage) HardState() (raftpb.HardState, error)
  • func (rds *RaftDiskStorage) InitialState() (hs raftpb.HardState, cs raftpb.ConfState, err error)
  • func (rds *RaftDiskStorage) LastIndex() (uint64, error)
  • func (rds *RaftDiskStorage) NumEntries() int
  • func (rds *RaftDiskStorage) Save(h *raftpb.HardState, entries []raftpb.Entry, snap *raftpb.Snapshot) error
  • func (rds *RaftDiskStorage) SaveEntries(h *raftpb.HardState, entries []raftpb.Entry, snap *raftpb.Snapshot) error
  • func (rds *RaftDiskStorage) SetUint(info MetaInfo, id uint64)
  • func (rds *RaftDiskStorage) SlotGe(index uint64) (int, int)
  • func (rds *RaftDiskStorage) Snapshot() (raftpb.Snapshot, error)
  • func (rds *RaftDiskStorage) Term(idx uint64) (uint64, error)
  • func (rds *RaftDiskStorage) TrySync() error
  • func (rds *RaftDiskStorage) Uint(info MetaInfo) uint64
• type SnapShotter
  • func (s *SnapShotter) TryToUpdateCommittedIndex(index uint64)

Constants

const (
	RaftIdOffset          = 0
	GroupIdOffset         = 8
	CheckpointIndexOffset = 16
)

Variables

var NewFile = errors.New("Create a new file")

Functions

func IsValidSnapshot

func IsValidSnapshot(snapshot raftpb.Snapshot) bool

Types

type DataType

type DataType uint32

const (
	Normal DataType = iota
	Snapshot
	ClearEntryLog
)

type DataWrapper

type DataWrapper struct {
	Data      []byte
	DataType  DataType
	Identity  string // raftNode identity
	ProposeId uint64 // propose seq id
}

func Unmarshal

func Unmarshal(dst []byte) (*DataWrapper, error)

func (*DataWrapper) GetData

func (d *DataWrapper) GetData() []byte

func (*DataWrapper) GetDataType

func (d *DataWrapper) GetDataType() DataType

func (*DataWrapper) Marshal

func (d *DataWrapper) Marshal() []byte

type FileWrap

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

FileWrap represents a file and includes both the buffer to the data and the file descriptor.

func (*FileWrap) Close

func (fw *FileWrap) Close() error

func (*FileWrap) Delete

func (fw *FileWrap) Delete() error

func (*FileWrap) GetEntryData

func (fw *FileWrap) GetEntryData(start, end int) []byte

GetEntryData returns entry data

func (*FileWrap) Name

func (fw *FileWrap) Name() string

func (*FileWrap) ReadSlice

func (fw *FileWrap) ReadSlice(offset int64) []byte

func (*FileWrap) Size

func (fw *FileWrap) Size() int

func (*FileWrap) SliceSize

func (fw *FileWrap) SliceSize(offset int) int

func (*FileWrap) Truncate

func (fw *FileWrap) Truncate(size int64) error

func (*FileWrap) TrySync

func (fw *FileWrap) TrySync() error

func (*FileWrap) Write

func (fw *FileWrap) Write(dat []byte) (int, error)

func (*FileWrap) WriteAt

func (fw *FileWrap) WriteAt(offset int64, dat []byte) (int, error)

func (*FileWrap) WriteSlice

func (fw *FileWrap) WriteSlice(offset int64, dat []byte) error

type FileWrapper

type FileWrapper interface {
	Name() string
	Size() int
	GetEntryData(start, end int) []byte
	Write(dat []byte) (int, error)
	WriteAt(offset int64, dat []byte) (int, error)
	WriteSlice(offset int64, dat []byte) error
	ReadSlice(offset int64) []byte
	SliceSize(offset int) int
	Truncate(size int64) error
	TrySync() error
	Delete() error
	Close() error
}

func OpenFile

func OpenFile(fpath string, flag int, maxSz int) (FileWrapper, error)

OpenFile opens an existing file or creates a new file. If the file is created, it would truncate the file to maxSz. In case the file is created, z.NewFile is returned.

type MetaInfo

type MetaInfo int

const (
	RaftId MetaInfo = iota
	GroupId
	CheckpointIndex
	SnapshotIndex
	SnapshotTerm
)

type RaftDiskStorage

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

RaftDiskStorage handles disk access and writing for the RAFT write-ahead log. Dir contains wal.meta file and <start idx zero padded>.entry files.

func Init

func Init(dir string) (*RaftDiskStorage, error)

Init initializes an instance of DiskStorage.

func (*RaftDiskStorage) Close

func (rds *RaftDiskStorage) Close() error

Close closes the DiskStorage.

func (*RaftDiskStorage) CreateSnapshot

func (rds *RaftDiskStorage) CreateSnapshot(i uint64, cs *raftpb.ConfState, data []byte) error

CreateSnapshot generates a snapshot with the given ConfState and data and writes it to disk.

func (*RaftDiskStorage) DeleteBefore

func (rds *RaftDiskStorage) DeleteBefore(index uint64)

func (*RaftDiskStorage) Entries

func (rds *RaftDiskStorage) Entries(lo, hi, maxSize uint64) ([]raftpb.Entry, error)

Entries returns a slice of log entries in the range [lo,hi). MaxSize limits the total size of the log entries returned, but Entries returns at least one entry if any.

func (*RaftDiskStorage) Exist

func (rds *RaftDiskStorage) Exist() bool

func (*RaftDiskStorage) FirstIndex

func (rds *RaftDiskStorage) FirstIndex() (uint64, error)

FirstIndex implements the Storage interface.

func (*RaftDiskStorage) FirstIndexWithSnap

func (rds *RaftDiskStorage) FirstIndexWithSnap() (uint64, error)

FirstIndex implements the Storage interface.

func (*RaftDiskStorage) GetFirstLast

func (rds *RaftDiskStorage) GetFirstLast() (uint64, uint64)

func (*RaftDiskStorage) HardState

func (rds *RaftDiskStorage) HardState() (raftpb.HardState, error)

func (*RaftDiskStorage) InitialState

func (rds *RaftDiskStorage) InitialState() (hs raftpb.HardState, cs raftpb.ConfState, err error)

InitialState returns the saved HardState and ConfState information.

func (*RaftDiskStorage) LastIndex

func (rds *RaftDiskStorage) LastIndex() (uint64, error)

func (*RaftDiskStorage) NumEntries

func (rds *RaftDiskStorage) NumEntries() int

func (*RaftDiskStorage) Save

func (rds *RaftDiskStorage) Save(h *raftpb.HardState, entries []raftpb.Entry, snap *raftpb.Snapshot) error

Save would write Entries, HardState and Snapshot to persistent storage in order, i.e. Entries first, then HardState and Snapshot if they are not empty. If persistent storage supports atomic writes then all of them can be written together. Note that when writing an Entry with Index i, any previously-persisted entries with Index >= i must be discarded.

func (*RaftDiskStorage) SaveEntries

func (rds *RaftDiskStorage) SaveEntries(h *raftpb.HardState, entries []raftpb.Entry, snap *raftpb.Snapshot) error

func (*RaftDiskStorage) SetUint

func (rds *RaftDiskStorage) SetUint(info MetaInfo, id uint64)

func (*RaftDiskStorage) SlotGe

func (rds *RaftDiskStorage) SlotGe(index uint64) (int, int)

func (*RaftDiskStorage) Snapshot

func (rds *RaftDiskStorage) Snapshot() (raftpb.Snapshot, error)

Snapshot returns the most recent snapshot. If snapshot is temporarily unavailable, it should return ErrSnapshotTemporarilyUnavailable, so raft state machine could know that Storage needs some time to prepare snapshot and call Snapshot later.

func (*RaftDiskStorage) Term

func (rds *RaftDiskStorage) Term(idx uint64) (uint64, error)

func (*RaftDiskStorage) TrySync

func (rds *RaftDiskStorage) TrySync() error

TrySync trys to write all the contents to disk.

func (*RaftDiskStorage) Uint

func (rds *RaftDiskStorage) Uint(info MetaInfo) uint64

type SnapShotter

type SnapShotter struct {
	RaftFlag       uint32 // 1 represents advancing committedIndex , other num represents suspending committedIndex
	RaftFlushC     chan bool
	CommittedIndex uint64
	// contains filtered or unexported fields
}

func (*SnapShotter) TryToUpdateCommittedIndex

func (s *SnapShotter) TryToUpdateCommittedIndex(index uint64)