Repository navigation
Conversation
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
commented
Sep 25, 2026
snarfable
left a comment
Contributor
Author
There was a problem hiding this comment.
Good implementation of a ring buffer.
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Replace the linear, memmove-compacting connection buffer with a
fixed-capacity single-producer/single-consumer ring buffer.
Buffer
masked on access, so full and empty states are unambiguous.
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.
to the wrap point), Consume, Drain, Available, Capacity, Empty, Full.
Shift() is kept as an alias for Consume().
Fixes in the old Buffer
left from a partial AMQP frame.
IConnectionHandler
the readable bytes wrap the ring, linearize them into a scratch buffer
so frames are always contiguous.
Write(); fix savail being unsigned so the < 0 error check can fire.
Tests and benchmark
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.
enabled with -DBUILD_BENCHMARKS=ON.