Параллельное программирование на С++ в действии полностью

Изменения в try_pop() минимальны. Во-первых, мы сравниваем head с tail (3), а не с NULL, потому что благодаря наличию фиктивного узла head никогда не может обратиться в NULL. Поскольку head имеет тип std::unique_ptr, для сравнения необходимо вызывать head.get(). Во-вторых, так как в node теперь хранится указатель на данные (1), то можно извлекать указатель непосредственно (4) без конструирования нового экземпляра T. Наиболее серьезные изменения претерпела функция push(): мы должны сначала создать новый экземпляр T в куче и передать владение им std::shared_ptr<> (7) (обратите внимание на использование функции std::make_shared, чтобы избежать накладных расходов на второе выделение памяти под счетчик ссылок). Вновь созданный узел станет новым фиктивным узлом, поэтому передавать конструктору значение new_value необязательно (8). Вместо этого мы записываем в старый фиктивный узел значение только что созданной копии — new_value (9). Наконец, первоначальный фиктивный узел следует создать в конструкторе (2).

Уверен, теперь вы задаетесь вопросом, что мы выиграли от всех этих изменений и как они помогут сделать код потокобезопасным. Разберемся. Функция 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();