wal

Documentation

Index

• Constants
• Variables
• func WalFileName(dir string, fid uint32, ext string) string
• type ChunkHeader
• type ChunkIterator
• type ChunkPos
• type ChunkType
• type F
  • func (f F) Less(nf F) bool
• type File
• type FileIterator
  • func (f *FileIterator) IndexFid() uint32
  • func (f *FileIterator) IndexPos() ChunkPos
  • func (f *FileIterator) NextData() ([]byte, ChunkPos, error)
  • func (f *FileIterator) NextFile()
• type LogFile
  • func (log *LogFile) Close() error
  • func (log *LogFile) Fid() uint32
  • func (log *LogFile) Iterator() (ChunkIterator, error)
  • func (log *LogFile) MarkImmutable()
  • func (log *LogFile) Read(block uint32, chunkOffset int64) ([]byte, error)
  • func (log *LogFile) Remove() error
  • func (log *LogFile) Size() int64
  • func (log *LogFile) Sync() error
  • func (log *LogFile) Write(data []byte) (ChunkPos, error)
  • func (log *LogFile) WriteAll(data [][]byte) ([]ChunkPos, error)
• type LogFileChunkIterator
  • func (l *LogFileChunkIterator) File() File
  • func (l *LogFileChunkIterator) Index() ChunkPos
  • func (l *LogFileChunkIterator) Next() ([]byte, ChunkPos, error)
• type Option
• type Wal
  • func Open(option Option) (*Wal, error)
  • func (w *Wal) ActiveFid() uint32
  • func (w *Wal) Close() error
  • func (w *Wal) IsEmpty() bool
  • func (w *Wal) Iterator(minFid, maxFid uint32, pos ChunkPos) (FileIterator, error)
  • func (w *Wal) PendingClean()
  • func (w *Wal) PendingPush() ([]ChunkPos, error)
  • func (w *Wal) PendingWrite(data []byte)
  • func (w *Wal) Purge() error
  • func (w *Wal) Read(pos ChunkPos) ([]byte, error)
  • func (w *Wal) Rotate() error
  • func (w *Wal) Sync() error
  • func (w *Wal) Write(data []byte) (ChunkPos, error)

Constants

const ChunkHeaderSize = 7

ChunkHeaderSize +-----------+--------------+------------+------------+ | check sum | chunk length | chunk type | chunk data | +-----------+--------------+------------+------------+ | 4 Bytes | 2 Bytes | 1 Bytes | <= 32KB-7B | +-----------+--------------+------------+------------+

const DefaultWalSuffix = "wal"

DefaultWalSuffix is the extension name of wal file, default is wal. eg: 001.wal

const MaxBlockSize = 32 * file.KB

Variables

var (
	ErrAlreadyClosed    = errors.New("file already closed")
	ErrBlockExceeded    = errors.New("block size exceeded")
	ErrWriteToImmutable = errors.New("write to immutable")
)

var (
	ErrFileNotFound   = errors.New("wal file not found")
	ErrDataExceedFile = errors.New("data size has exceeded the max file limit")
)

Functions

func WalFileName

func WalFileName(dir string, fid uint32, ext string) string

WalFileName returns the name of wal file

Types

type ChunkHeader

type ChunkHeader struct {
	Length    uint16
	ChunkType ChunkType
	// contains filtered or unexported fields
}

type ChunkIterator

type ChunkIterator interface {
	// File returns file that iterator belongs to
	File() File
	// Index returns current chunk position
	Index() ChunkPos
	// Next returns the next chunk data bytes, chunk info, and error
	Next() ([]byte, ChunkPos, error)
}

ChunkIterator iterate chunk data for a log file

type ChunkPos

type ChunkPos struct {
	// file id
	Fid uint32
	// chunk in which block
	Block uint32
	// chunk offset in block
	Offset int64
	// chunk size
	Size int64
}

type ChunkType

type ChunkType = byte

const (
	ChunkFullType ChunkType = 1 + iota
	ChunkFirstType
	ChunkMiddleType
	ChunkLastType
)

type F

type F struct {
	Fid  uint32
	File File
}

func (F) Less

func (f F) Less(nf F) bool

type File

type File interface {
	// Fid return file id
	Fid() uint32
	// Read bytes with specified block and offset
	// chunkOffset is offset by the start of the specified block
	Read(block uint32, chunkOffset int64) ([]byte, error)
	// Write bytes and returns ChunkPos
	Write(data []byte) (ChunkPos, error)
	// WriteAll write bytes in batch and returns ChunkPos
	WriteAll(data [][]byte) ([]ChunkPos, error)
	// Iterator iterate per chunk in file
	Iterator() (ChunkIterator, error)
	// Size return size of file
	Size() int64
	// Remove remove file
	Remove() error
	// Sync buffer to disk
	Sync() error
	// MarkImmutable mark this file as immutable
	MarkImmutable()
	// Close file and release resource
	Close() error
}

File represents a wal file

type FileIterator

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

FileIterator iterates over all the files and chunks

func (*FileIterator) IndexFid

func (f *FileIterator) IndexFid() uint32

IndexFid returns the current fid

func (*FileIterator) IndexPos

func (f *FileIterator) IndexPos() ChunkPos

IndexPos returns curren index position

func (*FileIterator) NextData

func (f *FileIterator) NextData() ([]byte, ChunkPos, error)

func (*FileIterator) NextFile

func (f *FileIterator) NextFile()

type LogFile

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

func (*LogFile) Close

func (log *LogFile) Close() error

func (*LogFile) Fid

func (log *LogFile) Fid() uint32

func (*LogFile) Iterator

func (log *LogFile) Iterator() (ChunkIterator, error)

func (*LogFile) MarkImmutable

func (log *LogFile) MarkImmutable()

func (*LogFile) Read

func (log *LogFile) Read(block uint32, chunkOffset int64) ([]byte, error)

func (*LogFile) Remove

func (log *LogFile) Remove() error

func (*LogFile) Size

func (log *LogFile) Size() int64

func (*LogFile) Sync

func (log *LogFile) Sync() error

func (*LogFile) Write

func (log *LogFile) Write(data []byte) (ChunkPos, error)

func (*LogFile) WriteAll

func (log *LogFile) WriteAll(data [][]byte) ([]ChunkPos, error)

type LogFileChunkIterator

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

func (*LogFileChunkIterator) File

func (l *LogFileChunkIterator) File() File

func (*LogFileChunkIterator) Index

func (l *LogFileChunkIterator) Index() ChunkPos

func (*LogFileChunkIterator) Next

func (l *LogFileChunkIterator) Next() ([]byte, ChunkPos, error)

type Option

type Option struct {
	// data dir
	DataDir string
	// max log file size
	MaxFileSize int64
	// log file extension
	Ext string
	// how many block will be cached, off if it is 0
	BlockCache uint32
	// sync buffer to disk per write, if not enabled data maybe not save when machine crashed
	FsyncPerWrite bool
	// specify the written threshold for triggering sync
	FsyncThreshold int64
}

type Wal

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

Wal instance represents a data directory and its sub-files that store write-ahead-log format data

func Open

func Open(option Option) (*Wal, error)

Open returns a new wal instance with the give Option

func (*Wal) ActiveFid

func (w *Wal) ActiveFid() uint32

ActiveFid returns current active file id

func (*Wal) Close

func (w *Wal) Close() error

Close will close all the immutable files sync the active file and close it

func (*Wal) IsEmpty

func (w *Wal) IsEmpty() bool

IsEmpty check wal data if is empty

func (*Wal) Iterator

func (w *Wal) Iterator(minFid, maxFid uint32, pos ChunkPos) (FileIterator, error)

Iterator returns an FileIterator maxFid determines the file range for the iterator pos determines the chunk position for the iterator

func (*Wal) PendingClean

func (w *Wal) PendingClean()

func (*Wal) PendingPush

func (w *Wal) PendingPush() ([]ChunkPos, error)

PendingPush writes pendingBytes to active file

func (*Wal) PendingWrite

func (w *Wal) PendingWrite(data []byte)

PendingWrite writes data to pendingBytes

func (*Wal) Purge

func (w *Wal) Purge() error

Purge close all the file and remove them

func (*Wal) Read

func (w *Wal) Read(pos ChunkPos) ([]byte, error)

Read reads data at the specified position

func (*Wal) Rotate

func (w *Wal) Rotate() error

func (*Wal) Sync

func (w *Wal) Sync() error

Sync call fsync on the active file

func (*Wal) Write

func (w *Wal) Write(data []byte) (ChunkPos, error)

Write writes data bytes to the active file