Skip to content

Rewrite agent::Buffer as a lock-free SPSC ring buffer - #44

Open
snarfable wants to merge 2 commits into
mathemaphysics:mainfrom
snarfable:spsc-ring-buffer
Open

snarfable wants to merge 2 commits into
mathemaphysics:mainfrom
snarfable:spsc-ring-buffer

Conversation

@snarfable

Copy link
Copy Markdown
Contributor

Replace the linear, memmove-compacting connection buffer with a
fixed-capacity single-producer/single-consumer ring buffer.

Buffer

  • Power-of-two capacity; indices are monotonically increasing counters
    masked on access, so full and empty states are unambiguous.
  • Lock-free: the producer publishes with a release store to head, the
    consumer frees space with a release store to tail; the opposite index
    is read with acquire. Each side caches the other's index and refreshes
    it only when the buffer looks full/empty, and head and tail live on
    separate cache lines to avoid false sharing.
  • New API: Write, Space, Peek, Read, ContiguousData (zero-copy view up
    to the wrap point), Consume, Drain, Available, Capacity, Empty, Full.
    Shift() is kept as an alias for Consume().
  • Non-copyable/non-movable (owns atomics).

Fixes in the old Buffer

  • Write() copied every chunk to offset 0, overwriting any unparsed bytes
    left from a partial AMQP frame.
  • Write() refused a write that would exactly fill the buffer.

IConnectionHandler

  • Feed AMQP::Connection::parse() directly from ContiguousData(); when
    the readable bytes wrap the ring, linearize them into a scratch buffer
    so frames are always contiguous.
  • Only read from the socket as many bytes as the ring can hold.
  • Check receiveBytes() for errors instead of passing a negative count to
    Write(); fix savail being unsigned so the < 0 error check can fire.
  • _sendDataFromBuffer() now consumes the bytes that were actually sent.

Tests and benchmark

  • tests/buffer_t.cpp: capacity rounding, append semantics, full/empty,
    peek/read/consume/drain, wrap-around, contiguous views, a randomized
    20,000-operation comparison against a reference model, and two
    producer/consumer thread stress tests (copying and zero-copy paths).
    Clean under ThreadSanitizer and AddressSanitizer/UBSan.
  • benchmarks/buffer_bench.cpp: SPSC throughput across chunk sizes,
    enabled with -DBUILD_BENCHMARKS=ON.

Replace the linear, memmove-compacting connection buffer with a
fixed-capacity single-producer/single-consumer ring buffer.

Buffer
- Power-of-two capacity; indices are monotonically increasing counters
  masked on access, so full and empty states are unambiguous.
- Lock-free: the producer publishes with a release store to head, the
  consumer frees space with a release store to tail; the opposite index
  is read with acquire. Each side caches the other's index and refreshes
  it only when the buffer looks full/empty, and head and tail live on
  separate cache lines to avoid false sharing.
- New API: Write, Space, Peek, Read, ContiguousData (zero-copy view up
  to the wrap point), Consume, Drain, Available, Capacity, Empty, Full.
  Shift() is kept as an alias for Consume().
- Non-copyable/non-movable (owns atomics).

Fixes in the old Buffer
- Write() copied every chunk to offset 0, overwriting any unparsed bytes
  left from a partial AMQP frame.
- Write() refused a write that would exactly fill the buffer.

IConnectionHandler
- Feed AMQP::Connection::parse() directly from ContiguousData(); when
  the readable bytes wrap the ring, linearize them into a scratch buffer
  so frames are always contiguous.
- Only read from the socket as many bytes as the ring can hold.
- Check receiveBytes() for errors instead of passing a negative count to
  Write(); fix savail being unsigned so the < 0 error check can fire.
- _sendDataFromBuffer() now consumes the bytes that were actually sent.

Tests and benchmark
- tests/buffer_t.cpp: capacity rounding, append semantics, full/empty,
  peek/read/consume/drain, wrap-around, contiguous views, a randomized
  20,000-operation comparison against a reference model, and two
  producer/consumer thread stress tests (copying and zero-copy paths).
  Clean under ThreadSanitizer and AddressSanitizer/UBSan.
- benchmarks/buffer_bench.cpp: SPSC throughput across chunk sizes,
  enabled with -DBUILD_BENCHMARKS=ON.

@snarfable snarfable left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good implementation of a ring buffer.

@snarfable snarfable closed this Sep 25, 2026
@snarfable snarfable reopened this Sep 25, 2026
onData() used to write straight to the socket, so _outbuffer was never
filled. Outgoing AMQP bytes are now queued in _outbuffer and sent by the
connection loop thread.

AMQP-CPP calls onData() from more than one thread: from the loop thread
for acks, heartbeats and replies produced inside parse(), and from user
threads via IAMQPWorker::AddMessage() -> publish(). To keep the ring's
single-producer contract:

- Producers are serialized with _outmutex; the consumer stays lock-free.
- Each call's bytes are queued contiguously, so frames from different
  threads never interleave.
- Non-loop threads queue a call only when it fits entirely, and never
  wait while holding the lock; otherwise they release it, yield, and
  retry (giving up if the worker is quitting). This prevents a deadlock
  with the loop thread's own onData().
- The loop thread, which is the ring's only consumer, makes room for
  itself when the ring is full by flushing to the socket.
- Calls larger than the ring's capacity are rejected with an error.

_sendDataFromBuffer() now drains the whole ring (two sends when the
pending bytes wrap), releases only what the socket accepted, and reports
send failures. operator()() records its thread id so onData() can tell
which thread it is on.

Adds a test that mirrors this pattern: four producer threads push
sequenced records through a mutex while one consumer checks that every
record arrives intact and in per-producer order. Clean under TSan and
ASan/UBSan.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant