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

Уверен, теперь вы задаетесь вопросом, что мы выиграли от всех этих изменений и как они помогут сделать код потокобезопасным. Разберемся. Функция push()теперь обращается только к tail, но не к head, и это, безусловно, улучшение. try_pop()обращается и к head, и к tail, но tailнужен только для начального сравнения, так что блокировка удерживается очень недолго. Основной выигрыш мы получили за счет того, что из-за наличия фиктивного узла try_pop()и push()никогда не оперируют одним и тем же узлом, так что нам больше не нужен всеохватывающий мьютекс. Стало быть, мы можем завести по одному мьютексу для headи tail. Но где расставить блокировки?

Мы хотим обеспечить максимум возможностей для распараллеливания, поэтому блокировки должны освобождаться как можно быстрее. С функцией push()всё просто: мьютекс должен быть заблокирован на протяжении всех обращений к tail, а это означает, что мы захватываем его после выделения памяти для нового узла (8)и перед тем, как записать данные в текущий последний узел (9). Затем блокировку следует удерживать до конца функции.

С try_pop()сложнее. Прежде всего, нам нужно захватить мьютекс для headи удерживать его, пока мы не закончим работать с head. По сути дела, этот мьютекс определяет, какой поток производит извлечение из очереди, поэтому захватить его надо в самом начале. После того как значение headизменено (5), мьютекс можно освободить; в момент, когда возвращается результат (6), он уже не нужен. Остается разобраться с защитой доступа к tail. Поскольку мы обращаемся к tailтолько один раз, то можно захватить мьютекс на время, требуемое для чтения. Проще всего сделать это, поместив операцию доступа в отдельную функцию. На самом деле, поскольку участок кода, в котором мьютекс для headдолжен быть заблокирован, является частью одной функции-члена, то будет правильнее завести отдельную функцию и для него тоже. Окончательный код приведён в листинге 6.6.

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

template

class threadsafe_queue {

private:

struct node {

std::shared_ptr data;

std::unique_ptr next;

};

std::mutex head_mutex;

std::unique_ptr head;

std::mutex tail_mutex;

node* tail;

node* get_tail() {

std::lock_guard tail_lock(tail_mutex);

return tail;

}

std::unique_ptr pop_head() {

std::lock_guard head_lock(head_mutex);

if (head.get() == get_tail()) {

return nullptr;

}

std::unique_ptr old_head = std::move(head);

head = std::move(old_head->next);

return old_head;

}

public:

threadsafe_queue():

head(new node), tail(head.get()) {}

threadsafe_queue(const threadsafe_queue& other) = delete;

threadsafe_queue& operator=(

const threadsafe_queue& other) = delete;

std::shared_ptr try_pop() {

std::unique_ptr old_head = pop_head();

return old_head ? old_head->data : std::shared_ptr();

}

void push(T new_value) {

std::shared_ptr new_data(

std::make_shared(std::move(new_value)));

std::unique_ptr p(new node);

node* const new_tail = p.get();

std::lock_guard tail_lock(tail_mutex);

tail->data = new_data;

tail->next = std::move(p);

tail = new_tail;

}

};

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

• tail->next == nullptr.

• tail->data == nullptr.

• head == tailозначает, что список пуст.

• Для списка с одним элементом head->next==tail.

• Для каждого узла xсписка, для которого x!=tail, x->dataуказывает на экземпляр T, a x->next— на следующий узел списка. Если x->next==tail, то x — последний узел списка.

• Если проследовать по указателям next, начиная с головы списка, то рано или поздно мы достигнем его хвоста.

Сама по себе, функция push()очень проста: все модификации данных защищены мьютексом tail_mutex, и инвариант при этом сохраняется, потому что новый хвостовой узел пуст и правильно установлены указатели dataи nextдля старого хвостового узла, который теперь стал настоящим последним узлом списка.

Самое интересное происходит в функции try_pop(). Как выясняется, мьютекс tail_mutexнужен не только для защиты чтения самого указателя tail, но и чтобы предотвратить гонку при чтении данных из головного узла. Не будь этого мьютекса, могло бы получиться, что один поток вызывает try_pop(), а другой одновременно вызывает push(), и эти операции никак не упорядочиваются. Хотя каждая функция-член удерживает мьютекс, но это разные мьютексы , а функции могут обращаться к одним и тем же данным — ведь все данные появляются в очереди только благодаря push(). Раз потоки потенциально могут обращаться к одним и тем же данным без какого бы то ни было упорядочения, то возможна гонка за данными и, как следствие (см. главу 5), неопределенное поведение. К счастью, блокировка мьютекса tail_mutexв get_tail()решает все проблемы. Поскольку внутри get_tail()захватывается тот же мьютекс, что в push(), то оба вызова оказываются упорядоченными. Либо обращение к функции get_tail() происходит раньше обращения к push()— тогда get_tail()увидит старое значение tail— либо после обращения к push()— и тогда она увидит новое значение tail и новые данные, присоединенные к прежнему значению tail .