Lock-free queue with doubly linked list correctness

Question

I need a lock-free queue which is implemented by doubly linked list.

Is my code correct? Are there any errors? Are there any improvements to be made?

linkedlist.h

/*
 * linkedlist.h
 *
 * Created on: 2014. 5. 10.
 * Author: dlaru_000
 */
#ifndef LINKEDLIST_H_
#define LINKEDLIST_H_
/*
 * circular doubly linked list
 * PushBack / PopFront: single-producer, single-customer lock-free queue
 * Erase: thread-unsafe
 */
#include <stddef.h>
#include <stdint.h>
typedef struct tagLinkedListNode
{
 struct tagLinkedListNode * volatile pNext;
 struct tagLinkedListNode * volatile pPrev;
} LinkedListNode;
typedef struct tagLinkedList
{
 LinkedListNode dummy;
 /*
 * It can be invalid while some function is running on the other thread.
 * However it must valid after/before the function is finished.
 */
 uint32_t size;
} LinkedList;
void ckLinkedListInit(LinkedList *pList);
static inline LinkedListNode *ckLinkedListHead(LinkedList *pList) { return pList->dummy.pNext; }
static inline LinkedListNode *ckLinkedListTail(LinkedList *pList) { return pList->dummy.pPrev; }
void ckLinkedListPushBack(LinkedList *pList, void *pNode);
LinkedListNode *ckLinkedListPopFront(LinkedList *pList);
void ckLinkedListErase(LinkedList *pList, void *pNode);
#endif /* LINKEDLIST_H_ */

linkedlist.c

/*
 * linkedlist.c
 *
 * Created on: 2014. 5. 10.
 * Author: dlaru_000
 */
#include "linkedlist.h"
void ckLinkedListInit(LinkedList *pList)
{
 // circular linked list
 pList->dummy.pNext = &pList->dummy;
 pList->dummy.pPrev = &pList->dummy;
 pList->size = 0;
}
void ckLinkedListPushBack(LinkedList *pList, void *pNode)
{
 __sync_add_and_fetch(&pList->size, 1);
 LinkedListNode *dummy = &pList->dummy;
 LinkedListNode *ptr = (LinkedListNode *)pNode;
 LinkedListNode *tail;
 ptr->pNext = dummy;
 while (1)
 {
 tail = dummy->pPrev;
 ptr->pPrev = tail;
 if (__sync_bool_compare_and_swap(&dummy->pPrev, tail, ptr))
 {
 tail->pNext = ptr;
 break;
 }
 }
}
LinkedListNode *ckLinkedListPopFront(LinkedList *pList)
{
 LinkedListNode *dummy = &pList->dummy;
 LinkedListNode *ptr, *next, *dnext;
 ptr = dummy->pNext;
 if (ptr == dummy)
 {
 return NULL;
 }
 while (1)
 {
 next = ptr->pNext;
 if (__sync_bool_compare_and_swap(&next->pPrev, ptr, dummy))
 {
 dnext = dummy->pNext;
 __sync_bool_compare_and_swap(&dummy->pNext, dnext, next);
 break;
 }
 }
 __sync_sub_and_fetch(&pList->size, 1);
 return ptr;
}
void ckLinkedListErase(LinkedList *pList, void *pNode)
{
 LinkedListNode *ptr = (LinkedListNode *)pNode;
 LinkedListNode *prev = ptr->pPrev;
 LinkedListNode *next = ptr->pNext;
 prev->pNext = next;
 next->pPrev = prev;
 pList->size--;
}

The type of the second parameter of PushBack is void *, because the list actually contains such a thing like the following:

struct SomeData
{
 LinkedListNode _node;
 ...
};

I'm using GCC 4.8.2 with cygwin.

---

Um, I've noticed that "__sync" functions, which the program is using, are legacy, and there are new functions __atomic. How can I migrate to __atomic?

Answer 1

C11 atomic library

I you want to avoid the "legacy" functions, one solution would be to use the C11 atomic operations library. It is not widely supported yet. I know that GCC has been working on it, but I don't know if the 4.9 release already implements all the operations. Anyway, had you a compliant compiler, you could use standard atomic types and atomic operations.

For example, here are your functions ckLinkedListPushBack and ckLinkedListPopFront rewritten with the C11 atomic module:

void ckLinkedListPushBack(LinkedList *pList, void *pNode)
{
 atomic_fetch_add(&pList->size, 1);
 LinkedListNode *dummy = &pList->dummy;
 LinkedListNode *ptr = (LinkedListNode *)pNode;
 LinkedListNode *tail;
 ptr->pNext = dummy;
 while (1)
 {
 tail = dummy->pPrev;
 ptr->pPrev = tail;
 if (atomic_compare_exchange_strong(&dummy->pPrev, tail, ptr))
 {
 tail->pNext = ptr;
 break;
 }
 }
}
LinkedListNode *ckLinkedListPopFront(LinkedList *pList)
{
 LinkedListNode *dummy = &pList->dummy;
 LinkedListNode *ptr, *next, *dnext;
 ptr = dummy->pNext;
 if (ptr == dummy)
 {
 return NULL;
 }
 while (1)
 {
 next = ptr->pNext;
 if (atomic_compare_exchange_strong(&next->pPrev, ptr, dummy))
 {
 dnext = dummy->pNext;
 atomic_compare_exchange_strong(&dummy->pNext, dnext, next);
 break;
 }
 }
 atomic_fetch_sub(&pList->size, 1);
 return ptr;
}

In order for it to work, you would have to declare size to be an instance of atomic_uint_least32_t (fixed size types are not provided for atomics).

typedef struct

You can make clearer the LinkedListNode declaration by separating the typedef from the actual declaration:

typedef struct tagLinkedListNode LinkedListNode;
struct tagLinkedListNode 
{
 LinkedListNode * volatile pNext;
 LinkedListNode * volatile pPrev;
};

• ikh: there's no <stdatomic.h> in GCC 4.8 ... Anyway, I didn't know I can make clear the declaration.. Thank you!