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

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() {← (1)

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

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

return old_head;

}

std::unique_lock wait_for_data() {← (2)

std::unique_lock head_lock(head_mutex);

data_cond.wait(

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

return std::move(head_lock); ← (3)

}

std::unique_ptr wait_pop_head() {

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

return pop_head();

}

std::unique_ptr wait_pop_head(T& value) {

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

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() (1)(модификация списка в результате удаления головного элемента) и wait_for_data() (2)(ожидание появления данных в очереди). Особенно стоит отметить функцию wait_for_data(), потому что она не только ждет условную переменную, используя лямбда-функцию в качестве предиката, но и возвращает объект блокировки вызывающей программе (3). Тем самым мы гарантируем, что та же самая блокировка удерживается, пока данные модифицируются в соответствующем перегруженном варианте wait_pop_head() (4), (5). Функция 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).

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

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