A ring buffer (circular buffer, circular queue, cyclic buffer) implemented in pure python. Includes flexible locking, waiting, and Disruptor variants.
A pure python implementation of a ring buffer with optional factory for alternate memory allocation. Variants included are Blocking (with a read cursor) and Locked (all manipulations are secured with a lock) ring buffers.
You may not call it a ring buffer, they also go by other names like circular buffer, circular queue or cyclic buffer.
python3 -m pip install pyring
Included are several variations of a ring buffer:
- RingBuffer - The most basic, non-blocking ring, supports optional RLocks as constructor argument.
- LockedRingBuffer - The same as
RingBufferbut secured by a lock (handy for multithread / multiproc)
- BlockingRingBuffer - Extension of
RingBufferwith a read method
next()which increments a read cursor and the writer cannot advance past the read cursor
- BlockingLockedRingBuffer - The same as
BlockingRingBufferbut secured by a lock (handy for multithread / multiproc)
- WaitingBlockingRingBuffer - The same as
BlockingRingBufferbut calls to next and put block and wait (with optional timeout arg)
- SingleProducerDisruptor - The same as
WaitingBlockingRingBufferbut allows multiple subscribers to a single ring buffer by calling
subscribe()which returns a
DisruptorSubscriberobject. No next method on disruptor, instead it is on the
Basic usage with default size and factory
import pyring # create ring buffer ring_buffer = pyring.RingBuffer() # add to ring ring_buffer.put("Something new!") # get latest from ring sequence, value = ring_buffer.get_latest() print(sequence, value) # 0 Something new!
Custom ring size
import pyring ring_buffer = pyring.RingBuffer(size=128) # size must be a power of 2
import pyring # custom factory that takes lists of ints and returns the sum # must declare the get and set methods class CustomSumFactory(pyring.RingFactory): value: typing.List[int] =  def get(self): return sum(self.value) def set(self, values): self.value = values ring_buffer = pyring.RingBuffer(factory=CustomSumFactory) ring_buffer.put([1,2,3,4,5]) sequence, value = ring_buffer.get_latest() print(sequence, value) # 0 15
Multiprocess C-Types Example
The below is a demonstration of using a RingBuffer across multiple processes, this requires firstly a LockedRingBuffer (with a multiproc lock), in addition to a custom cursor_position_value to increment sequence counts across processes.
This is more a demonstration on how flexible this ring buffer implementation can be rather than where you should use it, the below approach would most likely (with caveats on datasize) be better handled with threads reading off the ring buffer and passing messages via queues to worker processes.
import multiprocessing as mp import time from pyring import LockedRingBuffer, RingFactory # using r lock due to the reuse for the cursor_position_value mp_lock = mp.RLock() # note if using the same lock it must be recursive lock otherwise you will get deadlocks cursor_position_value = mp.Value("i", 0, lock=mp_lock) # using multiproc compatible c-types. class MultiprocFactory(RingFactory): def __init__(self): self.value = mp.Value("i") def set(self, v: int): self.value.value = v def get(self): return self.value.value ring_buffer = LockedRingBuffer( factory=MultiprocFactory, lock=mp_lock, cursor_position_value=cursor_position_value ) def worker_routine(worker_ring_buffer: LockedRingBuffer): for i in range(10): worker_ring_buffer.put(i) proc = mp.Process(target=worker_routine, args=(ring_buffer,)) proc.start() time.sleep(0.01) for i in range(10): sequence, value = ring_buffer.get(i) print(sequence, value) proc.join()
Single Producer Disruptor (Multiple Subscribers)
from pyring import SingleProducerDisruptor disruptor = SingleProducerDisruptor() subscriber_one = disruptor.subscribe() subscriber_two = disruptor.subscribe() for i in range(100): disruptor.put(i ** 2) sequence_one, res_one = subscriber_one.next() sequence_two, res_two = subscriber_two.next() # releases the subscribers barriers and allows disruptor to continue subscriber_one.unregister() subscriber_two.unregister()
Examples of Usage
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
Release history Release notifications | RSS feed
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.