используется несколькими процессами). Выглядит это обычно следующим образом:

блокировать п1и1ех(...): критическая область разблокировать п1и1ех(...);

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

Взаимные исключения по стандарту Posix объявлены как переменные с типом pthread mutex t. Если переменная-исключение выделяется статически, ее можно инициализировать константой PTHREAD MUTEX INITIALIZER: static pthreacl niutex t lock=PTHREAD MUTEX INITIALIZER:

При динамическом выделении памяти под взаимное исключение (например, вызовом mai 1 ос) или при помещении его в разделяемую память мы должны инициализировать эту переменную во время выполнения, вызвав функцию pthread mutex init, как показано в разделе 7.7.

ПРИМЕЧАНИЕ -

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

Следующие три функции используются для установки и снятия блокировки взаимного исключения: #inclucle <pthreacl.h>

int pthreacl niutexJock(pthreacl niutex t *mptr): int pthreacl niutex trylock(pthreacl niutex t *nptr): int pthreacl niutex unlock(pthreacl niutex t *wptr)\

I* Bee три возвращают О в случае успешного завершения, положительное значение Еххх -в случае ошибки */

При попытке заблокировать взаимное исключение, которое уже заблокировано другим потоком, функция pthread mutex lock будет ожидать его разблокирования, а pthread mutex trylock (неблокируемая функция) вернет ошибку с кодом EBUSY.

ПРИМЕЧАНИЕ-

Если несколько процессов ожидают освобождения взаимного исключения, какой из них начнет выполняться первым? Одна из возможностей, добавленных стандартом 1003.1Ь-1993, заключается в установке приоритета потоков. Мы не будем углубляться в эту область, отметим лишь, что разным потокам могут быть присвоены разные значения приоритета и функции синхронизации (работающие с взаимными исключениями, блокировками чтения-записи и семафорами) будут передавать управление заблокированному потоку с наивысшим значением приоритета. Раздел 5.5 книги [3] описывает возможности планирования выполнения в Posix.l более подробно.

Хотя мы говорим о защите критической области кода программы, на самом деле речь идет о защите данных, с которыми работает эта часть кода. То есть вза-

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

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

7.3. Схема производитель-потребитель

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

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

grep pattern chapters.* wc -1 является примером такой задачи. Программа grep выступает как производитель (единственный), а wc - как потребитель (тоже единственный). Канал используется как форма IPC. Требуемая синхронизация между производителем и потребителем обеспечивается ядром, обрабатывающим команды write производителя и read покупателя. Если производитель опережает потребителя (канал переполняется), ядро приостанавливает производителя при вызове write, пока в канале не появится место. Если потребитель опережает производителя (канал опустошается), ядро приостанавливает потребителя при вызове read, пока в канале не появятся данные.

Такой тип синхронизации называется неявным; производитель и потребитель не знают о том, что синхронизация вообще осуществляется. Если бы мы использовали очередь сообщений Posix или System V в качестве средства IPC между производителем и потребителем, ядро снова взяло бы на себя обеспечение синхронизации.

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

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

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

поток-производитель

поток-производитель

поток-произво-дитель

отправка элементов. в буфер

ЬиЩО]: ЬиЩ1]: ЬиЩ2]; ЬиЩЗ]:

bufflnitems-1]:

nitems-1

получение элементов из буфера

поток-потребитель

процесс

I--------------------------------------------------

Рис. 7.1. Производители и потребитель

Листинг 7.1Функция main

mutex/prodconsZ.c

#inclucle unpipc.h

4 5 б 7 8 9 10 11 12

#clefine MAXNITEMS #define MAXNTHREADS

1000000

int nitems: struct {

pthreacl niutex t mutex:

int buff[MAXNITEMS];

int nput:

int nval: } shared = {

PTHREAD MUTEX INITIALIZER

/* только для чтения потребителем и производителем */

13 void *produce(void *). *consume(void *);

14 int

15 main(int argc. char **argv)

16 {

17 int i. nthreads. count[MAXNTHREADS]:

18 pthread t tidj)roduce[MAXNTHREADS]. tid consume:

19 if (argc !- 3)

20 err quit( usage: prodcons2 <#items> <#threads> );

21 nitems = min(atoi(argv[l]). MAXNITEMS);

22 nthreads - min(atoi(argv[2]). MAXNTHREADS);

23 Set concurrency(nthreads):

24 /* запуск всех потоков-производителей */

25 for (i = 0; i < nthreads; i++) {

26 count[i] = 0:

27 Pthread create(&tid produce[i]. NULL, produce. &count[i]):