rsm

package
v3.3.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 25, 2023 License: Apache-2.0 Imports: 30 Imported by: 0

Documentation

Overview

Package rsm implements State Machines used in Dragonboat.

This package is internally used by Dragonboat, applications are not expected to import this package.

Index

Constants

View Source 
const (
	EEHeaderSize uint8 = 1
	EEVersion    uint8 = 0 << 4
	EEV0         uint8 = 0 << 4

	// for V0 format, entries with empty payload will cause panic as such
	// entries always have their TYPE value set to ApplicationEntry
	EENoCompression uint8 = 0 << 1
	EESnappy        uint8 = 1 << 1

	EENoSession  uint8 = 0
	EEHasSession uint8 = 1
	// uncompressed size is binary.Uvarint encoded
	EEV0SizeOffset int = 1
)

Entry Cmd format when Type = pb.EncodedEntry

------------------------------------- |Version|CompressionFlag|SessionFlag| | 4Bits | 3Bits | 1Bit | -------------------------------------

View Source 
const (
	// ChunkSize is the size of each snapshot chunk.
	ChunkSize = settings.SnapshotChunkSize
)
View Source 
const (
	// EmptyClientSessionLength defines the length of an empty sessions instance.
	EmptyClientSessionLength uint64 = 16
)

Variables

View Source 
var (
	// ErrClusterClosed indicates that the cluster has been closed
	ErrClusterClosed = errors.New("raft cluster already closed")
)
View Source 
var (
	// ErrRestoreSnapshot indicates there is error when trying to restore
	// from a snapshot
	ErrRestoreSnapshot = errors.New("failed to restore snapshot")
)
View Source 
var (
	// LRUMaxSessionCount is the largest number of client sessions that can be
	// concurrently managed by a LRUSession instance.
	LRUMaxSessionCount = settings.Hard.LRUMaxSessionCount
)

Functions

func GetDefaultChecksum 

func GetDefaultChecksum() hash.Hash

GetDefaultChecksum returns the default hash.Hash instance.

func GetEmptyLRUSession 

func GetEmptyLRUSession() []byte

GetEmptyLRUSession returns an marshaled empty sessions instance.

func GetEncoded added in v3.3.0 

func GetEncoded(ct dio.CompressionType, cmd []byte, dst []byte) []byte

GetEncoded returns the encoded payload using the specified compression type and the default encoded entry version.

func GetMaxBlockSize added in v3.2.0 

func GetMaxBlockSize(ct config.CompressionType) uint64

GetMaxBlockSize returns the maximum block length supported by the specified compression type.

func GetPayload added in v3.3.0 

func GetPayload(e pb.Entry) []byte

GetPayload returns the payload of the entry ready to be applied into the state machine.

func GetV2PayloadChecksum 

func GetV2PayloadChecksum(fp string, fs vfs.IFS) (crc []byte, err error)

GetV2PayloadChecksum calculates the payload checksum of the specified snapshot file.

func GetV2PayloadSize 

func GetV2PayloadSize(sz uint64) uint64

GetV2PayloadSize returns the actual on disk size for the input user payload size.

func GetWitnessSnapshot added in v3.2.0 

func GetWitnessSnapshot(fs vfs.IFS) ([]byte, error)

GetWitnessSnapshot returns the content of a witness snapshot.

func IsShrunkSnapshotFile added in v3.3.0 

func IsShrunkSnapshotFile(fp string, fs vfs.IFS) (shrunk bool, err error)

IsShrunkSnapshotFile returns a boolean flag indicating whether the specified snapshot file is already shrunk.

func ReplaceSnapshot added in v3.3.0 

func ReplaceSnapshot(newFp string, fp string, fs vfs.IFS) error

ReplaceSnapshot replace the specified snapshot file with the shrunk version atomically.

func ShrinkSnapshot 

func ShrinkSnapshot(fp string, newFp string, fs vfs.IFS) (err error)

ShrinkSnapshot shrinks the specified snapshot file and save the generated shrunk version to the path specified by newFp.

func ToDioType added in v3.3.0 

func ToDioType(ct config.CompressionType) dio.CompressionType

ToDioType converts the CompressionType type defined in the config package to the CompressionType value defined in the dio package.

Types

type BlockWriter 

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

BlockWriter is a writer type that writes the input data to the underlying storage with checksum appended at the end of each block.

func NewBlockWriter 

func NewBlockWriter(blockSize uint64,
	nb func(data []byte, crc []byte) error, t pb.ChecksumType) *BlockWriter

NewBlockWriter creates and returns a block writer.

func (*BlockWriter) Close added in v3.3.0 

func (bw *BlockWriter) Close() error

Close closes the writer by passing all in memory buffered data to the underlying onNewBlock function.

func (*BlockWriter) GetPayloadChecksum 

func (bw *BlockWriter) GetPayloadChecksum() []byte

GetPayloadChecksum returns the checksum for the entire payload.

func (*BlockWriter) Write 

func (bw *BlockWriter) Write(bs []byte) (int, error)

Write writes the specified data using the block writer.

type ChunkWriter 

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

ChunkWriter is an io.WriteCloser type that streams snapshot chunks to its intended remote nodes.

func NewChunkWriter 

func NewChunkWriter(sink pb.IChunkSink, meta SSMeta) *ChunkWriter

NewChunkWriter creates and returns a chunk writer instance.

func (*ChunkWriter) Close 

func (cw *ChunkWriter) Close() error

Close closes the chunk writer.

func (*ChunkWriter) Write 

func (cw *ChunkWriter) Write(data []byte) (int, error)

Write writes the specified input data.

type ConcurrentStateMachine 

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

ConcurrentStateMachine is an IStateMachine type capable of taking concurrent snapshots.

func NewConcurrentStateMachine 

func NewConcurrentStateMachine(s sm.IConcurrentStateMachine) *ConcurrentStateMachine

NewConcurrentStateMachine creates a new ConcurrentStateMachine instance.

func (*ConcurrentStateMachine) Close 

func (s *ConcurrentStateMachine) Close() error

Close closes the state machine.

func (*ConcurrentStateMachine) Concurrent added in v3.3.0 

func (s *ConcurrentStateMachine) Concurrent() bool

Concurrent returns a boolean flag indicating whether the state machine is capable of taking concurrent snapshot.

func (*ConcurrentStateMachine) GetHash 

func (s *ConcurrentStateMachine) GetHash() (uint64, error)

GetHash returns the uint64 hash value representing the state of a state machine.

func (*ConcurrentStateMachine) Lookup 

func (s *ConcurrentStateMachine) Lookup(query interface{}) (interface{}, error)

Lookup queries the state machine.

func (*ConcurrentStateMachine) NALookup 

func (s *ConcurrentStateMachine) NALookup(query []byte) ([]byte, error)

NALookup queries the state machine.

func (*ConcurrentStateMachine) OnDisk added in v3.3.0 

func (s *ConcurrentStateMachine) OnDisk() bool

OnDisk returns a boolean flag indicating whether this is a on disk state machine.

func (*ConcurrentStateMachine) Open 

func (s *ConcurrentStateMachine) Open(stopc <-chan struct{}) (uint64, error)

Open opens the state machine.

func (*ConcurrentStateMachine) Prepare added in v3.3.0 

func (s *ConcurrentStateMachine) Prepare() (interface{}, error)

Prepare makes preparations for taking concurrent snapshot.

func (*ConcurrentStateMachine) Recover added in v3.3.0 

func (s *ConcurrentStateMachine) Recover(r io.Reader,
	fs []sm.SnapshotFile, stopc <-chan struct{}) error

Recover recovers the state machine from a snapshot.

func (*ConcurrentStateMachine) Save added in v3.3.0 

func (s *ConcurrentStateMachine) Save(ctx interface{},
	w io.Writer, fc sm.ISnapshotFileCollection, stopc <-chan struct{}) error

Save saves the snapshot.

func (*ConcurrentStateMachine) Sync 

func (s *ConcurrentStateMachine) Sync() error

Sync synchronizes all in-core state with that on disk.

func (*ConcurrentStateMachine) Type added in v3.3.0 

func (s *ConcurrentStateMachine) Type() pb.StateMachineType

Type returns the type of the state machine.

func (*ConcurrentStateMachine) Update 

func (s *ConcurrentStateMachine) Update(entries []sm.Entry) ([]sm.Entry, error)

Update updates the state machine.

type Files 

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

Files is a collection of external files specified by the SaveSnapshot method of the state machine type.

func NewFileCollection 

func NewFileCollection() *Files

NewFileCollection creates and returns a Files instance.

func (*Files) AddFile 

func (fc *Files) AddFile(fileID uint64,
	path string, metadata []byte)

AddFile adds the specified file to the external file collection.

func (*Files) GetFileAt 

func (fc *Files) GetFileAt(idx uint64) *pb.SnapshotFile

GetFileAt returns the specified file.

func (*Files) PrepareFiles 

func (fc *Files) PrepareFiles(tmpdir string,
	finaldir string) ([]*pb.SnapshotFile, error)

PrepareFiles finalize the external files added to the collection.

func (*Files) Size 

func (fc *Files) Size() uint64

Size returns the number of external files already added to the external file collection.

type IBlockWriter 

type IBlockWriter interface {
	io.WriteCloser
	GetPayloadChecksum() []byte
}

IBlockWriter is the interface for writing checksumed data blocks.

type ILoadable added in v3.3.0 

type ILoadable interface {
	LoadSessions(io.Reader, SSVersion) error
}

ILoadable is the interface for types that can load client session state from a snapshot.

type IManagedStateMachine 

type IManagedStateMachine interface {
	Open() (uint64, error)
	Update(sm.Entry) (sm.Result, error)
	BatchedUpdate([]sm.Entry) ([]sm.Entry, error)
	Lookup(interface{}) (interface{}, error)
	ConcurrentLookup(interface{}) (interface{}, error)
	NALookup([]byte) ([]byte, error)
	NAConcurrentLookup([]byte) ([]byte, error)
	Sync() error
	GetHash() (uint64, error)
	Prepare() (interface{}, error)
	Save(SSMeta, io.Writer, []byte, sm.ISnapshotFileCollection) (bool, error)
	Recover(io.Reader, []sm.SnapshotFile) error
	Stream(interface{}, io.Writer) error
	Offloaded() bool
	Loaded()
	Close() error
	DestroyedC() <-chan struct{}
	Concurrent() bool
	OnDisk() bool
	Type() pb.StateMachineType
}

IManagedStateMachine is the interface used for managed state machine. A managed state machine contains a user state machine plus its engine state.

type INode added in v3.2.0 

type INode interface {
	StepReady()
	RestoreRemotes(pb.Snapshot)
	ApplyUpdate(pb.Entry, sm.Result, bool, bool, bool)
	ApplyConfigChange(pb.ConfigChange, uint64, bool)
	NodeID() uint64
	ClusterID() uint64
	ShouldStop() <-chan struct{}
}

INode is the interface of a dragonboat node.

type IRecoverable added in v3.3.0 

type IRecoverable interface {
	Recover(io.Reader, []sm.SnapshotFile) error
}

IRecoverable is the interface for types that can have its state restored from snapshots.

type ISavable 

type ISavable interface {
	Save(SSMeta, io.Writer, []byte, sm.ISnapshotFileCollection) (bool, error)
}

ISavable is the interface for types that can its content saved as snapshots.

type ISnapshotter 

type ISnapshotter interface {
	GetSnapshot(uint64) (pb.Snapshot, error)
	GetMostRecentSnapshot() (pb.Snapshot, error)
	Stream(IStreamable, SSMeta, pb.IChunkSink) error
	Shrunk(ss pb.Snapshot) (bool, error)
	Save(ISavable, SSMeta) (pb.Snapshot, SSEnv, error)
	Load(pb.Snapshot, ILoadable, IRecoverable) error
	IsNoSnapshotError(error) bool
}

ISnapshotter is the interface for the snapshotter object.

type IStateMachine 

type IStateMachine interface {
	Open(<-chan struct{}) (uint64, error)
	Update(entries []sm.Entry) ([]sm.Entry, error)
	Lookup(query interface{}) (interface{}, error)
	NALookup(query []byte) ([]byte, error)
	Sync() error
	Prepare() (interface{}, error)
	Save(interface{},
		io.Writer, sm.ISnapshotFileCollection, <-chan struct{}) error
	Recover(io.Reader, []sm.SnapshotFile, <-chan struct{}) error
	Close() error
	GetHash() (uint64, error)
	Concurrent() bool
	OnDisk() bool
	Type() pb.StateMachineType
}

IStateMachine is an adapter interface for underlying sm.IStateMachine, sm.IConcurrentStateMachine and sm.IOnDIskStateMachine instances.

type IStreamable 

type IStreamable interface {
	Stream(interface{}, io.Writer) error
}

IStreamable is the interface for types that can be snapshot streamed.

type ITestFS added in v3.2.0 

type ITestFS interface {
	SetTestFS(fs config.IFS)
}

ITestFS is an interface implemented by test SMs.

type IVReader 

type IVReader interface {
	Read(data []byte) (int, error)
	Sum() []byte
}

IVReader is the interface for versioned snapshot reader.

type IVValidator 

type IVValidator interface {
	AddChunk(data []byte, chunkID uint64) bool
	Validate() bool
}

IVValidator is the interface for versioned validator.

type IVWriter 

type IVWriter interface {
	io.WriteCloser
	GetVersion() SSVersion
	GetPayloadSum() []byte
	GetPayloadSize(uint64) uint64
}

IVWriter is the interface for versioned snapshot writer.

type InMemStateMachine added in v3.3.0 

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

InMemStateMachine is a regular state machine not capable of concurrent access from multiple goroutines.

func NewInMemStateMachine added in v3.3.0 

func NewInMemStateMachine(s sm.IStateMachine) *InMemStateMachine

NewInMemStateMachine creates a new InMemStateMachine instance.

func (*InMemStateMachine) Close added in v3.3.0 

func (i *InMemStateMachine) Close() error

Close closes the state machine.

func (*InMemStateMachine) Concurrent added in v3.3.0 

func (i *InMemStateMachine) Concurrent() bool

Concurrent returns a boolean flag indicating whether the state machine is capable of taking concurrent snapshot.

func (*InMemStateMachine) GetHash added in v3.3.0 

func (i *InMemStateMachine) GetHash() (uint64, error)

GetHash returns the uint64 hash value representing the state of a state machine.

func (*InMemStateMachine) Lookup added in v3.3.0 

func (i *InMemStateMachine) Lookup(query interface{}) (interface{}, error)

Lookup queries the state machine.

func (*InMemStateMachine) NALookup added in v3.3.0 

func (i *InMemStateMachine) NALookup(query []byte) ([]byte, error)

NALookup queries the state machine.

func (*InMemStateMachine) OnDisk added in v3.3.0 

func (i *InMemStateMachine) OnDisk() bool

OnDisk returns a boolean flag indicating whether this is an on disk state machine.

func (*InMemStateMachine) Open added in v3.3.0 

func (i *InMemStateMachine) Open(stopc <-chan struct{}) (uint64, error)

Open opens the state machine.

func (*InMemStateMachine) Prepare added in v3.3.0 

func (i *InMemStateMachine) Prepare() (interface{}, error)

Prepare makes preparations for taking concurrent snapshot.

func (*InMemStateMachine) Recover added in v3.3.0 

func (i *InMemStateMachine) Recover(r io.Reader,
	fs []sm.SnapshotFile, stopc <-chan struct{}) error

Recover recovers the state machine from a snapshot.

func (*InMemStateMachine) Save added in v3.3.0 

func (i *InMemStateMachine) Save(ctx interface{},
	w io.Writer, fc sm.ISnapshotFileCollection, stopc <-chan struct{}) error

Save saves the snapshot.

func (*InMemStateMachine) Sync added in v3.3.0 

func (i *InMemStateMachine) Sync() error

Sync synchronizes all in-core state with that on disk.

func (*InMemStateMachine) Type added in v3.3.0 

func (i *InMemStateMachine) Type() pb.StateMachineType

Type returns the type of the state machine.

func (*InMemStateMachine) Update added in v3.3.0 

func (i *InMemStateMachine) Update(entries []sm.Entry) ([]sm.Entry, error)

Update updates the state machine.

type ManagedStateMachineFactory 

type ManagedStateMachineFactory func(clusterID uint64,
	nodeID uint64, stopc <-chan struct{}) IManagedStateMachine

ManagedStateMachineFactory is the factory function type for creating an IManagedStateMachine instance.

type NativeSM added in v3.2.0 

type NativeSM struct {
	OffloadedStatus
	// contains filtered or unexported fields
}

NativeSM is the IManagedStateMachine object used to manage native data store in Golang.

func NewNativeSM added in v3.2.0 

func NewNativeSM(config config.Config, ism IStateMachine,
	done <-chan struct{}) *NativeSM

NewNativeSM creates and returns a new NativeSM object.

func (*NativeSM) BatchedUpdate added in v3.2.0 

func (ds *NativeSM) BatchedUpdate(ents []sm.Entry) ([]sm.Entry, error)

BatchedUpdate applies committed entries in a batch to hide latency.

func (*NativeSM) Close added in v3.3.0 

func (ds *NativeSM) Close() error

Close closes the underlying user state machine and set the destroyed flag.

func (*NativeSM) Concurrent added in v3.3.0 

func (ds *NativeSM) Concurrent() bool

Concurrent returns a boolean flag to indicate whether the managed state machine instance is capable of doing concurrent snapshots.

func (*NativeSM) ConcurrentLookup added in v3.3.0 

func (ds *NativeSM) ConcurrentLookup(query interface{}) (interface{}, error)

ConcurrentLookup queries the data store without obtaining the NativeSM.mu.

func (*NativeSM) DestroyedC added in v3.3.0 

func (ds *NativeSM) DestroyedC() <-chan struct{}

DestroyedC returns a chan struct{} used to indicate whether the SM has been fully offloaded.

func (*NativeSM) GetHash added in v3.2.0 

func (ds *NativeSM) GetHash() (uint64, error)

GetHash returns an integer value representing the state of the data store.

func (*NativeSM) Loaded added in v3.2.0 

func (ds *NativeSM) Loaded()

Loaded marks the statemachine as loaded by the specified component.

func (*NativeSM) Lookup added in v3.2.0 

func (ds *NativeSM) Lookup(query interface{}) (interface{}, error)

Lookup queries the data store.

func (*NativeSM) NAConcurrentLookup added in v3.3.0 

func (ds *NativeSM) NAConcurrentLookup(query []byte) ([]byte, error)

NAConcurrentLookup queries the data store without obtaining the NativeSM.mu.

func (*NativeSM) NALookup added in v3.2.0 

func (ds *NativeSM) NALookup(query []byte) ([]byte, error)

NALookup queries the data store.

func (*NativeSM) Offloaded added in v3.2.0 

func (ds *NativeSM) Offloaded() bool

Offloaded offloads the data store from a user component.

func (*NativeSM) OnDisk added in v3.3.0 

func (ds *NativeSM) OnDisk() bool

OnDisk returns a boolean flag indicating whether the state machine is an on disk state machine.

func (*NativeSM) Open added in v3.2.0 

func (ds *NativeSM) Open() (uint64, error)

Open opens on disk state machine.

func (*NativeSM) Prepare added in v3.3.0 

func (ds *NativeSM) Prepare() (interface{}, error)

Prepare makes preparation for concurrently taking snapshot.

func (*NativeSM) Recover added in v3.3.0 

func (ds *NativeSM) Recover(r io.Reader, files []sm.SnapshotFile) error

Recover recovers the state of the data store from the specified reader.

func (*NativeSM) Save added in v3.3.0 

func (ds *NativeSM) Save(meta SSMeta,
	w io.Writer, session []byte, c sm.ISnapshotFileCollection) (bool, error)

Save saves the state of the data store to the specified writer.

func (*NativeSM) Stream added in v3.3.0 

func (ds *NativeSM) Stream(ctx interface{}, w io.Writer) error

Stream creates and streams snapshot to a remote node.

func (*NativeSM) Sync added in v3.2.0 

func (ds *NativeSM) Sync() error

Sync synchronizes state machine's in-core state with that on disk.

func (*NativeSM) Type added in v3.3.0 

func (ds *NativeSM) Type() pb.StateMachineType

Type returns the state machine type.

func (*NativeSM) Update added in v3.2.0 

func (ds *NativeSM) Update(e sm.Entry) (sm.Result, error)

Update updates the data store.

type OffloadedStatus 

type OffloadedStatus struct {
	DestroyedC chan struct{}
	// contains filtered or unexported fields
}

OffloadedStatus is used for tracking whether the managed data store has been offloaded from various system components.

func (*OffloadedStatus) Destroyed 

func (o *OffloadedStatus) Destroyed() bool

Destroyed returns a boolean value indicating whether the belonging object has been destroyed.

func (*OffloadedStatus) SetDestroyed 

func (o *OffloadedStatus) SetDestroyed()

SetDestroyed set the destroyed flag to be true

func (*OffloadedStatus) SetLoaded 

func (o *OffloadedStatus) SetLoaded()

SetLoaded marks the managed data store as loaded by a user component.

func (*OffloadedStatus) SetOffloaded 

func (o *OffloadedStatus) SetOffloaded() uint64

SetOffloaded marks the managed data store as offloaded from a user component.

type OnDiskStateMachine 

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

OnDiskStateMachine is the type to represent an on disk state machine.

func NewOnDiskStateMachine 

func NewOnDiskStateMachine(s sm.IOnDiskStateMachine) *OnDiskStateMachine

NewOnDiskStateMachine creates and returns an on disk state machine.

func (*OnDiskStateMachine) Close 

func (s *OnDiskStateMachine) Close() error

Close closes the state machine.

func (*OnDiskStateMachine) Concurrent added in v3.3.0 

func (s *OnDiskStateMachine) Concurrent() bool

Concurrent returns a boolean flag indicating whether the state machine is capable of taking concurrent snapshot.

func (*OnDiskStateMachine) GetHash 

func (s *OnDiskStateMachine) GetHash() (uint64, error)

GetHash returns the uint64 hash value representing the state of a state machine.

func (*OnDiskStateMachine) Lookup 

func (s *OnDiskStateMachine) Lookup(query interface{}) (interface{}, error)

Lookup queries the state machine.

func (*OnDiskStateMachine) NALookup 

func (s *OnDiskStateMachine) NALookup(query []byte) ([]byte, error)

NALookup queries the state machine.

func (*OnDiskStateMachine) OnDisk added in v3.3.0 

func (s *OnDiskStateMachine) OnDisk() bool

OnDisk returns a boolean flag indicating whether this is an on disk state machine.

func (*OnDiskStateMachine) Open 

func (s *OnDiskStateMachine) Open(stopc <-chan struct{}) (uint64, error)

Open opens the state machine.

func (*OnDiskStateMachine) Prepare added in v3.3.0 

func (s *OnDiskStateMachine) Prepare() (interface{}, error)

Prepare makes preparations for taking concurrent snapshot.

func (*OnDiskStateMachine) Recover added in v3.3.0 

func (s *OnDiskStateMachine) Recover(r io.Reader,
	fs []sm.SnapshotFile, stopc <-chan struct{}) error

Recover recovers the state machine from a snapshot.

func (*OnDiskStateMachine) Save added in v3.3.0 

func (s *OnDiskStateMachine) Save(ctx interface{},
	w io.Writer, fc sm.ISnapshotFileCollection, stopc <-chan struct{}) error

Save saves the snapshot.

func (*OnDiskStateMachine) SetTestFS added in v3.2.0 

func (s *OnDiskStateMachine) SetTestFS(fs config.IFS)

SetTestFS injects the specified fs to the test SM.

func (*OnDiskStateMachine) Sync 

func (s *OnDiskStateMachine) Sync() error

Sync synchronizes all in-core state with that on disk.

func (*OnDiskStateMachine) Type added in v3.3.0 

func (s *OnDiskStateMachine) Type() pb.StateMachineType

Type returns the type of the state machine.

func (*OnDiskStateMachine) Update 

func (s *OnDiskStateMachine) Update(entries []sm.Entry) ([]sm.Entry, error)

Update updates the state machine.

type RaftClientID 

type RaftClientID uint64

RaftClientID is the type used as client id in sessions.

func (*RaftClientID) Compare 

func (a *RaftClientID) Compare(b llrb.Comparable) int

Compare implements the llrb.Comparable interface.

type RaftSeriesID 

type RaftSeriesID uint64

RaftSeriesID is the type used as series id in sessions.

type SMFactoryFunc 

type SMFactoryFunc func(clusterID uint64,
	nodeID uint64, done <-chan struct{}) IManagedStateMachine

SMFactoryFunc is the function type for creating an IStateMachine instance

type SSEnv added in v3.3.0 

type SSEnv = server.SSEnv

SSEnv is the snapshot environment type.

type SSMeta added in v3.2.0 

type SSMeta struct {
	From            uint64
	Index           uint64
	Term            uint64
	OnDiskIndex     uint64 // applied index of IOnDiskStateMachine
	Request         SSRequest
	Membership      pb.Membership
	Type            pb.StateMachineType
	Session         *bytes.Buffer
	Ctx             interface{}
	CompressionType config.CompressionType
}

SSMeta is the metadata of a snapshot.

type SSReqType added in v3.2.0 

type SSReqType uint64

SSReqType is the type of a snapshot request. 

const (
	// Periodic is the value to indicate periodic snapshot.
	Periodic SSReqType = iota
	// UserRequested is the value to indicate user requested snapshot.
	UserRequested
	// Exported is the value to indicate exported snapshot.
	Exported
	// Streaming is the value to indicate snapshot streaming.
	Streaming
)

type SSRequest added in v3.2.0 

type SSRequest struct {
	Type               SSReqType
	Key                uint64
	Path               string
	OverrideCompaction bool
	CompactionOverhead uint64
}

SSRequest is the type for describing the details of a snapshot request.

func (*SSRequest) Exported added in v3.3.0 

func (r *SSRequest) Exported() bool

Exported returns a boolean value indicating whether the snapshot request is to create an exported snapshot.

func (*SSRequest) Streaming added in v3.3.0 

func (r *SSRequest) Streaming() bool

Streaming returns a boolean value indicating whether the snapshot request is to stream snapshot.

type SSVersion added in v3.2.0 

type SSVersion uint64

SSVersion is the snapshot version value type. 

const (
	// V1 is the value of snapshot version 1.
	V1 SSVersion = 1
	// V2 is the value of snapshot version 2.
	V2 SSVersion = 2
	// DefaultVersion is the snapshot binary format version.
	DefaultVersion SSVersion = V2
	// HeaderSize is the size of snapshot in number of bytes.
	HeaderSize = settings.SnapshotHeaderSize

	// DefaultChecksumType is the default checksum type.
	DefaultChecksumType = defaultChecksumType
)

type Session 

type Session struct {
	ClientID      RaftClientID
	RespondedUpTo RaftSeriesID
	History       map[RaftSeriesID]sm.Result
}

Session is the session object maintained on the raft side.

func (*Session) AddResponse 

func (s *Session) AddResponse(id RaftSeriesID, result sm.Result)

AddResponse adds a response.

type SessionManager 

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

SessionManager is the wrapper struct that implements client session related functionalites used in the IManagedStateMachine interface.

func NewSessionManager 

func NewSessionManager() *SessionManager

NewSessionManager returns a new SessionManager instance.

func (*SessionManager) AddResponse 

func (ds *SessionManager) AddResponse(session *Session,
	seriesID uint64, result sm.Result)

AddResponse adds the specified result to the session.

func (*SessionManager) ClientRegistered 

func (ds *SessionManager) ClientRegistered(clientID uint64) (*Session, bool)

ClientRegistered returns whether the specified client exists in the system.

func (*SessionManager) GetSessionHash 

func (ds *SessionManager) GetSessionHash() uint64

GetSessionHash returns an uint64 integer representing the state of the session manager.

func (*SessionManager) LoadSessions 

func (ds *SessionManager) LoadSessions(reader io.Reader, v SSVersion) error

LoadSessions loads and restores sessions from io.Reader.

func (*SessionManager) MustHaveClientSeries 

func (ds *SessionManager) MustHaveClientSeries(session *Session,
	seriesID uint64)

MustHaveClientSeries checks whether the session manager contains a client session identified as clientID and whether it has seriesID responded.

func (*SessionManager) RegisterClientID 

func (ds *SessionManager) RegisterClientID(clientID uint64) sm.Result

RegisterClientID registers a new client, it returns the input client id if it is previously unknown, or 0 when the client has already been registered.

func (*SessionManager) SaveSessions 

func (ds *SessionManager) SaveSessions(writer io.Writer) error

SaveSessions saves the sessions to the provided io.writer.

func (*SessionManager) UnregisterClientID 

func (ds *SessionManager) UnregisterClientID(clientID uint64) sm.Result

UnregisterClientID removes the specified client session from the system. It returns the client id if the client is successfully removed, or 0 if the client session does not exist.

func (*SessionManager) UpdateRequired 

func (ds *SessionManager) UpdateRequired(session *Session,
	seriesID uint64) (sm.Result, bool, bool)

UpdateRequired return a tuple of request result, responded before, update required.

func (*SessionManager) UpdateRespondedTo 

func (ds *SessionManager) UpdateRespondedTo(session *Session,
	respondedTo uint64)

UpdateRespondedTo updates the responded to value of the specified client session.

type SnapshotReader 

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

SnapshotReader is an io.Reader for reading from snapshot files.

func NewSnapshotReader 

func NewSnapshotReader(fp string, fs vfs.IFS) (*SnapshotReader, error)

NewSnapshotReader creates a new snapshot reader instance.

func (*SnapshotReader) Close 

func (sr *SnapshotReader) Close() error

Close closes the snapshot reader instance.

func (*SnapshotReader) GetHeader 

func (sr *SnapshotReader) GetHeader() (pb.SnapshotHeader, error)

GetHeader returns the snapshot header instance.

func (*SnapshotReader) Read 

func (sr *SnapshotReader) Read(data []byte) (int, error)

Read reads up to len(data) bytes from the snapshot file.

func (*SnapshotReader) ValidatePayload 

func (sr *SnapshotReader) ValidatePayload(header pb.SnapshotHeader)

ValidatePayload validates whether the snapshot content matches the checksum recorded in the header.

type SnapshotValidator 

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

SnapshotValidator is the validator used to check incoming snapshot chunks.

func NewSnapshotValidator 

func NewSnapshotValidator() *SnapshotValidator

NewSnapshotValidator creates and returns a new SnapshotValidator instance.

func (*SnapshotValidator) AddChunk 

func (v *SnapshotValidator) AddChunk(data []byte, chunkID uint64) bool

AddChunk adds a new snapshot chunk to the validator.

func (*SnapshotValidator) Validate 

func (v *SnapshotValidator) Validate() bool

Validate validates the added chunks and return a boolean flag indicating whether the snapshot chunks are valid.

type SnapshotWriter 

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

SnapshotWriter is an io.Writer used to write snapshot file.

func NewSnapshotWriter 

func NewSnapshotWriter(fp string,
	ct pb.CompressionType, fs vfs.IFS) (*SnapshotWriter, error)

NewSnapshotWriter creates a new snapshot writer instance.

func (*SnapshotWriter) Close 

func (sw *SnapshotWriter) Close() error

Close closes the snapshot writer instance.

func (*SnapshotWriter) GetPayloadChecksum 

func (sw *SnapshotWriter) GetPayloadChecksum() []byte

GetPayloadChecksum returns the payload checksum.

func (*SnapshotWriter) GetPayloadSize 

func (sw *SnapshotWriter) GetPayloadSize(sz uint64) uint64

GetPayloadSize returns the payload size.

func (*SnapshotWriter) Write 

func (sw *SnapshotWriter) Write(data []byte) (int, error)

Write writes the specified data to the snapshot.

type StateMachine 

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

StateMachine is the state machine component in the replicated state machine scheme.

func NewStateMachine 

func NewStateMachine(sm IManagedStateMachine,
	snapshotter ISnapshotter,
	cfg config.Config, node INode, fs vfs.IFS) *StateMachine

NewStateMachine creates a new application state machine object.

func (*StateMachine) Close added in v3.3.0 

func (s *StateMachine) Close() error

Close closes the state machine.

func (*StateMachine) Concurrent added in v3.2.0 

func (s *StateMachine) Concurrent() bool

Concurrent returns a boolean flag indicating whether the state machine is capable of taking concurrent snapshot.

func (*StateMachine) DestroyedC added in v3.3.0 

func (s *StateMachine) DestroyedC() <-chan struct{}

DestroyedC return a chan struct{} used to indicate whether the SM has been fully unloaded.

func (*StateMachine) GetHash 

func (s *StateMachine) GetHash() (uint64, error)

GetHash returns the state machine hash.

func (*StateMachine) GetLastApplied 

func (s *StateMachine) GetLastApplied() uint64

GetLastApplied returns the last applied value.

func (*StateMachine) GetMembership 

func (s *StateMachine) GetMembership() pb.Membership

GetMembership returns the membership info maintained by the state machine.

func (*StateMachine) GetMembershipHash 

func (s *StateMachine) GetMembershipHash() uint64

GetMembershipHash returns the hash of the membership instance.

func (*StateMachine) GetSessionHash 

func (s *StateMachine) GetSessionHash() uint64

GetSessionHash returns the session hash.

func (*StateMachine) GetSyncedIndex 

func (s *StateMachine) GetSyncedIndex() uint64

GetSyncedIndex returns the index value that is known to have been synchronized.

func (*StateMachine) Handle 

func (s *StateMachine) Handle(batch []Task, apply []sm.Entry) (Task, error)

Handle pulls the committed record and apply it if there is any available.

func (*StateMachine) Loaded 

func (s *StateMachine) Loaded()

Loaded marks the state machine as loaded from the specified compone.

func (*StateMachine) Lookup 

func (s *StateMachine) Lookup(query interface{}) (interface{}, error)

Lookup queries the local state machine.

func (*StateMachine) NALookup 

func (s *StateMachine) NALookup(query []byte) ([]byte, error)

NALookup queries the local state machine.

func (*StateMachine) Offloaded 

func (s *StateMachine) Offloaded() bool

Offloaded marks the state machine as offloaded from the specified compone. It returns a boolean value indicating whether the node has been fully unloaded after unloading from the specified compone.

func (*StateMachine) OnDiskStateMachine 

func (s *StateMachine) OnDiskStateMachine() bool

OnDiskStateMachine returns a boolean flag indicating whether it is an on disk state machine.

func (*StateMachine) OpenOnDiskStateMachine 

func (s *StateMachine) OpenOnDiskStateMachine() (uint64, error)

OpenOnDiskStateMachine opens the on disk state machine.

func (*StateMachine) ReadyToStream added in v3.3.0 

func (s *StateMachine) ReadyToStream() bool

ReadyToStream returns a boolean flag to indicate whether the state machine is ready to stream snapshot. It can not stream a full snapshot when membership state is catching up with the all disk SM state. Meta only snapshot can be taken at any time.

func (*StateMachine) Recover added in v3.3.0 

func (s *StateMachine) Recover(t Task) (uint64, error)

Recover applies the snapshot.

func (*StateMachine) Save added in v3.3.0 

func (s *StateMachine) Save(req SSRequest) (pb.Snapshot, SSEnv, error)

Save creates a snapshot.

func (*StateMachine) SetLastApplied added in v3.3.0 

func (s *StateMachine) SetLastApplied(index uint64)

SetLastApplied sets the last applied index to the specified value. This method is only used in tests.

func (*StateMachine) Stream added in v3.3.0 

func (s *StateMachine) Stream(sink pb.IChunkSink) error

Stream starts to stream snapshot from the current SM to a remote node targeted by the provided sink.

func (*StateMachine) Sync 

func (s *StateMachine) Sync() error

Sync synchronizes state machine's in-core state with that on disk.

func (*StateMachine) TaskChanBusy 

func (s *StateMachine) TaskChanBusy() bool

TaskChanBusy returns whether the TaskC chan is busy. Busy is defined as having more than half of its buffer occupied.

func (*StateMachine) TaskQ 

func (s *StateMachine) TaskQ() *TaskQueue

TaskQ returns the task queue.

func (*StateMachine) Type added in v3.3.0 

func (s *StateMachine) Type() pb.StateMachineType

Type returns the state machine type.

type Task 

type Task struct {
	ClusterID    uint64
	NodeID       uint64
	Index        uint64
	Recover      bool
	Initial      bool
	Save         bool
	Stream       bool
	PeriodicSync bool
	NewNode      bool
	SSRequest    SSRequest
	Entries      []pb.Entry
}

Task describes a task that need to be handled by StateMachine.

func (*Task) IsSnapshotTask 

func (t *Task) IsSnapshotTask() bool

IsSnapshotTask returns a boolean flag indicating whether it is a snapshot task.

type TaskQueue 

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

TaskQueue is a queue of tasks to be processed by the state machine.

func NewTaskQueue 

func NewTaskQueue() *TaskQueue

NewTaskQueue creates and returns a new task queue.

func (*TaskQueue) Add 

func (tq *TaskQueue) Add(task Task)

Add adds a new task to the queue.

func (*TaskQueue) Get 

func (tq *TaskQueue) Get() (Task, bool)

Get returns a task from the queue if there is any.

func (*TaskQueue) GetAll added in v3.3.0 

func (tq *TaskQueue) GetAll() []Task

GetAll returns all tasks currently in the queue.

func (*TaskQueue) MoreEntryToApply 

func (tq *TaskQueue) MoreEntryToApply() bool

MoreEntryToApply returns a boolean value indicating whether it is ok to queue more entries to apply.

func (*TaskQueue) Size 

func (tq *TaskQueue) Size() uint64

Size returns the number of queued tasks.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.