Потоки, блокировки и условные переменные в C++11 [Часть 1]

В первой части этой статьи основное внимание будет уделено потокам и блокировкам в С++11, условные переменные во всей своей красе будут подробно рассмотрены во второй части…

Потоки

В C++11, работа с потокам осуществляется по средствам класса std::thread (доступного из заголовочного файла <thread>), который может работать с регулярными функциями, лямбдами и функторами. Кроме того, он позволяет вам передавать любое число параметров в функцию потока.

#include <thread>
 
void threadFunction()
{
     // do smth
}
 
int main()
{
     std::thread thr(threadFunction);
     thr.join();
     return 0;
}

В этом примере, thr — это объект, представляющий поток, в котором будет выполняться функция threadFunction(). Вызов join блокирует вызывающий поток (в нашем случае — поток main) до тех пор, пока thr (а точнее threadFunction()) не выполнит свою работу. Если функция потока возвращает значение — оно будет проигнорировано. Однако принять функция может любое количество параметров.

void threadFunction(int i, double d, const std::string &s)
{
     std::cout << i << ", " << d << ", " << s << std::endl;
}
 
int main()
{
     std::thread thr(threadFunction, 1, 2.34, "example");
     thr.join();
     return 0;
}

Несмотря на то, что передавать можно любое число параметров, все они были переданы по значению Если в функцию необходимо передать параметры по ссылке, они должны быть обернуты в std::ref или std::cref, как в примере:

void threadFunction(int &a)
{
     a++;
}
 
int main()
{
     int a = 1;
     std::thread thr(threadFunction, std::ref(a));
     thr.join();
     std::cout << a << std::endl; 
     return 0;
}

Программа напечатает в консоль 2. Если не использовать std::ref, то результатом работы программы будет 1.

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

int main()
{
     std::thread thr(threadFunction);
     thr.detach();
     return 0;
}

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

try
{
     std::thread thr1(threadFunction);
     std::thread thr2(threadFunction);
 
     thr1.join();
     thr2.join();
}

catch (const std::exception &ex)
{
     std::cout << ex.what() << std::endl;
}

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

std::mutex                       g_mutex;
std::vector<std::exception_ptr>  g_exceptions;

void throw_function()
{
     throw std::exception("something wrong happened");
}

void threadFunction()
{
     try
     {
          throw_function();
     }
     catch (...)
     {
          std::lock_guard<std::mutex> lock(g_mutex);
          g_exceptions.push_back(std::current_exception());
     }
}

int main()
{
     g_exceptions.clear();
     std::thread thr(threadFunction);
     thr.join();

     for(auto &e: g_exceptions)
     {
          try 
          {
               if(e != nullptr)
                    std::rethrow_exception(e);
          }
          catch (const std::exception &e)
          {
               std::cout << e.what() << std::endl;
          }
     }
     return 0;
}

Прежде, чем двигаться дальше, хочу отметить некоторые полезные функции, предоставляемые <thread>, в пространстве имен std::this_thread:

• get_id: возвращает id текущего потока
• yield: говорит планировщику выполнять другие потоки, может использоваться при активном ожидании
• sleep_for: блокирует выполнение текущего потока в течение установленного периода
• sleep_until: блокирует выполнение текущего потока, пока не будет достигнут указанный момент времени

Блокировки

В последнем примере, я должен был синхронизировать доступ к вектору g_exceptions, чтобы быть уверенным, что только один поток одновременно может вставить новый элемент. Для этого я использовал мьютекс и блокировку на мьютекс. Мьютекс — базовый элемент синхронизации и в С++11 представлен в 4 формах в заголовочном файле <mutex>:

• mutex: обеспечивает базовые функции lock() и unlock() и не блокируемый метод try_lock()
• recursive_mutex: может войти «сам в себя»
• timed_mutex: в отличие от обычного мьютекса, имеет еще два метода: try_lock_for() и try_lock_until()
• recursive_timed_mutex: это комбинация timed_mutex и recursive_mutex

Приведу пример использования std::mutex с упомянутыми ранее функциями-помощниками get_id() и sleep_for():

#include <iostream>
#include <chrono>
#include <thread>
#include <mutex>
 
std::mutex g_lock;
 
void threadFunction()
{
     g_lock.lock();
 
     std::cout << "entered thread " << std::this_thread::get_id() << std::endl;
     std::this_thread::sleep_for(std::chrono::seconds(rand()%10));
     std::cout << "leaving thread " << std::this_thread::get_id() << std::endl;
 
     g_lock.unlock();
}
 
int main()
{
     srand((unsigned int)time(0));
     std::thread t1(threadFunction);
     std::thread t2(threadFunction);
     std::thread t3(threadFunction);
     t1.join();
     t2.join();
     t3.join();
     return 0;
}

Программа должна выдавать примерно следующее:

entered thread 10144
leaving thread 10144
entered thread 4188
leaving thread 4188
entered thread 3424
leaving thread 3424

Перед обращением к общим данным, мьютекс должен быть заблокирован методом lock, а после окончания работы с общими данными — разблокирован методом unlock.

Следующий пример показывает простой потокобезопасный контейнер (реализованный на базе std::vector), имеющий методы add() для добавления одного элемента и addrange() для добавления нескольких элементов.
Примечание: и всё же этот контейнер не является полностью потокобезопасным по нескольким причинам, включая использование va_args. Также, метод dump() не должен принадлежать контейнеру, а должен быть автономной функцией. Цель этого примера в том, что показать основные концепции использования мьютексов, а не не сделать полноценный, безошибочный, потокобезопасный контейнер.

template <typename T>
class container 
{
     std::mutex _lock;
     std::vector<T> _elements;
public:
     void add(T element) 
     {
          _lock.lock();
          _elements.push_back(element);
          _lock.unlock();
     } 
     void addrange(int num, ...)
     {
          va_list arguments;
          va_start(arguments, num);
          for (int i = 0; i < num; i++)
          {
               _lock.lock();
               add(va_arg(arguments, T));
               _lock.unlock();
          }
          va_end(arguments); 
     }
     void dump()
     {
          _lock.lock();
          for(auto e: _elements)
          std::cout << e << std::endl;
          _lock.unlock();
     }
};
 
void threadFunction(container<int> &c)
{
     c.addrange(3, rand(), rand(), rand());
}
 
int main()
{
     srand((unsigned int)time(0));
     container<int> cntr;
     std::thread t1(threadFunction, std::ref(cntr));
     std::thread t2(threadFunction, std::ref(cntr));
     std::thread t3(threadFunction, std::ref(cntr));
     t1.join();
     t2.join();
     t3.join();
     cntr.dump();
     return 0;
}

При выполнении этой программы произойдет deadlock (взаимоблокировка, т.е. заблокированный поток так и останется ждать). Причиной является то, что контейнер пытается получить мьютекс несколько раз до его освобождения (вызова unlock), что невозможно. Здесь и выходит на сцену std::recursive_mutex, который позволяет получать тот же мьютекс несколько раз. Максимальное количество получения мьютекса не определено, но если это количество будет достигно, то lock бросит исключение std::system_error. Поэтому, решение проблемы в коде выше (кроме изменения реализации addrange(), чтобы не вызывались lock и unlock), заключается в замене мьютекса на std::recursive_mutex.

template <typename T>
class container 
{
     std::recursive_mutex _lock;
     // ...
};

Теперь, результат работы программы будет следующего вида:

6334
18467
41
6334
18467
41
6334
18467
41

Вы, наверное, заметили, что при вызове threadFunction(), генерируются одни и те же числа. Это происходит потому, что функция void srand (unsigned int seed); инициализирует seed только для потока main. В других потоках, генератор псевдо-случайных чисел не инициализируется и получаются каждый раз одни и те же числа.
Явная блокировка и разблокировка могут привести к ошибкам, например, если вы забудете разблокировать поток или, наоборот, будет неправильный порядок блокировок — все это вызовет deadlock. Std предоставляет несколько классов и функций для решения этой проблемы.
Классы «обертки» позволяют непротиворечиво использовать мьютекс в RAII-стиле с автоматической блокировкой и разблокировкой в рамках одного блока. Эти классы:

• lock_guard: когда объект создан, он пытается получить мьютекс (вызывая lock()), а когда объект уничтожен, он автоматически освобождает мьютекс (вызывая unlock())
• unique_lock: в отличие от lock_guard, также поддерживает отложенную блокировку, временную блокировку, рекурсивную блокировку и использование условных переменных

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

template <typename T>
class container 
{
     std::recursive_mutex _lock;
     std::vector<T> _elements;
public:
     void add(T element) 
     {
          std::lock_guard<std::recursive_mutex> locker(_lock);
          _elements.push_back(element);
     }
     void addrange(int num, ...)
     {
          va_list arguments;
          va_start(arguments, num);
          for (int i = 0; i < num; i++)
          {
               std::lock_guard<std::recursive_mutex> locker(_lock);
               add(va_arg(arguments, T));
          }
          va_end(arguments); 
     }
     void dump()
     {
          std::lock_guard<std::recursive_mutex> locker(_lock);
          for(auto e: _elements)
               std::cout << e << std::endl;
     }
};

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

‘std::lock_guard<_Mutex>::lock_guard(_Mutex &)' : cannot convert parameter 1 from ‘const std::recursive_mutex' 
                                                  to ‘std::recursive_mutex &'

Мьютекс (не зависимо от формы реализации), должен быть получен и освобожден, а это подразумевает использование не константных методов lock() и unlock(). Таким образом, аргумент lock_guard не может быть константой. Решение этой проблемы заключается в том, чтобы сделать мьютекс mutable, тогда спецификатор const будет игнорироваться и это позволит изменять состояние из константных функций.

template <typename T>
class container 
{
     mutable std::recursive_mutex _lock;
     std::vector<T> _elements;
public:
     void dump() const
     {
          std::lock_guard<std::recursive_mutex> locker(_lock);
          for(auto e: _elements)
               std::cout << e << std::endl;
     }
};

Конструкторы классов «оберток» могут принимать параметр, определяющий политику блокировки:

• defer_lock типа defer_lock_t: не получать мьютекс
• try_to_lock типа try_to_lock_t: попытаться получить мьютекс без блокировки
• adopt_lock типа adopt_lock_t: предполагается, что у вызывающего потока уже есть мьютекс

Объявлены они следующим образом:

struct defer_lock_t { };
struct try_to_lock_t { };
struct adopt_lock_t { };
 
constexpr std::defer_lock_t defer_lock = std::defer_lock_t();
constexpr std::try_to_lock_t try_to_lock = std::try_to_lock_t();
constexpr std::adopt_lock_t adopt_lock = std::adopt_lock_t();

Помимо «оберток» для мьютексов, std также предоставляет несколько методов для блокировки одного или нескольких мьютексов:

• lock: блокирует мьютекс, используя алгоритм избегания deadlock'ов (используя lock(), try_lock() и unlock())
• try_lock: пытается блокировать мьютексы в порядке, в котором они были указаны

Вот типичный пример возникновения взаимоблокировки (deadlock): у нас есть некий контейнер с элементами и функция exchange(), которая меняет местами два элемента разных контейнеров. Для потокобезопасности, функция синхронизирует доступ к этим контейнерам, получая мьютекс, связанный с каждым контейнером.

template <typename T>
class container 
{
public:
     std::mutex _lock;
     std::set<T> _elements;
     void add(T element) 
     {
          _elements.insert(element);
     }
     void remove(T element) 
     {
          _elements.erase(element);
     }
};
 
void exchange(container<int> &c1, container<int> &c2, int value)
{
     c1._lock.lock();
     std::this_thread::sleep_for(std::chrono::seconds(1)); // симулируем deadlock
     c2._lock.lock();    
     c1.remove(value);
     c2.add(value);
     c1._lock.unlock();
     c2._lock.unlock();
}

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

int main()
{
     srand((unsigned int)time(NULL));
     container<int> cntr1; 
     cntr1.add(1);
     cntr1.add(2);
     cntr1.add(3);
     container<int> cntr2; 
     cntr2.add(4);
     cntr2.add(5);
     cntr2.add(6);
     std::thread t1(exchange, std::ref(cntr1), std::ref(cntr2), 3);
     std::thread t2(exchange, std::ref(cntr2), std::ref(cntr1), 6);
     t1.join();
     t2.join();
     return 0;
}

Для решения этой проблемы можно использовать std::lock, который гарантирует блокировку безопасным (с точки зрения взаимоблокировки) способом:

void exchange(container<int> &c1, container<int> &c2, int value)
{
     std::lock(c1._lock, c2._lock); 
     c1.remove(value);
     c2.add(value);
     c1._lock.unlock();
     c2._lock.unlock();
}

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

Автор: Renzo

Источник