Гунько А. В. Системное программное обеспечение

3.5. Средства синхронизации потоков в Linux

Взаимные исключения (mutual exclusion — mutex) и условные переменные (conditional variables) являются основными средствами синхронизации действий нескольких программных потоков или процессов. Обычно это требуется для предоставления нескольким потокам или процессам совместного доступа к данным.

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

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

Взаимное исключение (mutex) является простейшей формой синхронизации. Оно используется для защиты критической области (critical region), предотвращая одновременное выполнение участка кода несколькими потоками или процессами:

блокировать_mutex(…);
критическая область
разблокировать_mutex(…);

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

Взаимные исключения по стандарту Posix объявлены как переменные с типом pthread_mutex_t. Если переменная-исключение выделяется статически, ее можно инициализировать константой PTHREAD_MUTEX_INITIALIZER:

static pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;

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

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

#include

int pthread_mutex_lock(pthread_mutex_t *mptr);

int pthread_mutex_trylock (pthread_mutex_t *mptr);

int pthread_mutex_unlock (pthread_mutex_t *mptr);

Все три возвращают 0 в случае успешного завершения или положительное значение Еххх в случае ошибки.

При попытке заблокировать взаимное исключение, которое уже заблокировано другим потоком, функция pthread_mutex_lock будет ожидать его разблокирования, a pthread_mutex_trylock (неблокируемая функция) вернет ошибку с кодом BUSY.

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

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

В одном процессе у нас имеется несколько потоков-производителей и один поток-потребитель. Целочисленный массив buff содержит производимые и потребляемые данные (данные совместного пользования) Для простоты производители просто устанавливают значение buff[0] в 0, buff[1] в 1 и т. д. Потребитель перебирает элементы массива, проверяя правильность записей.

Рис. 3.16. Схема задачи производителей – потребителей

В этом первом примере важна синхронизация между потоками-производителями. Поток-потребитель не будет запущен, пока все производители не завершат свою работу. В программе prodcons_on.c, доступной на сайте в папке «Mutex», приведен текст примера.

Буфер buff и переменные nput, nval совместно используются потоками. Они объединены в структуру shared вместе с взаимным исключением, чтобы подчеркнуть, что доступ к ним можно получить только вместе с ним. Переменная nput хранит индекс следующего элемента массива buff, подлежащего обработке, a nval содержит следующее значение, которое должно быть в него помещено. Под структуру выделяется память и инициализируется взаимное исключение, используемое для синхронизации потоков-производителей.

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

Каждый из создаваемых потоков-производителей вызывает функцию produce. Идентификаторы потоков хранятся в массиве tid_produce. Аргументом каждого потока-производителя является указатель на элемент массива count. Счетчики инициализируются значением 0, и каждый поток увеличивает значение своего счетчика на 1 при помещении очередного элемента в буфер. Содержимое массива счетчиков затем выводится на экран, так что можно узнать, сколько элементов было помещено в буфер каждым из потоков.

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

Критическая область кода производителя состоит из проверки на достижение конца массива (завершение работы):

if (shared.nput >= nitems)

и строк, помещающих очередное значение в массив:

shared.buff[shared.nput] = shared.nval;

shared.nput++; shared.nval++;

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

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

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

Если убрать из этого примера (см. prodcons_off.c) блокировку с помощью взаимного исключения, он перестанет работать, как предполагается. Потребитель обнаружит ряд элементов buff[i], значения которых будут отличны от i. Другой вариант его работы - несколько производителей будут формировать одни и те же элементы данных.

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

Продемонстрируем, что взаимные исключения предназначены для блокирования, но не для ожидания. Изменим (см. prodcons1.c) пример так, чтобы потребитель запускался сразу после запуска всех производителей, что позволит ему обрабатывать данные сразу по мере их формирования. Теперь придется синхронизовать потребителя с производителями, чтобы первый обрабатывал только данные, уже сформированные последними.

Начало кода (до объявления функции main) не претерпело никаких изменений. Поток-потребитель создается сразу же после создания потоков-производителей. Функция produce не изменяется. Функция consume, вызывает новую функцию consume_wait. Единственное изменение в функции consume заключается в добавлении вызова consume_wait перед обработкой следующего элемента массива.

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

Возникает вопрос, что делать, если нужный элемент еще не готов. Здесь мы повторяем операции в цикле, устанавливая и снимая блокировку и проверяя значение индекса. Это называется опросом (spinning или polling) и является лишней тратой времени процессора.

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

Взаимное исключение используется для блокирования, а условная переменная — для ожидания. Это два различных средства синхронизации, и оба они нужны. Условная переменная представляет собой переменную типа pthread_cond_t. Для работы с такими переменными предназначены две функции:

#include

int pthread_cond_wait(pthread_cond_t *cptr, pthread_mutex_t *mptr);

int pthread_cond_signal(pthread_cond_t *cptr);

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

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

В программе prodcons2.c две переменные nput и nval ассоциируются с mutex, и мы объединяем их в структуру с именем put. Эта структура используется производителями.

Другая структура, nready, содержит счетчик, условную переменную и взаимное исключение, инициализируемое с помощью PTHREAD_COND_INITIALIZER. Функция main по сравнению с предыдущим листингом не изменяется.

Функции produce и consume претерпевают некоторые изменения. Для блокирования критической области в потоке-производителе теперь используется исключение put.mutex. Там же увеличивается счетчик nready.nready, в котором хранится количество элементов, готовых для обработки потребителем. Перед его увеличением проверяется, было ли значение счетчика нулевым, и если нет, то вызывается функция pthread_cond_signal, позволяющая возобновить выполнение всех потоков (здесь - потребителя), ожидающих установки ненулевого значения этой переменной.

Счетчик используется совместно потребителем и производителями, поэтому доступ к нему осуществляется с блокировкой соответствующего взаимного исключения (nready.mutex).

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

При этом выполняются два атомарных действия:

Разблокируется nready.mutex.
Выполнение потока приостанавливается, пока какой-нибудь другой поток не вызовет pthread_cond_signal.

Перед возвращением управления потоку функция pthread_cond_wait блокирует nready.mutex. Если после возвращения из функции обнаруживается, что счетчик имеет ненулевое значение, то этот счетчик уменьшается (зная, что взаимное исключение заблокировано) и разблокируется взаимное исключение. После возвращения из pthread_cond_wait всегда заново проверяется условие, поскольку может произойти ложное пробуждение.

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

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

int dosignal;

pthread_mutex_lock(nready.mutex);

doslgnal= (nready.nready == 0);

nready.nready++;

pthread_mutex_unlock(&nready.mutex;);

if (doslgnal)

pthread_cond_signal(&nready.cond;);

Здесь сигнал условной переменной отправляется только после разблокирования взаимного исключения. Это разрешено стандартом Posix: поток, вызывающий pthread_cond_signal, не обязательно должен в этот момент блокировать связанное с переменной взаимное исключение. Однако Posix говорит, что если требуется предсказуемое поведение при одновременном выполнении потоков, это взаимное исключение должно быть заблокировано процессом, вызывающим pthread_cond_signal.

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

#include

int pthread_cond_broadcast (pthread_cond_t *aptr);

int pthread_cond_timewait (pthread_cond_t, *cptr, pthread_mutex_t *mpfr, const struct timespec *abstime);

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

Функция pthread_cond_timedwait позволяет установить ограничение на время блокирования процесса. Аргумент abstime представляет собой структуру timespec:

struct timespec {

ttme_t tv_sec; /* секунды */

long tv_nsec; /*наносекунды. Заявлен на перспективу */

};

Эта структура задает конкретный момент системного времени, в который происходит возврат из функции, даже если сигнал по условной переменной еще не будет получен. В этом случае возвращается ошибка с кодом ETIMEDOUT. Этот момент представляет собой абсолютное значение времени, а не промежуток. Аргумент abstime задает количество секунд (наносекунды реально не поддерживаются) с 1 января 1970 UTC до того момента времени, в который должен произойти возврат из функции.

Это отличает функцию от select, pselect и poll, которые в качестве аргумента принимают некоторое количество долей секунды, спустя которое должен произойти возврат. (Функция select принимает количество секунд и микросекунд, pselect — секунд, a poll — миллисекунд.) Преимущество использования абсолютного времени заключается в том, что если функция возвратится до ожидаемого момента (например, при перехвате сигнала), ее можно будет вызвать еще раз, не изменяя содержимого структуры timespec.

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

Инициализировать и удалять взаимное исключение и условную переменную можно с помощью функций

#include

int pthread_mutex_init(pthread_mutex_t *mptr, const pthread_mutexattr_t *attr);

int pthread_mutex_destroy (pthread_mutex_t *mptr);

int pthread_cond_init(pthread_cond_t *cptr, const pthread_condattr_t *attr);

int pthread_cond_destroy (pthread_cond_t *cptr);

Все четыре функции возвращает 0 в случае успешного завершения работы или положительное значение Еххх в случае ошибки.

Аргумент mptr должен указывать на переменную типа pthread_mutex_t, для которой должна быть уже выделена память, и тогда функция pthread_mutex_init инициализирует это взаимное исключение. Значение типа pthread_mutexattr_t, на которое указывает второй аргумент функции pthread_mutex_init (attr), задает атрибуты этого исключения. Если этот аргумент содержит нулевой указатель, используются значения атрибутов по умолчанию.

Атрибуты взаимного исключения имеют тип pthread_mutexattr_t, а условной переменной — pthread_condattr_t, и инициализируются и уничтожаются с помощью следующих функций:

#include

int pthread_mutexattr_init(pthread_mutexattr_t *attr);

int pthread_mutexattr_destroy (pthread_mutexattr_t *attr);

int pthread_condattr_init (pthread_condattr_t *attr);

int pthread_condattr_destroy (pthread_condattr_t *attr);

Все четыре функции возвращают 0 в случае успешного завершения или положительное значение Еххх в случае ошибки.

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

#include

int pthread_mutexattr_getpshared(const pthread_mutexattr_t *attr, int *valptr);

int pthread_mutexattr_setpshared (pthread_muutexattr_t *attr, int value);

int pthread_condattr_getpshared(const pthread_condattr_t *attr, int *valptr);

int pthread_condattr_setpshared (pthread_condattr_t *attr, int value);

Две функции get* возвращают текущее значение атрибута через целое, на которое указывает valptr, а две функции set* устанавливают значение атрибута равным значению value. Значение value может быть либо PTHREAD_PROCESS_PRIVATE, либо PTHREAD_PROCESS_SHARED. Последнее также называется атрибутом совместного использования процессами.

Вот как нужно инициализировать взаимное исключение для совместного использования нескольким процессам:

pthread_mutex_t *mptr;

pthread_mutexattr_t mattr;

mptr = malloc(sizeof (pthread_mutex_t));/* адрес */

pthread_mutexattr_init(&mattr;);

pthread mutexattr_setpshared (&mattr;, PTHREAD_PROCESS_SHARED);

pthread_mutex_init(mptr, &mattr;);

Здесь объявляется переменная mattr типа pthread_mutexattr_t, инициализируется значениями атрибутов по умолчанию, а затем устанавливаем атрибут PTRREAD_PROCESS_SHARED, позволяющий совместно использовать взаимное исключение нескольким процессам. Затем pthread_mutex_init инициализирует само исключение с соответствующими атрибутами.

Такая же последовательность команд (с заменой mutex на cond) позволяет установить атрибут PTHREAD_PROCESS_SHARED для условной переменной, хранящейся в разделяемой процессами памяти.

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

Поток также может быть завершен в момент работы с заблокированным ресурсом, если его выполнение отменит (pthread_cancel) другой поток или он сам вызовет pthread_exit. Последнее сомнительно, поскольку поток должен сам знать, блокирует ли он взаимное исключение в данный момент или нет, и в зависимости от этого вызывать pthread_exit. На случай отмены другим потоком можно предусмотреть обработчик сигнала, вызываемый при отмене потока. Если же для потока возникают фатальные условия, это обычно приводит к завершению работы всего процесса.

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

Литература

Гордеев А.В., Молчанов А.Ю. Системное программное обеспечение. - СПб.: Питер, 2002. – 736 с.
Эпплман Д. Windows API и Visual Basic. - М: «Русская редакция», 1999. –926 с.
Материалы по курсу «Системное программное обеспечение» (Гунько А.В.) [Электронный ресурс] // Гунько А.В. – Режим доступа: https://gun.cs.nstu.ru/ssw
Харт Дж. М. Системное программирование в среде Windows. М: Вильямс, 2005. – 592 с.
Стивенс У. UNIX: взаимодействие процессов. - СПб.: Питер, 2002. - 624 с.
Гунько А.В. Системное программное обеспечение. Метод. Указания к лаб. работам №3556. Новосибирск, НГТУ, 2008. – 36 с.