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
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.
Creates a new entry.
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
@type entry() :: {:entry, key :: any(), value :: any(), version :: integer(), updates :: non_neg_integer()}
Entry record
Functions
@spec buffer_size(atom()) :: non_neg_integer()
Returns the partition's buffer size.
Returns a specification to start this module under a supervisor.
See Supervisor.
Deletes an entry by key from the given partition.
Gets the value for the given key from the partition's current ETS table.
Returns default if the key is not found.
Creates a new entry.
Puts a batch of pre-built entries into the given partition.
Each entry must be created with new_entry/3.
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.
@spec start_link(keyword()) :: GenServer.on_start()
Starts a buffer partition.
@spec update_options( GenServer.server(), keyword() ) :: :ok
Updates the options for the partition.