fifodb

README

FifoDB - persistent fifo disk-based queue

Thread-safe FIFO-queue backend by a filesystem and acknowledgement opportunity.

Motivation

• LSM based storage (like Badger, RocksDB, LevelDB, etc) are very nice and powerful, but much strong and has overhead for LSM mechanics
• LSM based storage need more resources for each instance (arena size, io/compaction loops). So we can't get new instance for every queue, only for every virtual host and need to implement bucket system on keys like 'vhost-name.queue-name.'
• need simple embedded native FIFO disk storage
• have fun and learn a lot :)

Basic idea

Basic ideas of storage is

• stores data in 64MB segments for compaction performance (configurable)
• stores data strongly in FIFO order
• retrieves data strongly in FIFO order
• supports 4 main methods - Push, Pop, Ack

Usage

Internals

We need 2 additional and strange methods 'Ack' and 'Nack' to handle 'At least once' delivery guarantees after server crash or restart. User can simply 'Push' data into queue tail and 'Pop' data from queue head. 'Pop' always returns NEXT message from queue after 'Open' and ignores Nack-ed messages. 'Ack' and 'Nack' are only for compaction and rebuilding queue after server crash or restart.

Documentation

Index

• Variables
• type DB
  • func Open(opt Options) (db *DB, msgLeft uint64, err error)
  • func (db *DB) Ack(segID uint32, offset uint32) (err error)
  • func (db *DB) Clean() (err error)
  • func (db *DB) Close() (err error)
  • func (db *DB) GetPath() string
  • func (db *DB) Pop() (data []byte, segID uint32, offset uint32, err error)
  • func (db *DB) Push(data []byte) (segID uint32, offset uint32, err error)
  • func (db *DB) SetReadPos(fileID uint32, offset uint32) error
  • func (db *DB) Sync() error
• type Options

Variables

var DefaultOptions = Options{
	WriteBufferSize:    128 << 10,
	ReadBufferSize:     128 << 10,
	MaxBytesPerSegment: 128 << 20,
	Fsync:              false,
}

DefaultOptions contains options that should work for most applications

Types

type DB

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

DB represents thread-safe FIFO disk queue with WAL of queue operations.

func Open

func Open(opt Options) (db *DB, msgLeft uint64, err error)

func (*DB) Ack

func (db *DB) Ack(segID uint32, offset uint32) (err error)

func (*DB) Clean

func (db *DB) Clean() (err error)

Clean deletes all db files Thread-safe.

func (*DB) Close

func (db *DB) Close() (err error)

func (*DB) GetPath

func (db *DB) GetPath() string

func (*DB) Pop

func (db *DB) Pop() (data []byte, segID uint32, offset uint32, err error)

func (*DB) Push

func (db *DB) Push(data []byte) (segID uint32, offset uint32, err error)

func (*DB) SetReadPos

func (db *DB) SetReadPos(fileID uint32, offset uint32) error

func (*DB) Sync

func (db *DB) Sync() error

type Options

type Options struct {
	// Directory to store the data. Would be created if not exists.
	Path string

	// Size of bytes to keep in memory before persisting to avoid much write syscalls.
	WriteBufferSize uint32

	// Size of bytes buffered from file to avoid much read syscalls.
	ReadBufferSize uint32

	// Maximum size of each segment file.
	MaxBytesPerSegment uint32

	// Fsync each write
	Fsync bool
}

Options are params for creating DB object.