PartitionedBuffer.Partition (PartitionedBuffer v0.4.3)

Copy Markdown View Source

Buffer partition.

The implementation is based on OpenTelemetry Batch Processor. The use case is very similar. The "OTel batch processor" buffers spans (large/massive amounts of them) and then exports them to an external source after some time (via OTLP). It is designed and implemented for efficiently handling large workloads. The partitioned buffer takes inspiration from the "OTel batch processor" to leverage all these properties.

Double-Buffering Processing Cycle

Phase 1 — Buffering          Phase 2 — Processing
===================          ====================

Client writes                 Timer fires :processing
     |                              |
     v                              v
+-----------+                 Swap pointer (A -> B)
| Table A   |<-- write        via persistent_term
| (current) |                       |
+-----------+                       v
| Table B   |                 +-----------+
| (idle)    |                 | Table B   |<-- write (new current)
+-----------+                 +-----------+
                              | Table A   |-- give_away --> Task
                              +-----------+       |
                                                  v
                                            :ets.select (batches)
                                                  |
                                            processor(batch)
                                                  |
                                            :ets.delete(Table A)
                                                  |
                                            :processing_completed

Phase 3 — Reset
========================
Recreate Table A (empty)
Ready for next swap

Summary

Types

Entry record

Functions

Returns the partition's buffer size.

Returns a specification to start this module under a supervisor.

Deletes an entry by key from the given partition.

Gets the value for the given key from the partition's current ETS table.

Puts a batch of pre-built entries into the given partition.

Puts a batch of versioned entries into the given partition.

Starts a buffer partition.

Updates the options for the partition.

Types

entry()

@type entry() ::
  {:entry, key :: any(), value :: any(), version :: integer(),
   updates :: non_neg_integer()}

Entry record

Functions

buffer_size(partition)

@spec buffer_size(atom()) :: non_neg_integer()

Returns the partition's buffer size.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

delete(partition, key)

@spec delete(atom(), any()) :: :ok

Deletes an entry by key from the given partition.

get(partition, key, default \\ nil)

@spec get(atom(), any(), any()) :: any()

Gets the value for the given key from the partition's current ETS table.

Returns default if the key is not found.

new_entry(key, value, version \\ 0)

@spec new_entry(any(), any(), integer()) :: entry()

Creates a new entry.

put(partition, entries)

@spec put(atom(), [entry()]) :: :ok

Puts a batch of pre-built entries into the given partition.

Each entry must be created with new_entry/3.

put_newer(partition, entries)

@spec put_newer(atom(), [entry()]) :: :ok

Puts a batch of versioned entries into the given partition.

Uses "newer version wins" semantics: an entry is only written if:

  • The key doesn't exist (insert), or
  • The new version is greater than the existing version (conditional update)

Each entry must be created with new_entry/3.

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

Starts a buffer partition.

update_options(server, opts)

@spec update_options(
  GenServer.server(),
  keyword()
) :: :ok

Updates the options for the partition.