Энтони Уильямс - Параллельное программирование на С++ в действии. Практика разработки многопоточных программ

С другой стороны, если мы все-таки хотим, чтобы несколько потоков могли одновременно вызывать pop(), то нужно каким-то образом отслеживать момент, когда удаление узла становится безопасным. По сути дела, это означает, что необходимо написать специализированный сборщик мусора для узлов node. Звучит пугающе, и эта задача действительно не самая простая, но на практике все не так плохо: мы проверяем только указатели на nodeи только те узлы, к которым обращается pop(). Нас не интересуют узлы внутри push(), потому что они доступны только одному потоку, пока не окажутся в стеке. А вот внутри pop()к одному и тому же узлу могут одновременно иметь доступ несколько потоков.

Если потоков, вызывающих pop(), нет вообще, то можно без опаски удалить все узлы, ожидающие удаления. Поэтому, если после извлечения данных помещать узлы в список «подлежат удалению», то их можно будет удалить одним махом в момент, когда не будет потоков, вызывающих pop(). Но как узнать, что потоков, вызывающих pop(), действительно нет? Очень просто — подсчитывать их. Если увеличивать счетчик при входе и уменьшать при выходе, то удалять узлы из списка «подлежащих удалению» можно будет, когда счетчик становится равным нулю. Разумеется, сам счетчик должен быть атомарным, чтобы к нему можно было безопасно обращаться из нескольких потоков. В листинге 7.4 показала исправленная функция pop(), а в листинге 7.5 — вспомогательные функции, используемые в ее реализации.

Листинг 7.4.Освобождение занятой узлами памяти в момент, когда нет потоков, вызывающих pop()

template

class lock_free_stack {

private:

std::atomic threads_in_pop;←┐ Атомарная

void try_reclaim(node* old_head); (1) переменная

public:

std::shared_ptr pop()

{ (2) Увеличить счетчик

++threads_in_pop; ←┤ перед тем, как что-то

node* old_head = head.load();│ делать

while (old_head &&

!head.compare_exchange_weak(old_head, old_head->next));

std::shared_ptr res;

if (old_head)

{ (3) He копировать

res.swap(old_head->data);←┤ указатель, а извлечь

} │ данные из узла

try_reclaim(old_head);←┐ Освободить удаленные

return res; (4) узлы, если получится

}

};

Атомарная переменная threads_in_pop (1)нужна для подсчета потоков, которые в данный момент пытаются извлечь элемент из стека. Она увеличивается на единицу в начале pop() (2)и уменьшается на единицу внутри функции try_reclaim(), которая вызывается после изъятия узла из списка (4). Поскольку мы откладываем удаление самого узла, то можно с помощью swap()переместить из него данные (3), а не просто скопировать указатель; тогда данные будут автоматически удалены, когда в них отпадает необходимость, вместо того, чтобы занимать память только потому, что на них ссылается еще не удаленный узел. В следующем листинге показан код функции try_reclaim().

Листинг 7.5.Механизм освобождения памяти на основе подсчёта ссылок

template

class lock_free_stack {

private:

std::atomic to_be_deleted;

static void delete_nodes(node* nodes) {

while (nodes) {

node* next = nodes->next;

delete nodes;

nodes = next;

}

}

void try_reclaim(node* old_head) {

if (threads_in_pop == 1) ← (1)

{ Заявляем права на список подлежащих удалению узлов (2)

node* nodes_to_delete = to_be_deleted.exchange(nullptr);←┘

if (!--threads_in_pop)←┐ Я — единственный

{ (3) поток в pop()?

delete_nodes(nodes_to_delete); ← (4)

} else if(nodes_to_delete) { ← (5)

chain_pending_nodes(nodes_to_delete);← (6)

}

delete old_head; ← (7)

} else {

chain_pending_node(old_head); ← (8)

--threads_in_pop;

}

}

void chain_pending_nodes(node* nodes) {

node* last = nodes;

while (node* const next =

last->next) {←┐ По указателям

├ (9) next доходим до

last = next; │ конца списка

}

chain_pending_nodes(nodes, last);

}

void chain_pending_nodes(node* first, node* last) {

last->next = to_be_deleted;← (10)

while (

!to_be_deleted.compare_exchange_weak(←┐ цикл гарантиру-

last->next, first)); ├ (11)ет, что last->next

} │ корректно

void chain_pending_node(node* n) {

chain_pending_nodes(n, n);← (12)

}

};

Если при попытке освободить занятую узлом (1)память счетчик threads_in_popоказывается равен 1, то данный поток — единственный в pop(), и, значит, можно безопасно удалять только что исключенный из списка узел (7)и, быть может , также узлы, ожидающие удаления. Если счетчик не равен 1, то никакие узлы удалять нельзя, поэтому мы просто добавляем узел в список ожидающих (8).

Предположим, что threads_in_popравно 1. Тогда нам нужно освободить ожидающие узлы, потому что если этого не сделать, то они так и будут ожидать удаления до момента уничтожения стека. Для этого мы запрашиваем монопольные права на список с помощью атомарной операции exchange (2), после чего уменьшаем на единицу счетчик threads_in_pop (3). Если в результате счетчик оказался равным нулю, значит, больше ни один поток не работает со списком ожидающих удаления узлов. По ходу дела могли появиться новые ожидающие узлы, но сейчас — когда можно безопасно очистить список — нам это безразлично. Мы просто вызываем функцию delete_nodes, которая обходит список и удаляет узлы (4).

Если счетчик после уменьшения не равен нулю, то освобождать узлы небезопасно, поэтому если такие узлы есть (5), то нужно поместить их в начало списка ожидающих (6). Такое может случиться, если к структуре данных одновременно обращаются несколько потоков. Другие потоки могли вызвать pop()в промежутке между первой проверкой threads_in_pop (1)и «заявлением прав» на список (2)и добавить в список узлы, к которым все еще обращается один или несколько потоков. На рис. 7.1 поток С добавляет узел Y в список to_be_deleted, несмотря на то, что поток В все еще ссылается на него по указателю old_headи, значит, будет пробовать читать его указатель next. Поэтому поток А не может удалить узлы без риска вызвать неопределенное поведение в потоке В.

Рис. 7.1.На примере трех потоков, вызывающих pop(), видно, почему необходимо проверять threads_in_popпосле заявления прав на список узлов, ожидающих удаления, в try_reclaim()