Kepler.RingBuffer (Kepler v0.1.0)

Copy Markdown View Source

A fixed-size ring of recent terms, written in O(1) and read only when a watch fires.

An event that says "the queue passed 10,000" is a number going up. An event that also carries the last 25 things that went through the queue is actionable. That is what this is for.

Writes take an atomic sequence number and overwrite one ETS slot — no scanning, no trimming, no growth. Reads happen once, at fire time, and are allowed to be comparatively expensive.

Two writers racing for the same slot is not an error: the later write wins and the ring stays exactly size entries deep. Sequence numbers are stored alongside each entry so reads can order and discard stale slots.

iex> ring = Kepler.RingBuffer.new(3)
iex> Enum.each(1..5, &Kepler.RingBuffer.push(ring, &1))
iex> Kepler.RingBuffer.to_list(ring)
[3, 4, 5]

Summary

Types

t()

A ring buffer handle. Safe to copy into other processes.

Functions

Releases the backing table. Safe to call more than once.

Allocates a ring holding the last size terms.

Appends term, overwriting the oldest entry once the ring is full.

The ring's capacity. Constant for the ring's lifetime.

The count most recent entries, oldest first.

Every entry currently held, oldest first.

Types

t()

@type t() :: %Kepler.RingBuffer{
  index: :atomics.atomics_ref(),
  size: pos_integer(),
  table: :ets.table()
}

A ring buffer handle. Safe to copy into other processes.

Functions

destroy(ring_buffer)

@spec destroy(t()) :: :ok

Releases the backing table. Safe to call more than once.

new(size)

@spec new(pos_integer()) :: t()

Allocates a ring holding the last size terms.

The backing ETS table is owned by the calling process and dies with it, so create rings from a process that lives as long as the watches that use them.

push(ring_buffer, term)

@spec push(t(), term()) :: :ok

Appends term, overwriting the oldest entry once the ring is full.

This is the only part of a ring buffer that runs on a hot path, so it is one atomic add and one ETS insert. Opting a watch into context capture moves it up a cost tier — see Kepler.

size(ring_buffer)

@spec size(t()) :: pos_integer()

The ring's capacity. Constant for the ring's lifetime.

take(ring_buffer, count)

@spec take(t(), non_neg_integer()) :: [term()]

The count most recent entries, oldest first.

to_list(ring)

@spec to_list(t()) :: [term()]

Every entry currently held, oldest first.