elijahr / ringbuf

A lock-free ring buffer for Python and Cython

ringbuf

A lock-free, single-producer, single-consumer, ring buffer for Python and Cython.

Installation

OS X: brew install boost

Ubuntu: apt-get install libboost-all-dev

Windows: Install the latest version of Boost then set the BOOST_ROOT environment variable to point to its folder.

Then:

pip install ringbuf

Motivation

When working with realtime DSP in Python, we might be wrapping some external C/C++ library (for instance, PortAudio) which runs some user-provided callback function in realtime. The callback function shouldn't allocate/deallocate memory, shouldn't contain any critical sections (mutexes), and so forth, to prevent priority inversion. If the callback were to contain Python objects, we'd likely be allocating and deallocating, and at the very least, acquiring and releasing the GIL. So, the callback cannot interact with Python objects if we expect realtime performance. As such, there's a need for buffering data in a non-locking way between a C/C++ callback and Python.

Enter ringbuf, Cython wrappers for boost::lockfree::spsc_queue. Our Python code can read from and write to a ringbuf.RingBuffer object, and our C++ code can read from and write to that buffer's underlying spsc_queue, no GIL required.

Usage

Any Python object which supports the buffer protocol can be stored in ringbuf.RingBuffer. This includes, but is not limited to: bytes, bytearray, array.array, and numpy.ndarray.

NumPy

import numpy as np
from ringbuf import RingBuffer

buffer = RingBuffer(format='f', capacity=100)

data = np.linspace(-1, 1, num=100, dtype='f')

buffer.push(data)

popped = buffer.pop(100)

assert np.array_equal(data, popped)

bytes

from ringbuf import RingBuffer

buffer = RingBuffer(format='B', capacity=11)

buffer.push(b'hello world')

popped = buffer.pop(11)

assert bytes(popped) == b'hello world'

Interfacing with C/C++

mymodule.pxd:

# distutils: language = c++
cdef void callback(void* q)

mymodule.pyx:

# distutils: language = c++
from array import array

from ringbuf.boost cimport spsc_queue, void_ptr_to_spsc_queue_char_ptr
from ringbuf.ringbufcy cimport RingBuffer

from some_c_library cimport some_c_function


cdef void callback(void* q):
    cdef:
        # Cast the void* back to an spsc_queue.
        # The underlying queue always holds chars.
        spsc_queue[char] *queue = void_ptr_to_spsc_queue_char_ptr(q)
        double[5] to_push = [1.0, 2.0, 3.0, 4.0, 5.0]

    # Since the queue holds chars, you'll have to cast and adjust size accordingly.
    queue.push(<char*>to_push, sizeof(double) * 5)


def do_stuff():
    cdef:
        RingBuffer buffer = RingBuffer(format='d', capacity=100)
        void* queue = buffer.queue_void_ptr()

    # Pass our callback and a void pointer to the buffer's queue to some third party library.
    # Presumably, the C library schedules the callback and passes it the queue's void pointer.
    some_c_function(callback, queue)

    sleep(1)

    assert array.array('d', buffer.pop(5)) == array.array('d', range(1, 6))

Handling overflow & underflow

When RingBuffer.push() overflows, it returns the data that couldn't be pushed (or None, if all was pushed):

from ringbuf import RingBuffer

buffer = RingBuffer(format='B', capacity=10)
overflowed = buffer.push(b'spam eggs ham')
assert overflowed == b'ham'

When RingBuffer.pop() underflows, it returns whatever data could be popped:

from ringbuf import RingBuffer

buffer = RingBuffer(format='B', capacity=13)
buffer.push(b'spam eggs ham')
popped = buffer.pop(buffer.capacity * 100)
assert bytes(popped) == b'spam eggs ham'

For additional usage see the tests.

Supported platforms

GitHub Actions tests the following matrix:

• Linux:
  • CPython 3.7
  • CPython 3.8
  • CPython 3.9
  • CPython 3.10
• macOS:
  • CPython 3.10
• Windows:
  • CPython 3.10

Any platform with a C++11 compiler and boost installed should work.

Contributing

Pull requests are welcome, please file any issues you encounter.

The code is linted with lintball. There is a pre-commit hook to lint, configured by running:

npm install -g lintball
git config --local core.hooksPath .githooks

Changelog

v2.6.0 2022-09-27

• Move CI to GitHub Actions.
• Lint codebase with lintball
• Improve project structure

v2.5.0 2020-04-17

• Added experimental support for Windows.

v2.4.0 2020-03-23

• Added RingBuffer.reset() method to clear the buffer.

v2.3.0 2020-03-22

• Added concatenate function for joining multiple arbitrary Python objects that support the buffer protocol.

ringbuf's Issues

Peek and skip method

Hi, thank you for this library. This issue is a feature request.

I would like for the library to provide a peek method, which would behave like pop, but would not actually remove anything from the ring buffer (so, read_available and write_available would be left intact after a peek).

Motivating use case: I would like to read some data from the ring buffer to write it somewhere else, but that write might return short (meaning, not all data could be written). In this case, the "data that could not be written" should remain in the correct position in the ring buffer. However, had I used read and then push with the remaining data, then it would be in the wrong position!

# Initialize a sample ring buffer with 10 bytes
ring = RingBuffer(10)
ring.push(bytes(range(10))

# Read a smaller chunk of, say, 4 bytes (buffer is left with 6 bytes)
chunk = ring.pop(4)

# Try to write the chunk to some stream, which returns how much data could be written
n = stream.write(chunk)

# Error!
# If n returns short (say, 3), we're left with 4-3 = 1 bytes worth that should not have been removed from the buffer!
# This push will place them in the wrong position, so this is not a solution!
ring.push(chunk[n:])

With my proposed methods, this would not be an issue:

ring = RingBuffer(10)
ring.push(bytes(range(10))
chunk = ring.peek(4)  # Data is NOT removed from the buffer yet
n = stream.write(chunk)
ring.read(n)  # Only the actual amount written is removed from the buffer, success!

For the last read, it might be more efficient to provide a skip method. Although this is not necessary, it would save a few memory operations:

ring = RingBuffer(10)
ring.push(bytes(range(10))
chunk = ring.peek(4)
n = stream.write(chunk)
ring.skip(n)  # No data is returned, but we drop the right amount from the buffer