3 #define SEM MUTEX mutex /* аргументы pxjpc name() */

4 #define SEM NEMPTY nempty

5 #define SEM NSTOREO nstored

6 int nitems: /* read-only для производителя и потребителя */

7 struct { /* разделяемые производителем и потребителем данные */

8 int buff[NBUFF]:

9 sem t *mutex. *nempty. *nstored:

10 } shared:

11 void *produce(void *), *consume(vo1d *):

12 int

13 mainCint argc, char **argv)

14 {

15 pthread t tid produce, tid consume:

16 if (argc != 2)

17 err quit( usage: prodconsl <#items> ):

18 nitems - ato1(argv[l]);

19 /* создание трех семафоров */

20 shared.mutex - Sem open(Px ipc name(SEM MUTEX), 0 CREAT OJXCL,

21 FILE MODE, 1);

22 shared.nempty = Sem open(PxJpc name(SEM NEMPTY), 0 CREAT OJXCL,

23 FILE M00E7 NBUFF):

24 shared.nstored - Sem open(Px ipc name(SEM NSTORED). 0 CREAT OJXCL,

25 FILE MODE, 0);

26 /* создание одного потока-производителя и одного потока-потребителя */

27 Set concurrency(2)-,

28 Pthread create(&tid produce, NULL, produce, NULL):

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

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

31 PthreadJoin(tid produce, NULL):

32 PthreadJoin(tid consume, NULL):

33 /* удаление семафоров */

34 Sem unlink(Px ipc name(SEM MUTEX));

35 Sem unlink(PxJpc name(SEM NEMPTY));

36 Sem unlink(PxJpc name(SEM NSTORED)):

37 exit(O):

38 }

Глобальные переменные

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

Создание семафоров

19-25 Мы создаем три семафора, передавая их имена функции pxJ pc name. Флаг D EXCL мы указываем, для того чтобы гарантировать инициализацию каждого семафора

правильным значением. Если после преждевременно завершенного предыдущего запуска программы остались неудаленные семафоры, мы обработаем эту ситуацию, вызвав перед их созданием sein unl ink и игнорируя ошибки. Мы могли бы проверять возвращение ошибки EEXIST при вызове sein open с флагом D EXCL, а затем вызывать sein unl ink и еще раз sein open, но это усложнило бы программу. Если нам нужно проверить, что запущен только один экземпляр программы (что следует сделать перед созданием семафоров), можно обратиться к разделу 9.7, где описаны методы решения этой задачи.

Создание двух потоков

26- 29 Создаются два потока, один из которых является производителем, а другой - потребителем. При запуске никакие аргументы им не передаются.

30-36 Главный поток ждет завершения работы производителя и потребителя, а затем удаляет три семафора.

ПРИМЕЧАНИЕ-

Мы могли бы вызвать для каждого семафора sem cIose, но это делается автоматически при завершении процесса. А вот удалить имя семафора из файловой системы необходимо явно.

В листинге 10.9 приведен текст функций produce и consume.

Листинг 10.9. Функции produce и consume

pxsem/prodconsl.c

39 void *

40 produceCvoid *arg)

41 {

42 int i;

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

44 Sem wait(shared.nempty): /* ожидаем освобождения поля */

45 Sem wait(shared.mutex);

46 shared.buff[i % NBUFF] = i; /* помещаем i в циклический буфер */

47 Sem post(shared.mutex);

48 Sem post(shared.nstored): /* сохраняем еще 1 элемент */

49 }

50 return(NULL):

51 }

52 void *

53 consume(void *arg)

54 {

55 int i;

56 for (i = 0: i < nitems: i++) {

57 Sem wait(shared.nstored): /* ожидаем появления объекта в буфере */

58 Sem wait(shared.mutex):

59 if (shared.buff[i % NBUFF] != i)

60 printf( buff[*d] - *d\n\ i, shared.buff[i % NBUFF]);

61 Sem post(shared.mutex);

62 Sem post(shared.nempty); /* еще одно пустое поле */

63 }

64 return(NULL);

65 }

Производитель ожидает освобождения места в буфере

44 Производитель вызывает sein wai t для семафора nempty, ожидая появления свободного места. В первый раз при выполнении этой команды значение семафора nempty уменьшится с NBUFF до NBUFF-1.

Производитель помещает элемент в буфер

45-48 Перед помещением нового элемента в буфер производитель должен установить блокировку на семафор mutex. В нашем примере, где производитель просто сохраняет значение в элементе массива с индексом i % NBUFF, для описания состояния буфера не используется никаких разделяемых переменных (то есть мы не используем связный список, который нужно было бы обновлять каждый раз при помещении элемента в буфер). Следовательно, установка и снятие семафора mutex не являются обязательными. Тем не менее мы иллюстрируем эту технику, потому что обычно ее применение является необходимым в задачах такого рода (обновление буфера, разделяемого несколькими потоками).

После помещения элемента в буфер блокировка с семафора mutex снимается (его значение увеличивается с О до 1) и увеличивается значение семафора nstored. Первый раз при выполнении этой команды значение nstored изменится с начального значения О до 1.

Потребитель ожидает изменения семафора nstored

57-62 Если значение семафора nstored больше О, в буфере имеются объекты для обработки. Потребитель изымает один элемент из буфера и проверяет правильность его значения, защищая буфер в момент доступа к нему с помощью семафора mutex. Затем потребитель увеличивает значение семафора nempty, указывая производителю на наличие свободных полей.

Зависание

Что произойдет, если мы по ошибке поменяем местами вызовы Semwait в функции consumer (листинг 10.9)? Предположим, что первым запускается производитель (как в решении, предложенном для упражнения 10.1). Он помещает в буфер NBUFF элементов, уменьшая значение семафора nempty от NBUFF до О и увеличивая значение семафора nstored от О до NBUFF. Затем производитель блокируется в вызове Sem wait(shared.nempty), поскольку буфер полон и помещать элементы больше некуда.

Запускается потребитель и проверяет первые NBUFF элементов буфера. Это уменьшает значение семафора п sto red от NBUFF до О и увеличивает значение семафора nempty от О до NBUFF. Затем потребитель блокируется в вызове Sem wait( shared, nstored) после вызова Sem wait(shared.mutex). Производитель мог бы продолжать работу, поскольку значение семафора nempty уже отлично от О, но он вызвал Sem wait(shared. mutex) и его выполнение было приостановлено.