Реализация очереди Lockless заканчивается тем, что у меня есть стресс

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. Однако я подозреваю, что проблема может быть в том, как я нахожу конец списка.

Есть несколько мест, которые толкают запрос в очередь, но все они выглядят Generaly так:

// 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. Тело цикла в то время как работает, так что мы теперь имеем:

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, что составляет NULL. Он сравнивает req1 с NULL, сравнение успешно, то цикл в то время как работает снова, и теперь мы имеем:

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

На этот раз тест не пройден, выходы в то время как петли, и у вас есть цикл.

---

Это должно исправить:

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