Наконец, состояние состязания возникает в том случае, когда в программе делается предположение об определенном порядке выполнения блокировок или об их атомарности, что оказывается неверным. Например, рассмотрим пользователя класса
Queue
, который опрашивает первый элемент очереди и при определенном условии извлекает его из очереди с помощью функции
dequeue
.
if (q.getFront == "Cyrus") {
str = q.dequeue;
// ...
Этот фрагмент программного кода хорошо работает в однопоточной среде, потому что
q
не может быть модифицирован в промежутке между первой и второй строкой. Однако
в условиях многопоточной обработки, когда практически в любой момент другой поток может модифицировать
q
, следует исходить из предположения, что совместно используемые объекты модифицируются, когда поток не блокирует доступ к ним. После строки 1 другой поток, работая параллельно, может извлечь следующий элемент из
q
при помощи функции
dequeue
, что означает получение в строке 2 чего-то неожиданного или совсем ничего. Как функция
getFront
, так и функция
dequeue
блокирует один объект
mutex
, используемый для модификации
q
, но между их вызовами мьютекс разблокирован, и, если другой поток находится в ожидании выполнения блокировки, он может это сделать до того, как получит свой шанс строка 2.
Проблема состояния состязания в этом конкретном случае решается путем гарантирования сохранения блокировки на весь период выполнения операции. Создайте функцию-член
dequeueIfEquals
, которая извлекает следующий объект из очереди, если он равен аргументу. Функция
dequeueIfEquals
может использовать блокировку, как и всякая другая функция.
T dequeueIfEquals(const T& t) {
boost::mutex::scoped_lock lock(mutex_);
if (list_.front == t)
// ...
Существуют состояния состязания другого типа, но этот пример должен дать общее представление о том, чего следует остерегаться. По мере увеличения количества потоков и совместно используемых ресурсов состояния состязания оказываются более изощренными и обнаруживать их сложнее. Поэтому следует быть особенно осторожным на этапе проектирования, чтобы не допускать их.
В многопоточной обработке самое сложное — гарантировать сериализованный доступ к ресурсам, потому что если это сделано неправильно, отладка становится кошмаром. Поскольку многопоточная программа по своей сути недетерминирована (так как потоки могут выполняться в различной очередности и с различными квантами времени при каждом новом выполнении программы), очень трудно точно обнаружить место и способ ошибочной модификации чего-либо. Здесь еще в большей степени, чем в однопоточном программировании, надежный проект позволяет минимизировать затраты на отладку и переработку.
12.3. Уведомление одного потока другим
Проблема
Используется шаблон, в котором один поток (или группа потоков) выполняет какие-то действия, и требуется сделать так, чтобы об этом узнал другой поток (или группа потоков). Может использоваться главный поток, который передает работу подчиненным потокам, или может использоваться одна группа потоков для пополнения очереди и другая для удаления данных из очереди и выполнения чего-либо полезного.
Решение
Используйте объекты
mutex
и
condition
, которые объявлены в boost/thread/mutex.hpp и boost/thread/condition.hpp. Можно создать условие (
condition
) для каждой ожидаемой потоками ситуации и при возникновении такой ситуации уведомлять все ее ожидающие потоки. Пример 12.4 показывает, как можно обеспечить передачу уведомлений в модели потоков «главный/подчиненные».
Пример 12.4. Передача уведомлений между потоками
#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/mutex.hpp>
#include <list>
#include <string>
class Request { /*...*/ };
//
Простой класс очереди заданий; в реальной программе вместо этого класса
// используйте std::queue
template<typename T>
class JobQueue {
public:
JobQueue {}
~JobQueue {}
void submitJob(const T& x) {
boost::mutex::scoped_lock lock(mutex_);
list_.push_back(x);
workToBeDone_.notify_one;
}
T getJob {
boost::mutex::scoped_lock lock(mutex_);
workToBeDone_.wait(lock); // Ждать удовлетворения этого условия, затем
// блокировать мьютекс
T tmp = list_.front;
list_.pop_front;
return(tmp);
}
private:
std::list<T> list_;
boost::mutex mutex_;
boost::condition workToBeDone_;
};
JobQueue<Request> myJobQueue;
void boss {
for (;;) {
// Получить откуда-то запрос
Request req;
myJobQueue.submitJob(req);
}
}
void worker {
for (;;) {
Request r(myJobQueue.getJob);
// Выполнить какие-то действия с заданием...
}
}
int main {
boost::thread thr1(boss);
boost::thread thr2(worker);
boost::thread thr3(worker);
thr1.join;
thr2.join;
thr3.join;
}
Обсуждение
Объект условия использует мьютекс
mutex
и позволяет дождаться ситуации, когда он становится заблокированным. Рассмотрим пример 12.4, в котором представлена модифицированная версии класса
Queue
из примера 12.2. Я модифицировал очередь
Queue
, получая более специализированную очередь, а именно
JobQueue
, объекты которой являются заданиями, поступающими в очередь со стороны главного потока и обрабатываемыми подчиненными потоками.