Параллельные потоки — это очень удобно. Класс Thread неплохо делает свою работу на многих платформах. Однако, самостоятельно работающий поток, который не имеет связи с внешним миром (другими потоками), вряд ли полезен в реальном приложении. Вычисления для этого и распараллеливаются, чтобы, так сказать, в несколько рук (потоков) выполнить одну задачу. Возникает задача синхронизации потоков.
Например, в программе есть некоторая строковая переменная, хранящая описание текущего состояния. Это состояние может, например, выводиться в нижней полоске рабочего окна. Теперь представим, что в программе есть два параллельно работающих потока. Первый занимается получением данных из сети, а второй — обработкой базы данных. Допустим, настал некоторый момент времени, когда первый поток принял данные из сети и хочет об этом отрапортовать в строке состояния, записав туда “Принято 16384 байт”. Приблизительно в этот же момент второй поток завершил периодическую проверку базы данных и также желает об этом сообщить пользователю строкой “База данных проверена”. Операция копирования строки не является атомарной, то есть нет никакой гарантии, что во время ее выполнения процессор не переключится на какой-то другой поток, и операция копирования не будет прервана посреди работы. Итак, когда поток номер 1 успел записать в строку состояния слово “Принято”, может так случиться, что процессор активирует поток номер 2, который также начнет запись своей строки и добавит к уже записанному “Принято” строку “База данных про”, но будет прерван первым потоком и т.д. В итоге в переменная может содержать кашу типа “ПрияноБаза данных 1про6в3ерена84 байт”. Вывод такой — результат полностью непредсказуем.
Для решения подобного вроде проблем в мире параллельного программирования существует такое понятие, как блокировка. Суть ее в том, что когда один процесс захватывает блокировку, то все остальные процессы, пытающиеся ее захватить после, будут блокированы до тех пор, пока первый процесс ее не отпустит. Это похоже на дверь в комнату: представим, что наша переменная globalStatus находится в комнате с одной дверью и ключом внутри. Если дверь открыта (блокировка свободна), то в комнате никого нет (никто не работает с переменной). Когда процесс заходит в комнату, он запирает дверь изнутри (захватывает блокировку). После этого процесс может спокойно работать с переменной как угодно долго, так как гарантированно никто другой не войдет в комнату, так как она заперта изнутри, и не будет мешать ему работать с переменной.
Это была идея простейшей блокировки, которую часто называют мьютекс (mutex). Сейчас мы рассмотрим реализацию такой блокировки на С++, которая будет работать в Windows и UNIX. Как я писал в статье про параллельные потоки, в мире UNIX стандартом де-факто является библиотека pthread (POSIX Threads). Имеено ее мы и будем использовать для UNIX-версии. Для Windows будет отдельная реализация.
Класс Mutex получился весьма простой, в виде единственного файла mutex.h. Пространство имен (namespace) называется ext для простоты. Переименуйте его, если это требуется для вашего проекта.
mutex.h:
#ifndef _EXT_MUTEX_H #define _EXT_MUTEX_H #ifdef WIN32 #define WIN32_LEAN_AND_MEAN #define NOGDI #include <windows.h> #else #include <stdlib.h> #include <pthread.h> #endif namespace ext { #ifdef WIN32 typedef CRITICAL_SECTION MutexType; #else typedef pthread_mutex_t MutexType; #endif // Интерфейс класса Mutex. // Класс задумывался как маленький и быстрый, поэтому все // определено прямо в заголовочном файле, и все функции // объявлены принудительно inline. Это должно уберечь // от ошибок и предупреждений о двойных символах при // включении mutex.h в несколько модулей. class Mutex { public: inline Mutex(); // Деструктор объявлен как не виртуальный из-за тех же // соображений эффективности. Если вы планируете // наследоваться от этого класса, то лучше сделать // деструктор виртуальным, так как наследование от класса // с не виртуальным деструктором потенциально опасно // с точки зрения утечек памяти и является одним из // больших "no-no" в С++. inline ~Mutex(); // Функция захвата блокировки (вход в комнату и запирание // двери ключом изнутри). inline void Lock(); // Функция освобождения блокировки (отпирание двери и // выход из комнаты) inline void Unlock(); private: MutexType __mutex; // Защита от случайного копирования объекта данного класса. // Экземпляр этого класса с трудом может быть нормально // скопирован, так как он жестко привязан к системному // ресурсу. Mutex(const Mutex&); void operator=(const Mutex&); }; #ifdef WIN32 // Реализация через Windows API Mutex::Mutex() { InitializeCriticalSection(&__mutex); } Mutex::~Mutex() { DeleteCriticalSection(&__mutex); } void Mutex::Lock() { EnterCriticalSection(&__mutex); } void Mutex::Unlock() { LeaveCriticalSection(&__mutex); } #else // WIN32 // UNIX версия через pthread Mutex::Mutex() { pthread_mutex_init(&__mutex, NULL); } Mutex::~Mutex() { pthread_mutex_destroy(&__mutex); } void Mutex::Lock() { pthread_mutex_lock(&__mutex); } void Mutex::Unlock() { pthread_mutex_unlock(&__mutex); } #endif // WIN32 } // ext #endif
Касаемо техники “защиты” объекта в С++ от случайного копирования я уже писал ранее.
Я не стал проверять коды возвратов данных функций для упрощения класса. Могу сказать, что если хоть одна из них завершиться с ошибкой, то это значит, что-то конкретно не так в вашей системе, и приложение по любому не будет работать нормально еще по миллиарду причин.
Пощупаем класс в работе. И конечно, используя unit-тестирование.
Традиционно, для компиляции тестов нам нужна Google Test Framework. Как я уже писал, вы можете скачать мою модификацию этой библиотеки, которая сокращена до двух необходимых файлов gtest/gtest.h и gtest-all.cc.
Для теста нам также потребуются файлы thread.cpp и thread.h из статьи про параллельные потоки в С++.
mutex_unittest.cpp:
#include "gtest/gtest.h" #include "mutex.h" #include "thread.h" // Макрос для осуществления задержки в миллисекундах #ifdef WIN32 #include <windows.h> #define msleep(x) Sleep(x) #else #include <unistd.h> #define msleep(x) usleep((x)*1000) #endif // Определим параллельный поток, который будет // "конкурировать" с основным потоком за блокировку. // Данный поток будет пытаться захватить блокировку, // изменить значение флага и освободить затем // блокировку. class A: public ext::Thread { public: // Передаем в конструкторе ссылку на флаг и // ссылку на блокировку. A(volatile int& flag, ext::Mutex& mutex) : __flag(flag), __mutex(mutex) {} virtual void Execute() { // Захват блокировки (1) __mutex.Lock(); // Изменяет флаг на 1 __flag = 1; // Освобождаем блокировку __mutex.Unlock(); } private: volatile int& __flag; ext::Mutex& __mutex; }; TEST(MutexTest, Generic) { // Начальное значение флага - 0. volatile int flag = 0; // Создаем объект-блокировку ext::Mutex mutex; // Захватываем блокировку. mutex.Lock(); // Создаем параллельный поток выполнения. A a(flag, mutex); // Запускаем его. a.Start(); // Ждем для проформы десятую секунды, чтобы дать // время параллельному потоку создаться и успеть // дойти до строки (1), то есть до захвата блокировки. msleep(100); // Значение флага должно быть все еще 0, так как // параллельный поток должен быть блокирован на // строке (1), так как мы захватили блокировку еще // до его создания. EXPECT_EQ(0, flag); // Освобождаем блокировку, тем самым давая // параллельному потоку выполняться дальше и // изменить значение флага на 1. mutex.Unlock(); // Ждем завершения параллельного потока. a.Join(); // Так как параллельный поток завершился, то // флаг теперь точно должен быть равен 1. EXPECT_EQ(1, flag); }
Для запуска тестов нам нужен стандартный файл запуска runner.cpp:
#include "gtest/gtest.h" int main(int argc, char* argv[]) { testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); }
Компилируем все вместе.
В Visual Studio:
cl /EHsc /I. /Femutex_unittest_vs2008.exe /DWIN32 runner.cpp mutex_unittest.cpp thread.cpp gtest\gtest-all.cc
Или если вы используете gcc:
g++ -I. -o mutex_unittest_cygwin.exe runner.cpp mutex_unittest.cpp thread.cpp gtest/gtest-all.cc
Запускаем mutex_unittest_vs2008.exe или mutex_unittest_cygwin.exe:
[==========] Running 1 test from 1 test case.
[----------] Global test environment set-up.
[----------] 1 test from MutexText
[ RUN ] MutexText.Generic
[ OK ] MutexText.Generic
[----------] Global test environment tear-down
[==========] 1 test from 1 test case ran.
[ PASSED ] 1 test.
Вроде работает как надо.
Теперь внесем в исходный текст класса “случайную” ошибку, заменив строку:
void Mutex::Lock() { EnterCriticalSection(&__mutex); }
на
void Mutex::Lock() { /* EnterCriticalSection(&__mutex); */ }
Этой “ошибкой” мы просто отключили создание блокировки. Перекомпилируем все заново и запустим:
[==========] Running 1 test from 1 test case.
[----------] Global test environment set-up.
[----------] 1 test from MutexText
[ RUN ] MutexText.Generic
mutex_unittest.cpp:41: Failure
Value of: flag
Actual: 1
Expected: 0
[ FAILED ] MutexText.Generic
[----------] Global test environment tear-down
[==========] 1 test from 1 test case ran.
[ PASSED ] 0 tests.
[ FAILED ] 1 test, listed below:
[ FAILED ] MutexText.Generic
1 FAILED TEST
Видно, что флаг был изменен в параллельном потоке вне зависимости от блокировки (и понятно почему, мы ж ее “сломали”).
Итак, можно вернуть исправленную строку в исходное состояние. Класс работает, и тесты на это подтвердили.
При использовании класса Mutex у себя в проекте не забудьте включить файл mutex_unittest.cpp в ваш набор unit-тестов.
В завершении могу сказать, что данный класс успешно работает и проверен мной лично на Windows (32- и 64-бит), Linux 2.6 (32- и 64-бит Intel и SPARC), AIX 5.3 и 6, SunOS 5.2 64-bit SPARC, HP-UX и HP-UX IA64.
Другие посты по теме: