# Lock free MPMC Ring buffer implementation in C

I have written a lock free MPMC FIFO in C based on a ring buffer. It uses gcc's atomic built-ins to achieve thread safety. The queue is designed to return -1 if it's full on enqueue or empty on dequeue.

After some feedback, I've changed by design. I have tested this code to work, but testing multithreaded code is hardly enough to prove it's correct.

struct queue
{
void** buf;
size_t num;
uint64_t writePos;
};

queue_t* createQueue(size_t num)
{
struct queue* new = xmalloc(sizeof(*new));
new->buf = xmalloc(sizeof(void*) * num);
memset(new->buf, 0, sizeof(void*)*num);
new->writePos = 0;
new->num = num;
return new;
}

void destroyQueue(queue_t* queue)
{
if(queue)
{
xfree(queue->buf);
xfree(queue);
}
}

int enqueue(queue_t* queue, void* item)
{
for(int i = 0; i < kNumTries; i++)
{
if(__sync_bool_compare_and_swap(&queue->buf[queue->writePos % queue->num], NULL, item))
{
return 0;
}
}
return -1;
}

void* dequeue(queue_t* queue)
{
for(int i = 0; i < kNumTries; i++)
{
void* value = queue->buf[queue->readPos % queue->num];
if(value && __sync_bool_compare_and_swap(&queue->buf[queue->readPos % queue->num], value, NULL))
{
return value;
}
}
return NULL;
}


-
Also check out Relacy for thread-safeness-testing. No guarantees on its accuracy -- 1024cores.net/home/relacy-race-detector –  Ross Aiken Feb 21 '14 at 18:54

Did you test it?

I think you have a problem if a thread calls enqueue and is interrupted just before the memcpy, having calculated pos to be slot-1. If another thread now calls enqueue and completes it will write to slot-2 and increment the pendingRead counter, leaving slot-1 still unwritten. If a third thread now reads before the first thread can complete its write, the reader will read from the unwritten slot-1. This is becase the second thread correctly set the queue to say there is a pending read (slot-2) but dequeue assumes the data must be in the next available slot according to readPos.

1. queue_t is defined (elsewhere) as a pointer type. I prefer pointers to be explicit.

2. You have mixed use of struct queue * and queue_t

3. Return from malloc is not checked.

4. buf in struct queue should be void* not char **

5. Why are writePos and readPos of type uint64_t and not size_t? After all, num is size_t. Why the difference?

6. Loss of precision assigning num in createQueue

-
I rewrote the queue in an attempt to solve the problem you mentioned. writePos and readPos are uint64_t because even on 32 bit systems they need to be 64 bit. –  charliehorse55 Feb 21 '13 at 7:18
It would be nice if you could take a look at the new code. –  charliehorse55 Feb 21 '13 at 7:18
In dequeue, You need to cache the value of queue->readPos % queue->num to protect against a change in readPos between assigning value and the following if(...). In addition, you are using a busy-loop to wait for space/new-data; this is not usually considered a good idea but can sometimes be done - it depends upon your application (but personally I'd be suspicious of code like that if I met it in the wild). I don't see any other problem, but as you say, spotting race conditions is difficult; there might still be something wrong... –  William Morris Feb 21 '13 at 22:26

It seems that your enqueue() does not guarantee progress by at least one thread, in case the thread that just enqueued into the writePos, before adjusting the index is being suspended by some interrupt or system scheduler. In such case all threads trying to enqueue() will wait until the first one wake up and increment the writePos.

You might want to advance the writePos in case the CAS failed, and it is smaller than readPos - 1.

-