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

На простейшем уровне это означает, что нужно спроектировать структуру данных, к которой смогут одновременно обращаться несколько потоков для выполнения одних и тех же или разных операций, причём каждый поток должен видеть согласованное состояние структуры. Данные не должны теряться или искажаться, все инварианты должны быть соблюдены, и никаких проблематичных состояний гонки не должно быть. Такая структура данных называется потокобезопасной. В общем случае структура данных безопасна только относительно определенных типов параллельного доступа. Не исключено, что несколько потоков могут одновременно выполнять какую-то одну операцию над структурой, а для выполнения другой необходим монопольный доступ. Наоборот, бывает, что несколько потоков могут одновременно и безопасно выполнять различные операции, но при выполнении одной и той же операции в разных потоках возникают проблемы.

Но проектирование с учетом параллелизма этим не исчерпывается: задача заключается в том, чтобы предоставить возможность распараллеливания потокам, обращающимся к структуре данных. По природе своей, мьютекс означает взаимное исключение: в каждый момент времени только один поток может захватить мьютекс. Следовательно, мьютекс защищает структуру данных, явным образом предотвращая истинно параллельный доступ к ней.

Это называется сериализацией: потоки обращаются к защищенным мьютексом данным по очереди, то есть последовательно, а не параллельно. Чтобы обеспечить истинно параллельный доступ, нужно тщательно продумывать внутреннее устройство структуры данных. Одни структуры данных оставляют больший простор для распараллеливания, чем другие, но идея остается неизменной: чем меньше защищаемая область, тем меньше операций приходится сериализовывать и тем больше потенциал для распараллеливания.

Прежде чем знакомиться с конкретными структурами данных, приведём несколько простых рекомендаций, полезных при проектировании с учетом параллелизма.

6.1.1. Рекомендации по проектированию структур данных для параллельного доступа

Как я уже отмечал, при проектировании структур данных для параллельного доступа нужно учитывать два аспекта: обеспечить безопасность доступа и разрешить истинно параллельный доступ. Как сделать структуру потокобезопасной, я уже рассказывал в главе 3.

• Гарантировать, что ни один поток не может увидеть состояние, в котором инварианты структуры данных нарушены действиями со стороны других потоков.

• Позаботиться о предотвращении состояний гонки, внутренне присущих структуре данных, предоставив такие функции, которые выполняли бы операции целиком, а не частями.

• Обращать внимание на том, как ведет себя структура данных при наличии исключений, — не допускать нарушения инвариантов и в этом случае.

• Минимизировать шансы возникновения взаимоблокировки, ограничивая область действия блокировок и избегая но возможности вложенных блокировок.

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

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

swap()

Второй аспект, нуждающийся в рассмотрении, — обеспечение истинно параллельного доступа. Тут я не могу предложить конкретных рекомендаций, а вместо этого перечислю несколько вопросов, которые должен задать себе проектировщик структуры данных.

• Можно ли ограничить область действия блокировок, так чтобы некоторые части операции выполнялись не под защитой блокировки?

• Можно ли защитить разные части структуры данных разными мьютексами?

• Все ли операции нуждаются в одинаковом уровне защиты?

• Можно ли с помощью простого изменения структуры данных расширить возможности распараллеливания, не затрагивая семантику операций?

В основе всех этих вопросов лежит одна и та же мысль: как свести к минимуму необходимую сериализацию и обеспечить максимально возможную степень истинного параллелизма? Часто бывает так, что структура данных допускает одновременный доступ из нескольких потоков для чтения, но поток, желающий модифицировать данные, должен получать монопольный доступ. Такое требование поддерживает класс

boost::shared_mutex

В простейших потокобезопасных структурах данных обычно для защиты используются мьютексы и блокировки. Хотя, как мы видели в главе 3, им свойственны некоторые проблемы, но гарантировать с их помощью, что в каждый момент времени доступ к данным будет иметь только один поток, сравнительно легко. Мы будем знакомиться с проектированием потокобезопасных структур данных постепенно, и в этой главе рассмотрим только структуры на основе блокировок. А разговор о параллельных структурах данных без блокировок отложим до главы 7.

Проектирование параллельных структур данных с блокировками сводится к тому, чтобы захватить нужный мьютекс при доступе к данным и удерживать его минимально возможное время. Это довольно сложно, даже когда имеется только один мьютекс, защищающий всю структуру. Как мы видели в главе 3, требуется гарантировать, что к данным невозможно обратиться без защиты со стороны мьютекса и что интерфейс свободен от внутренне присущих состояний гонки. Если для защиты отдельных частей структуры применяются разные мьютексы, то проблема еще усложняется, поскольку в случае, когда некоторые операции требуют захвата нескольких мьютексов, появляется возможность взаимоблокировки. Поэтому к проектированию структуры данных с несколькими мьютексами следует подходить еще более внимательно, чем при наличии единственного мьютекса. В этом разделе мы применим рекомендации из раздела 6.1.1 к проектированию нескольких простых структур данных, защищаемых мьютексами. В каждом случае мы будем искать возможности повысить уровень параллелизма, обеспечивая в то же время потокобезопасность.

Начнем с реализации стека, приведённой в главе 3; это одна из самых простых структур данных, к тому же в ней используется всего один мьютекс. Но является ли она потокобезопасной? И насколько она хороша с точки зрения достижения истинного распараллеливания?

6.2.1. Потокобезопасный стек с блокировками

В следующем листинге воспроизведен код потокобезопасного стека из главы 3. Задача состояла в том, чтобы реализовать потокобезопасную структуру данных наподобие

std::stack<>

Листинг 6.1. Определение класса потокобезопасного стека

#include 

struct empty_stack: std::exception {

 const char* what() const throw();

};

template

class threadsafe_stack {

private:

 std::stack data;

 mutable std::mutex m;

public:

 threadsafe_stack(){}

 threadsafe_stack(const threadsafe_stack& other) {

 std::lock_guard lock(other.m);

  data = other.data;

 }

 threadsafe_stack& operator=(const threadsafe_stack&) = delete;

 void push(T new_value) {

  std::lock_guard lock(m);

  data.push(std::move(new_value)); ←

 }

 std::shared_ptr pop() {

  std::lock_guard lock(m);

  if (data.empty()) throw empty_stack(); ←

  std::shared_ptr const res(

  std::make_shared(std::move(data.top())));←

  data.pop(); ←

  return res;

 }

 void pop(T& value) {

  std::lock_guard lock(m);

  if (data.empty()) throw empty_stack();

  value = std::move(data.top()); ←

  data.pop(); ←

 }

 bool empty() const {

  std::lock_guard lock(m);

  return data.empty();

 }

};

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

m

Во-вторых, существует потенциальная гонка между

empty()

pop()

pop()

pop()

top()

pop()

std::stack<>

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

std::lock_guard<>

Вызов

data.push()

std::stack<>

В первом перегруженном варианте

pop()

empty_stack

res

std::make_shared

data.pop()

pop()

Второй перегруженный вариант

pop()

std::shared_ptr

data.pop()

Наконец, функция

empty()

В этом коде есть две возможности для взаимоблокировки из-за того, что пользовательский код вызывается, когда удерживается блокировка: копирующий или перемещающий конструктор (1), (3) и копирующий или перемещающий оператор присваивания (5) хранимых в стеке данных. И еще —

operator new

Поскольку все функции-члены используют для защиты данных класс

std::lock_guard<>

Хотя благодаря блокировке несколько потоков могут одновременно вызывать функции-члены стека, в каждый момент времени с ним реально работает не более одного потока. Такая сериализация потоков может снизить производительность приложения в случае, когда имеется значительная конкуренция за стек, — пока поток ожидает освобождения блокировки, он не выполняет никакой полезной работы. Кроме того, стек не предоставляет средств, позволяющих ожидать добавления элемента. Следовательно, если потоку нужно получить из стека элемент, то он должен периодически опрашивать его с помощью

empty()

pop()

empty_stack

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

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

В листинге 6.2 воспроизведен код потокобезопасной очереди из главы 4. Если стек построен по образцу

std::stack<>

std::queue<>

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

template

class threadsafe_queue {

private:

 mutable std::mutex mut;

 std::queue data_queue;

 std::condition_variable data_cond;

public:

 threadsafe_queue() {}

 void push(T new_value) {

  std::lock_guard lk(mut);

  data_queue.push(std::move(data));

  data_cond.notify_one(); ←

 }

 void wait_and_pop(T& value) { ←

  std::unique_lock lk(mut);

  data_cond.wait(lk, [this]{return !data_queue.empty();});

  value = std::move(data_queue.front());

  data_queue.pop();

 }

 std::shared_ptr wait_and_pop() ←

  std::unique_lock lk(mut);

  data_cond.wait(lk, [this] {return !data_queue.empty();});←

  std::shared_ptr res(

  std::make_shared(std::move(data_queue.front())));

  data_queue.pop();

  return res;

 }

 bool try_pop(T& value) {

  std::lock_guard lk(mut);

  if (data_queue.empty())

  return false;

  value = std::move(data_queue.front());

  data_queue.pop();

  return true;

 }

 std::shared_ptr try_pop() {

  std::lock_guard lk(mut);

  if (data_queue.empty())

  return std::shared_ptr(); ←

  std::shared_ptr res(

  std::make_shared(std::move(data_queue.front())));

  data_queue.pop();

  return res;

 }

 bool empty() const {

  std::lock_guard lk(mut);

  return data_queue.empty();

 }

};

Структурно очередь в листинге 6.2 реализована аналогично стеку в листинге 6.1, отличие только в обращениях к функции

data_cond.notify_one()

push()

wait_and_pop()

try_pop()

pop()

NULL

wait_and_pop()

Новые функции

wait_and_pop()

empty()

wait_and_pop()

data_cond.wait()

wait_and_pop()

В части безопасности относительно исключений есть мелкая неприятность — если помещения данных в очередь ожидают несколько потоков, то лишь один из них будет разбужен в результате вызова

data_cond.notify_one()

wait_and_pop()

std::shared_ptr<>

notify_one()

data_cond.notify_all()

wait_and_pop()

notify_one()

std::shared_ptr<>

push()

std::shared_ptr<>

std::shared_ptr<>

std::queue<>

wait_and_pop()

Листинг 6.3. Потокобезопасная очередь, в которой хранятся объекты

std::shared_ptr

template

class threadsafe_queue {

private:

 mutable std::mutex mut;

 std::queue > data_queue;

 std::condition_variable data_cond;

public:

 threadsafe_queue() {}

 void wait_and_pop(T& value) {

  std::unique_lock lk(mut);

  data_cond.wait(lk, [this]{return !data_queue.empty();});

  value = std::move(*data_queue.front()); ←

  data_queue.pop();

 }

 bool try_pop(T& value) {

  std::lock_guard lk(mut);

  if (data_queue.empty())

  return false;

  value = std::move(*data_queue.front()); ←

  data_queue.pop();

  return true;

 }

 std::shared_ptr wait_and_pop() {

  std::unique_lock lk(mut);

  data_cond.wait(lk, [this]{return !data_queue.empty();});

  std::shared_ptr res = data_queue.front(); ←

  data_queue.pop();

  return res;

 }

 std::shared_ptr try_pop() {

  std::lock_guard lk(mut);

  if (data_queue.empty())

  return std::shared_ptr();

  std::shared_ptr res = data_queue.front(); ←

  data_queue.pop();

  return res;

 }

 void push(T new_value) {

  std::shared_ptr data(

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

  std::lock_guard lk(mut);

  data_queue.push(data);

  data_cond.notify_one();

 }

 bool empty() const {

  std::lock_guard lk(mut);

  return data_queue.empty();

 }

};

Последствия хранения данных, обернутых в

std::shared_ptr<>

pop

std::shared_ptr<>

У хранения данных в виде

std::shared_ptr<>

push()

pop()

Как и в примере стека, применение мьютекса для защиты всей структуры данных ограничивает возможности распараллеливания работы с очередью; хотя ожидать доступа к очереди могут несколько потоков, выполняющих разные функции, в каждый момент лишь один совершает какие-то действия. Однако это ограничение отчасти проистекает из того, что мы пользуемся классом

std::queue<>

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

В листингах 6.2 и 6.3 имеется только один защищаемый элемент данных (

data_queue

Проще всего реализовать очередь в виде односвязного списка, как показано на рис. 6.1. Указатель head направлен на первый элемент списка, и каждый элемент указывает на следующий. Когда данные извлекаются из очереди, в head записывается указатель на следующий элемент, после чего возвращается элемент, который до этого был в начале.

Добавление данных производится с другого конца. Для этого нам необходим указатель tail, направленный на последний элемент списка. Чтобы добавить узел, мы записываем в поле next в последнем элементе указатель на новый узел, после чего изменяем указатель tail, так чтобы он адресовал новый элемент. Если список пуст, то оба указателя head и tail равны

NULL

В следующем листинге показана простая реализация такой очереди с урезанным по сравнению с листингом 6.2 интерфейсом; мы оставили только функцию

try_pop()

wait_and_pop()

Рис. 6.1. Очередь, представленная в виде односвязного списка

Листинг 6.4. Простая реализация однопоточной очереди

template

class queue {

private:

 struct node {

  T data;

  std::unique_ptr next;

  node(T data_):

  data(std::move(data_)) {}

 };

 std::unique_ptr head;←

 node* tail;        ←

public:

 queue() {}

 queue(const queue& other) = delete;

 queue& operator=(const queue& other) = delete;

 std::shared_ptr try_pop() {

  if (!head) {

  return std::shared_ptr();

  }

  std::shared_ptr const res(

  std::make_shared(std::move(head->data)));

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

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

  return res;

 }

 void push(T new_value) {

  std::unique_ptr p(new node(std::move(new_value)));

  node* const new_tail = p.get();

  if (tail) {

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

  } else {

  head = std::move(p);    ←

  }

  tail = new_tail;      ←

}

};

Прежде всего отметим, что в листинге 6.4 для управления узлами используется класс

std::unique_ptr

delete

head

tail

В однопоточном контексте эта реализация прекрасно работает, но при попытке ввести мелкогранулярные блокировки в многопоточном контексте возникают две проблемы. Учитывая наличие двух элементов данных (

head

tail

head

tail

Самая очевидная проблема заключается в том, что

push()

head

tail

push()

pop()

next

push()

tail->next

try_pop()

head->next

head==tail

head->next

tail->next

head

tail

push()

try_pop()

Решить проблему можно, заранее выделив фиктивный узел, не содержащий данных, и тем самым гарантировать, что в очереди всегда есть хотя бы один узел, отделяющий голову от хвоста. В случае пустой очереди

head

tail

NULL

try_pop()

head->next

head

tail

head->next

tail->next

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

template

class queue {

private:

 struct node {

  std::shared_ptr data;←

  std::unique_ptr next;

 };

 std::unique_ptr head;

 node* tail;

public:

 queue():

 head(new node), tail(head.get())←

 {}

 queue(const queue& other) = delete;

 queue& operator=(const queue& other) = delete;

 std::shared_ptr try_pop() {

  if (head.get() ==tail) ←

  {

  return std::shared_ptr();

 }

  std::shared_ptr const res(head->data);←

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

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

  return res; ←

 }

 void push(T new_value) {

                std::shared_ptr new_data(
            

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

  std::unique_ptr p(new node); ←

  tail->data = new_data; ←

  node* const new_tail = p.get();

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

  tail = new_tail;

 }

};

Изменения в

try_pop()

head

tail

NULL

head

NULL

head

std::unique_ptr

head.get()

node

T

push()

T

std::shared_ptr<>

std::make_shared

new_value

new_value

Уверен, теперь вы задаетесь вопросом, что мы выиграли от всех этих изменений и как они помогут сделать код потокобезопасным. Разберемся. Функция

push()

tail

head

try_pop()

head

tail

tail

try_pop()

push()

head

tail

Мы хотим обеспечить максимум возможностей для распараллеливания, поэтому блокировки должны освобождаться как можно быстрее. С функцией

push()

tail

С

try_pop()

head

head

head

tail

tail

head

Листинг 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

x->next

x->next==tail

x

• Если проследовать по указателям

next

Сама по себе, функция

push()

tail_mutex

data

next

Самое интересное происходит в функции

try_pop()

tail_mutex

tail

try_pop()

push()

push()

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()←┘

{                  

                   │

 node* const old_tail = get_tail();←┘

 std::lock_guard head_lock(head_mutex);

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

  return nullptr;

 }

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

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

 return old_head;

}

При такой — некорректной — реализации в случае, когда

get_tail()

head

tail

head_mutex

head

old_tail

head

head

tail

get_tail()

head_mutex

head

tail

push()

head

get_tail()

После того как

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()

Ответ, разумеется, — да, только вот как это сделать? Модифицировать

push()

data_cond.notify_one()

notify_one()

Функция

wait_and_pop()

head != tail

head_mutex

tail_mutex

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

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;

 std::condition_variable data_cond;

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();

 bool try_pop(T& value);

 std::shared_ptr wait_and_pop();

 void wait_and_pop(T& value);

 void push(T new_value);

 void empty();

};

Код, помещающий новые узлы в очередь, прост — его реализация (показанная в листинге ниже) близка к той, что мы видели раньше.

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

template

void threadsafe_queue::push(T new_value) {

 std::shared_ptr new_data(

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

 std::unique_ptr p(new node);

 {

  std::lock_guard tail_lock(tail_mutex);

  tail->data = new_data;

  node* const new_tail = p.get();

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

  tail = new_tail;

 }

 data_cond.notify_one();

}

Как уже отмечалось, вся сложность сосредоточена в части pop. В листинге ниже показана реализация функции-члена

wait_and_pop()

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

wait_and_pop

template

class threadsafe_queue {

private:

 node* get_tail() {

  std::lock_guard tail_lock(tail_mutex);

  return tail;

 }

 std::unique_ptr pop_head() {←

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

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

  return old_head;

 }

 std::unique_lock wait_for_data() {←

  std::unique_lock head_lock(head_mutex);

  data_cond.wait(

  head_lock, [&]{return head.get() != get_tail();});

  return std::move(head_lock);         ←

 }

 std::unique_ptr wait_pop_head() {

  std::unique_lock head_lock(wait_for_data());←

  return pop_head();

 }

 std::unique_ptr wait_pop_head(T& value) {

  std::unique_lock head_lock(wait_for_data());←

  value = std::move(*head->data);

  return pop_head();

 }

public:

 std::shared_ptr wait_and_pop() {

  std::unique_ptr const old_head = wait_pop_head();

  return old_head->data;

 }

 void wait_and_pop(T& value) {

  std::unique_ptr const old_head = wait_pop_head(value);

 }

};

В реализации извлечения из очереди используется несколько небольших вспомогательных функций, которые упрощают код и позволяют устранить дублирование, например:

pop_head()

wait_for_data()

wait_for_data()

wait_pop_head()

pop_head()

try_pop()

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

try_pop()

empty()

template

class threadsafe_queue {

private:

 std::unique_ptr try_pop_head() {

  std::lock_guard head_lock(head_mutex);

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

  return std::unique_ptr();

  }

  return pop_head();

 }

 std::unique_ptr try_pop_head(T& value) {

  std::lock_guard head_lock(head_mutex);

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

  return std::unique_ptr();

  }

  value = std::move(*head->data);

  return pop_head();

 }

public:

 std::shared_ptr try_pop() {

  std::unique_ptr old_head = try_pop_head();

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

 }

 bool try_pop(T& value) {

  std::unique_ptr const old_head = try_pop_head(value);

  return old_head;

 }

 void empty() {

  std::lock_guard head_lock(head_mutex);

  return (head.get() == get_tail());

 }

};

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

Показанную реализацию неограниченной очереди легко преобразовать в очередь с ограниченной длиной, введя ожидание условной переменной в функцию

push()

pop()

Стеки и очереди — простые структуры данных, их интерфейс крайне ограничен, и используются они в очень узких целях. Но не все структуры данных так просты, как правило они поддерживают более широкий набор операций. В принципе, это может открыть больше возможностей для распараллеливания, но и проблема защиты данных сильно усложняется, так как приходится учитывать много разных способов доступа. При проектировании структур данных, предназначенных для параллельного доступа, важно принимать во внимание характер выполняемых операций.

Чтобы понять, какие трудности возникают на этом пути, рассмотрим справочную таблицу (lookup table).

6.3.1. Разработка потокобезопасной справочной таблицы с блокировками

Справочная таблица, или словарь позволяет ассоциировать значения одного типа (типа ключа) со значениями того же или другого типа (ассоциированного типа). В общем случае задача такой структуры — запрашивать данные, ассоциированные с данным ключом. В стандартной библиотеке С++ для этой цели предназначены ассоциативные контейнеры:

std::map<>

std::multimap<>

std::unordered_map<>

std::unordered_multimap<>

Справочная таблица используется иначе, чем стек или очередь. Если большинство операций со стеком и очередью каким-то образом модифицируют структуру данных, то есть либо добавляют, либо удаляют элементы, то справочная таблица изменяется сравнительно редко. Примером может служить простой DNS-кэш из листинга 3.13, предоставляющий интерфейс, сильно урезанный по сравнению с

std::map<>

Самая серьезная с точки зрения распараллеливания проблема в интерфейсе контейнера

std::map<>

std::map<>

Справочной таблице нужно всего несколько основных операций:

• добавить новую пару ключ/значение;

• изменить значение, ассоциированное с данным ключом;

• удалить ключ и ассоциированное с ним значение;

• получить значение, ассоциированное с данным ключом, если такой ключ существует.

Есть также несколько полезных операций, относящихся к контейнеру в целом, например: проверить, пуст ли контейнер, получить полный текущий список ключей или полное множество пар ключ/ значение.

Если придерживаться базовых рекомендаций, касающихся потокобезопасности, например, не возвращать ссылки и защищать мьютексом все тело каждой функции-члена, то все эти операции будут безопасны. Угроза возникновения гонки возникает прежде всего при добавлении новой пары ключ/значение; если два потока одновременно добавляют новое значение, то лишь один из них будет первым, а второй, следовательно, получит ошибку. Один из способов решить эту проблему состоит в том, чтобы объединить операции добавления и изменения в одной функции-члене, как мы проделали в DNS-кэше в листинге 3.13.

С точки зрения интерфейса, остается еще лишь один интересный момент: фраза «если такой ключ существует» в описании операции получения ассоциированного значения. Можно, например, предоставить пользователю возможность задать некий результат «по умолчанию», который будет возвращен, если указанного ключа нет.

mapped_type get_value(

 key_type const& key, mapped_type default_value);

В таком случае можно использовать сконструированный по умолчанию экземпляр типа

mapped_type

default_value

mapped_type

std::pair

bool

NULL

Как уже отмечалось, после определения интерфейса (в предположении, что состояний гонки в нем нет), обеспечить потокобезопасность можно, заведя единственный мьютекс, который дет захватываться в начале каждой функции-члена для защиты внутренней структуры данных. Однако тем самым мы сведем на нет все возможности распараллеливания, которые могли бы открыться в результате наличия отдельных функций для чтения и модификации структуры данных. Отчасти исправить ситуацию можно было бы, воспользовавшись мьютексом, который разрешает нескольким потокам читать данные, но только одному — изменять их, например, классом

boost::shared_mutex

Как и в случае очереди (см. раздел 6.2.3), чтобы ввести мелкогранулярные блокировки, нужно внимательно изучить особенности структуры данных, а не пытаться обернуть уже имеющийся контейнер, например

std::map<>

• двоичное, например красно-черное, дерево;

• отсортированный массив;

• хеш-таблица.

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

Отсортированный массив еще хуже, потому что заранее невозможно сказать, в каком месте массива может оказаться требуемое значение, поэтому придется заводить одну блокировку на весь массив. Остается только хеш-таблица. В предположении, что число кластеров фиксировано, вопрос о том, в каком кластере находится ключ, является свойством только лишь самого ключа и хеш-функции. А это значит, что мы сможем завести по одной блокировке на каждый кластер. Если еще и использовать мьютекс, который поддерживает несколько читателей и одного писателя, то коэффициент распараллеливания увеличится в N раз, где N — число кластеров. Недостаток в том, что нужна хорошая хеш-функция для ключа. В стандартной библиотеке С++ имеется шаблон

std::hash<>

int

std::string

std::hash<>

Теперь обратимся собственно к коду. Как могла бы выглядеть реализация потокобезопасной справочной таблицы? Один из вариантов показан ниже.

Листинг 6.11. Потокобезопасная справочная таблица

template

      typename Hash = std::hash >

class threadsafe_lookup_table {

private:

 class bucket_type {

 private:

  typedef std::pair bucket_value;

  typedef std::list bucket_data;

  typedef typename bucket_data::iterator bucket_iterator;

  bucket_data data;

  mutable boost::shared_mutex mutex;←

  bucket_iterator find_entry_for(Key const& key) const {←

  return std::find_if(data.begin(), data.end(),

            [&](bucket_value const& item)

  { return item.first == key; });

  }

 public:

  Value value_for(

  Key const& key, Value const& default_value) const {

  boost::shared_lock lock(mutex);←

  bucket_iterator const found_entry = find_entry_for(key);

  return (found_entry==data.end()) ?

      default_value : found_entry->second;

  }

  void add_or_update_mapping(

  Key const& key, Value const& value) {

  std::unique_lock lock(mutex);←

  bucket_iterator const found_entry = find_entry_for(key);

  if (found_entry == data.end()) {

   data.push_back(bucket_value(key, value));

  } else {

   found_entry->second = value;

  }

  }

 void remove_mapping(Key const& key) {

  std::unique_lock lock(mutex);←

  bucket_iterator const found_entry = find_entry_for(key);

  if (found_entry != data.end()) {

   data.erase(found_entry);

  }

  }

 };

 std::vector> buckets;←

 Hash hasher;

 bucket_type& get_bucket(Key const& key) const {← 

  std::size_t const bucket_index = hasher(key)%buckets.size();

  return *buckets[bucket_index];

 }

public:

 typedef Key key_type;

 typedef Value mapped_type;

 typedef Hash hash_type;

 threadsafe_lookup_table(

  unsigned num_buckets = 19,

  Hash const& hasher_ = Hash()):

 buckets(num_buckets), hasher(hasher_) {

  for (unsigned i = 0; i < num_buckets; ++i) {

  buckets[i].reset(new bucket_type);

  }

 }

 threadsafe_lookup_table(

  threadsafe_lookup_table const& other) = delete;

 threadsafe_lookup_table& operator=(

  threadsafe_lookup_table const& other) = delete;

 Value value_for(Key const& key,

  Value const& default_value = Value()) const {

  return get_bucket(key).value_for(key, default_value);←

 }

 void add_or_update_mapping(Key const& key,Value const& value) {

  get_bucket(key).add_or_update_mapping(key, value);←

 }

 void remove_mapping(Key const& key) {

  get_bucket(key).remove_mapping(key);←

 }

};

std::vector>

boost::shared_mutex

get_bucket()

find_entry_for()

std::list<>

value_for

remove_mapping

erase

add_or_update_mapping

if

push_back

std::map<>

threadsafe_lookup_table

std::map<>

std::map threadsafe_lookup_table::get_map() const {

 std::vector > locks;

 for (unsigned i = 0; i < buckets.size(); ++i) {

  locks.push_back(

  std::unique_lock(buckets[i].mutex));

 }

 std::map res;

 for (unsigned i = 0; i < buckets.size(); ++i) {

  for (bucket_iterator it = buckets[i].data.begin();

  it != buckets[i].data.end(); ++it) {

  res.insert(*it);

  }

 }

 return res;

}

boost::shared_mutex