wal

Documentation

Index

• Variables
• func AddExistingWALFileToChan(c chan string, archiveWalFolder string) error
• type ArchivedFileFuncter
• type File
  • func ReadFileFromPath(path string) (*File, error)
  • func (wf *File) ColdReplay(activeFolder, archiveFolder string) wutils.ErrorList
• type PersistentState
  • func InitPersistentFileFromDisk(config config.Config, shardIndex int) (*PersistentState, error)
  • func (p *PersistentState) Save() error
• type Replicator
  • func InitReplicator(archiveEventChans []chan string, activeFolder, archiveFolder string, ...) *Replicator
  • func InitReplicatorWithOneFile(path string, activeFolder, archiveFolder string, archiveCmd string, ...) *Replicator
  • func (r *Replicator) Execute() error
  • func (r *Replicator) Start()
  • func (r *Replicator) Stop()
• type ShardWAL
  • func InitShardWAL(config config.Config, logger *zap.Logger, ...) (*ShardWAL, error)
  • func (swa *ShardWAL) CloseAll() wutils.ErrorList
  • func (swa *ShardWAL) CloseShardIndex(shardIndex uint32) error
  • func (swa *ShardWAL) ExecRsyncCommand(params map[string][]string) ([]byte, error)
  • func (swa *ShardWAL) FlushAll() (errOpsCount int, errors wutils.ErrorList)
  • func (swa *ShardWAL) FlushShardIndex(shardIndex uint32) (errOpsCount int, err error)
  • func (swa *ShardWAL) GetArchiveEventChan() []chan string
  • func (swa *ShardWAL) GetWalForShardIndex(shardIndex uint32) *WAL
  • func (swa *ShardWAL) LockShardIndex(shardIndex uint32)
  • func (swa *ShardWAL) UnlockShardIndex(shardIndex uint32)
• type WAL
  • func InitWAL(fileExecutor *fileop.BucketFileOperationner, c config.Config, shardIndex int, ...) (*WAL, error)
  • func (w *WAL) AppendWrite(cf config.ContainerFile, buffers ...[]byte) error
  • func (w *WAL) Archive(cf config.ContainerFile) error
  • func (w *WAL) Close() error
  • func (w *WAL) Flush() (errOpsCount int, err error)
  • func (w *WAL) GetArchiveEventChan() chan string
  • func (w *WAL) GetArchiveFileCreatedEventChan() chan string
  • func (w *WAL) GetFileBuffer(cf config.ContainerFile, fileBuf *wutils.Buffer) error
  • func (w *WAL) Truncate(cf config.ContainerFile, offset int64) error
  • func (w *WAL) Write(cf config.ContainerFile, fileOffset int64, fileSize int64, buffers ...[]byte) error

Variables

var ErrBadWalFileCrcCommand = errors.New("bad CRC for command")

ErrBadWalFileCrcCommand means a command has a bad CRC and will be discarded

Functions

func AddExistingWALFileToChan

func AddExistingWALFileToChan(c chan string, archiveWalFolder string) error

AddExistingWALFileToChan add wal file in order to this channel

Types

type ArchivedFileFuncter

type ArchivedFileFuncter interface {
	Do(path string, file config.ContainerFile)
	Close()
}

ArchivedFileFuncter func called when a new archived file is created

type File

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

File content

func ReadFileFromPath

func ReadFileFromPath(path string) (*File, error)

ReadFileFromPath read a wal file from a specific path

func (*File) ColdReplay

func (wf *File) ColdReplay(activeFolder, archiveFolder string) wutils.ErrorList

ColdReplay will replay all the cmds of the wal file

type PersistentState

type PersistentState struct {
	WalIndex uint64
	// contains filtered or unexported fields
}

PersistentState save information needed between restart

func InitPersistentFileFromDisk

func InitPersistentFileFromDisk(config config.Config, shardIndex int) (*PersistentState, error)

InitPersistentFileFromDisk restore from file

func (*PersistentState) Save

func (p *PersistentState) Save() error

Save state to file

type Replicator

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

Replicator use the archive wal file to make a live copy of the files or will archive the wal in another folder or both. As it's completely unrelated to the bdd, we can copy file on a slow fs.

func InitReplicator

func InitReplicator(archiveEventChans []chan string, activeFolder, archiveFolder string, archiveCmd string, logger *zap.Logger) *Replicator

InitReplicator init a replicator

func InitReplicatorWithOneFile

func InitReplicatorWithOneFile(path string, activeFolder, archiveFolder string, archiveCmd string, logger *zap.Logger) *Replicator

InitReplicatorWithOneFile use when reloading actual wal file

func (*Replicator) Execute

func (r *Replicator) Execute() error

Execute synchronously the replicator

func (*Replicator) Start

func (r *Replicator) Start()

Start execute the replicator

func (*Replicator) Stop

func (r *Replicator) Stop()

Stop execute the replicator and wait for it to finish current wal

type ShardWAL

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

ShardWAL is a shard of wal to maximise cpu bound operation and and spread checpoint operations

func InitShardWAL

func InitShardWAL(config config.Config, logger *zap.Logger, archivedFileFuncter ArchivedFileFuncter) (*ShardWAL, error)

InitShardWAL init a shard wal

func (*ShardWAL) CloseAll

func (swa *ShardWAL) CloseAll() wutils.ErrorList

CloseAll wal file. It's a blocking operation.

func (*ShardWAL) CloseShardIndex

func (swa *ShardWAL) CloseShardIndex(shardIndex uint32) error

CloseShardIndex Close shard index. It's a blocking operation.

func (*ShardWAL) ExecRsyncCommand

func (swa *ShardWAL) ExecRsyncCommand(params map[string][]string) ([]byte, error)

ExecRsyncCommand will clean up, pause, execute the rsync command and resume

func (*ShardWAL) FlushAll

func (swa *ShardWAL) FlushAll() (errOpsCount int, errors wutils.ErrorList)

FlushAll wal file. It's a blocking operation.

func (*ShardWAL) FlushShardIndex

func (swa *ShardWAL) FlushShardIndex(shardIndex uint32) (errOpsCount int, err error)

FlushShardIndex flush shard index. It's a blocking operation.

func (*ShardWAL) GetArchiveEventChan

func (swa *ShardWAL) GetArchiveEventChan() []chan string

GetArchiveEventChan get all archive envent chan

func (*ShardWAL) GetWalForShardIndex

func (swa *ShardWAL) GetWalForShardIndex(shardIndex uint32) *WAL

GetWalForShardIndex get wal for shard index

func (*ShardWAL) LockShardIndex

func (swa *ShardWAL) LockShardIndex(shardIndex uint32)

LockShardIndex lock

func (*ShardWAL) UnlockShardIndex

func (swa *ShardWAL) UnlockShardIndex(shardIndex uint32)

UnlockShardIndex unlock

type WAL

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

WAL is used to make sure, we can restore state after a power failure or software failure

func InitWAL

func InitWAL(fileExecutor *fileop.BucketFileOperationner, c config.Config, shardIndex int, logger *zap.Logger, currentCheckpointShardIndex *int32) (*WAL, error)

InitWAL init the wal file

func (*WAL) AppendWrite

func (w *WAL) AppendWrite(cf config.ContainerFile, buffers ...[]byte) error

AppendWrite append write to a file

func (*WAL) Archive

func (w *WAL) Archive(cf config.ContainerFile) error

Archive file

func (*WAL) Close

func (w *WAL) Close() error

Close flush current wal file and close all files

func (*WAL) Flush

func (w *WAL) Flush() (errOpsCount int, err error)

Flush current wal file

func (*WAL) GetArchiveEventChan

func (w *WAL) GetArchiveEventChan() chan string

GetArchiveEventChan for this wal

func (*WAL) GetArchiveFileCreatedEventChan

func (w *WAL) GetArchiveFileCreatedEventChan() chan string

GetArchiveFileCreatedEventChan when an archive file is created the ContainerFile key is send to this channel

func (*WAL) GetFileBuffer

func (w *WAL) GetFileBuffer(cf config.ContainerFile, fileBuf *wutils.Buffer) error

GetFileBuffer for file

func (*WAL) Truncate

func (w *WAL) Truncate(cf config.ContainerFile, offset int64) error

Truncate file

func (*WAL) Write

func (w *WAL) Write(cf config.ContainerFile, fileOffset int64, fileSize int64, buffers ...[]byte) error