Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrDataTooBig = errors.New("data size is too big")
ErrDataTooBig is returned if the given data is bigger than the predefined limit
var ErrSizeLimitOutOfRange = errors.New("size limit is, for now, file = [256KB, 2MB], itemSize <= 128KB")
ErrSizeLimitOutOfRange is returned when the given size is bigger than our limit
var ErrWalFileCorrupted = errors.New("wal file checksum mismatch, the file is considered corrupted")
ErrWalFileCorrupted is returned when checksum check fails
var ErrWalManagerClosed = errors.New("wal Manager is already closed")
ErrWalManagerClosed is returned when `Record()` is called after Close is called
var FILE_TERMINAL_CODE = []byte{0}
finding this means this file has no more records
var NEW_RECORD_CODE = []byte{1}
finding this means it is a new record
var RECORD_METADATA_SIZE = 9
RECORD_METADATA_SIZE is other data recorded together with data
consists of:
1. 1 byte code (1 for new record)
2. 4 byte data length
3. 4 byte crc32 checksum
var WAL_BATCH_LOWER_LIMIT = 128 * 1024
too small data fsync'ed waste the call
128KB seems like a good balance
var WAL_BATCH_UPPER_LIMIT = 4 * 1024 * 1024
a WAL file is 32MB, and we allow 4 pending in the fsync chan. to guarantee only 1 new file not yet created, we need to bind the hard limit to also be 32/4 = 8 MB.
Just to allow more into a file, divide by 2, so 4 MB
var WAL_FILE_FLAG = os.O_CREATE | os.O_RDWR
var WAL_FILE_MODE = os.FileMode(0644)
var WAL_FILE_SIZE = 32 * 1024 * 1024
var WAL_ITEM_SIZE_LIMIT = 256 * 1024
It is supposed to be binary data, and because aimed for business case, 256KB should be much more than enough
For comparison:
1. AWS SQS allows 256KB, but priced for each 256KB
2. Facebook's FOQS allows only 10KB message size
var WRITE_CHAN_SIZE = 4
WRITE_CHAN_SIZE limits the number of waiting to be written and/or fsync'ed As doing more, we gonna just put in the backlog, while the OS/disk is having problem keeping-up
Functions ¶
func PutWalBuffer ¶
func PutWalBuffer(buf []byte)
PutWalBuffer returns wal buffer to pool. Can be used to maintain memory usage during snapshot/recovery.
No published `GetWalBuffer` method, because mostly the buffer is used by `Load()` walManager calls
Types ¶
type AntriWalManager ¶
type AntriWalManager struct {
// contains filtered or unexported fields
}
AntriWalManager handles how antri interacts with filesystem, and implements `walManagerInterface`. The goal is to allow antri amortizes the cost of fsync across multiple request.
This implementation allows `record()` call while fsync is on progress. To do that, that means fsync should be outside the lock, but still need reliable way to notify the fsync goroutine. This is done via fsync chan, buffered (hardcoded to 4, for now). If chan buffer is full, it is okay to still hold main lock (and block others), as it means the storage layer is not keeping up. (My decision, as of the time of writing)
This implementation only uses `fsync`, and not `O_[D]SYNC`. While O_SYNC is usually faster, it doesn't work on windows, so writes on windows may not be synced at all. With fsync, both linux variant and windows behave the same, resulting in easy portability :)
Important notes (probably need to be implemented at some point):
1. Aligned write (4KB page), to reduce stalls (stable write), while still being safe. Also with O_DIRECT, bypass kernel
2. fsync the directory after file creation (how to do this?)
func NewWalManager ¶
func NewWalManager( dataDir string, itemSizeLimit int, batchSize int, timeLimit time.Duration, fsyncOnWrite bool) (*AntriWalManager, error)
NewWalManager creates our AntriWalManager object
func (*AntriWalManager) Close ¶
func (w *AntriWalManager) Close()
Close this instance, rejecting subsequent request
func (*AntriWalManager) Load ¶
func (w *AntriWalManager) Load(fileCounter uint64) ([][]byte, error)
Load file of the given counter
This function is not latency critical, and should be easy to parallelize (with bit of change)
For now, it uses defer for readability.
Any other errors happening here, probably the file is corrupted so bad
func (*AntriWalManager) Record ¶
func (w *AntriWalManager) Record(data []byte) (RecordHandle, error)
Record the given data to the batch, to be handled by committer goroutine The format is:
1. 1 byte `1`, indicating a new data
2. 4 bytes data length
3. data itself
4. a crc32 checksum. This also acts as commit record (as it is under page size, which usually is 4 KB)
func (*AntriWalManager) Run ¶
func (w *AntriWalManager) Run(initialCounter uint64)
Run our WalManager, also all corresponding background goroutines
type RecordHandle ¶
type RecordHandle struct {
// contains filtered or unexported fields
}
RecordHandle is our `future` implementation which will notify its caller the result of the batch.
we separate this implementation from batchHandle so we can easily add more attribute to it later, if needed
func (*RecordHandle) GetBatchId ¶
func (rh *RecordHandle) GetBatchId() uint64
GetBatchId allows the caller to track which batch the returned value corresponds to
func (*RecordHandle) GetFileNumber ¶
func (rh *RecordHandle) GetFileNumber() uint64
GetFileNumber waits until the batch is finished, then return in which wal file the batch is
type WalManagerInterface ¶
type WalManagerInterface interface {
Run(int) error
Record([]byte) (RecordHandle, error)
Load(int) ([]byte, error)
}
WalManagerInterface should be implemented by core implementation of WalManager
Source Files