Repository files navigation

Lock-free multi-producer single-consumer (MPSC) ring buffer which supportscontiguous range operations and which can be conveniently used for messagepassing. The implementation is written in C11 and distributed under the2-clause BSD license.

• int ringbuf_setup(ringbuf_t *rbuf, unsigned nworkers, size_t length)

  • Setup a new ring buffer of a givenlength. Therbuf is a pointerto the opaque ring buffer object; the caller is responsible to allocatethe space for this object. Typically, the object would be allocateddynamically if using threads or reserved in a shared memory blocked ifusing processes. The allocation size for the object shall be obtainedusing theringbuf_get_sizes function. Returns 0 on success and -1on failure.

• void ringbuf_get_sizes(unsigned nworkers, size_t *ringbuf_obj_size, size_t *ringbuf_worker_size)

  • Returns the size of the opaqueringbuf_t and, optionally,ringbuf_worker_t structures.The size of theringbuf_t structure depends on the number of workers,specified by thenworkers parameter.

• ringbuf_worker_t *ringbuf_register(ringbuf_t *rbuf, unsigned i)

  • Register the current worker (thread or process) as a producer. Eachproducer MUST register itself. Thei is a worker number, startingfrom zero (i.e. shall be thannworkers used in the setup). On success,returns a pointer to an opaqueringbuf_worker_t structured, which isa part of theringbuf_t memory block. On failure, returnsNULL.

• void ringbuf_unregister(ringbuf_t *rbuf, ringbuf_worker_t *worker)

  • Unregister the specified worker from the list of producers.

• ssize_t ringbuf_acquire(ringbuf_t *rbuf, ringbuf_worker_t *worker, size_t len)

  • Request a space of a given length in the ring buffer. Returns theoffset at which the space is available or -1 on failure. Once the datais ready (typically, when writing to the ring buffer is complete), theringbuf_produce function must be called to indicate that. Nestedacquire calls are not allowed.

• void ringbuf_produce(ringbuf_t *rbuf, ringbuf_worker_t *worker)

  • Indicate that the acquired range in the buffer is produced and is readyto be consumed.

• size_t ringbuf_consume(ringbuf_t *rbuf, size_t *offset)

  • Get a contiguous range which is ready to be consumed. Returns zeroif there is no data available for consumption. Once the data isconsumed (typically, when reading from the ring buffer is complete),theringbuf_release function must be called to indicate that.

• void ringbuf_release(ringbuf_t *rbuf, size_t nbytes)

  • Indicate that the consumed range can now be released and may now bereused by the producers.

The consumer will return a contiguous block of ranges produced i.e. theringbuf_consume call will not return partial ranges. If you think ofproduced range as a message, then consumer will return a block of messages,always ending at the message boundary. Such behaviour allows us to usethis ring buffer implementation as a message queue.

The implementation was extensively tested on a 24-core x86 machine,seethe stress test for the details on the technique.It also provides an example how the mechanism can be used for messagepassing.

This ring buffer implementation always provides a contiguous range ofspace for the producer. It is achieved by an early wrap-around if therequested range cannot fit in the end. The implication of this is thattheringbuf_acquire call may fail if the requested range is greaterthan half of the buffer size. Hence, it may be necessary to ensure thatthe ring buffer size is at least twice as large as the maximum productionunit size.

It should also be noted that one of the trade-offs of such design is thatthe 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);}