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

Мы хотим обеспечить максимум возможностей для распараллеливания, поэтому блокировки должны освобождаться как можно быстрее. С функцией 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 .

Важно также, что обращение к get_tail()производится под защитой захваченного мьютекса head_mutex. Если бы это было не так, то между вызовом get_tail()и захватом head_mutexмог бы вклиниться вызов pop_head(), потому что другой поток вызвал try_pop()(и, следовательно, pop_head()) и захватил мьютекс первым, не давая первому потоку продолжить исполнение:

│ Эта реализация

std: :unique_ptr pop_head()←┘ некорректна

{ (1) Старое значение tail

│ получено не

node* const old_tail = get_tail();←┘ под защитой head_mutex

std::lock_guard head_lock(head_mutex);

if (head.get() == old_tail) ← (2)

return nullptr;

}

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

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

return old_head;

}

При такой — некорректной — реализации в случае, когда get_tail() (1)вызывается вне области действия блокировки, может оказаться, что и head, и tailизменились к моменту, когда первому потоку удалось захватить head_mutex, и теперь возвращенный хвост мало того что больше не является хвостом, но и вообще не принадлежит списку. Тогда сравнение headс old_tail (2)не прошло бы, хотя headв действительности является последним узлом. Следовательно, после обновления (3)узел headмог бы оказаться в списке дальше tail, то есть за концом списка, что полностью разрушило бы структуру данных. В корректной реализации, показанной в листинге 6.6, вызов get_tail()производится под защитой head_mutex. Это означает, что больше никакой поток не сможет изменить head, a tailбудет только отодвигаться от начала списка (по мере добавления новых узлов с помощью push()) — это вполне безопасно. Указатель headникогда не сможет оказаться дальше значения, возвращенного get_tail(), так что инварианты соблюдаются.