How do you implement a circular buffer in C?

Viewed 242251

I have a need for a fixed-size (selectable at run-time when creating it, not compile-time) circular buffer which can hold objects of any type and it needs to be very high performance. I don't think there will be resource contention issues since, although it's in a multi-tasking embedded environment, it's a co-operative one so the tasks themselves can manage that.

My initial thought was to store a simple struct in the buffer which would contain the type (simple enum/define) and a void pointer to the payload but I want this to be as fast as possible so I'm open to suggestions that involve bypassing the heap.

Actually I'm happy to bypass any of the standard library for raw speed - from what I've seen of the code, it's not heavily optimized for the CPU : it looks like they just compiled C code for things like strcpy() and such, there's no hand-coded assembly.

Any code or ideas would be greatly appreciated. The operations required are:

  • create a buffer with specific size.
  • put at the tail.
  • get from the head.
  • return the count.
  • delete a buffer.
9 Answers

The simplest solution would be to keep track of the item size and the number of items, and then create a buffer of the appropriate number of bytes:

typedef struct circular_buffer
{
    void *buffer;     // data buffer
    void *buffer_end; // end of data buffer
    size_t capacity;  // maximum number of items in the buffer
    size_t count;     // number of items in the buffer
    size_t sz;        // size of each item in the buffer
    void *head;       // pointer to head
    void *tail;       // pointer to tail
} circular_buffer;

void cb_init(circular_buffer *cb, size_t capacity, size_t sz)
{
    cb->buffer = malloc(capacity * sz);
    if(cb->buffer == NULL)
        // handle error
    cb->buffer_end = (char *)cb->buffer + capacity * sz;
    cb->capacity = capacity;
    cb->count = 0;
    cb->sz = sz;
    cb->head = cb->buffer;
    cb->tail = cb->buffer;
}

void cb_free(circular_buffer *cb)
{
    free(cb->buffer);
    // clear out other fields too, just to be safe
}

void cb_push_back(circular_buffer *cb, const void *item)
{
    if(cb->count == cb->capacity){
        // handle error
    }
    memcpy(cb->head, item, cb->sz);
    cb->head = (char*)cb->head + cb->sz;
    if(cb->head == cb->buffer_end)
        cb->head = cb->buffer;
    cb->count++;
}

void cb_pop_front(circular_buffer *cb, void *item)
{
    if(cb->count == 0){
        // handle error
    }
    memcpy(item, cb->tail, cb->sz);
    cb->tail = (char*)cb->tail + cb->sz;
    if(cb->tail == cb->buffer_end)
        cb->tail = cb->buffer;
    cb->count--;
}

Can you enumerate the types needed at the time you code up the buffer, or do you need to be able to add types at run time via dynamic calls? If the former, then I would create the buffer as a heap-allocated array of n structs, where each struct consists of two elements: an enum tag identifying the data type, and a union of all the data types. What you lose in terms of extra storage for small elements, you make up in terms of not having to deal with allocation/deallocation and the resulting memory fragmentation. Then you just need to keep track of the start and end indices that define the head and tail elements of the buffer, and make sure to compute mod n when incrementing/decrementing the indices.

A simple implementation could consist of:

  • A buffer, implemented as an array of size n, of whatever type you need
  • A read pointer or index (whichever is more efficient for your processor)
  • A write pointer or index
  • A counter indicating how much data is in the buffer (derivable from the read and write pointers, but faster to track it separately)

Every time you write data, you advance the write pointer and increment the counter. When you read data, you increase the read pointer and decrement the counter. If either pointer reaches n, set it to zero.

You can't write if counter = n. You can't read if counter = 0.

@Adam Rosenfield's solution, although correct, could be implemented with a more lightweight circular_buffer structure that does not involve count and capacity.

The structure could only hold the following 4 pointers:

  • buffer: Points to the start of the buffer in memory.
  • buffer_end: Points to the end of the buffer in memory.
  • head: Points to the end of stored data.
  • tail: Points to the start of stored data.

We could keep the sz attribute to allow the parametrisation of the unit of storage.

Both the count and the capacity values should be derive-able using the above pointers.

Capacity

capacity is straight forward, as it can be derived by dividing the distance between the buffer_end pointer and the buffer pointer by the unit of storage sz (snippet below is pseudocode):

capacity = (buffer_end - buffer) / sz

Count

For count though, things get a bit more complicated. For example, there is no way to determine whether the buffer is empty or full, in the scenario of head and tail pointing to the same location.

To tackle that, the buffer should allocate memory for an additional element. For example, if the desired capacity of our circular buffer is 10 * sz, then we need to allocate 11 * sz.

Capacity formula will then become (snippet below is pseudocode):

capacity_bytes = buffer_end - buffer - sz
capacity = capacity_bytes / sz

This extra element semantic allows us to construct conditions that evaluate whether the buffer is empty or full.

Empty state conditions

In order for the buffer to be empty, the head pointer points to the same location as the tail pointer:

head == tail

If the above evaluates to true, the buffer is empty.

Full state conditions

In order for the buffer to be full, the head pointer should be 1 element behind the tail pointer. Thus, the space needed to cover in order to jump from the head location to the tail location should be equal to 1 * sz.

if tail is larger than head:

tail - head == sz

If the above evaluates to true, the buffer is full.

if head is larger than tail:

  1. buffer_end - head returns the space to jump from the head to the end of the buffer.
  2. tail - buffer returns the space needed to jump from the start of the buffer to the `tail.
  3. Adding the above 2 should equal to the space needed to jump from the head to the tail
  4. The space derived in step 3, should not be more than 1 * sz
(buffer_end - head) + (tail - buffer) == sz
=> buffer_end - buffer - head + tail == sz
=> buffer_end - buffer - sz == head - tail
=> head - tail == buffer_end - buffer - sz
=> head - tail == capacity_bytes

If the above evaluates to true, the buffer is full.

In practice

Modifying @Adam Rosenfield's to use the above circular_buffer structure:

#include <string.h>

#define CB_SUCCESS 0        /* CB operation was successful */
#define CB_MEMORY_ERROR 1   /* Failed to allocate memory */
#define CB_OVERFLOW_ERROR 2 /* CB is full. Cannot push more items. */
#define CB_EMPTY_ERROR 3    /* CB is empty. Cannot pop more items. */

typedef struct circular_buffer {
  void *buffer;
  void *buffer_end;
  size_t sz;
  void *head;
  void *tail;
} circular_buffer;

int cb_init(circular_buffer *cb, size_t capacity, size_t sz) {
  const int incremented_capacity = capacity + 1; // Add extra element to evaluate count
  cb->buffer = malloc(incremented_capacity * sz);
  if (cb->buffer == NULL)
    return CB_MEMORY_ERROR;
  cb->buffer_end = (char *)cb->buffer + incremented_capacity * sz;
  cb->sz = sz;
  cb->head = cb->buffer;
  cb->tail = cb->buffer;
  return CB_SUCCESS;
}

int cb_free(circular_buffer *cb) {
  free(cb->buffer);
  return CB_SUCCESS;
}

const int _cb_length(circular_buffer *cb) {
  return (char *)cb->buffer_end - (char *)cb->buffer;
}

int cb_push_back(circular_buffer *cb, const void *item) {
  const int buffer_length = _cb_length(cb);
  const int capacity_length = buffer_length - cb->sz;

  if ((char *)cb->tail - (char *)cb->head == cb->sz ||
      (char *)cb->head - (char *)cb->tail == capacity_length)
    return CB_OVERFLOW_ERROR;

  memcpy(cb->head, item, cb->sz);

  cb->head = (char*)cb->head + cb->sz;
  if(cb->head == cb->buffer_end)
    cb->head = cb->buffer;

  return CB_SUCCESS;
}

int cb_pop_front(circular_buffer *cb, void *item) {
  if (cb->head == cb->tail)
    return CB_EMPTY_ERROR;

  memcpy(item, cb->tail, cb->sz);

  cb->tail = (char*)cb->tail + cb->sz;
  if(cb->tail == cb->buffer_end)
    cb->tail = cb->buffer;

  return CB_SUCCESS;
}

Extending adam-rosenfield's solution, i think the following will work for multithreaded single producer - single consumer scenario.

int cb_push_back(circular_buffer *cb, const void *item)
{
  void *new_head = (char *)cb->head + cb->sz;
  if (new_head == cb>buffer_end) {
      new_head = cb->buffer;
  }
  if (new_head == cb->tail) {
    return 1;
  }
  memcpy(cb->head, item, cb->sz);
  cb->head = new_head;
  return 0;
}

int cb_pop_front(circular_buffer *cb, void *item)
{
  void *new_tail = cb->tail + cb->sz;
  if (cb->head == cb->tail) {
    return 1;
  }
  memcpy(item, cb->tail, cb->sz);
  if (new_tail == cb->buffer_end) {
    new_tail = cb->buffer;
  }
  cb->tail = new_tail;
  return 0;
}
Related