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

Листинг 7.8.Простая реализация функций освобождения узлов

template

void do_delete(void* p) {

delete static_cast(p);

}

struct data_to_reclaim {

void* data;

std::function deleter;

data_to_reclaim* next;

template

data_to_reclaim(T* p) : ← (1)

data(p),

deleter(&do_delete), next(0) {}

~data_to_reclaim() {

deleter(data); ← (2)

}

};

std::atomic nodes_to_reclaim;

void add_to_reclaim_list(data_to_reclaim* node) {← (3)

node->next = nodes_to_reclaim.load();

while (

!nodes_to_reclaim.compare_exchange_weak(node->next, node));

}

template

void reclaim_later(T* data) { ← (4)

add_to_reclaim_list(new data_to_reclaim(data));← (5)

}

void delete_nodes_with_no_hazards() {

data_to_reclaim* current =

nodes_to_reclaim.exchange(nullptr); ← (6)

while(current) {

data_to_reclaim* const next = current->next;

if (!outstanding_hazard_pointers_for(current->data)) {← (7)

delete current; ← (8)

} else {

add_to_reclaim_list(current); ← (9)

}

current = next;

}

}

Полагаю, вы обратили внимание, что reclaim_later()— шаблон функции, а не обычная функция (4). Объясняется это тем, что указатели опасности — это универсальный механизм, поэтому не стоит ограничивать его только узлами стека. Ранее для хранения указателей мы использовали тип std::atomic. Поэтому мы должны обрабатывать произвольный указательный тип, но просто указать void*нельзя, так как мы собираемся удалять данные по указателю, а оператору deleteнужно знать реальный тип указателя. Как мы скоро увидим, конструктор data_to_reclaimпрекрасно справляется с этой проблемой: reclaim_later()просто создает новый экземпляр data_to_reclaimдля переданного указателя и добавляет его в список отложенного освобождения (5). Сама функция add_to_reclaim_list() (3)— не более чем простой цикл по compare_exchange_weak()для головного элемента списка; мы уже встречались с такой конструкцией раньше.

Но вернёмся к конструктору data_to_reclaim (1), который также является шаблоном. Он сохраняет подлежащие удалению данные в виде указателя void*в члене data, после чего запоминает указатель на подходящую конкретизацию do_delete() — простую функцию, которая приводит тип void*к типу параметризованного указателя, а затем удаляет объект, на который он указывает. Шаблон std::function<>безопасно обертывает этот указатель на функцию, так что впоследствии деструктору data_to_reclaimдля удаления данных нужно всего лишь вызвать запомненную функцию (2).

Деструктор data_to_reclaimне вызывается, когда мы добавляем узлы в список; он вызывается только, когда на узел не ссылается ни один указатель опасности. За это отвечает функция delete_nodes_with_no_hazards().

Эта функция сначала заявляет права на владение всем списком подлежащих освобождению узлов, вызывая exchange() (6). Это простое, но крайне важное действие гарантирует, что данный конкретный набор узлов будет освобождать только один поток. Другие потоки вправе добавлять в список новые узлы и даже пытаться освободить их, никак не затрагивая работу этого потока.

Далее мы по очереди просматриваем все узлы списка, проверяя, ссылаются ли на них не сброшенные указатели опасности (7). Если нет, то запись можно удалить (очистив хранящиеся в ней данные) (8). В противном случае мы возвращаем элемент в список, чтобы освободить его позже (9).

Хотя эта простая реализация справляется с задачей безопасного освобождения удаленных узлов, она заметно увеличивает накладные расходы. Для просмотра массива указателей опасности требуется проверить max_hazard_pointersатомарных переменных, и это делается при каждом вызове pop(). Атомарные операции по необходимости работают медленно — зачастую в 100 раз медленнее эквивалентных обычных операций на настольном ПК, — поэтому pop()оказывается дорогостоящей операцией. Мало того что приходится просматривать список указателей опасности для исключаемого из списка узла, так еще надо просмотреть его для каждого узла в списке ожидающих освобождения. Понятно, что это не слишком удачная идея. В списке может храниться max_hazard_pointersузлов, и каждый из них нужно сравнить с max_hazard_pointersхранимых указателей опасности. Черт! Должно существовать решение получше.

И оно, конечно же, существует. Показанное выше решение — это простая и наивная реализация указателей опасности, которую я привел, только чтобы объяснить идею. Первое, что можно сделать, — пожертвовать памятью ради быстродействия. Вместо того чтобы проверять каждый узел в списке освобождения при каждом обращении к pop(), мы вообще не будем пытаться освобождать узлы, пока их число в списке не превысит max_hazard_pointers. Тогда мы гарантированно сможем освободить хотя бы один узел. Но если просто ждать, пока в списке накопится max_hazard_pointers+1узлов, то выиграем мы немного. После того как число узлов достигает max_hazard_pointers, мы будем пытаться освобождать их почти при каждом вызове pop(), так что проблема лишь немного отодвинулась во времени. Но если дождаться, пока в списке наберётся 2*max_hazard_pointersузлов, то мы гарантированно сможем освободить по крайней мере max_hazard_pointersузлов, и тогда следующую попытку нужно будет делать не раньше, чем через max_hazard_pointersобращений к pop(). Это уже гораздо лучше. Вместо того чтобы просматривать max_hazard_pointersузлов при каждом вызове pop()(и, возможно, ничего не освободить), мы проверяем 2*max_hazard_pointersчерез каждые max_hazard_pointersвызовов pop()и освобождаем не менее max_hazard_pointers. Получается, что в среднем мы проверяем два узла при каждом вызове pop(), и один из них точно освобождается.

Но и у этого решения есть недостаток (помимо увеличенного расхода памяти): теперь мы должны подсчитывать узлы в списке освобождения, то есть использовать атомарный счетчик, а, кроме того, за доступ к самому списку конкурируют несколько потоков. Если память позволяет, то можно предложить еще более эффективную схему освобождения: каждый поток хранит собственный список освобождения в поточно-локальной памяти. Тогда ни для счетчика, ни для доступа к списку не понадобятся атомарные переменные. Но в обмен придется выделить память для max_hazard_pointers * max_hazard_pointersузлов. Если поток завершается прежде, чем освобождены все его узлы, то оставшиеся можно перенести в глобальный список, как и раньше, а затем добавить в локальный список следующего потока, пожелавшего выполнить процедуру освобождения.