无锁队列答案

【问题标题】：Lock-free queue无锁队列
【发布时间】：2011-08-30 15:34:16
【问题描述】：

我也在做一个c 的实现，目前有队列的结构：

typedef struct queueelem {
    queuedata_t data;
    struct queueelem *next;
} queueelem_t;

typedef struct queue {
    int capacity;
    int size;
    queueelem_t *head;
    queueelem_t *tail;
} queue_t;

queue_t *
queue_init(int capacity)
{
    queue_t *q = (queue_t *) malloc(sizeof(queue_t));
    q->head = q->tail = NULL;
    q->size = 0;
    q->capacity = capacity;
    return q;
}

int CompareAndExchange (void **a, void *comparand,void *new) {
    int success = 0;
    pthread_mutex_lock(&CE_MUTEX);
    if ((*a) != comparand) {
       (*a) = new;
       //return     TRUE
       success = 1;
    }
    pthread_mutex_unlock(&CE_MUTEX);     
   //return     FALSE
    return success;
 }

但不确定如何继续，使用队列和出队功能......

代码看起来如何？

【问题讨论】：

Compare&Swap 和 Fetch&Add 是 CPU 的扩展函数，可能内置在您的编译器/库中
我正在执行 CompareAndExchange，请参阅我的更新。
无锁队列是独角兽，您必须使用运行时或操作系统提供的低锁原语。示例算法中的 FetchAndAdd、CompareAndSwap。这就是责任停止的地方，您没有记录您的运行时环境。
@darkcminor：您知道，您正在使用 lock 来实现队列的基本实用功能，因此它不是 lock-free 排队！
对于比较和交换和获取和添加，您可以使用 GCC 的原子内置函数：gcc.gnu.org/onlinedocs/gcc/Atomic-Builtins.html

标签： c queue lock-free

【解决方案1】：

您的伪代码可能（并且很可能确实）受到ABA problem 的影响，因为只检查了指针，而不是随附的唯一标记，您会发现this paper 在这方面很有用，并且作为无锁队列实现的一般指南及其缺陷。

在处理无锁编程时，阅读 Herb Sutter 的作品也是一个好主意，因为他对需要什么、为什么需要它以及它潜在的弱点给出了很好、有见地的解释（尽管要注意他的一些年长的发现包含一些隐藏/未预见问题的出版物/文章）。

【讨论】：

有时 ABA 并不重要，如果一个对象的地址是它的永久标识符。例如，我尝试使用无锁队列的一个地方是在堆管理器中，其中对象（堆上的空闲块）的位置是它的标识。如果您正在处理其他类型的长寿命对象，这些对象只是从不同的列表中取出而从未被释放或分配，则同样适用。

【解决方案2】：

还有最近关于这个主题的 boost'con 讨论： https://github.com/boostcon/2011_presentations/raw/master/wed/lockfree_2011_slides.pdf

【讨论】：

【解决方案3】：

前段时间，我发现a nice solution 解决了这个问题。我相信它是迄今为止发现的最小的。

存储库有一个示例，说明如何使用它创建 N 个线程（读取器和写入器），然后共享一个席位。

我在测试示例上做了一些基准测试，得到了以下结果（以百万次操作/秒为单位）：

按缓冲区大小

按线程数

请注意线程数不会改变吞吐量。

我认为这是这个问题的最终解决方案。它的工作原理令人难以置信的快速和简单。即使有数百个线程和单个位置的队列。它可以用作线程之间的管道，在队列内部分配空间。

存储库有一些用 C# 和 pascal 编写的早期版本。我正在努力使一些东西更加完善以显示其真正的力量。

我希望你们中的一些人可以验证这项工作或提供一些想法。或者至少，你能打破它吗？

【讨论】：

【解决方案4】：

（暂时离开这里，但请参阅编辑。）

你知道 C 中无锁队列的实现吗？

我最近写了无锁队列（http://www.ideone.com/l2QRp）。我实际上不能保证它正常工作，但我无法找到任何错误，并且我已经在几个单线程程序中使用它没有任何问题，所以它没有什么太明显的错误.

简单的用法示例：

queue_t queue;
int val = 42;
queue_init(&queue,sizeof val);
queue_put(&queue,&val);
val = 0; 
queue_pop(&queue,&val);
printf("%i\n",val); // 42
queue_destroy(&queue);

编辑：

正如@Alexey Kukanov 所指出的，如果tmp 被弹出、释放、再次分配，并在检查空值和交换之间再次放置，queue_pop 可能会失败：

    if(!tmp->next) return errno = ENODATA;
    /* can fail here */
    } while(!sync_swap(q->head,tmp,tmp->next));

我还不确定如何解决这个问题，但我会（希望）在我弄明白后更新它。暂时不要理会这个。

【讨论】：

单线程代码的无锁队列有什么意义？在多线程代码中，它会因为 ABA 问题而中断。
@Alexey Kukanov，它的设计（不一定实现，但设计）不需要关心“A”背后的内容，只要 A 在那里，它应该可以正常工作，即使数据后面的A已经成。不过，我确实需要仔细检查并确认这是真的。
Joel Falcou 的回答中引用的幻灯片解释了为什么您的 queue_pop 患有 ABA。读取 tmp->next 和 CAS 之间有一个时间窗口；无论多么小，都足以使您阅读的价值变得陈旧。
如果您总是将新项目添加到列表中，则没有 ABA。只有当相同的项目被删除然后重新添加时，你才会有这个问题。
我们总是将项目添加到列表的头部，而不是尾部。事实上，我们没有尾部指针。因此，您只有一个 CAS，没有比赛。

【解决方案5】：

你可以试试这个库，它是用 c native 构建的。 lfqueue

举例

int* int_data;
lfqueue_t my_queue;

if (lfqueue_init(&my_queue) == -1)
    return -1;

/** Wrap This scope in other threads **/
int_data = (int*) malloc(sizeof(int));
assert(int_data != NULL);
*int_data = i++;
/*Enqueue*/
 while (lfqueue_enq(&my_queue, int_data) == -1) {
    printf("ENQ Full ?\n");
}

/** Wrap This scope in other threads **/
/*Dequeue*/
while  ( (int_data = lfqueue_deq(&my_queue)) == NULL) {
    printf("DEQ EMPTY ..\n");
}

// printf("%d\n", *(int*) int_data );
free(int_data);
/** End **/

lfqueue_destroy(&my_queue);

【讨论】：