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

std::unique_lock lk(mut);

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

value = data_queue.front();

data_queue.pop();

}

};

threadsafe_queue data_queue; ← (1)

void data_preparation_thread() {

while (more_data_to_prepare()) {

data_chunk const data = prepare_data();

data_queue.push(data); ← (2)

}

}

void data_processing_thread() {

while (true) {

data_chunk data;

data_queue.wait_and_pop(data); ← (3)

process(data);

if (is_last_chunk(data))

break;

}

}

Теперь мьютекс и условная переменная находятся в экземпляре threadsafe_queue, поэтому не нужно ни отдельных переменных (1), ни внешней синхронизации при обращении к функции push() (2). Кроме того, wait_and_pop()берет на себя заботу об ожидании условной переменной (3).

Второй перегруженный вариант wait_and_pop()тривиален, а остальные функции можно почти без изменений скопировать из кода стека в листинге 3.5. Ниже приведена окончательная реализация.

Листинг 4.5. Полное определение класса потокобезопасной очереди на базе условных переменных

#include

#include

#include

#include

template

class threadsafe_queue {

private:

mutable std::mutex mut;← (1) Мьютекс должен быть изменяемым

std::queue data_queue;

std::condition_variable data_cond;

public:

threadsafe_queue() {}

threadsafe_queue(threadsafe_queue const& other) {

std::lock_guard lk(other.mut);

data_queue = other.data_queue;

}

void push(T new_value) {

std::lock_guard lk(mut);

data_queue.push(new_value);

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 = 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(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 = 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(data_queue.front()));

data_queue.pop();

return res;

}

bool empty() const {

std::lock_guard lk(mut);

return data_queue.empty();

}

};

Хотя empty()— константная функция-член, а параметр копирующего конструктора — const-ссылка, другие потоки могут хранить неконстантные ссылки на объект и вызывать изменяющие функции-члены, которые захватывают мьютекс. Поэтому захват мьютекса — это изменяющая операция, следовательно, член mutнеобходимо пометить как mutable (1), чтобы его можно было захватить в функции empty()и в копирующем конструкторе.

Условные переменные полезны и тогда, когда есть несколько потоков, ожидающих одного события. Если потоки используются для разделения работы и, следовательно, на извещение должен реагировать только один поток, то применима точно такая же структура программы, как в листинге 4.1; нужно только запустить несколько потоков обработки данных. При поступлении новых данных функция notify_one()разбудит только один поток, который проверяет условие внутри wait(), и этот единственный поток вернет управление из wait()(в ответ на помещение нового элемента в очередь data_queue). Заранее нельзя сказать, какой поток получит извещение и есть ли вообще ожидающие потоки (не исключено, что все они заняты обработкой ранее поступивших данных).

Альтернативный сценарий — когда несколько потоков ожидают одного события, и отреагировать должны все. Так бывает, например, когда инициализируются разделяемые данные, и все работающие с ними потоки должны ждать, пока инициализация завершится (хотя для этого случая существуют более подходящие механизмы, см. раздел 3.3.1 главы 3), или когда потоки должны ждать обновления разделяемых данных, например, в случае периодической повторной инициализации. В таких ситуациях поток, отвечающий за подготовку данных, может вызвать функцию-член notify_all()условной переменной вместо notify_one(). Эта функция извещает все потоки, ожидающие внутри функции wait(), о том, что они должны проверить ожидаемое условие.

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