UNIX: взаимодействие процессов, Стивенс Уильям Ричард

UNIX: взаимодействие процессов

на обложку

Стивенс Уильям Ричард

Шрифт:

Листинг 10.12. Функция main задачи с несколькими производителями

//pxsem/prodcons3.c

1 #include "unpipc.h"

2 #define NBUFF 10

3 #define MAXNTHREADS 100

4 int nitems, nproducers; /* только для чтения производителем и потребителем */

5 struct { /* общие данные */

6 int buff[NBUFF];

7 int nput;

8 int nputval;

9 sem_t mutex, nempty, nstored; /* семафоры, а не указатели */

10 } shared;

11 void *produce(void *), *consume(void *);

12 int

13 main(int argc, char **argv)

14 {

15 int i, count[MAXNTHREADS];

16 pthread_t tid_produce[MAXNTHREADS], tid_consume;

17 if (argc != 3)

18 err_quit("usage: prodcons3 <#items> <#producers>");

19 nitems = atoi(argv[1]);

20 nproducers = min(atoi(argv[2]), MAXNTHREADS);

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

22 Sem_init(&shared.mutex, 0, 1);

23 Sem_init(&shared.nempty, 0, NBUFF);

24 Sem_init(&shared.nstored, 0, 0);

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

26 Set_concurrency(nproducers + 1);

27 for (i = 0; i < nproducers; i++) {

28 count[i] = 0;

29 Pthread_create(&tid_produce[i], NULL, produce, &count[i]);

30 }

31 Pthread_create(&tid_consume, NULL, consume, NULL);

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

33 for (i = 0; i < nproducers; i++) {

34 Pthread_join(tid_produce[i], NULL);

35 printf("count[%d] = %d\n", i, count[i]);

36 }

37 Pthread_join(tid_consume, NULL);

38 Sem_destroy(&shared.mutex);

39 Sem_destroy(&shared.nempty);

40 Sem_destroy(&shared.nstored);

41 exit(0);

42 }

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

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

Общая структура

5-10 В структуру shared добавляются два новых элемента: nput, обозначающий индекс следующего элемента, куда должен быть помещен объект (по модулю BUFF), и nputval —следующее значение, которое будет помещено в буфер. Эти две переменные взяты из нашего решения в листингах 7.1 и 7.2. Они нужны для синхронизации нескольких потоков-производителей.

Новые аргументы командной строки

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

Запуск всех потоков

21-41 Инициализируем семафоры и запускаем потоки-производители и поток-потребитель. Затем ожидается завершение работы потоков. Эта часть кода практически идентична листингу 7.1.

В листинге 10.13 приведен текст функции produce, которая выполняется каждым потоком-производителем.

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

//pxsem/prodcons3.c

43 void *

44 produce(void *arg)

45 {

46 for (;;) {

47 Sem_wait(&shared.nempty); /* ожидание освобождения поля */

48 Sem_wait(&shared.mutex);

49 if (shared.nput >= nitems) {

50 Sem_post(&shared.nempty);

51 Sem_post(&shared.mutex);

52 return(NULL); /* готово */

53 }

54 shared.buff[shared.nput % NBUFF] = shared.nputval;

55 shared.nput++;

56 shared.nputval++;

57 Sem_post(&shared.mutex);

58 Sem_post(&shared.nstored); /* еще один элемент */

59 *((int *) arg) += 1;

60 }

61 }

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

49-53 Отличие от листинга 10.8 в том, что цикл завершается, когда nitems объектов будет помещено в буфер всеми потоками. Обратите внимание, что потоки-производители могут получить семафор nempty в любой момент, но только один производитель может иметь семафор mutex. Это защищает переменные nput и nval от одновременного изменения несколькими производителями.

Завершение производителей

50-51 Нам нужно аккуратно обработать завершение потоков-производителей. После того как последний объект помещен в буфер, каждый поток выполняет

Sem_wait(&shared.nempty); /* ожидание пустого поля */

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