Andrey Gulin

Я не ручаюсь, но lock free очередь была у Саттера в цикле статей на Dr. Dobb's. Чесно ли будет смотреть туда?
ЗЫ. У меня тоже точно такая бутылка есть, кстати)

Хотя пардон, даже прочитав трезво ваш пост, я было подумал что надо предложить свою реализацию, а не подправтиь вашу, поспешил, увы.

А известно, что тут есть ошибка, или предположительно? :)
Впрочем я конечно слабо верю в безбажный lock free код на 4 экрана почти без комментариев...

вот-вот. И я не верю :) Доказательства, что ошибки нет тоже принимаются, но не эмпирические :)

а какие комментарии тебе нужны? Очередь реализована как 2 стека. В один добавляем (PushQueue), из второго достаем (PopQueue). Когда стек PopQueue кончается мы в него инвертируем PushQueue стек это ListInvertor. Против ABA защищаемся тем, что пока есть больше одного активного треда (FreememCounter > 1) мы не зовем delete. Начало/конец работы с рутом - AsyncRef()/AsyncUnref()

Для ускорения инвертирования в условиях contention при неудачном инвертировании (добавили новый элемент, пока мы инвертировали) мы не выбрасываем инвертированный список, а просто добавляем в него вновь добавленные элементы. Для этого запоминаем какой был первый элемент в ListInvertor::PrevFirst и саму копию ListInvertor::Copy-ListInvertor::Tail.

Два вопроса:
Ты не веришь в реализацию любого lock-free алгоритма? Или только для стека.

а что, намного быстрее получается, чем традиционно?

традиционно это с локом? Вообще-то в среднем - медленнее, чем с локом :) Но worst case на порядки лучше.

Есть неплохая штука для верификации l/f алгоритмов - Relacy race detector http://groups.google.com/group/relacy

прикольная штука, спасибо. Правда, насчет именно верификации не уверен. Я так понял даже с "fair_full_search_scheduler_type" она сможет проверить только ограниченное число тредов producer/consumer, правильно?

Т.е. найти ошибку она может, а доказать, что ее нет - походу не может

Допустим, есть два треда: один выходит из Dequeue в строке 180, вызывая AsyncUnref(curRoot, tail), другой выходит, допустим, в строке 187, вызвая AsyncUnref(). Главное, что случается два вызова AsyncUnref, хотя бы один из которых - параметризованный.

Допустим, первый тред попадает на строку 65 (cas в функции AsyncDel), второй - на строку 40 (cas в функции TryToFreeAsyncMemory). Пусть выигрывает компетицию первый, т.е. после выхода обоих тредов из их операций в очереди остаётся непустой FreePtr со своими кишками.

Затем очередь убивают. Случается вызов деструктора, и там TryToFreeAsyncMemory().

К сожалению, FreePtr и его замыкание не будут удалены, поскольку в деструкторе FreememCounter == 0.

Имеем memory leak.

Лечение, само собой, тривиально.

Да, похоже в диструкторе забыли TryToFreeAsyncMemory() обернуть в atomic_add(FreememCounter, 1) и atomic_add(FreememCounter, -1);

Фактический факт :) Предлагай способ вручить тебе приз, пиши в личку.

Других явных багов, как ни странно, не вижу.

Но смущают строки 167-168.

Допустим, компилятор решил сделать такую совершенно валидную в контексте одного треда трансформацию (ну, регаллок тупанул, а instruction scheduler / щелевик поменяли местами загрузку локалов):

167. TRootNode *curRoot = JobQueue;
168. TListNode *tail = curRoot->PopQueue;

-->

167. TListNode *tail = JobQueue->PopQueue;
168. TRootNode *curRoot = JobQueue;

Пусть есть два треда, одновременно выполняющих Dequeue.

Один заснул между 167 и 168, а другой в это время, пройдя через обе эти строки, далее сделал

179. tail->Next = 0;

В такой ситуации первый тред, проснувшись, на строке 175 запишет 0 в newRoot->PopQueue.

Да, в теории это выглядит натянуто, но важен самый факт того, что ты опираешься на последовательность загрузки локалов из шаренной переменной. В случае неудачного выбора компилятора и/или целевого камня вся эта красота может сломаться.

В принципе, слово volatile на локалах curRoot и tail должно снять (до конца ли?) потенциальную проблему.

Кстати, сказанное относится не только к строкам 167-168, но _ко_всем_ местам, где ты доступаешься до кишков рута до вызова CAS.

Например,

151. TRootNode *curRoot = JobQueue;
152. newRoot->PushQueue = newNode;
153. newNode->Next = curRoot->PushQueue;

вполне может быть заменено компилятором на

151'. TLinkNode *_t1 = JobQueue->PushQueue;
151. TRootNode *curRoot = JobQueue;
152. newRoot->PushQueue = newNode;
153. newNode->Next = _t1;