无锁队列实现最终会在压力下产生循环

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问题.

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

我想我怎么能破坏我的功能,但是我没有想到一种方法来结束循环.

所以问题是:有人能看到这样的方式可以打破吗？有人可以证明这不可以吗？

最终我会解决这个问题(可能通过使用尾指针或其他一些解决方案 - 锁定将是一个问题,因为发布的线程不应该被锁定,虽然我手边有一个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条件比较之前,推送被中断并且弹出开始运行.

pop正常运行完成,弹出req1- 这意味着我们有:

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

推动重启.它现在提取request->next进行比较 - 并且它获取新值req2->next,即NULL.它比较req1用NULL的比较成功,而循环再次运行,现在我们有:

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);
}