Lock-free ring buffer (MPSC)
Lock-free multi-producer single-consumer (MPSC) ring buffer which supports contiguous range operations and which can be conveniently used for message passing. The implementation is written in C11 and distributed under the 2-clause BSD license.
int ringbuf_setup(ringbuf_t *rbuf, unsigned nworkers, size_t length)
rbuf
is a pointer
to the opaque ring buffer object; the caller is responsible to allocate
the space for this object. Typically, the object would be allocated
dynamically if using threads or reserved in a shared memory blocked if
using processes. The allocation size for the object shall be obtained
using the ringbuf_get_sizes
function. Returns 0 on success and -1
on failure.void ringbuf_get_sizes(unsigned nworkers, size_t *ringbuf_obj_size, size_t *ringbuf_worker_size)
ringbuf_t
and, optionally, ringbuf_worker_t
structures.
The size of the ringbuf_t
structure depends on the number of workers,
specified by the nworkers
parameter.ringbuf_worker_t *ringbuf_register(ringbuf_t *rbuf, unsigned i)
i
is a worker number, starting
from zero (i.e. shall be than nworkers
used in the setup). On success,
returns a pointer to an opaque ringbuf_worker_t
structured, which is
a part of the ringbuf_t
memory block. On failure, returns NULL
.void ringbuf_unregister(ringbuf_t *rbuf, ringbuf_worker_t *worker)
ssize_t ringbuf_acquire(ringbuf_t *rbuf, ringbuf_worker_t *worker, size_t len)
ringbuf_produce
function must be called to indicate that. Nested
acquire calls are not allowed.void ringbuf_produce(ringbuf_t *rbuf, ringbuf_worker_t *worker)
size_t ringbuf_consume(ringbuf_t *rbuf, size_t *offset)
ringbuf_release
function must be called to indicate that.void ringbuf_release(ringbuf_t *rbuf, size_t nbytes)
The consumer will return a contiguous block of ranges produced i.e. the
ringbuf_consume
call will not return partial ranges. If you think of
produced range as a message, then consumer will return a block of messages,
always ending at the message boundary. Such behaviour allows us to use
this ring buffer implementation as a message queue.
The implementation was extensively tested on a 24-core x86 machine, see the stress test for the details on the technique. It also provides an example how the mechanism can be used for message passing.
This ring buffer implementation always provides a contiguous range of
space for the producer. It is achieved by an early wrap-around if the
requested range cannot fit in the end. The implication of this is that
the ringbuf_acquire
call may fail if the requested range is greater
than half of the buffer size. Hence, it may be necessary to ensure that
the ring buffer size is at least twice as large as the maximum production
unit size.
It should also be noted that one of the trade-offs of such design is that the consumer currently performs an O(n) scan on the list of producers.
Producers:
if ((w = ringbuf_register(r, worker_id)) == NULL)
err(EXIT_FAILURE, "ringbuf_register")
...
if ((off = ringbuf_acquire(r, w, len)) != -1) {
memcpy(&buf[off], payload, len);
ringbuf_produce(r, tls);
}
Consumer:
if ((len = ringbuf_consume(r, &off)) != 0) {
process(&buf[off], len);
ringbuf_release(r, len);
}