Documentation ¶
Overview ¶
Package bigqueue provides embedded, fast and persistent queue written in pure Go using memory mapped file. bigqueue is currently not thread safe. To use bigqueue in parallel context, a Write lock needs to be acquired (even for Read APIs).
Create or open a bigqueue:
bq, err := bigqueue.NewQueue("path/to/queue") defer bq.Close()
bigqueue persists the data of the queue in multiple Arenas. Each Arena is a file on disk that is mapped into memory (RAM) using mmap syscall. Default size of each Arena is set to 128MB. It is possible to create a bigqueue with custom Arena size:
bq, err := bigqueue.NewQueue("path/to/queue", bigqueue.SetArenaSize(4*1024)) defer bq.Close()
bigqueue also allows setting up the maximum possible memory that it can use. By default, the maximum memory is set to [3 x Arena Size].
bq, err := bigqueue.NewQueue("path/to/queue", bigqueue.SetArenaSize(4*1024), bigqueue.SetMaxInMemArenas(10)) defer bq.Close()
In this case, bigqueue will never allocate more memory than `4KB*10=40KB`. This memory is above and beyond the memory used in buffers for copying data.
Bigqueue allows to set periodic flush based on either elapsed time or number of mutate (enqueue/dequeue) operations. Flush syncs the in memory changes of all memory mapped files with disk. *This is a best effort flush*. Elapsed time and number of mutate operations are only checked upon an enqueue/dequeue.
This is how we can set these options:
bq, err := bigqueue.NewQueue("path/to/queue", bigqueue.SetPeriodicFlushOps(2))
In this case, a flush is done after every two mutate operations.
bq, err := bigqueue.NewQueue("path/to/queue", bigqueue.SetPeriodicFlushDuration(time.Minute))
In this case, a flush is done after one minute elapses and an Enqueue/Dequeue is called.
Write to bigqueue:
err := bq.Enqueue([]byte("elem")) // size = 1
bigqueue allows writing string data directly, avoiding conversion to `[]byte`:
err := bq.EnqueueString("elem") // size = 2
Read from bigqueue:
elem, err := bq.Dequeue()
we can also read string data from bigqueue:
elem, err := bq.DequeueString()
Check whether bigqueue has non zero elements:
isEmpty := bq.IsEmpty()
bigqueue allows reading data from bigqueue using consumers similar to Kafka. This allows multiple consumers from reading data at different offsets (not in thread safe manner yet). The offsets of each consumer are persisted on disk and can be retrieved by creating a consumer with the same name. Data will be read from the same offset where it was left off.
We can create a new consumer as follows. The offsets of a new consumer are set at the start of the queue wherever the first non-deleted element is.
consumer, err := bq.NewConsumer("consumer")
We can also copy an existing consumer. This will create a consumer that will have the same offsets into the queue as that of the existing consumer.
copyConsumer, err := bq.FromConsumer("copyConsumer", consumer)
Now, read operations can be performed on the consumer:
isEmpty := consumer.IsEmpty() elem, err := consumer.Dequeue() elem, err := consumer.DequeueString()
Index ¶
- Variables
- type Consumer
- type MmapQueue
- func (q *MmapQueue) Close() error
- func (q *MmapQueue) Dequeue() ([]byte, error)
- func (q *MmapQueue) DequeueString() (string, error)
- func (q *MmapQueue) Enqueue(message []byte) error
- func (q *MmapQueue) EnqueueString(message string) error
- func (q *MmapQueue) Flush() error
- func (q *MmapQueue) FromConsumer(name string, from *Consumer) (*Consumer, error)
- func (q *MmapQueue) IsEmpty() bool
- func (q *MmapQueue) NewConsumer(name string) (*Consumer, error)
- type Option
Constants ¶
This section is empty.
Variables ¶
var ( // ErrInvalidArenaSize is returned when persisted arena size // doesn't match with desired arena size. ErrInvalidArenaSize = errors.New("mismatch in arena size") // ErrDifferentQueues is returned when caller wants to copy // offsets from a consumer from a different queue. ErrDifferentQueues = errors.New("consumers from different queues") )
var ( // ErrTooSmallArenaSize is returned when arena size is smaller than OS page size. ErrTooSmallArenaSize = errors.New("too small arena size") // ErrTooFewInMemArenas is returned when number of arenas allowed in memory < 3. ErrTooFewInMemArenas = errors.New("too few in memory arenas") )
var ( // ErrEmptyQueue is returned when dequeue is performed on an empty queue. ErrEmptyQueue = errors.New("queue is empty") )
var ( // ErrIncompatibleVersion is returned when file format is older/newer. ErrIncompatibleVersion = errors.New("incompatible format of the code and data") )
Functions ¶
This section is empty.
Types ¶
type Consumer ¶ added in v0.4.0
type Consumer struct {
// contains filtered or unexported fields
}
Consumer is a bigqueue consumer that allows reading data from bigqueue. A consumer is represented using just a base offset into the metadata
func (*Consumer) Dequeue ¶ added in v0.4.0
Dequeue removes an element from the queue and returns it.
func (*Consumer) DequeueString ¶ added in v0.4.0
DequeueString removes a string element from the queue and returns it.
type MmapQueue ¶ added in v0.3.0
type MmapQueue struct {
// contains filtered or unexported fields
}
MmapQueue implements Queue interface.
func NewMmapQueue ¶ added in v0.3.0
NewMmapQueue constructs a new persistent queue.
func (*MmapQueue) Dequeue ¶ added in v0.3.0
Dequeue removes an element from the queue and returns it. This function uses the default consumer to consume from the queue.
func (*MmapQueue) DequeueString ¶ added in v0.4.0
DequeueString removes a string element from the queue and returns it. This function uses the default consumer to consume from the queue.
func (*MmapQueue) Enqueue ¶ added in v0.3.0
Enqueue adds a new slice of byte element to the tail of the queue.
func (*MmapQueue) EnqueueString ¶ added in v0.3.0
EnqueueString adds a new string element to the tail of the queue.
func (*MmapQueue) FromConsumer ¶ added in v0.4.0
FromConsumer creates a new consumer or finds an existing one with same name. It also copies the offsets from the given consumer to this consumer.
type Option ¶ added in v0.2.0
type Option func(*bqConfig) error
Option is function type that takes a bqConfig object and sets various config parameters in the object.
func SetArenaSize ¶ added in v0.2.0
SetArenaSize returns an Option closure that sets the arena size.
func SetMaxInMemArenas ¶ added in v0.2.0
SetMaxInMemArenas returns an Option closure that sets maximum number of Arenas that could reside in memory (RAM) at any time. By default, all the arenas reside in memory and Operating System takes care of memory by paging in and out the pages from disk. A recommended value of maximum arenas that should be in memory should be chosen such that:
maxInMemArenas > (# of consumers) * 2 + 1
func SetPeriodicFlushDuration ¶ added in v0.3.0
SetPeriodicFlushDuration returns an Option that sets a periodic flush every given duration after which the queue's in-memory changes will be flushed to disk. This is a best effort flush and elapsed time is checked upon an enqueue/dequeue only. If the value is set to <= 0, no periodic flush will be performed.
For durability this value should be low. For performance this value should be high.
func SetPeriodicFlushOps ¶ added in v0.3.0
SetPeriodicFlushOps returns an Option that sets the number of mutate operations (enqueue/dequeue) after which the queue's in-memory changes will be flushed to disk. This is a best effort flush and number of mutate operations is checked upon an enqueue/dequeue. If the value is set to <= 0, no periodic flush will be performed.
For durability this value should be low. For performance this value should be high.