19 if (argc != 4)

20 err quit( usage: prodcons4 <#items> <#producers> <#consumers> ):

21 nitems - atoi(argv[l]):

22 nproducers = min(atoi(argv[2]), MAXNTHREADS):

23 nconsumers = mln(atoi(argv[3]), MAXNTHREADS):

24 /* инициализация трех семафоров */

25 Sem init(&shared.mutex, 0, 1);

26 Sem init(&shared.nempty, 0, NBUFF):

27 Sem init(&shared.nstored, 0, 0):

28 /* создание производителей и потребителей */

29 Set concurrency(nproducers + nconsumers):

30 for (i - 0: i < nproducers: i++) {

31 prodcount[i] - 0:

32 Pthread create(&tid produce[i], NULL, produce. &prodcount[i]);

33 }

34 for (i = 0: i < nconsumers; i++) {

35 conscount[i] = 0:

36 Pthread create(&tid consume[i]. NULL, consume. &conscount[1]):

37 }

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

39 for (1 = 0; 1 < nproducers: i++) {

40 PthreadJo1n(tid produce[1]. NULL):

41 printf( producer count[d] = %А\п . 1, prodcount[1]):

42 }

43 for (1 = 0: 1 < nconsumers; i++) {

44 PthreadJo1n(tid consume[i], NULL):

45 pri ntf ( consumer count[M] = d\n , 1, conscount[i]):

46 }

47 Sem destroy(&shared.mutex):

48 Sem destroy(&shared.nempty);

49 Sem destroy(&shared.nstored);

50 exit(O);

51 }

Функция produce содержит одну новую строку по сравнению с листингом 10.13. В части кода, относящейся к заверщению потока-производителя, появляется строка, отмеченная знаком +;

if (shared.nput > nitems) { + Sem post(&shared.nstored); /*даем возможность потребителям завершить работу */

Sem post(&shared.nempty):

Sem post(&shared.mutex); return(NULL): /* готово */ }

Снова нам нужно быть аккуратными при обработке завершения процессов-производителей и потребителей. После обработки всех объектов в буфере все потребители блокируются в вызове

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

Производителям приходится увеличивать семафор nstored для разблокирования потребителей, чтобы они узнали, что работа завершена.

Функция consume приведена в листинге 10.17.

Листинг 10.17. Функция, выполняемая всеми потоками-потребителями

pxsem/prodcons4.c

72 void *

73 consume(void *arg)

74 {

75 int i:

76 for ( ; : ) {

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

78 Sem wait(&shared.mutex):

79 if (shared.nget >= nitems) {

80 Sem post(&shared.nstored):

81 Sem post(&shared.mutex);

82 return(NULL): /* готово */

83 }

84 i = shared.nget % NBUFF:

85 if (shared.buff[l] !- shared.ngetval)

86 printfCerror: buff[M] - %й\п . i. shared.buff[i]);

87 shared.nget++;

88 shared.ngetva1++;

89 Sem post(&sha red.mutex):

90 Sem post(&shared.nempty); /* освобождается место для элемента */

91 *((int *) arg) +- 1:

92 }

93 }

Завершение потоков-потребителей

79-83 Функция consume сравнивает nget и nitems, чтобы узнать, когда следует остановиться (аналогично функции produce). Обработав последний объект в буфере, потоки-потребители блокируются, ожидая изменения семафора nstored. Когда завершается очередной поток-потребитель, он увеличивает семафор nstored, давая возможность завершить работу другому потоку-потребителю.

10.11. Несколько буферов

Во многих программах, обрабатывающих какие-либо данные, можно встретить цикл вида

while ( (п - read(fdin. buff. BUFFSIZE)) > 0) { /* обработка данных */ write(fdout, buff, n):

Например, программы, обрабатывающие текстовые файлы, считывают строку из входного файла, выполняют с ней некоторые действия, а затем записывают строку в выходной файл. Для текстовых файлов вызовы read и wri te часто заменяются на функции стандартной библиотеки ввода-вывода fgets и fputs.

На рис. 10.11 изображена иллюстрация к такой схеме. Здесь функция reader считывает данные из входного файла, а функция writer записывает данные в выходной файл. Используется один буфер.

входной

файл

выходной файл

входной

файл

выходной файл

25

31

V read ()

р> write ()

У read ()

> write ()

время

Рис. 10.10. Процесс считывает данные в буфер, а потом записывает его содержимое в другой файл