Ring Buffers¶

Lock-free SPSC (single-producer, single-consumer) ring buffers backed by a virtual-memory double-mapping, so the consumer always sees a contiguous window across the wrap boundary.

Every block of code on this page is a region of a script under src/doppler/examples/ that CI runs and that checks its own results, so what you read here is what was executed.

script	what it shows
`buffers_demo`	three widths, one shape; the wrap; two threads
`ring_lifecycle_demo`	owning a ring, and owning a view of one
`ring_chunking_demo`	the producer's block size is not the consumer's
`ring_nonblocking_demo`	one thread: `peek` / `write_some` / `space` / `reset`
`ring_iq16_demo`	16-bit I/Q as a record, without a copy
`ring_refusals_demo`	everything a ring says no to, and how it says it
`ring_interrupt_demo`	stopping a consumer blocked in `wait()`

The C face has its own, under native/examples/: ring_threaded_demo, ring_write_policy_demo, ring_drip_feed_demo, ring_chunking_demo, ring_element_view_demo and ring_backed_demo (a ring whose samples are a file, which the Python face does not expose).

Three widths, one shape¶

write() copies a block in; wait(n) lends the consumer a view of the next n samples; consume() gives them back. Every count is in samples and every view is 1-D, on every width — so this loop is written once:

import threading

import numpy as np

from doppler.buffer import F32Buffer, F64Buffer, I16Buffer

IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])


# A ramp in each width's own element, so a sample says where it came from.
def ramp_complex(dtype):
    return lambda n: np.arange(n).astype(dtype)


def ramp_iq16(n):
    x = np.zeros(n, dtype=IQ16)
    x["i"] = np.arange(n)
    return x


WIDTHS = [
    (F32Buffer, np.dtype(np.complex64), ramp_complex(np.complex64)),
    (F64Buffer, np.dtype(np.complex128), ramp_complex(np.complex128)),
    (I16Buffer, IQ16, ramp_iq16),
]

for cls, dtype, ramp in WIDTHS:
    with cls(1024) as buf:
        cap = buf.capacity
        assert cap == 1024, "exactly what was asked, on every machine"

        assert buf.write(ramp(100))
        assert (buf.available, buf.space) == (100, cap - 100)

        view = buf.wait(64)
        assert view.shape == (64,), "1-D, one element per sample"
        assert view.dtype == dtype
        assert np.array_equal(view, ramp(64))
        buf.consume()
        assert buf.available == 36

The double mapping is what makes a lent frame contiguous even when it straddles the physical end of the ring — no copy, and no special case in the consumer:

# The point of the double mapping: a frame that straddles the end of the
# ring's memory still comes back as ONE contiguous array. Any capacity will
# do -- 1000 here, not a power of two -- and a 256-sample frame taken at a
# hop of 100 lands across the end sooner or later, wherever the end is.
with F32Buffer(1000) as buf:
    assert buf.capacity == 1000
    for pos in range(40):  # frame k starts at stream position k
        buf.write_some(np.arange(pos, pos + 256, dtype=np.complex64))
        frame = buf.peek(256)
        assert frame.flags["C_CONTIGUOUS"]
        assert np.array_equal(frame, np.arange(pos, pos + 256))
        buf.consume()
        buf.write_some(np.zeros(100, dtype=np.complex64))  # move on by 100
        buf.peek(100)
        buf.consume()

Two threads, and an end¶

wait() releases the GIL while it spins, so a producer thread runs. write() never blocks — it refuses — so backpressure is the producer's: wait for space. And an empty ring cannot tell a slow producer from a finished one, so the producer says so with close(), which ends the consumer's wait() in EOFError once the ring is drained:

# Two threads. wait() releases the GIL while it spins, so the producer
# runs; close() is how the consumer learns there is no more.
buf = F64Buffer(4096)
got = []


def producer():
    for k in range(8):
        while buf.space < 512:  # write() never blocks: wait for room
            pass
        buf.write(np.full(512, k, dtype=np.complex128))
    buf.close()


t = threading.Thread(target=producer)
t.start()
try:
    while True:
        got.append(int(buf.wait(512)[0].real))
        buf.consume()
except EOFError:
    pass  # closed and drained: the normal end of a stream
t.join()

assert got == list(range(8)), "every block, once, in order"
assert buf.dropped == 0
buf.destroy()

Owning a ring, and a view of one¶

A ring is a mapping, so with is the spelling that cannot leak it; after it, every member raises rather than touching memory that is gone:

import numpy as np

from doppler.buffer import F32Buffer

with F32Buffer(capacity=1024) as buf:
    assert buf.capacity == 1024  # exactly what was asked for
    buf.write(np.arange(8, dtype=np.complex64))

# Out of the block the mapping is gone, and the object says so.
try:
    _ = buf.available
    raise AssertionError("a destroyed ring answered")
except RuntimeError:
    pass
buf.destroy()  # again, by hand: harmless

# Any size from 1 up is a ring of exactly that many samples; it need not be
# a power of two. A ring of nothing is refused up front, as a ValueError.
assert F32Buffer(1000).capacity == 1000
try:
    F32Buffer(0)
    raise AssertionError("a ring of nothing was mapped")
except ValueError:
    pass

A view is a loan: zero-copy, read-only, and it keeps the ring alive. consume() with no argument releases exactly what was lent, so the count is written once. Copy anything you want to keep before releasing it:

buf = F32Buffer(1024)
buf.write(np.arange(8, dtype=np.complex64))

view = buf.wait(4)
assert view.base is buf, "the view keeps the ring alive"
assert not view.flags.owndata, "zero-copy: these are the ring's bytes"
assert not view.flags.writeable, "a consumer reads what it was lent"

keep = view.copy()  # anything wanted after consume() is copied first
buf.consume()  # releases the 4 that were lent -- the count written once
assert buf.available == 4
assert np.array_equal(keep, [0, 1, 2, 3])

# Nothing is on loan now, so there is no count to default to.
try:
    buf.consume()
    raise AssertionError("consume() released a view nobody holds")
except RuntimeError:
    pass

Releasing less than was lent keeps the rest, which is how overlapped frames are read:

# A hop smaller than the frame: release 1, keep 3, and the next frame
# overlaps the last by three samples.
starts = []
while (frame := buf.peek(4)) is not None:
    starts.append(int(frame[0].real))
    buf.consume(1)
assert starts == [4], "only one whole frame of 4 was left"

buf.write(np.arange(8, 16, dtype=np.complex64))
while (frame := buf.peek(4)) is not None:
    starts.append(int(frame[0].real))
    buf.consume(1)
assert starts == [4, 5, 6, 7, 8, 9, 10, 11, 12]

The producer's block size is not the consumer's¶

This is what the ring is for, and it works in both directions. The double-mapping is what makes it free: wait(n) returns a contiguous view of n samples even when those n straddle the physical end of the ring, so a consumer with a fixed block size hands the view straight to the next stage with no copy and no special case for the wrap.

Large in, fixed out. A capture device gives you whatever its driver batched; an FFT of length N cannot take N-1. One thread, five lines: feed what fits, take every whole frame, repeat until the block is gone. write_some never refuses, so there is no room to make first — and no block too large: two of these are bigger than the whole ring. peek never blocks, so the same thread can do both jobs.

import threading

import numpy as np

from doppler.buffer import F32Buffer
from doppler.spectral import FFT


# A ramp, so every sample says where in the stream it came from: a frame
# that is off by one, duplicated or dropped shows up as a break in the
# sequence rather than as a plausible-looking block of signal.
def stream(start: int, n: int) -> np.ndarray:
    idx = np.arange(start, start + n)
    return (idx + 1j * idx).astype(np.complex64)


FFT_N = 1024
ring = F32Buffer(4096)  # four frames. SMALLER than the largest block below.

# What the device hands over: nothing is a multiple of FFT_N, and two of
# them are larger than the whole ring.
DRIVER_BLOCKS = [3000, 5000, 1700, 4096, 777, 2048, 6000, 1234]

fft = FFT(FFT_N, -1)
spectra: list[np.ndarray] = []
frames: list[np.ndarray] = []
produced = 0

for _ in range(6):
    for n in DRIVER_BLOCKS:
        block = stream(produced, n)
        produced += n

        # THE PATTERN. Feed what fits, take every whole frame, repeat until
        # the block is gone. `write_some` never refuses, so there is no
        # room to make first and no block too large; `peek` never blocks,
        # so one thread can do both jobs.
        fed = 0
        while fed < len(block):
            fed += ring.write_some(block[fed:])
            while (frame := ring.peek(FFT_N)) is not None:
                spectra.append(fft.execute_cf32(frame))  # straight in: no copy
                frames.append(frame.copy())  # kept past consume(): copied
                ring.consume()  # releases the frame that was lent

The ring need only hold one frame, not one block.

Small in, drain when full. A chatty producer costs one wake-up per write unless something batches for it. The consumer asks for a whole batch and wait() is what tells it one is ready — it never polls available. When the producer closes the ring, EOFError is what tells it the stream ended, and that is the one moment to read available: the ring is closed, so the count can no longer grow, and whatever is there is the tail.

BATCH = 2048
drip = F32Buffer(4096)
assert drip.capacity >= BATCH, "a larger ask raises: compare to capacity"
DRIP_BLOCKS = [16, 48, 32, 9, 64, 24, 100, 7]
TOTAL = 60_000

batches: list[int] = []
occupancy: list[int] = []


def producer() -> None:
    """Write small irregular blocks until TOTAL, then say so."""
    sent = 0
    i = 0
    while sent < TOTAL:
        n = min(DRIP_BLOCKS[i % len(DRIP_BLOCKS)], TOTAL - sent)
        # Wait for ROOM, rather than retrying a write that will be refused.
        # `space` read from the producer side is a LOWER bound -- the
        # consumer can only free more -- so a write sized by it always
        # fits. Retrying instead would work, but see the note below about
        # what it does to `dropped`.
        while drip.space < n:
            pass
        assert drip.write(stream(sent, n)), "waited for room, so it fits"
        sent += n
        i += 1
    drip.close()  # the consumer's only way to tell "slow" from "finished"


t = threading.Thread(target=producer)
t.start()

received = 0
while True:
    try:
        # One wake-up per BATCH samples instead of one per 7..100-sample
        # write -- which is the entire point of buffering a chatty producer.
        view = drip.wait(BATCH)
    except EOFError:
        break  # closed AND drained: whatever is left is under a batch
    occupancy.append(drip.available)
    assert len(view) == BATCH, "a drain is the whole batch"
    assert np.array_equal(np.asarray(view), stream(received, BATCH)), (
        "the batch is not the next BATCH samples of the stream"
    )
    received += BATCH
    batches.append(BATCH)
    drip.consume(BATCH)

t.join()

# The tail under one batch is still in the ring: close() does not discard it,
# and a consumer that only ever asks for BATCH will never see it.
#
# EOFError is what says WHEN to look. The consumer never polls `available`:
# wait() is the notification while the stream runs, and EOFError is the
# notification that it ended. After it the ring is closed, so the count
# can no longer grow -- this one read is exact, and it is the only one.
tail = drip.available
if tail:
    view = drip.wait(tail)
    assert np.array_equal(np.asarray(view), stream(received, tail))
    drip.consume(tail)
    received += tail

Three things that bite, all of them measurable:

write is all-or-nothing and never blocks. A block that does not fit is rejected whole. For a stream that is the wrong call — use write_some, as above. write is for a frame that is meaningless in part, and there the producer waits for space first; backpressure is the caller's.
dropped counts the length of every rejected call, not samples lost. A producer that spins on write until it succeeds inflates it without losing anything: a 60,000-sample run written that way reported 5,960,438 "dropped". Wait for room if you want the counter to mean what it says.
Size a block from capacity, not from the producer's chunk. wait(n) with n > capacity can never be satisfied — the ring cannot hold that many — so it raises ValueError naming both numbers rather than waiting for something that cannot arrive. capacity is exactly what you passed the constructor, and it can be any size: pick the one your largest frame needs.

The runnable version of both directions, with the assertions that keep it honest, is src/doppler/examples/ring_chunking_demo.py.

One thread: `peek` and `write_some`¶

wait(n) spins until a producer on another thread delivers, so a caller that is its own producer — a read loop, a callback, a notebook cell — would wait forever. peek(n) is the same zero-copy view without the wait: the frame if it is there, None if it is not yet.

Drip-feed until a frame is there. Arrivals smaller than a frame; ask after each one:

import numpy as np

from doppler.buffer import F32Buffer


def stream(start: int, n: int) -> np.ndarray:
    idx = np.arange(start, start + n)
    return (idx + 1j * idx).astype(np.complex64)


ring = F32Buffer(4096)
CAP = ring.capacity

PIECE, FRAME, ARRIVALS = 100, 1024, 25
sent = frames = 0
for _ in range(ARRIVALS):
    assert ring.write_some(stream(sent, PIECE)) == PIECE
    sent += PIECE
    frame = ring.peek(FRAME)  # None until FRAME samples have accumulated
    if frame is not None:
        assert np.array_equal(frame, stream(frames * FRAME, FRAME))
        ring.consume(FRAME)
        frames += 1

A chunk larger than the ring, read as overlapped frames. write is all-or-nothing, so it can never accept this chunk. write_some takes what fits and says how much; alternate it with the drain. consume(HOP) with HOP < NFFT releases a hop and keeps the overlap:

ring.reset()
NFFT, HOP = 1000, 250
assert CAP % NFFT, "frames must straddle the wrap for this to show anything"
chunk = stream(0, 5 * CAP)
assert not ring.write(chunk[: CAP + 1]), "write() refuses what cannot fit"
refused = ring.dropped

fed = taken = 0
while fed < len(chunk):
    k = ring.write_some(chunk[fed:])
    fed += k
    drained = 0
    while (frame := ring.peek(NFFT)) is not None:
        assert frame.flags["C_CONTIGUOUS"], "the double-mapping's promise"
        assert np.array_equal(frame, chunk[taken * HOP :][:NFFT]), (
            f"frame {taken} is not the stream at hop {taken * HOP}"
        )
        ring.consume(HOP)  # release a hop, keep the overlap
        taken += 1
        drained += 1
    # A full ring holds a whole frame, so one side always moves. Without
    # this, a regression in either call is a hang rather than a failure.
    assert k or drained, "neither side made progress"

End of stream, then reuse. None only ever means not yet. Once the ring is closed a frame that cannot arrive raises EOFError; what is there can still be read, and reset() reopens the same mapping for a second stream:

ring.close()
tail = ring.available
assert 0 < tail < NFFT
try:
    ring.peek(NFFT)
    raise AssertionError("a closed ring answered 'not yet'")
except EOFError:
    pass  # the rest of that frame is never coming: not None, an error
last = ring.peek(tail)  # what IS there can still be read
assert np.array_equal(last, chunk[-tail:])
ring.consume()

ring.reset()
assert not ring.closed and ring.available == 0 and ring.space == CAP
assert ring.write_some(stream(0, 8)) == 8
assert np.array_equal(ring.peek(8), stream(0, 8))

space is the room a write is guaranteed to find — use it instead of deriving capacity - available. From Python this pair is also the cheaper one per frame: wait releases and retakes the GIL, peek has no reason to. Time it with the benchmarks in src/doppler/buffer/benchmarks/.

16-bit I/Q: a record, not a pair of columns¶

numpy has no complex-integer dtype, so I16Buffer speaks a record — one element per sample, like its siblings, and exactly the four bytes the hardware delivered:

import numpy as np

from doppler.buffer import I16Buffer
from doppler.cvt import I16ToF32

IQ16 = np.dtype([("i", "<i2"), ("q", "<i2")])

# What the hardware gives you: bytes. Eight samples, I = k, Q = -k.
adc = np.empty(16, dtype=np.int16)
adc[0::2] = np.arange(8)
adc[1::2] = -np.arange(8)
capture = adc.tobytes()

samples = np.frombuffer(capture, dtype=IQ16)  # zero-copy: 8 records
assert samples.shape == (8,)

buf = I16Buffer(1024)
assert buf.write(samples)
assert buf.available == 8, "counted in SAMPLES, like every other width"

Fields, the interleaved form and the record are three views of the same bytes; none of these lines copies:

view = buf.wait(8)
assert view.dtype == IQ16 and view.shape == (8,)

i, q = view["i"], view["q"]  # strided int16 views of the ring -- no copy
assert not i.flags.owndata and not q.flags.owndata
assert i.tolist() == list(range(8))
assert q.tolist() == [-k for k in range(8)]

flat = view.view(np.int16)  # back to I, Q, I, Q, ... -- also no copy
assert np.array_equal(flat, adc)

To floating point with doppler's own converter:

# To floating point with doppler's own converter, which takes the
# interleaved form: full scale becomes +-1.0.
to_float = I16ToF32()  # 1/32768
f = to_float.steps(flat)
z = f[0::2] + 1j * f[1::2]
assert f.dtype == np.float32 and z.dtype == np.complex64
assert np.allclose(z, (np.arange(8) - 1j * np.arange(8)) / 32768.0)
buf.consume()

A record rather than two int16 packed into an int32, because arithmetic on the packed form carries across the I/Q boundary and corrupts I silently. A record refuses instead — and so does the ring, given a bare int16 array:

buf.write(samples)
view = buf.wait(8)

# Arithmetic on a record is refused, where a packed int32 would corrupt I.
try:
    _ = view + 1
    raise AssertionError("arithmetic on a record went through")
except TypeError:
    pass

# And a bare int16 array is not an array of samples -- flat or (n, 2).
for not_samples in (adc, adc.reshape(-1, 2)):
    try:
        buf.write(not_samples)
        raise AssertionError("a bare int16 array was accepted")
    except TypeError:
        pass
assert buf.available == 8, "a refused write takes nothing"

What a ring says no to¶

Each refusal is a distinct exception, raised before anything is copied. A ring exists to avoid copies, so an input that would need casting, flattening or compacting is refused rather than quietly fixed:

import threading

import numpy as np

from doppler.buffer import F32Buffer


def refusal(exc, call):
    """The message `call` raises, which must be an `exc`."""
    try:
        call()
    except exc as e:
        return str(e)
    raise AssertionError(f"expected {exc.__name__}")


buf = F32Buffer(1024)
cap = buf.capacity
# A ring exists to avoid copies, so it will not quietly make one: an input
# that would need casting, flattening or compacting is refused instead.
msg = refusal(TypeError, lambda: buf.write(np.zeros(8, dtype=np.float32)))
assert "float32" in msg, "the message says what it was given"

grid = np.zeros((8, 2), dtype=np.complex64)
refusal(ValueError, lambda: buf.write(grid))  # 2-D
refusal(ValueError, lambda: buf.write(grid[:, 0]))  # strided
assert buf.available == 0 and buf.dropped == 0, "nothing was taken"

Full is not an error. write() says False, write_some() says how much it took. dropped adds up refused samples — the caller still holds them, so it is a loss count only if they are then thrown away:

# Full is not an error. write() is all-or-nothing and says False;
# write_some() takes what fits and says how much.
assert buf.write(np.zeros(cap, dtype=np.complex64)) is True
assert buf.write(np.zeros(1, dtype=np.complex64)) is False
assert buf.dropped == 1, "counts what was REFUSED -- the caller still has it"
assert buf.write_some(np.zeros(8, dtype=np.complex64)) == 0
assert buf.dropped == 1, "write_some never refuses, so never counts"
buf.reset()

So is releasing more than is there — refused, and nothing released, because the two positions are all a ring knows about itself:

# Releasing more than is there is refused too, and releases NOTHING. The
# ring's two positions are all it knows about itself: let the read position
# pass the write position and every later count would describe a ring that
# does not exist.
buf.write(np.zeros(10, dtype=np.complex64))
msg = refusal(ValueError, lambda: buf.consume(11))
assert buf.available == 10 and buf.space == cap - 10, "still a ring"
buf.consume(10)

A request nothing could ever satisfy is a caller bug, and says so with both numbers instead of waiting forever. "Not yet" and "never" are different answers — None and EOFError:

# A request no producer could ever satisfy is a caller bug, and is said
# plainly -- with both numbers -- rather than waited on forever.
msg = refusal(ValueError, lambda: buf.wait(cap + 1))
assert str(cap + 1) in msg and str(cap) in msg
refusal(ValueError, lambda: buf.peek(cap + 1))
# "Not yet" and "never" are different answers. peek() says None for the
# first and raises EOFError for the second, so a poll loop cannot mistake
# a finished stream for a slow one.
buf.write(np.zeros(100, dtype=np.complex64))
assert buf.peek(512) is None  # not yet
buf.close()
refusal(EOFError, lambda: buf.peek(512))  # never
refusal(EOFError, lambda: buf.wait(512))  # and wait() does not spin
assert len(buf.peek(100)) == 100, "what IS there is still readable"
buf.consume()

Which is why a producer closes the ring in finally. If it raises, the consumer's wait() ends instead of spinning:

# The pattern that keeps a consumer from waiting forever: the producer
# closes the ring in `finally`. If it raises -- here, by writing the wrong
# type -- the consumer's wait() ends in EOFError instead of never.
buf.reset()
failed = []


def producer():
    try:
        buf.write(np.zeros(256, dtype=np.float32))  # wrong type: raises
    except TypeError as e:
        failed.append(e)
    finally:
        buf.close()


t = threading.Thread(target=producer)
t.start()
refusal(EOFError, lambda: buf.wait(256))
t.join()
assert len(failed) == 1

Stopping a blocked `wait()`¶

wait() spins in C with the GIL released, so nothing in that loop is running Python and an ordinary flag cannot stop it. doppler.interrupt is the stop it listens to — process-wide, so a guard made anywhere reaches a wait blocked anywhere — and Interrupt([signal.SIGINT]) is how Ctrl-C is wired to it:

import threading
import time

import numpy as np

from doppler.buffer import F32Buffer
from doppler.interrupt import Interrupt

buf = F32Buffer(1024)
# `[]` installs no signal handlers: this guard is only a handle to the
# flag. `Interrupt([signal.SIGINT])` is the same with Ctrl-C attached.
with Interrupt([]) as stop:
    # Nothing will ever write 512 samples, so this wait would never return.
    # Another thread asks for a stop a fifth of a second in.
    threading.Timer(0.2, stop.interrupt).start()

    t0 = time.monotonic()
    try:
        buf.wait(512)
        raise AssertionError("wait() returned with nothing written")
    except KeyboardInterrupt:
        waited = time.monotonic() - t0
    assert 0.1 < waited < 5.0, f"stopped after {waited:.2f} s"
    assert stop.interrupted()
    # The flag stays set until it is cleared, so a second wait would stop
    # at once. resume() clears it, and the ring is as it was.
    stop.resume()
    assert not stop.interrupted()
    buf.write(np.arange(4, dtype=np.complex64))
    assert len(buf.wait(4)) == 4
    buf.consume()

A consumer loop usually wants both endings:

# The consumer loop with both endings. close() ends it here; a stop would
# end it the same way, through the other except.
frames = 0
stopped = False


def producer():
    for _ in range(5):
        while buf.space < 256:
            pass
        buf.write(np.zeros(256, dtype=np.complex64))
    buf.close()


with Interrupt([]):
    t = threading.Thread(target=producer)
    t.start()
    try:
        while True:
            buf.wait(256)
            frames += 1
            buf.consume()
    except EOFError:
        pass  # the producer finished
    except KeyboardInterrupt:
        stopped = True  # somebody asked the process to stop
    t.join()
assert frames == 5 and not stopped, "ended by close(), not by a stop"

Buffer types¶

Type	Import	NumPy dtype	Notes
`F32Buffer`	`doppler.buffer`	`complex64`	CF32 IQ pairs
`F64Buffer`	`doppler.buffer`	`complex128`	CF64 IQ pairs
`I16Buffer`	`doppler.buffer`	`(i, q)` int16 record	`view["i"]`, `view["q"]`

Any capacity from 1 up, a power of two or not, and capacity is exactly what you asked for on every machine. The mapping behind it is rounded up to a power of two and to whole pages — address space, never room.