7.4. Блокировка и ожидание

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

В листинге 7.3 приведен текст функции main. Начало кода (до объявления функции main) не претерпело никаких изменений по сравнению с листингом 7.1.

Листинг 7.3. Функция main: запуск потребителя сразу после запуска производителей

mutex/prodconsS.c

14 int

15 niain(int argc. char **argv)

16 {

17 int i. nthreads. count[MAXNTHREAOS]:

18 pthread t tid produce[MAXNTHREADS]. tid consunie:

19 if (argc !- 3)

20 err quit( usage: prodcons3 <#itenis> <#threads> ):

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

22 nthreads = min(atoi(argv[2]). MAXNTHREADS):

23 /* создание всех производителей и одного потребителя */

24 Set concurrency(nthreads + 1):

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

26 countEi] = 0:

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

28 }

29 Pthread create(&tid consume. NULL, consume. NULL):

30 /* ожидание завершения производителей и потребителя */

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

32 PthreadJoin(tid produce[i]. NULL);

33 printf( count[d] = d\n . i. count[i]);

34 }

35 PthreadJoin(tid consume. NULL);

36 exit(O);

37 }

24 Мы увеличиваем уровень параллельного выполнения на единицу, чтобы учесть поток-потребитель, выполняемый параллельно с производителями.

25-29 Поток-потребитель создается сразу же после создания потоков-производителей.

Функция produce по сравнению с листингом 7.2 не изменяется, в листинге 7.4 приведен текст функции consume, вызывающей новую функцию consume wait.

Листинг 7.4. Функции consume и consume wait

mutex/prodconsS.c

54 void

55 consunie wait(int i)

56 {

57 for ( : ; ) {

58 Pthread niutex lock(&shared.mutex);

59 if (i < shared.nput) {

60 Pthread niutex unlock(&shared.mutex);

61 return: /* элемент готов */

62 }

63 Pthread mutex unlock(&shared.mutex);

64 }

65 }

66 void *

67 consume(void *arg)

68 {

69 int i;

70 for (i = 0; i < nitems; i++) {

71 consume wait(i);

72 if (shared.buffEi] != i)

73 printf( buff[M] - M\n . i. Shared.buffEi]);

74 }

75 return(NULL);

76 }

Потребитель должен ждать

71 Единственное изменение в функции consume заключается в добавлении вызова consume wait перед обработкой следующего элемента массива.

Ожидание производителей

57-64 Наша функция consume wait должна ждать, пока производители не создадут i -й элемент. Для проверки этого условия производится блокировка взаимного исключения и значение i сравнивается с индексом производителя nput. Блокировка необходима, поскольку значение nput может быть изменено одним из производителей в момент его проверки.

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

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

7.5. Условные переменные: ожидание и сигнализация

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

pthread cond t. Для работы с такими переменными предназначены две функции:

finclude <pthread.h>

int pthread cond wait(pthread cond t *cptr. pthread niutex t *mptr)\ int pthread cond signal(pthread cond t *cptr)\

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

Слово signal в имени второй функции не имеет никакого отношения к сигналам Unix SIGxxx.

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

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

Мы проиллюстрируем использование условных переменных, переписав пример из предыдущего раздела. В листинге 7.5 объявляются глобальные переменные.

Переменные производителя и взаимное исключение объединяются в структуру

7-13 Две переменные nput и nval ассоциируются с mutex, и мы объединяем их в структуру с именем put. Эта структура используется производителями.

14-20 Другая структура, nready, содержит счетчик, условную переменную и взаимное исключение. Мы инициализируем условную переменную с помощью PTHREAD COND INITIALIZER.

Функция main по сравнению с листингом 7.3 не изменяется.

Листинг 7.5. Глобальные переменные: использование условной переменной

mutex/prodconse.c