UNIX Network Programming. Volume 2 Second Edition. Interprocess Communications

46    return(NULL); /* массив полный, готово */

47   }

48   shared.buff[shared.nput] = shared.nval;

49   shared.nput++;

50   shared.nval++;

51   Pthread_mutex_unlock(shared.mutex);

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

53  }

54 }

55 void *

56 consume(void *arg)

57 {

58  int i;

59  for (i = 0; i nitems; i++) {

60   if (shared.buff[i] != i)

61    printf("buff[%d] = %d\n", i, shared.buff[i]);

62  }

63  return(NULL);

64 }

Формирование данных

42-53 Критическая область кода производителя состоит из проверки на достижение конца массива (завершение работы)

if (shared.nput = nitems)

и трех строк, помещающих очередное значение в массив:

shared.buff[shared.nput] = shared.nval;

shared.nput++;

shared.nval++;

Мы защищаем эту область с помощью взаимного исключения, не забыв разблокировать его после завершения работы. Обратите внимание, что увеличение элемента count (через указатель arg) не относится к критической области, поскольку у каждого потока счетчик свой (массив count в функции main). Поэтому мы не включаем эту строку в блокируемую взаимным исключением область. Один из принципов хорошего стиля программирования заключается в минимизации объема кода, защищаемого взаимным исключением.

Потребитель проверяет содержимое массива

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

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

solaris % prodcons2 1000000 5

count[0] = 167165

count[1] = 249891

count[2] = 194221

count[3] = 191815

count[4] = 196908

Как мы отмечали ранее, если убрать вызов set_concurrency, в системе Solaris 2.6 значение count[0] будет 1000000, а все остальные счетчики будут нулевыми.

Если убрать из этого примера блокировку с помощью взаимного исключения, он перестанет работать, как и предполагается. Потребитель обнаружит множество элементов buff[i], значения которых будут отличны от i. Также мы можем убедиться, что удаление блокировки ничего не изменит, если будет выполняться только один поток.

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

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

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

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

//mutex/prodcons3.c

14 int

15 main(int argc, char **argv)

16 {

17  int i, nthreads, count[MAXNTHREADS];

18  pthread_t tid_produce[MAXNTHREADS], tid_consume;

19  if (argc != 3)

20   err_quit("usage: prodcons3 #items #threads");

21  nitems = min(atoi(argv[1]), MAXNITEMS);

22  nthreads = min(atoi(argv[2]), MAXNTHREADS);

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

24  Set_concurrency(nthreads + 1);

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

26   count[i] = 0;

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

28  }

29  Pthread_create(tid_consume, NULL, consume, NULL);

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