cascadeq

package module
v0.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 21, 2026 License: MIT Imports: 19 Imported by: 0

README

cascadeq

GoDoc Build Status codecov License

cascadeq (cascade-queue) is a persistent, disk-backed FIFO queue for []byte items. It runs entirely in memory under normal load and spills to disk only when the configured memory limits are reached, making it fast when the queue is shallow and able to grow to any depth, limited only by disk space. Data in the queue is persisted to disk when the queue is closed and is available for reading when the queue is opened again.

Features

  • Memory-first: items stay in memory until configurable byte or item-count limits are hit
  • Disk overflow: tailQ is flushed to sequentially numbered .dat files; files are reloaded into headQ as items are consumed
  • Persisted: in-memory state is persisted on Close; existing files are picked up automatically on restart
  • Optional compression: pass WithGzip(true) to compress overflow files
  • Idle snapshots: pass WithSnapshotInterval to periodically persist in-memory data when the queue is idle, to increase crash safety with minimal throughput cost.
  • Channel-based API: consume items with a plain <-q.Out() in a select
  • Batch I/O: PutBatch enqueues a slice of items in one round-trip; Drain dequeues up to N items in one round-trip

Installation

go get github.com/gammazero/cascadeq

Usage

package main

import (
    "fmt"
    "log"

    "github.com/gammazero/cascadeq"
)

func main() {
    q, err := cascadeq.New("/tmp/myqueue")
    if err != nil {
        log.Fatal(err)
    }
    defer q.Close()

    // Write items.
    for i := range 5 {
        if err := q.Put([]byte(fmt.Sprintf("item-%d", i))); err != nil {
            log.Fatal(err)
        }
    }

    // Read items until the queue is empty.
    for {
        select {
        case item := <-q.Out():
            fmt.Println(string(item))
        case <-q.Empty():
            return
        }
    }
}

Batch I/O — enqueue or dequeue multiple items in a single round-trip:

// Enqueue a batch.
items := [][]byte{[]byte("a"), []byte("b"), []byte("c")}
if err := q.PutBatch(items); err != nil {
    log.Fatal(err)
}

// Drain up to 64 items at once.
dst := make([][]byte, 64)
n := q.Drain(dst)
for _, item := range dst[:n] {
    fmt.Println(string(item))
}
Options
Option Default Description
WithMaxMemory(n) 1 MiB Maximum bytes held across both in-memory queues
WithMaxMemItems(n) 4096 Maximum item count held across both in-memory queues
WithMaxItemSize(n) 64 KiB Reject items larger than this
WithMinItemSize(n) 0 Reject items smaller than this
WithGzip(true) disabled Compress overflow files with gzip
WithSnapshotInterval(d) disabled Write in-memory state to disk after this much idle time
WithLogger(l) JSON→stderr Replace the default slog.Logger

Design

Dual-queue memory model

The queue maintains two in-memory deques at all times:

  • headQ — the read-from queue; holds the oldest unconsumed items
  • tailQ — the write-to queue; holds the newest items

The configured memory/item limits are divided equally between headQ and tailQ. Stats.MaxQBytes and Stats.MaxQLen reflect this per-queue half.

Write path

New items go to headQ when both queues are empty and headQ has space. Otherwise they go to tailQ. When tailQ is full:

  • If no overflow files exist and headQ has room, shift items from tailQ into headQ.
  • Otherwise, flush tailQ to the next numbered .dat file and clear tailQ.
Read path

Items are consumed from the front of headQ. When headQ empties:

  1. Load the next numbered file into headQ, then delete the file.
  2. If no files exist, swap tailQ ↔ headQ (O(1)).
  3. If neither, signal Empty() and stop sending on the output channel.
File format and naming

Files are stored in the directory passed to New and named cq-{hexnum}.dat (or .dat.gz when compression is enabled). Each file is a sequence of big-endian uint32 length-prefixed byte records. File number 0 is reserved for the headQ snapshot written on Close or on an idle snapshot tick; higher numbers are tailQ overflow files written in sequence. Corrupt files are renamed with a .bad extension rather than deleted. The gzip option can be toggled between runs without losing data: when loading, if a file is not found under the current setting, the opposite extension is tried.

Single-goroutine event loop

All state mutation happens inside one goroutine via a select over the input channel (Put and PutBatch), output channel (Out), drain, clear, and stats request channels, the empty signal channel, an optional snapshot ticker, and the close signal. A single sync.RWMutex only protects the closed flag, gating Put/PutBatch/Drain/Clear/Stats from racing with Close. No other synchronization is needed.

Snapshot feature

When WithSnapshotInterval is set, a ticker fires at half the configured interval. If no items have been written or read since the previous tick (idle), the in-memory queues are written to disk. On Close, a synchronous snapshot is always written for any non-empty queue.

Documentation

Overview

Package cascadeq implements a persistent, disk-backed FIFO queue for []byte items. It operates primarily in memory and spills to disk only when the configured memory limits are reached, making it efficient for workloads that fit in memory while remaining durable under backpressure.

Memory model

The queue maintains two in-memory deques at all times:

  • headQ: the read-from queue, holding the oldest unconsumed items
  • tailQ: the write-to queue, holding the newest items

The configured memory/item limits are divided equally between headQ and tailQ. When tailQ is full, its contents are flushed to a numbered .dat file on disk unless headQ has room to absorb them directly. Items are read back by loading the next numbered file into headQ; when no files remain, tailQ and headQ are swapped in O(1).

Usage

Create a Queue with New, write items with Put or PutBatch, and consume them via the channel returned by Out or by calling Drain: 

q, err := cascadeq.New("/var/data/queues/myqueue")
if err != nil {
    log.Fatal(err)
}
defer q.Close()

// Single-item producer
if err := q.Put([]byte("hello")); err != nil {
    log.Println(err)
}

// Batch producer — one round-trip for the whole slice
items := [][]byte{[]byte("a"), []byte("b"), []byte("c")}
if err := q.PutBatch(items); err != nil {
    log.Println(err)
}

// Channel consumer
for {
    select {
    case item := <-q.Out():
        process(item)
    case <-q.Empty():
        return // no more items right now
    case <-q.Done():
        return // queue closed
    }
}

// Batch consumer — drain up to N items in one round-trip
dst := make([][]byte, 64)
if n := q.Drain(dst); n > 0 {
    process(dst[:n])
}

Options

Behaviour is configured via functional options passed to New:

Persistence and file format

Overflow files are written to the directory supplied to New and named cq-{hexnum}.dat (or .dat.gz when compression is enabled). Each file is a sequence of big-endian int32 length-prefixed byte records. File number 0 is reserved for the headQ snapshot written on close or idle snapshot; higher numbers are sequential tailQ overflow files. On restart, New re-discovers these files and resumes from where the queue left off. Corrupt files are renamed with a .bad extension rather than deleted.

Concurrency

All state mutations run inside a single goroutine. Put, Clear, and Stats are safe to call from multiple goroutines concurrently. Close is idempotent and safe to call from any goroutine.

Example (Basic)

Example_basic demonstrates creating a queue, writing items, and reading them back until the queue is empty. 

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/gammazero/cascadeq"
)

func main() {
	dir, err := os.MkdirTemp("", "cascadeq-example1-*")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(dir)

	q, err := cascadeq.New(dir)
	if err != nil {
		log.Fatal(err)
	}
	defer q.Close()

	for i := range 3 {
		if err := q.Put(fmt.Appendf(nil, "item-%d", i)); err != nil {
			log.Fatal(err)
		}
	}

	for {
		select {
		case item := <-q.Out():
			fmt.Println(string(item))
		case <-q.Empty():
			return
		}
	}
}

Output:
item-0
item-1
item-2
Example (Batch)

Example_batch demonstrates PutBatch and Drain for high-throughput batch I/O. 

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/gammazero/cascadeq"
)

func main() {
	dir, err := os.MkdirTemp("", "cascadeq-example2-*")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(dir)

	q, err := cascadeq.New(dir)
	if err != nil {
		log.Fatal(err)
	}
	defer q.Close()

	items := [][]byte{[]byte("alpha"), []byte("beta"), []byte("gamma")}
	if err := q.PutBatch(items); err != nil {
		log.Fatal(err)
	}

	dst := make([][]byte, 10)
	n := q.Drain(dst)
	for _, item := range dst[:n] {
		fmt.Println(string(item))
	}
}

Output:
alpha
beta
gamma
Example (Options)

Example_options shows configuring memory limits and gzip compression. 

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/gammazero/cascadeq"
)

func main() {
	dir, err := os.MkdirTemp("", "cascadeq-example3-*")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(dir)

	q, err := cascadeq.New(dir,
		cascadeq.WithMaxMemory(64*1024),  // 64 KiB in-memory budget
		cascadeq.WithMaxMemItems(128),    // at most 128 items in memory
		cascadeq.WithMaxItemSize(4*1024), // reject items larger than 4 KiB
		cascadeq.WithGzip(true),          // compress overflow files
	)
	if err != nil {
		log.Fatal(err)
	}
	defer q.Close()

	if err := q.Put([]byte("hello")); err != nil {
		log.Fatal(err)
	}

	item := <-q.Out()
	fmt.Println(string(item))
}

Output:
hello

Index

Examples

Constants

View Source 
const (
	DefaultMaxMemory   = 1024 * 1024
	DefaultMaxMemItems = 4096
	DefaultMaxItemSize = 65536
)
View Source 
const (
	BadFileExt = ".bad"
)

Variables

View Source 
var ErrClosed = errors.New("closed")

ErrClosed is returned when I/O is attempted on a closed Queue.

View Source 
var ErrIsDirectory = errors.New("queue file is a directory")

ErrIsDirectory is returned when trying to read a queue file that is a directory.

Functions

func WithGzip 

func WithGzip(enable bool) func(*Queue)

WithGzip enables, if passed true, gzip compression of buffer files.

func WithLogger added in v0.0.2 

func WithLogger(logger *slog.Logger) func(*Queue)

WithLogger sets the slog.Logger instance to use for logging. This replaces the default cascadeq slog.Logger, which writes JSON to stderr. A nil logger is ignored, leaving the default in place.

func WithMaxItemSize 

func WithMaxItemSize(maxSize int) func(*Queue)

WithMaxItemSize specifies the maximum allowed size of a single []byte item in the queue.

func WithMaxMemItems 

func WithMaxMemItems(maxItems int) func(*Queue)

WithMaxMemItems sets the maximum number of items that the queue keeps in memory before items are written to disk. The value is rounded up to the next power of two, so the effective limit may be larger than requested. The effective limit is never less than 32.

func WithMaxMemory 

func WithMaxMemory(maxBytes int) func(*Queue)

WithMaxMemory sets the maximum amount of memory used by all items in the queue before items are written to disk.

func WithMinItemSize 

func WithMinItemSize(minSize int) func(*Queue)

WithMinItemSize specifies the minimum allowed size of a single []byte item in the queue.

func WithSnapshotInterval 

func WithSnapshotInterval(d time.Duration) func(*Queue)

WithSnapshotInterval enables snapshots and sets the amount of time that the queue must be idle before saving a snapshot of the items stored in memory. The queue must be idle for at least half of the specified time and at most the entire specified time. Idle snapshots are disabled by default and are enabled when a positive value is specified for this option.

Types

type Queue 

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

Queue implements a filesystem backed FIFO queue.

func New 

func New(dir string, options ...func(*Queue)) (*Queue, error)

New creates a new file-backed FIFO queue instance. Files for this queue are stored in the specified directory. A queue that stores different items must use a different directory.

func (*Queue) Clear 

func (q *Queue) Clear() error

Clear removes all items from the queue.

func (*Queue) Close 

func (q *Queue) Close() error

Close stops the queue's internal goroutine and prevents any more input or output with the queue. After calling Close, any attempted input or output results in an error.

func (*Queue) Dir 

func (q *Queue) Dir() string

Dir returns the directory where queued data files are stored.

func (*Queue) Done 

func (q *Queue) Done() <-chan struct{}

Done returns a channel that is closed when the Queue is closed.

func (*Queue) Drain added in v0.1.0 

func (q *Queue) Drain(dst [][]byte) int

Drain fills dst with up to len(dst) items currently available in the queue and returns the number placed. Returns 0 if the queue is empty or closed, or if dst is nil or zero-length.

func (*Queue) Empty 

func (q *Queue) Empty() <-chan struct{}

Empty returns a channel that is signaled when the queue is empty. This is useful for exiting a select when there are currently no more queued items to read.

func (*Queue) Name 

func (q *Queue) Name() string

Name returns the base of the directory path of this Queue, which serves as the queue's name.

func (*Queue) Out 

func (q *Queue) Out() <-chan []byte

Out returns the receive-only []byte channel for reading data.

func (*Queue) Put 

func (q *Queue) Put(item []byte) error

Put writes a []byte to the queue. A nil item is ignored and returns nil without being enqueued.

func (*Queue) PutBatch added in v0.1.0 

func (q *Queue) PutBatch(items [][]byte) error

PutBatch enqueues all items in a single event-loop visit. If items is nil or empty, it returns nil immediately. Nil items are skipped. Returns the first error encountered, whether from size validation or from writing overflow to disk; items before the failing one are already enqueued.

func (*Queue) Stats 

func (q *Queue) Stats() Stats

Stats retrieves information about queue internal data.

type Stats 

type Stats struct {
	MaxQBytes  int
	MaxQLen    int
	HeadQBytes int
	HeadQLen   int
	TailQBytes int
	TailQLen   int
	Files      []string
	Closed     bool
}

Stats holds information about the Queue internal state.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.