lock-free linked list

[kivkiv]

Я реализовал lock-free добавление и удаление в односвязный кольцевой список:

#define CAS(var, oldValue, newValue) __sync_bool_compare_and_swap(&(var), oldValue, newValue)

typedef struct Item Item;
struct Item {
	Item * volatile next;
	...
};

Item * volatile head = NULL;

void addToList(Item *item) {
	while (true) {
		item->next = listHead;
		if (item->next == NULL) {
			item->next = item;
			if (CAS(listHead, NULL, item)) {
				break;
			}
		} else {
			if (CAS(listHead, item->next, item)) {
				break;
			}
		}
	}
}

void removeFromList(Item *item) {
	while (true) {
		Item *prev = listHead;
		while (prev->next != item) {
			prev = prev->next;
		}
		if (CAS(prev->next, item, item->next)) {
				if (CAS(head, item, item->next)) {
					CAS(head, item, NULL);
				}
				break;
			}
		}
	}
}

Я не уверен, что этот код полностью thread-safe. Особенно беспокоит функция удаления (цикл поиска prev). Помогите разобраться, пожалуйста.
P.S.: Вопрос освобождения памяти после удаления элемента меня сейчас не интересует. У меня есть возможность убедиться, что элемент больше никто не попытается добавить в список перед его уничтожением.

[waleri]

Вставка еще туда-сюда.
Удаление сломано, от слова "совсем".
Начинаем от того, что идем по списку без синхронизации и кончаем ABA проблемой.
Например при вызове CAS можете передать один item и item->next от другого.

Ref:
https://en.wikipedia.org/wiki/ABA_problem

[kivkiv]

Например при вызове CAS можете передать один item и item->next от другого.

Нет, не могу. Потому что item - аргумент функции и его никто подменить не может (гарантируется, что только один процесс будет одновременно пытаться добавить/удалить один и тот же элемент).
Про ABA проблему, насколько я понимаю, это когда нашу функцию прервали, освободили память из-под элемента и выделили её под новый, а адрес оказался такой же. Либо если элемент удалят и добавят в другой список.
В моём случае список только один (элемент после удаления могут добавить только в тот же самый список), а освобождение памяти от элементов не происходит вообще (точнее происходит, но только тогда, когда все операции добавления/удаления уже давно окончены и никто элемент не тронет). Даже если элемент убрали из списка, его указатель next никто не затёр. Так что мы сможем пройти по нему дальше в цикле. Если его добавят в список заново, то мы просто перескочим с середины списка на начало (next указывает на старую голову списка после добавления). Небольшая потеря времени, но не критичная.

[Croessmah]

1) Зачем volatile?
2) Если один поток позовет addToList, а другой в это время позовет removeFromList, то что получим?
3) Если пока выполняется addToList, другой поток изменит элемент, указатель на который передан?

Ну и не знаю как там со списком, но вот Lock-free Dynamically Resizable Arrays от Страуструпа. Можно что-то почерпнуть

[waleri]

Сообщение от kivkiv

Нет, не могу.

Ну как скажете...
Гарантий того, что вы описываете в коде нет, а комментировать я могу только то, что вижу.

[kivkiv]

#define CAS(var, oldValue, newValue) __sync_bool_compare_and_swap(&(var), oldValue, newValue)

typedef struct ListItem;
struct ListItem {
	ListItem * volatile next;
	bool locked;
	bool removed;
};

typedef struct List {
	ListItem * volatile head;
	ListItem * volatile current;
} List;

List list;

void insertListItem(ListItem *item) {
	if (!CAS(item->locked, false, true)) return;
	if (item->removed == true) {
		item->removed = false;
		while (true) {
			ListItem *head = list.head;
			assert(item != head); // Во время многопоточных тестов иногда срабатывает этот assert, без него получался циклический список и зависала функция удаления.
			item->next = head;
			if (CAS(list.head, item->next, item)) {
				break;
			}
		}
	}
	item->locked = false;
}

void removeListItem(ListItem *item) {
	if (!CAS(item->locked, false, true)) return;
	if (item->removed == false) {
		item->removed = true;
		while (true) {
			ListItem *prev = list.head;
			if (prev == item) {
				ListItem *next = item->next;
				if (CAS(list.head, item, next)) {
					break;
				} else {
					continue;
				}
			}
			while (prev->next != item) {
				ListItem *next = prev->next;
				prev = next;
			}
			if (CAS(prev->next, item, item->next)) {
				ListItem *next = item->next;
				CAS(list.head, item, next);
				break;
			}
		}
	}
	item->locked = false;
}

Вот полная версия кода. С помощью item->locked гарантируется, что никто не тронет элемент, когда его добавляют/удаляют (я принимаю, что тот кто первый захотел что-то сделать с элементом, тот и прав). А для защиты от повторного добавления/удаления помогает item->removed.

То, что память из-под элемента никто не освободит гарантирует отсутствие вызовов free в основном коде. Просто специфика задачи такова, что элементы не уничтожаются.

[waleri]

Зачем делать CAS(prev->next, item, item->next))?
Если кто-то удаляет предыдущий элемент на next это не влияет а текущий элемент может удалять только один поток, так что вполне можно делать prev->next = item->next...

Все это работает, потому что элементы не меняются а в таком случае зачем вообще их удалять...

[kivkiv]

Хм... тут вы правы. Можно немного оптимизировать код, убрав CAS. Однако на работоспособность это влиять не должно.

Вообще, в функции удаления есть какая-то проблема. Сейчас добавил захват mutex перед входом в функцию удаления и освобождение перед выходом. То есть функция добавления может прерывать саму себя и функцию удаления. А функция удаления сама себя прерывать не может, только функцию добавления.

Тестирование я провожу следующим образом. Создаю N элементов и добавляю их все в список. Также указатели на все элементы сохраняются в массиве. Затем запускаю 4 потока. Каждый поток без всяких задержек и т. п., пока стоит глобальный флаг testing генерирует 2 случайных числа. Одно от 0 до N - 1, второе от 0 до 1. Если второе число равно 0, то он вызывает функцию добавления в список для элемента с выбранным номером. Иначе вызывает функцию удаления. И его совершенно не волнует есть элемент в списке или нет - это должно быть корректно отработано. Главный поток ждёт несколько сотен секунд, а затем убирает флаг testing и ожидает завершения всех потоков. Внутри функций удаления находится ряд assert, проверяющих, что данные не повреждены.

Тест считается успешным, если не сработают assert и ни один поток не зависнет. При этом счётчики показывают, что функции вызываются несколько миллиардов раз (тест выполняется минут 5-10).

Так вот. Если вызывать функции полностью lock-free, то срабатывает assert(head != item) в функции добавления в какой-то момент. Если вызывать добавление lock-free, в удаление разрешить только одно в один момент времени, то тест успешно проходится вне зависимости от количества итераций и N (пробовал маленькие значения типа 1 и 2 и большие типа 10 и 100).

Получается, что единственная ситуация, которую мой код некорректно обрабатывает - когда во время удаления одного элемента запускается удаление другого. Надо копать в эту сторону...

[kivkiv]

Произвёл некоторое упрощение кода удаления.

void removeFromList(ListItem *item) {
	if (!CAS(item->locked, false, true)) return;
	if (item->removed == false) {
		while (true) {
			ListItem *prev = (ListItem*)(&(list.head)); // next - первое поле структуры ListItem, так что мы имеем право так сделать.
			while (prev->next != item) {
				prev = prev->next;
			}
			if (CAS(prev->next, item, item->next)) {
				break;
			}
		}
		item->removed = true;
	}
	item->locked = false;
}

CAS для prev->next пришлось вернуть, ведь в некоторых случаях prev->next может являться list.head, а list.head может быть модифицирован функцией добавления. Ошибку это упрощение кода не убрало (но код стал работать не хуже), зато сделало код понятнее. По-прежнему, если removeFromList не lock-free, а addToList lock-free всё работает (кстати, наоборот не получается - если add использует mutex, а remove нет, то ошибка всё равно проявляется).