c - 无锁队列实现最终在压力下有一个循环

Question

我有一个用 C 语言编写的无锁队列，其形式为链表，其中包含来自多个线程的请求，这些线程发布到单个线程并在其中处理。经过几个小时的压力，我最终让最后一个请求的下一个指针指向它自己，这创建了一个无限循环并锁定了处理线程。

应用程序在 Linux 和 Windows 上运行(并失败)。我在 Windows 上调试，我的 COMPARE_EXCHANGE_PTR 映射到 InterlockedCompareExchangePointer .

这是将请求推送到列表头部的代码，并从多个线程调用:

void push_request(struct request * volatile * root, struct request * request)
{
    assert(request);

    do {
        request->next = *root;
    } while(COMPARE_EXCHANGE_PTR(root, request, request->next) != request->next);
}

这是从列表末尾获取请求的代码，并且仅由处理它们的单个线程调用:

struct request * pop_request(struct request * volatile * root)
{
    struct request * volatile * p;
    struct request * request;

    do {
        p = root;
        while(*p && (*p)->next) p = &(*p)->next; // <- loops here
        request = *p;
    } while(COMPARE_EXCHANGE_PTR(p, NULL, request) != request);

     assert(request->next == NULL);

     return request;
}

请注意，我没有使用尾指针，因为我想避免必须在 push_request 中处理尾指针的复杂情况。但是我怀疑问题可能出在我找到列表末尾的方式上。

有几个地方可以将请求推送到队列中，但它们大体上看起来都是这样的:

// device->requests is defined as struct request * volatile requests;
struct request * request = malloc(sizeof(struct request));
if(request) {
    // fill out request fields
    push_request(&device->requests, request);
    sem_post(device->request_sem);
}

处理请求的代码做的不止这些，但本质上是在循环中进行的:

if(sem_wait_timeout(device->request_sem, timeout) == sem_success) {
    struct request * request = pop_request(&device->requests);
    // handle request
    free(request);
}

我还刚刚添加了一个功能，在每次操作前后检查列表是否重复，但我担心这个检查会改变时间，这样我就永远不会遇到失败的地方。 (我在写这篇文章时正在等待它崩溃。)

当我中断挂起的程序时，处理程序线程在标记位置的 pop_request 中循环。我有一个包含一个或多个请求的有效列表，最后一个的下一个指针指向它自己。请求队列通常很短，我从未见过超过 10 个，而且只有 1 和 3 两次我可以在调试器中查看此故障。

我尽可能多地考虑了这一点，并得出结论，除非我两次推送相同的请求，否则我永远不会在我的列表中出现循环。我很确定这永远不会发生。我也很确定(虽然不完全)它不是 ABA problem .

我知道我可能会同时弹出多个请求，但我相信这在这里无关紧要，而且我从未见过这种情况发生。 (我也会解决这个问题)

我对如何破坏我的功能进行了长期而艰苦的思考，但我没有找到以循环结束的方法。

所以问题是:有人能看出破解方法吗？有人能证明这不能吗？

最终我会解决这个问题(也许通过使用尾指针或其他一些解决方案 - 锁定将是一个问题，因为发布的线程不应该被锁定，虽然我手头有一个 RW 锁)但我想确保更改列表确实解决了我的问题(而不是因为时间不同而降低了它的可能性)。

Answer 1

这很微妙，但确实存在竞争条件。

从一个包含一个元素的列表开始，req1。所以我们有:

device->requests == req1;
req1->next == NULL;

现在，我们推送一个新元素 req2，同时尝试弹出队列。 req2 的推送首先开始。 while 循环体运行，所以我们现在有:

device->requests == req1;
req1->next == NULL;
req2->next == req1;

然后 COMPARE_EXCHANGE_PTR 运行，所以我们有:

device->requests == req2;
req1->next == NULL;
req2->next == req1;

...COMPARE_EXCHANGE_PTR() 返回 req1。现在，此时，在 while 条件下进行比较之前，推送被中断，弹出开始运行。

弹出正确运行到完成，弹出 req1 - 这意味着我们有:

device->requests == req2;
req2->next == NULL;

推送重新开始。它现在获取 request->next 来进行比较 - 它获取 req2->next 的 new 值，即 空。它将 req1 与 NULL 进行比较，比较成功，while 循环再次运行，现在我们有:

device->requests == req2;
req2->next == req2;

这次测试失败，while 循环退出，你有你的循环。

---

这应该可以解决:

void push_request(struct request * volatile * root, struct request * request)
{
    struct request *oldroot;

    assert(request);

    do {
        request->next = oldroot = *root;
    } while(COMPARE_EXCHANGE_PTR(root, request, oldroot) != oldroot);
}