C++17 STL Стандартная библиотека шаблонов

В этой главе:

□ автоматическое распараллеливание кода, использующего стандартные алгоритмы;

□ приостановка программы на конкретный промежуток времени;

□ запуск и приостановка потоков;

□ выполнение устойчивой к исключениям общей блокировки с помощью

std::unique_lock

std::shared_lock

□ избегание взаимных блокировок с применением

std::scoped_lock

□ синхронизация конкурентного использования

std::cout

□ безопасное откладывание инициализации с помощью

std::call_once

□ отправка выполнения задач в фоновый режим с применением

std::async

□ реализация идиомы «производитель/потребитель» с использованием

std::condition_variable

□ реализация идиомы «несколько потребителей/производителей» с помощью

std::condition_variable

□ распараллеливание отрисовщика множества Мандельброта в ASCII с применением

std::a

□ реализация небольшой автоматической библиотеки для распараллеливания с использованием

std::future

Введение

До C++11 язык C++ не поддерживал параллельные вычисления. Это не значило, что запуск, управление, остановка и синхронизация потоков были невыполнимы, но для каждой операционной системы требовались специальные библиотеки, поскольку потоки по своей природе связаны с ОС.

С появлением C++11 мы получили библиотеку

std::thread

std::condition_variable

Кроме того, интересными дополнениями являются

std::async

std::future

std::async

std::future

В данной главе мы пройдемся по всем указанным дополнениям, чтобы узнать их самые важные особенности. После этого у нас будет достаточно информации о поддержке параллелизации в STL версии C++17. Мы не станем рассматривать все свойства, только самые важные. Информация, полученная из этой книги, позволит быстро понять остальную часть механизмов распараллеливания, которую можно найти в Интернете в документации к STL версии C++17.

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

Автоматическое распараллеливание кода, использующего стандартные алгоритмы

В C++17 появилось одно действительно крупное расширение для параллелизма: политики выполнения для стандартных алгоритмов. Шестьдесят девять алгоритмов были расширены и теперь принимают политики выполнения, чтобы работать параллельно на нескольких ядрах и даже при включенной векторизации.

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

В данном разделе мы реализуем простую программу (с не самым серьезным сценарием применения), которая генерирует несколько вызовов алгоритмов STL. При этом увидим, как легко использовать политики выполнения C++17, чтобы запустить их в нескольких потоках. В последних подразделах мы более подробно рассмотрим разные политики выполнения.

Как это делается

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

1. Сначала включим некоторые заголовочные файлы и объявим об использовании пространства имен

std

execution

#include 

#include 

#include 

#include 

#include 

using namespace std;

2. В качестве примера создадим функцию-предикат, которая говорит, является ли число четным. Мы воспользуемся ею далее.

static bool odd(int n) { return n%2; }

3. Сначала определим в нашей функции

main

int main()

{

  vector d (50000000);

4. Чтобы получить большое количество случайных данных для вектора, создадим генератор случайных чисел и распределение и упакуем их в вызываемый объект. Если это кажется странным, пожалуйста, взгляните сначала на примеры, в которых рассматриваются генераторы случайных чисел и распределения в главе 8.

  mt19937 gen;

  uniform_int_distribution dis(0, 100000);

  auto rand_num ([=] () mutable { return dis(gen); });

5. Теперь воспользуемся стандартным алгоритмом

std::generate

std::par

  generate(execution::par, begin(d), end(d), rand_num);

6. Метод

std::sort

  sort(execution::par, begin(d), end(d));

7. Это верно и для

std::reverse

  reverse(execution::par, begin(d), end(d));

8. Затем воспользуемся функцией

std::count_if

  auto odds (count_if(execution::par, begin(d), end(d), odd));

9. Вся эта программа не выполняет никакой реальной научной работы, поскольку мы просто смотрим, как можно распараллелить стандартные алгоритмы, но выведем что-нибудь на экран в конце работы.

  cout << (100.0 * odds / d.size())

    << "% of the numbers are odd.\n";

}

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

seq

par

par_vec

$ ./auto_parallel

50.4% of the numbers are odd.

Как это работает

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

Взглянем на вызовы и ответим на три основных вопроса:

generate(execution::par, begin(d), end(d), rand_num);

sort(   execution::par, begin(d), end(d));

reverse( execution::par, begin(d), end(d));

auto odds (count_if(execution::par, begin(d), end(d), odd));

Какие алгоритмы STL можно распараллелить таким образом?

Шестьдесят девять существующих алгоритмов STL получили поддержку параллелизма по стандарту C++17. Появились также семь новых алгоритмов, которые тоже поддерживают параллелизм. Несмотря на то что такое улучшение может оказаться довольно инвазивным для реализации, с точки зрения их интерфейса изменилось не очень много: все они получили дополнительный аргумент

ExecutionPolicy&& policy

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

std::adjacent_difference

std::adjacent_find

std::all_of

std::any_of

std::copy

std::copy_if

std::copy_n

std::count

std::count_if

std::equal

std::exclusive_scan

std::fill

std::fill_n

std::find

std::find_end

std::find_first_of

std::find_if

std::find_if_not

std::for_each

std::for_each_n

std::generate

std::generate_n

std::includes

std::inclusive_scan

std::inner_product

std::inplace_merge

std::is_heap

std::is_heap_until

std::is_partitioned

std::is_sorted

std::is_sorted_until

std::lexicographical_compare

std::max_element

std::merge std::min_element

std::minmax_element

std::mismatch std::move

std::none_of

std::nth_element

std::partial_sort

std::partial_sort_copy

std::partition

std::partition_copy

std::remove

std::remove_copy

std::remove_copy_if

std::remove_if

std::replace

std::replace_copy

std::replace_copy_if

std::replace_if

std::reverse

std::reverse_copy

std::rotate

std::rotate_copy

std::search

std::search_n

std::set_difference

std::set_intersection

std::set_symmetric_difference

std::set_union

std::sort

std::stable_partition

std::stable_sort

std::swap_ranges

std::transform

std::transform_exclusive_scan

std::transform_inclusive_scan

std::transform_reduce

std::uninitialized_copy

std::uninitialized_copy_n

std::uninitialized_fill

std::uninitialized_fill_n

std::unique

std::unique_copy

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

Однако вместо того, чтобы разрабатывать собственные сложные параллельные алгоритмы с помощью

std::thread

std::async

Как работают эти политики выполнения

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

Следующие три типа политик существуют в пространстве имен

std::execution

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

□ все элементы функций доступа, используемые параллелизованными алгоритмами, не должны вызывать взаимных блокировок и гонок;

□ в случае параллелизации и векторизации все функции получения доступа не должны использовать блокирующую синхронизацию.

До тех пор, пока подчиняемся этим правилам, мы не столкнемся с ошибками, которые могут появиться в параллельных версиях алгоритмов STL.

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

Что означает понятие «векторизация»

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

std::vector v {1, 2, 3, 4, 5, 6, 7 /*...*/};

int sum {std::accumulate(v.begin(), v.end(), 0)};

Компилятор в конечном счете сгенерирует цикл из вызова

accumulate

int sum {0};

for (size_t i {0}; i < v.size(); ++i) {

 sum += v[i];

}

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

N*4

int sum {0};

for (size_t i {0}; i < v.size() / 4; i += 4) {

 sum += v[i] + v[i+1] + v[i + 2] + v[i + 3];

}

// если операция v.size()/4 имеет остаток,

// в реальном коде также нужно это обработать.

Зачем это делать? Многие процессоры предоставляют инструкции, которые могут выполнять математические операции наподобие

sum += v[i]+v[i+1]+v[i+2]+v[i+3];

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

Приостанавливаем программу на конкретный промежуток времени

Простая и удобная возможность управления потоками добавлена в С++11. В данной версии появилось пространство имен

this_thread

В этом примере мы сконцентрируемся на том, как приостанавливать потоки на определенный промежуток времени.

Как это делается

В этом примере мы напишем короткую программу, которая приостанавливает основной поток на определенные промежутки времени.

1. Сначала включим все необходимые заголовочные файлы и объявим об использовании пространств имен

std

chrono_literals

#include 

#include 

#include 

using namespace std;

using namespace chrono_literals;

2. Сразу же приостановим основной поток на 5 секунд 300 миллисекунд. Благодаря пространству имен

chrono_literals

int main()

{

  cout << "Going to sleep for 5 seconds"

      " and 300 milli seconds.\n";

  this_thread::sleep_for(5s + 300ms);

3. Последним выражением приостановки являлось

relative

absolute

  cout << "Going to sleep for another 3 seconds.\n";

 this_thread::sleep_until(

   chrono::high_resolution_clock::now() + 3s);

4. Перед завершением программы выведем на экран какое-нибудь сообщение, что укажет на окончание второго периода приостановки.

  cout << "That's it.\n";

}

5. Компиляция и запуск программы дадут следующие результаты. В Linux, Mac и других UNIX-подобных операционных системах имеется команда

time

time

$ time ./sleep

Going to sleep for 5 seconds and 300 milli seconds.

Going to sleep for another 3 seconds.

That's it.

real 0m8.320s

user 0m0.005s

sys  0m0.003s

Как это работает

Функции

sleep_for

sleep_until

std::this_thread

Функция

this_thread::sleep_for

chrono::duration

1s

5s+300ms

std::chrono_literals;

Функция

this_thread::sleep_until

chrono::time_point

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

Приостановить выполнение потока на короткий промежуток времени также можно с помощью функции

this_thread::yield

yield

Запускаем и приостанавливаем потоки

Еще одним дополнением, появившимся в C++11, является класс

std::thread

В данном примере мы реализуем программу, которая запускает и останавливает потоки. Далее рассмотрим информацию о том, что с ними делать после запуска.

Как это делается

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

1. Сначала включим всего два заголовочных файла, а затем объявим об использовании пространств имен

std

chrono_literals

#include 

#include 

using namespace std;

using namespace chrono_literals;

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

i

cout

static void thread_with_param(int i)

{

  this_thread::sleep_for(1ms * i);

  cout << "Hello from thread " << i << '\n';

 this_thread::sleep_for(1s * i);

  cout << "Bye from thread " << i << '\n';

}

3. В функции

main

std::thread::hardware_concurrency

int main()

{

  cout << thread::hardware_concurrency()

    << " concurrent threads are supported.\n";

4. Наконец, начнем работать с потоками. Запустим три потока с разными идентификаторами. При создании экземпляра потока с помощью выражения наподобие

t{f, x}

f(x)

thread_with_param

  thread t1 {thread_with_param, 1};

  thread t2 {thread_with_param, 2};

  thread t3 {thread_with_param, 3};

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

join

  t1.join();

  t2.join();

6. Альтернативой присоединению является открепление. Если мы не вызовем функцию

join

detach

detach

thread

  t3.detach();

7. Перед завершением функции

main

  cout << "Threads joined.\n";

}

8. Компиляция и запуск программы дадут следующий результат. Мы можем увидеть, что моя машина имеет восемь ядер процессора. Далее сообщения hello видим из всех потоков, а сообщения bye — лишь из двух, которые мы объединили. Поток 3 все еще ожидает завершения трехсекундного ожидания, но вся программа уже завершилась после того, как поток 2 завершил свое двухсекундное ожидание. Таким образом, мы не можем увидеть прощальное сообщение потока 3, поскольку он был уничтожен:

$ ./threads

8 concurrent threads are supported.

Hello from thread 1

Hello from thread 2

Hello from thread 3

Bye from thread 1

Bye from thread 2

Threads joined.

Как это работает

Запуск и остановку потоков выполнить очень просто. Многопроцессорная обработка начинает усложняться в момент, когда потокам нужно работать вместе (делить ресурсы, ожидать завершения других потоков и т.д.).

Чтобы запустить поток, нужно иметь функцию, которую он будет выполнять. Функция не обязательно должна быть особенной, поскольку в потоке можно выполнить практически любую функцию. Напишем небольшую программу-пример, которая запускает поток и ожидает его завершения:

void f(int i) { cout << i << '\n'; }

int main()

{

  thread t {f, 123};

 t.join();

}

Вызов конструктора

std::thread

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

К этому моменту мы не контролируем следующие детали:

□ порядок, в котором потоки чередуются на одном ядре;

□ приоритет потока, указывающий, что один поток главнее другого;

□ распределение потоков между ядрами. Вполне возможна ситуация, когда все потоки будут выполняться на одном ядре, несмотря на то что машина имеет более 100 ядер.

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

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

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

main

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

Единственное, о чем следует волноваться, — это объединение. Когда поток вызывает функцию

x.join()

x

x.detach()

thread

std::terminate()

В момент, когда функция main возвращает значение, приложение заканчивает работу. Однако в это же время наш открепленный поток

t3

main

Выполняем устойчивую к исключениям общую блокировку с помощью td::unique_lock и std::shared_lock

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

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

Как это делается

В этом примере мы напишем программу, которая использует экземпляр класса

std::shared_mutex

lock

unlock

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

std

chrono_literal

#include 

#include 

#include 

#include 

using namespace std;

using namespace chrono_literals;

2. Вся программа строится вокруг одного общего мьютекса, поэтому для простоты объявим его глобальный экземпляр:

shared_mutex shared_mut;

3. Мы будем использовать вспомогательные функции RAII

std::shared_lock

std::unique_lock

using shrd_lck = shared_lock;

using uniq_lck = unique_lock;

4. Прежде чем начнем писать функцию main, определим две вспомогательные функции, которые пытаются заблокировать мьютекс в эксклюзивном режиме. Эта функция создаст экземпляр класса

unique_lock

defer_lock

try_lock

exclusive_lock

static void print_exclusive()

{

  uniq_lck l {shared_mut, defer_lock};

  if (l.try_lock()) {

   cout << "Got exclusive lock.\n";

  } else {

   cout << "Unable to lock exclusively.\n";

  }

}

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

unique_lock

static void exclusive_throw()

{

  uniq_lck l {shared_mut};

 throw 123;

}

6. Теперь перейдем к функции

main

shared_lock

int main()

{

  {

   shrd_lck sl1 {shared_mut};

   cout << "shared lock once.\n";

7. Откроем еще одну область видимости и создадим второй экземпляр типа

shared_lock

shared_lock

shared_lock

print_exclusive

   {

    shrd_lck sl2 {shared_mut};

    cout << "shared lock twice.\n";

   print_exclusive();

   }

8. После выхода из самой поздней области видимости деструктор объекта

sl2

shared_lock

print_exclusive

   cout << "shared lock once again.\n";

   print_exclusive();

  }

  cout << "lock is free.\n";

9. После выхода из второй области видимости все объекты типа

shared_lock

exclusive_throw

print_exclusive

exclusive_throw

unique_lock

exclusive_throw

print_exclusive

   try {

    exclusive_throw();

   } catch (int e) {

   cout << "Got exception " << e << '\n';

  }

  print_exclusive();

}

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

print_exclusive

print_exclusive

exclusive_throw

print_exclusive

$ ./shared_lock

shared lock once.

shared lock twice.

Unable to lock exclusively.

shared lock once again.

Unable to lock exclusively.

lock is free.

Got exception 123

Got exclusive lock.

Как это работает

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

Классы мьютексов

Термин mutex расшифровывается как mutual exclusion (взаимное исключение). Чтобы предотвратить неуправляемое изменение одного объекта несколькими конкурирующими потоками, способное привести к повреждению данных, можно использовать объекты мьютексов. STL предоставляет разные классы мьютексов, которые хороши в разных ситуациях. Все они похожи в том, что имеют методы

lock

unlock

Когда некий поток первым вызывает метод

lock()

lock

unlock

std::mutex

В STL существует множество разных классов мьютексов (табл. 9.2).

Классы блокировок

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

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

delete

Для управления памятью существуют вспомогательные классы

unique_ptr

shared_ptr

weak_ptr

std::lock_guard

void critical_function()

{

  lock_guard l {some_mutex};

  // критический раздел

}

Конструктор элемента

lock_guard

lock

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

В то время как

lock_guard

scoped_lock

unique_lock

shared_lock

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

shrd_lck

shared_lock

sl1

sl2

print_exclusive

Теперь перейдем к эксклюзивным функциям блокировки, которые появились позднее в функции

main

int main()

{

  {

   shrd_lck sl1 {shared_mut};

   {

    shrd_lck sl2 {shared_mut};

    print_exclusive();

   }

   print_exclusive();

  }

  try {

   exclusive_throw();

  } catch (int e) {

   cout << "Got exception " << e << '\n';

  }

  print_exclusive();

}

Важная деталь — после возвращения из

exclusive_throw

print_exclusive

exclusive_throw

Еще раз взглянем на функцию

print_exclusive

void print_exclusive()

{

  uniq_lck l {shared_mut, defer_lock};

  if (l.try_lock()) {

   // ...

  }

}

Мы предоставили

shared_mut

defer_lock

unique_lock

defer_lock

unique_lock

l.try_lock()

Избегаем взаимных блокировок с применением std::scoped_lock

Если бы взаимные блокировки происходили на дорогах, то выглядели бы так (рис. 9.2).

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

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

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

std::scoped_lock

Как это делается

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

1. Сначала включим все необходимые заголовочные файлы и объявим об использовании пространств имен

std

chrono_literals

#include 

#include 

#include 

using namespace std;

using namespace chrono_literals;

2. Затем создадим два объекта мьютексов, которые понадобятся для создания взаимной блокировки:

mutex mut_a;

mutex mut_b;

3. Чтобы создать взаимную блокировку с двумя ресурсами, нужны две функции. Одна пробует заблокировать мьютекс

А

В

'\n'

endl

cout

static void deadlock_func_1()

{

  cout << "bad f1 acquiring mutex A..." << endl;

 lock_guard la {mut_a};

 this_thread::sleep_for(100ms);

  cout << "bad f1 acquiring mutex B..." << endl;

  lock_guard lb {mut_b};

  cout << "bad f1 got both mutexes." << endl;

}

4. Как мы и говорили на предыдущем шаге, функция

deadlock_func_2

deadlock_func_1

A

B

static void deadlock_func_2()

{

  cout << "bad f2 acquiring mutex B..." << endl;

  lock_guard lb {mut_b};

  this_thread::sleep_for(100ms);

  cout << "bad f2 acquiring mutex A..." << endl;

  lock_guard la {mut_a};

  cout << "bad f2 got both mutexes." << endl;

}

5. Теперь напишем свободный от взаимных блокировок вариант этих функций. Они используют класс

scoped_lock

А

В

static void sane_func_1()

{

  scoped_lock l {mut_a, mut_b};

  cout << "sane f1 got both mutexes." << endl;

}

static void sane_func_2()

{

  scoped_lock l {mut_b, mut_a};

  cout << "sane f2 got both mutexes." << endl;

}

6. В функции

main

int main()

{

  {

   thread t1 {sane_func_1};

   thread t2 {sane_func_2};

   t1.join();

   t2.join();

  }

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

  {

   thread t1 {deadlock_func_1};

   thread t2 {deadlock_func_2};

   t1.join();

   t2.join();

  }

}

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

А

В

Это приложение нужно завершить снаружи, например нажав Ctrl+C:

$ ./avoid_deadlock

sane f1 got both mutexes

sane f2 got both mutexes

bad f2 acquiring mutex B...

bad f1 acquiring mutex A...

bad f1 acquiring mutex B...

bad f2 acquiring mutex A...

Как это работает

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

В подобных ситуациях поможет

scoped_lock

lock_guard

unique_lock

Класс

scoped_lock

std::lock

try_lock

scoped_lock

std::lock

Синхронизация конкурентного использования std::cout

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

Одной из структур данных, часто применяемых для вывода данных, является

std::cout

cout

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

cout

cout

Как это делается

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

1. Как и обычно, сначала укажем все директивы

include

std

#include 

#include 

#include 

#include 

#include 

using namespace std;

2. Далее реализуем вспомогательный класс, который назовем

pcout

pcout

stringstream

operator<<

pcout

stringstream

struct pcout : public stringstream {

 static inline mutex cout_mutex;

  ~pcout() {

   lock_guard l {cout_mutex};

   cout << rdbuf();

   cout.flush();

  }

};

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

cout

cout

pcout

operator<<

pcout

pcout

static void print_cout(int id)

{

  cout << "cout hello from " << id << '\n';

}

static void print_pcout(int id)

{

  pcout{} << "pcout hello from " << id << '\n';

}

4. Опробуем его. Сначала воспользуемся

print_cout

cout

int main()

{

  vector v;

  for (size_t i {0}; i < 10; ++i) {

   v.emplace_back(print_cout, i);

  }

  for (auto &t : v) { t.join(); }

5. Далее делаем то же самое для функции

print_pcout

 cout << "=====================\n";

  v.clear();

  for (size_t i {0}; i < 10; ++i) {

  v.emplace_back(print_pcout, i);

  }

  for (auto &t : v) { t.join(); }

}

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

cout

print_pcout

Как это работает

О’кей, мы создали эту «оболочку для cout», которая автоматически сериализует последовательные попытки вывода текста. Как она работает?

Выполним те же действия, что и

pcout

stringstream ss;

ss << "This is some printed line " << 123 << '\n';

Затем блокируем глобально доступный мьютекс:

{

  lock_guard l {cout_mutex};

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

cout.flush()

flush

  cout << ss.rdbuf();

 cout.flush();

}

О’кей, это достаточно просто, но утомительно писать, если нужно делать одно и то же раз за разом. Можно сократить создание объекта stringstream таким образом:

stringstream{} << "This is some printed line " << 123 << '\n';

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

stringstream

Мы не можем изменить методы-члены экземпляра

stringstream

struct pcout : public stringstream {

  ~pcout() {

   lock_guard l {cout_mutex};

   cout << rdbuf();

   cout.flush();

  }

};

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

cout

Кроме того, мы поместили объект

cout_mutex

pcout

Безопасно откладываем инициализацию с помощью std::call_once

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

Однако данный подход имеет следующие недостатки.

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

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

В данном примере мы рассмотрим вспомогательную функцию

std::call_once

Как это делается

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

1. Сначала включим все необходимые заголовочные файлы и объявим об использовании пространства имен

std

#include 

#include 

#include 

#include 

using namespace std;

2. Далее будем использовать функцию

std::call_once

once_flag

call_once

once_flag callflag;

3. Функция, которая должна быть выполнена всего раз, выглядит так. Она просто выводит один восклицательный знак:

static void once_print()

{

  cout << '!';

}

4. Все потоки будут выполнять функцию

print

once_print

std::call_once

call_once

callflag

static void print(size_t x)

{

  std::call_once(callflag, once_print);

 cout << x;

}

5. О’кей, теперь запустим десять потоков, все они будут использовать функцию

print

int main()

{

  vector v;

  for (size_t i {0}; i < 10; ++i) {

   v.emplace_back(print, i);

  }

  for (auto &t : v) { t.join(); }

 cout << '\n';

}

6. Компиляция и запуск дадут следующий результат. Сначала мы увидим восклицательный знак благодаря функции

once_print

call_once

once_print

once_print

$ ./call_once

!1239406758

Как это работает

Функция

std:call_once

call_once

Чтобы организовать этот небольшой «танцевальный номер», требуется переменная, на основе которой другие потоки могут определить, следует ли им ждать, а также время их освобождения. Именно для этого и предназначена переменная

once_flag callflag;

call_once

once_flag

Еще одна приятная деталь: если поток, который был выбран для выполнения функции

call_once

Отправляем выполнение задач в фоновый режим с применением std::async

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

std::thread t {my_function, arg1, arg2, ...};

// сделать что-то еще

t.join(); // подождать завершения потока

Но здесь начинаются неудобства:

t.join()

my_function

my_function

В C++11 появилась функция

std::async

std::async

Как это делается

В этом примере мы реализуем программу, которая делает несколько дел конкурентно, но вместо того, чтобы явно запускать потоки, мы используем

std::async

std::future

1. Сначала включим все необходимые заголовочные файлы и объявим об использовании пространства имен

std

#include 

#include 

#include 

#include 

#include 

#include 

#include 

using namespace std;

2. Реализуем три функции, которые не связаны с параллелизмом, а просто выполняют интересные задачи. Первая функция принимает строку и создает гистограмму включения всех символов внутри этой строки:

static map histogram(const string &s)

{

  map m;

  for (char c : s) { m[c] += 1; }

 return m;

}

3. Вторая функция также принимает строку и возвращает ее отсортированную копию:

static string sorted(string s)

{

  sort(begin(s), end(s));

 return s;

}

4. Третья функция подсчитывает, как много гласных находится внутри принимаемой строки:

static bool is_vowel(char c)

{

  char vowels[] {"aeiou"};

 return end(vowels) !=

   find(begin(vowels), end(vowels), c);

}

static size_t vowels(const string &s)

{

  return count_if(begin(s), end(s), is_vowel);

}

5. В функции

main

ios::skipws

pop_back

'\0'

int main()

{

  cin.unsetf(ios::skipws);

  string input {istream_iterator{cin}, {}};

 input.pop_back();

6. Теперь вернем значения из всех функций, которые реализовали ранее. Чтобы ускорить выполнение программы в случае получения очень больших входных данных, запустим их асинхронно. Функция

std::async

histogram

sorted

vowels

launch::async

  auto hist (async(launch::async,

          histogram, input));

  auto sorted_str (async(launch::async,

          sorted, input));

 auto vowel_count(async(launch::async,

          vowels, input));

7. Вызовы

async

hist

sorted_str

vowel_count

histogram

sorted

vowels

future

std::async

.get()

main

for (const auto &[c, count] : hist.get()) {

    cout << c << ": " << count << '\n';

}

cout << "Sorted string: "

   << quoted(sorted_str.get()) << '\n'

   << "Total vowels: "

   << vowel_count.get() << '\n';

}

8. Компиляция и запуск кода выглядят так. Мы использовали короткую строку, которую не стоит распараллеливать, но для примера выполнили ее конкурентно. Вдобавок общая структура программы не изменилась по сравнению с наивной последовательной реализацией:

$ echo "foo bar baz foobazinga" | ./async

 : 3

a: 4

b: 3

f: 2

g: 1

i: 1

n: 1

o: 4

r: 1

z: 2

Sorted string: " aaaabbbffginoooorzz"

Total vowels: 9

Как это работает

Если бы мы не использовали std::async, то последовательный нераспараллеленный код выглядел бы довольно просто:

auto hist     (histogram(input));

auto sorted_str  (sorted(  input));

auto vowel_count (vowels(  input));

for (const auto &[c, count] : hist) {

   cout << c << ": " << count << '\n';

}

cout << "Sorted string: " << quoted(sorted_str) << '\n';

cout << "Total vowels: " << vowel_count << '\n';

Чтобы распараллелить код, мы сделали следующее: обернули три вызова функций в вызовы

async(launch::async, ...)

async

auto hist     (async(launch::async, histogram, input));

auto sorted_str  (async(launch::async, sorted,   input));

auto vowel_count (async(launch::async, vowels,   input));

for (const auto &[c, count] : hist.get()) {

    cout << c << ": " << count << '\n';

}

cout << "Sorted string: "

   << quoted(sorted_str.get()) << '\n'

   << "Total vowels: "

   << vowel_count.get() << '\n';

В то время как, например, функция

histogram

async(..., histogram, ...)

future

histogram

future

get

Рассмотрим еще один короткий пример. Взгляните на этот фрагмент:

auto x (f(1, 2, 3));

cout << x;

Вместо предыдущего кода мы могли написать следующее:

auto x (async(launch::async, f, 1, 2, 3));

cout << x.get();

Вот, по сути, и все. Выполнение задач в фоновом режиме никогда не было проще по стандартам С++. Осталось разрешить один момент: что означает

launch::async

Вызов наподобие

async(f, 1, 2, 3)

async

Дополнительная информация

Следует рассмотреть последний момент. Предположим, мы пишем код следующим образом:

async(launch::async, f);

async(launch::async, g);

Это может привести к тому, что функции

f

g

Почему же код блокируется? Разве

async

future

async

launch::async

Это значит, что оба вызова

async

future

Реализуем идиому «производитель/потребитель» с использованием std::condition_variable

В этом примере мы реализуем типичную программу, работающую по принципу «производитель/потребитель», в которой запускается несколько потоков. Основная идея заключается в том, что существует один поток, который создает элементы и помещает их в очередь. Еще один поток потребляет (использует) эти элементы. Если создавать нечего, то поток-производитель приостанавливается. При отсутствии в очереди элементов приостанавливается потребитель.

Оба потока имеют доступ к очереди и могут изменить, поэтому ее нужно защитить с помощью мьютекса.

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

Для таких событий в C++11 предоставляется удобная структура данных:

std::condition_variable

Как это делается

Итак, в этом примере мы реализуем простую программу, работающую по принципу «производитель/потребитель», которая запускает одного производителя значений в отдельном потоке, а также одного потребителя в другом потоке.

1. Сначала выполним все необходимые директивы

include

std

#include 

#include 

#include 

#include 

#include 

using namespace std;

using namespace chrono_literals;

2. Создадим экземпляр очереди простых численных значений и назовем ее

q

condition_variable

cv

finished

queue    q;

mutex        mut;

condition_variable cv;

bool        finished {false};

3. Реализуем функцию-производитель. Она принимает аргумент

items

cv.notify_all()

static void producer(size_t items) {

 for (size_t i {0}; i < items; ++i) {

   this_thread::sleep_for(100ms);

   {

    lock_guard lk {mut};

   q.push(i);

   }

   cv.notify_all();

  }

4. После создания всех элементов снова блокируем мьютекс, поскольку нужно задать значение для бита

finished

cv.notify_all()

  {

   lock_guard lk {mut};

   finished = true;

  }

  cv.notify_all();

}

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

finished

finished

cv.wait

static void consumer() {

  while (!finished) {

   unique_lock l {mut};

   cv.wait(l, [] { return !q.empty() || finished; });

6. Вызов

cv.wait

finished

   while (!q.empty()) {

   cout << "Got " << q.front()

      << " from queue.\n"; q.pop();

   }

  }

}

7. В функции

main

int main()

{

  thread t1 {producer, 10};

 thread t2 {consumer};

 t1.join();

  t2.join();

  cout << "finished!\n";

}

8. Компиляция и запуск программы дадут следующий результат. При выполнении программы можно увидеть, что между появлением каждой строки проходит какое-то время (100 миллисекунд), поскольку создание элементов занимает время:

$ ./producer_consumer

Got 0 from queue.

Got 1 from queue.

Got 2 from queue.

Got 3 from queue.

Got 4 from queue.

Got 5 from queue.

Got 6 from queue.

Got 7 from queue.

Got 8 from queue.

Got 9 from queue.

finished!

Как это работает

В данном примере мы просто запустили два потока. Первый создает элементы и помещает их в очередь. Второй извлекает их из очереди. При взаимодействии с очередью один из этих потоков блокирует общий мьютекс

mut

Помимо очереди и мьютекса мы объявили четыре переменные, которые были включены во взаимодействие производителя/потребителя:

queue    q;

mutex        mut;

condition_variable cv;

bool        finished {false};

Переменную

finished

true

true

condition_variable cv

cv

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

mut

unique_lock

cv.wait

while (!finished) {

 unique_lock l {mut};

  cv.wait(l, [] { return !q.empty() || finished; });

  while (!q.empty()) {

   // consume

  }

}

Данный код чем-то похож на следующий эквивалентный код. Вскоре мы рассмотрим, почему эти фрагменты не похожи друг на друга:

while (!finished) {

 unique_lock l {mut};

  while (q.empty() && !finished) {

   l.unlock();

   l.lock();

  }

  while (!q.empty()) {

   // consume

  }

}

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

1. Есть ли элементы, которые можно использовать? В таком случае сохраняем блокировку, потребляем эти элементы, возвращаем блокировку и начинаем сначала.

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

Реальная причина, по которой строка

cv.wait

while (q.empty() && ...)

l.unlock(); l.lock();

Выражение наподобие

cv.wait(lock, predicate)

predicate()

true

wait

condition_variable

notify_one()

notify_all()

predicate()

Положительный момент таков: после вызова

wait

notify

На стороне производителя мы просто вызвали

cv.notify_all()

finished

true

Реализуем идиому «несколько производителей/потребителей» с помощью std::condition_variable

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

Таким образом, нужно приостанавливать не только потребителей, при отсутствии в очереди элементов, но и производителей, если элементов в ней слишком много.

Мы увидим, как решить эту проблему с помощью нескольких объектов типа

std::condition_variable

Как это делается

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

1. Сначала включим все необходимые заголовочные файлы и объявим об использовании пространств имен

std

chrono_literals

#include 

#include 

#include 

#include 

#include 

#include 

#include 

#include 

#include 

using namespace std;

using namespace chrono_literals;

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

struct pcout : public stringstream {

 static inline mutex cout_mutex;

  ~pcout() {

   lock_guard l {cout_mutex};

   cout << rdbuf();

  }

};

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

queue q;

mutex     q_mutex;

bool      production_stopped {false};

4. В этой программе задействуем две разные переменные

condition_variables

condition_variable

go_consume

go_produce

condition_variable go_produce;

condition_variable go_consume;

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

go_produce.wait

stock

static void producer(size_t id, size_t items, size_t stock)

{

  for (size_t i = 0; i < items; ++i) {

   unique_lock lock(q_mutex);

   go_produce.wait(lock,

    [&] { return q.size() < stock; });

6. После того как производитель будет возобновлен, он создаст элемент и поместит его в очередь. Значение, помещаемое в очередь, определяется на основе выражения

id*100+i

   q.push(id * 100 + i);

   pcout{} << " Producer " << id << " --> item "

       << setw(3) << q.back() << '\n';

7. После создания элемента можно возобновить приостановленных потребителей. Период приостановки, равный 90 миллисекундам, симулирует тот факт, что на создание элементов требуется какое-то время:

   go_consume.notify_all();

   this_thread::sleep_for(90ms);

  }

  pcout{} << "EXIT: Producer " << id << '\n';

}

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

static void consumer(size_t id)

{

  while (!production_stopped || !q.empty()) {

   unique_lock lock(q_mutex);

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

go_consume

wait

1s

wait

wait_for

false

   if (go_consume.wait_for(lock, 1s,

     [] { return !q.empty(); })) {

    pcout{} << " item "

        << setw(3) << q.front()

        << " --> Consumer "

        << id << '\n';

    q.pop();

10. После потребления элемента оповещаем производителей и приостанавливаем поток на 130 миллисекунд для симуляции того факта, что потребление элементов тоже требует времени:

    go_produce.notify_all();

    this_thread::sleep_for(130ms);

   }

  }

  pcout{} << "EXIT: Producer " << id << '\n';

}

11. В функции

main

int main()

{

  vector workers;

 vector consumers;

12. Далее порождаем три потока-производителя и пять потоков-потребителей:

  for (size_t i = 0; i < 3; ++i) {

   workers.emplace_back(producer, i, 15, 5);

  }

  for (size_t i = 0; i < 5; ++i) {

   consumers.emplace_back(consumer, i);

  }

13. Сначала позволим закончить работу потокам-производителям. Как только все из них вернут значения, установим значение флага

production_stopped;

  for (auto &t : workers) { t.join(); }

 production_stopped = true;

  for (auto &t : consumers) { t.join(); }

}

14. Компиляция и запуск программы дадут следующий результат. Сообщений получилось довольно много, поэтому мы приводим их в сокращенном виде. Как видите, производители приостанавливаются время от времени и позволяют потребителям использовать несколько элементов, чтобы снова получить возможность их производить. Интересно будет изменить время ожидания для производителей/потребителей, а также манипулировать количеством производителей/потребителей и максимальным количеством элементов в очереди, поскольку это значительно меняет шаблоны появления выходных сообщений:

$ ./multi_producer_consumer

  Producer 0 --> item 0

  Producer 1 --> item 100

          item 0 --> Consumer 0

  Producer 2 --> item 200

          item 100 --> Consumer 1

          item 200 --> Consumer 2

  Producer 0 --> item 1

  Producer 1 --> item 101

          item 1 --> Consumer 0

...

  Producer 0 --> item 14

EXIT: Producer 0

  Producer 1 --> item 114

EXIT: Producer 1

         item 14 --> Consumer 0

  Producer 2 --> item 214

EXIT: Producer 2

         item 114 --> Consumer 1

         item 214 --> Consumer 2

EXIT: Consumer 2

EXIT: Consumer 3

EXIT: Consumer 4

EXIT: Consumer 0

EXIT: Consumer 1

Как это работает

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

M

N

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

condition_variable

1.

go_produce

2.

go_consume

Таким образом производители заполняют очередь элементами и сигнализируют с помощью события

go_consume

if (go_consume.wait_for(lock, 1s, [] { return !q.empty(); })) {

  // получили событие без тайм-аута

}

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

go_produce.wait(lock, [&] { return q.size() < stock; });

Интересный момент: мы не позволяем потребителям ждать вечно. В вызове

go_consume.wait_for

timeout

1

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

Рассмотрим следующую ситуацию, которую позволяет элегантно решить

condition_variable:

go_produce

go_produce

Представим ситуацию, когда у нас в очереди имеется

(max-1)

Независимо от того, какой метод вызовет поток-потребитель —

go_produce.notify_ one()

go_produce.notify_all()

go_produce.wait

q.size()