parallel

Documentation

Overview

Package parallel contains implementations of various buffer types where the buffer can be consumed by any number of parallel consumer threads. Therefore, since it is possible for consumers to requeue a message if the propagation failed, it is possible for messages to be consumed out of sequence.

Index

• type AckFunc
• type Badger
  • func NewBadger(conf BadgerConfig) (*Badger, error)
  • func (b *Badger) Close()
  • func (b *Badger) CloseOnceEmpty()
  • func (b *Badger) NextMessage() (types.Message, AckFunc, error)
  • func (b *Badger) PushMessage(msg types.Message) (int, error)
• type BadgerConfig
  • func NewBadgerConfig() BadgerConfig
• type BadgerTuningParams
  • func NewBadgerTuningParams() BadgerTuningParams
• type Memory
  • func NewMemory(cap int) *Memory
  • func (m *Memory) Close()
  • func (m *Memory) CloseOnceEmpty()
  • func (m *Memory) NextMessage() (types.Message, AckFunc, error)
  • func (m *Memory) PushMessage(msg types.Message) (int, error)

Types

type AckFunc

type AckFunc func(ack bool) (int, error)

AckFunc is a func returned when a message is read from a parallel buffer. The func should be called when the message is finished with, and a flag indicates whether the message was successfully propagated and can be removed from the buffer. Returns the current backlog of the buffer in bytes, or an error if the message was not successfully removed.

If an error is returned it is safe to call the function again. Otherwise, it is not.

It is safe to call this func even if the buffer has closed.

type Badger

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

Badger is a parallel buffer implementation that allows multiple parallel consumers to read and purge messages from a Badger embedded key/value store, where messages are persisted to disk.

func NewBadger

func NewBadger(conf BadgerConfig) (*Badger, error)

NewBadger creates a Badger based parallel buffer.

func (*Badger) Close

func (b *Badger) Close()

Close closes the Buffer so that blocked readers or writers become unblocked.

func (*Badger) CloseOnceEmpty

func (b *Badger) CloseOnceEmpty()

CloseOnceEmpty closes the Buffer once the buffer has been emptied, this is a way for a writer to signal to a reader that it is finished writing messages, and therefore the reader can close once it is caught up. This call blocks until the close is completed.

func (*Badger) NextMessage

func (b *Badger) NextMessage() (types.Message, AckFunc, error)

NextMessage reads the next oldest message, the message is preserved until the returned AckFunc is called.

func (*Badger) PushMessage

func (b *Badger) PushMessage(msg types.Message) (int, error)

PushMessage adds a new message to the stack. Returns the backlog in bytes.

type BadgerConfig

type BadgerConfig struct {
	Directory    string             `json:"directory" yaml:"directory"`
	SizeCap      int64              `json:"max_bytes" yaml:"max_bytes"`
	SyncWrites   bool               `json:"sync_writes" yaml:"sync_writes"`
	GCIntervalMS int                `json:"gc_interval_ms" yaml:"gc_interval_ms"`
	Tuning       BadgerTuningParams `json:"tuning" yaml:"tuning"`
}

BadgerConfig contains configuration fields for a badger based buffer.

func NewBadgerConfig

func NewBadgerConfig() BadgerConfig

NewBadgerConfig creates a BadgerConfig with default values.

type BadgerTuningParams

type BadgerTuningParams struct {
	MaxTableSize            int64  `json:"max_table_size" yaml:"max_table_size"`
	LevelSizeMultiplier     int    `json:"level_size_multiplier" yaml:"level_size_multiplier"`
	MaxLevels               int    `json:"max_levels" yaml:"max_levels"`
	ValueThreshold          int    `json:"value_threshold" yaml:"value_threshold"`
	NumMemtables            int    `json:"num_memtables" yaml:"num_memtables"`
	NumLevelZeroTables      int    `json:"num_level_zero_tables" yaml:"num_level_zero_tables"`
	NumLevelZeroTablesStall int    `json:"num_level_zero_tables_stall" yaml:"num_level_zero_tables_stall"`
	LevelOneSize            int64  `json:"level_one_size" yaml:"level_one_size"`
	ValueLogFileSize        int64  `json:"value_log_file_size" yaml:"value_log_file_size"`
	ValueLogMaxEntries      uint32 `json:"value_log_max_entries" yaml:"value_log_max_entries"`
	NumCompactors           int    `json:"num_compactors" yaml:"num_compactors"`
}

BadgerTuningParams are more granular parameters for tuning the cache performance.

func NewBadgerTuningParams

func NewBadgerTuningParams() BadgerTuningParams

NewBadgerTuningParams returns a default set of tuning params.

type Memory

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

Memory is a parallel buffer implementation that allows multiple parallel consumers to read and purge messages from the buffer asynchronously.

func NewMemory

func NewMemory(cap int) *Memory

NewMemory creates a memory based parallel buffer.

func (*Memory) Close

func (m *Memory) Close()

Close closes the Buffer so that blocked readers or writers become unblocked.

func (*Memory) CloseOnceEmpty

func (m *Memory) CloseOnceEmpty()

CloseOnceEmpty closes the Buffer once the buffer has been emptied, this is a way for a writer to signal to a reader that it is finished writing messages, and therefore the reader can close once it is caught up. This call blocks until the close is completed.

func (*Memory) NextMessage

func (m *Memory) NextMessage() (types.Message, AckFunc, error)

NextMessage reads the next oldest message, the message is preserved until the returned AckFunc is called.

func (*Memory) PushMessage

func (m *Memory) PushMessage(msg types.Message) (int, error)

PushMessage adds a new message to the stack. Returns the backlog in bytes.