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

После того как pop_head()удалит узел из очереди, обновив head, мьютекс освобождается, и try_pop()может извлечь данные и удалить узел, если таковой был (или вернуть NULL-экземпляр класса std::shared_ptr<>, если узла не было), твердо зная, что она работает в единственном потоке, который имеет доступ к этому узлу.

Далее, внешний интерфейс является подмножеством интерфейса из листинга 6.2, поэтому ранее выполненный анализ остается в силе: в интерфейсе нет внутренне присущих состояний гонки.

Вопрос об исключениях более интересен. Поскольку мы изменили порядок выделения памяти, исключения могут возникать в других местах. Единственные операции в try_pop(), способные возбудить исключение, — это захваты мьютексов, но пока мьютексы не захвачены, данные не модифицируются. Поэтому try_pop()безопасна относительно исключений. С другой стороны, push()выделяет из кучи память для объектов Tи node, и каждая такая операция может возбудить исключение. Однако оба вновь созданных объекта присваиваются интеллектуальным указателям, поэтому в случае исключения память корректно освобождается. После того как мьютекс захвачен, ни одна из последующих операций внутри push()не может возбудить исключение, так что мы снова в безопасности.

Поскольку мы не изменяли интерфейс, то новых внешних возможностей для взаимоблокировки не возникло. Внутренних возможностей также нет; единственное место, где захватываются два мьютекса, — это функция pop_head(), но она всегда захватывает сначала head_mutex, а потом tail_mutex, так что взаимоблокировки не случится.

Осталось рассмотреть только один вопрос — в какой мере возможно распараллеливание. Эта структура данных предоставляет куда больше таких возможностей, чем приведенная в листинге 6.2, потому что гранулярность блокировок мельче, и больше работы выполняется не под защитой блокировок . Например, в push()память для нового узла и нового элемента данных выделяется, когда ни одна блокировка не удерживается. Это означает, что несколько потоков могут спокойно выделять новые узлы и элементы данных в одно и то же время. В каждый момент времени только один поток может добавлять новый узел в список, но выполняющий это действие код сводится к нескольким простым присваиваниям указателей, так что блокировка удерживается совсем недолго по сравнению с реализацией на основе std::queue<>, где мьютекс остается захваченным в течение всего времени, пока выполняются операции выделения памяти внутри std::queue<>.

Кроме того, try_pop()удерживает tail_mutexлишь на очень короткое время, необходимое для защиты чтения tail. Следовательно, почти все действия внутри try_pop()могут производиться одновременно с вызовом push(). Объем операций, выполняемых под защитой мьютекса head_mutexтакже совсем невелик; дорогостоящая операция delete(в деструкторе указателя на узел) производится вне блокировки. Это увеличивает потенциальное число одновременных обращений к try_pop(); в каждый момент времени только один поток может вызывать pop_head(), зато несколько потоков могут удалять старые узлы и безопасно возвращать данные.

Ну хорошо, код в листинге 6.6 дает в наше распоряжение потокобезопасную очередь с мелкогранулярными блокировками, но он поддерживает только функцию try_pop()(и к тому же всего в одном варианте). А как насчет таких удобных функций wait_and_pop(), которые мы написали в листинге 6.2? Сможем ли мы реализовать идентичный интерфейс, сохранив мелкогранулярные блокировки?

Ответ, разумеется, — да, только вот как это сделать? Модифицировать push()несложно: нужно лишь добавить вызов data_cond.notify_one()в конец функции, как и было в листинге 6.2. Но на самом деле не всё так просто; мы же связались с мелкогранулярными блокировками для того, чтобы увеличить уровень параллелизма. Если оставить мьютекс захваченным на все время вызова notify_one()(как в листинге 6.2), то поток, разбуженный до того, как мьютекс освобожден, должен будет ждать мьютекса. С другой стороны, если освободить мьютекс до обращения к notify_one(), то ожидающий поток сможет захватить его сразу, как проснётся (если, конечно, какой-то другой поток не успеет раньше). Это небольшое улучшение, но в некоторых случаях оно бывает полезно.

Функция wait_and_pop()сложнее, потому что мы должны решить, где поместить ожидание, какой задать предикат и какой мьютекс захватить. Мы ждем условия «очередь не пуста», оно представляется выражением head != tail. Если записать его в таком виде, то придется захватывать и head_mutex, и tail_mutex, но, разбирая код в листинге 6.6, мы уже поняли, что захватывать tail_mutexнужно только для чтения tail, а не для самого сравнения, та же логика применима и здесь. Если записать предикат в виде head != get_tail(), то нужно будет захватить только head_mutexи использовать уже полученную блокировку для защиты data_cond.wait(). Прочий код такой же, как в try_pop().

Второй перегруженный вариант try_pop()и соответствующий ему вариант wait_and_pop()нуждаются в тщательном осмыслении. Если просто заменить возврат указателя std::shared_ptr<>, полученного из old_head, копирующим присваиванием параметру value, то функция перестанет быть безопасной относительно исключений. В этот момент элемент данных уже удален из очереди и мьютекс освобожден, осталось только вернуть данные вызывающей программе. Однако, если копирующее присваивание возбудит исключение (а почему бы и нет?), то элемент данных будет потерян, потому что вернуть его в то же место очереди, где он был, уже невозможно.

Если фактический тип T, которым конкретизируется шаблон, обладает не возбуждающими исключений оператором перемещающего присваивания или операцией обмена ( swap), то так поступить можно, но ведь мы ищем общее решение, применимое к любому типу T. В таком случае следует поместить операции, способные возбудить исключения, в защищенную область перед тем, как удалять узел из списка. Это означает, что нам необходим еще один перегруженный вариант pop_head(), который извлекает сохраненное значение до модификации списка.

Напротив, модификация функции empty() тривиальна: нужно просто захватить head_mutexи выполнить проверку head == get_tail()(см. листинг 6.10). Окончательный код очереди приведён в листингах 6.7, 6.8, 6.9 и 6.10.

Листинг 6.7.Потокобезопасная очередь с блокировкой и ожиданием: внутренние данные и интерфейс

template

class threadsafe_queue {

private:

struct node {

std::shared_ptr data;