5
\$\begingroup\$

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?

\$\endgroup\$
3
\$\begingroup\$

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;
};
\$\endgroup\$
  • \$\begingroup\$ there's no <stdatomic.h> in GCC 4.8 ... Anyway, I didn't know I can make clear the declaration.. Thank you! \$\endgroup\$ – ikh May 25 '14 at 9:25

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.