persistentqueue

Documentation

Index

• Constants
• type FastQueue
  • func MustOpenFastQueue(path, name string, maxInmemoryBlocks int, maxPendingBytes int64, ...) *FastQueue
  • func MustOpenFastQueueWithOpts(path, name string, opts OpenFastQueueOpts) *FastQueue
  • func (fq *FastQueue) Dirname() string
  • func (fq *FastQueue) GetInmemoryQueueLen() int
  • func (fq *FastQueue) GetPendingBytes() uint64
  • func (fq *FastQueue) IsPersistentQueueDisabled() bool
  • func (fq *FastQueue) IsWriteBlocked() bool
  • func (fq *FastQueue) MustClose()
  • func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool)
  • func (fq *FastQueue) MustReadInMemoryBlock(dst []byte) ([]byte, bool)
  • func (fq *FastQueue) MustReadInMemoryBlockBlocking(dst []byte) ([]byte, bool)
  • func (fq *FastQueue) MustWriteBlockIgnoreDisabledPQ(block []byte)
  • func (fq *FastQueue) TryWriteBlock(block []byte) bool
  • func (fq *FastQueue) UnblockAllReaders()
• type OpenFastQueueOpts

Constants

const DefaultChunkFileSize = (MaxBlockSize + 8) * 16

DefaultChunkFileSize represents default chunk file size

const MaxBlockSize = 32 * 1024 * 1024

MaxBlockSize is the maximum size of the block persistent queue can work with.

Types

type FastQueue

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

FastQueue is fast persistent queue, which prefers sending data via memory.

It falls back to sending data via file when readers don't catch up with writers.

func MustOpenFastQueue

func MustOpenFastQueue(path, name string, maxInmemoryBlocks int, maxPendingBytes int64, isPQDisabled bool) *FastQueue

MustOpenFastQueue opens persistent queue at the given path.

It holds up to maxInmemoryBlocks in memory before falling back to file-based persistence.

if maxPendingBytes is 0, then the queue size is unlimited. Otherwise its size is limited by maxPendingBytes. The oldest data is dropped when the queue reaches maxPendingSize. if isPQDisabled is set to true, then write requests that exceed in-memory buffer capacity are rejected. in-memory queue part can be stored on disk during graceful shutdown.

func MustOpenFastQueueWithOpts

func MustOpenFastQueueWithOpts(path, name string, opts OpenFastQueueOpts) *FastQueue

MustOpenFastQueueWithOpts opens persistent queue at the given path with given opts

func (*FastQueue) Dirname

func (fq *FastQueue) Dirname() string

Dirname returns the directory name for persistent queue.

func (*FastQueue) GetInmemoryQueueLen

func (fq *FastQueue) GetInmemoryQueueLen() int

GetInmemoryQueueLen returns the length of inmemory queue.

func (*FastQueue) GetPendingBytes

func (fq *FastQueue) GetPendingBytes() uint64

GetPendingBytes returns the number of pending bytes in the fq.

func (*FastQueue) IsPersistentQueueDisabled

func (fq *FastQueue) IsPersistentQueueDisabled() bool

IsPersistentQueueDisabled returns true if persistent queue at fq is disabled.

func (*FastQueue) IsWriteBlocked

func (fq *FastQueue) IsWriteBlocked() bool

IsWriteBlocked checks if data can be pushed into fq

func (*FastQueue) MustClose

func (fq *FastQueue) MustClose()

MustClose unblocks all the readers.

It is expected no new writers during and after the call.

func (*FastQueue) MustReadBlock

func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool)

MustReadBlock reads the next block from fq into dst and returns it. It first reads from the file-based queue, then checks in-memory queue. It blocks until a block is available or the stop deadline is exceeded, in which case it returns (dst, false).

func (*FastQueue) MustReadInMemoryBlock

func (fq *FastQueue) MustReadInMemoryBlock(dst []byte) ([]byte, bool)

MustReadInMemoryBlock reads the next block from the in-memory queue into dst and returns it. It returns (dst, true) if a block was available, or (nil, false) if the in-memory queue is empty. It does not block waiting for new blocks.

func (*FastQueue) MustReadInMemoryBlockBlocking

func (fq *FastQueue) MustReadInMemoryBlockBlocking(dst []byte) ([]byte, bool)

MustReadInMemoryBlockBlocking reads the next block from the in-memory queue into dst and returns it. It blocks until a block is available or the stop deadline is exceeded, in which case it returns (dst, false).

func (*FastQueue) MustWriteBlockIgnoreDisabledPQ

func (fq *FastQueue) MustWriteBlockIgnoreDisabledPQ(block []byte)

MustWriteBlockIgnoreDisabledPQ unconditionally writes block to fq.

This method allows persisting in-memory blocks during graceful shutdown, even if persistence is disabled.

func (*FastQueue) TryWriteBlock

func (fq *FastQueue) TryWriteBlock(block []byte) bool

TryWriteBlock tries writing block to fq.

false is returned if the block couldn't be written to fq when the in-memory queue is full and the persistent queue is disabled.

func (*FastQueue) UnblockAllReaders

func (fq *FastQueue) UnblockAllReaders()

UnblockAllReaders unblocks all the readers.

type OpenFastQueueOpts

type OpenFastQueueOpts struct {
	// MaxInmemoryBlocks defines amount of blocks to hold in memory before falling back to file-based persistence.
	MaxInmemoryBlocks int
	// MaxPendingBytes limits file-based size of the queue.
	// If MaxPendingBytes is 0, then the queue size is unlimited.
	// The oldest data is dropped when the queue
	// reaches MaxPendingSize.
	MaxPendingBytes int64
	// IsPQDisabled defines whether file-based queue could be used.
	// If it is set to true, then write requests that exceed in-memory buffer capacity are rejected.
	// in-memory queue part can be stored on disk during graceful shutdown.
	IsPQDisabled bool
	// PrioritizeInMemoryData instructs FastQueue to write data into the in-memory queue
	// even if the file-based queue is not empty.
	// This is useful when data order doesn't matter and getting the most recent data
	// as fast as possible is more important.
	PrioritizeInmemoryData bool
}

OpenFastQueueOpts defines options for FastQueue