28 }

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

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

31 PthreadJo1n(tid produce[i]. NULL):

32 printf( countEd] = d\n , i. countEi]):

33 }

34 /* запуск и ожидание завершения потока-потребителя */

35 Pthread create(&tid consunie. NULL, consume. NULL):

36 PthreadJoin(tid consume. NULL):

37 exit(O):

38 }

Совместное использование глобальных переменных потоками

4-12 Эти переменные совместно используются потоками. Мы объединяем их в структуру с именем shared вместе с взаимным исключением, чтобы подчеркнуть, что доступ к ним можно получить только вместе с ним. Переменная nput хранит индекс следующего элемента массива buff, подлежащего обработке, а nval содержит следующее значение, которое должно быть в него помещено (О, 1, 2 и т. д.). Мы выделяем память под эту структуру и инициализируем взаимное исключение, используемое для синхронизации потоков-производителей.

ПРИМЕЧАНИЕ -

Мы всегда будем стараться размещать совместно используемые данные вместе со средствами синхронизации, к ним относящимися (взаимными исключениями, условными переменными, семафорами), в одной структуре, как мы сделали в этом примере. Это хороший стиль программирования. Однако во многих случаях совместно используемые данные являются динамическими, представляя собой, например, связный список. Мы, наверное, сможем поместить в структуру первый элемент списка вместе со средствами синхронизации (как в структуре mq hdr в листинге 5.16), но оставшаяся часть списка в структуру не попадет. Следовательно, это решение не всегда является идеальным.

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

19-22 Первый аргумент командной строки указывает количество элементов, которые будут произведены производителями, а второй - количество запускаемых потоков-производителей.

Установка уровня параллельности

23 Функция set concurrency (наша собственная) указывает подсистеме потоков количество одновременно выполняемых потоков. В Solaris 2.6 она просто вызывает thr setconcurrency, причем ее запуск необходим, если мы хотим, чтобы у нескольких процессов-производителей была возможность начать выполняться. Если мы не сделаем этого вызова в системе Solaris, будет запущен только первый поток. В Digital Unix 4.0В наша функция set concurrency не делает ничего, поскольку

в этой системе по умолчанию все потоки процесса имеют равные права на вычислительные ресурсы.

ПРИМЕЧАНИЕ -

Unix 98 требует наличия функции pthread setconcurrency, выполняющей это же действие. Эта функция требуется для тех реализаций, которые мультиплексируют пользовательские потоки (создаваемые функцией pthread create) на небольшое множество выполняемых потоков ядра. Такие реализации часто называются многие-к-немногим (many-to-few), двухуровневые (two-level) или М-на-N (M-to-N). В разделе 5.6 книги [3] отношения между пользовательскими потоками и потоками ядра рассматриваются более подробно.

Создание процессов-производителей

24-28 Создаются потоки-производители, каждый из которых вызывает функцию produce. Идентификаторы потоков хранятся в массиве tid produce. Аргументом каждого потока-производителя является указатель на элемент массива count. Счетчики инициализируются значением О, и каждый поток увеличивает значение своего счетчика на 1 при помещении очередного элемента в буфер. Содержимое массива счетчиков затем выводится на экран, так что мы можем узнать, сколько элементов было помещено в буфер каждым из потоков.

Ожидание завершения работы производителей, запуск потребителя

29-36 Мы ожидаем завершения работы всех потоков-производителей, выводя содержимое счетчика для каждого потока, а затем запускаем единственный процесс-потребитель. Таким образом (на данный момент) мы исключаем необходимость синхронизации между потребителем и производителями. Мы ждем завершения работы потребителя, а затем завершаем работу процесса. В листинге 7.2 приведен текст функций produce и consume.

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

niutex/proclcons2.c

39 void *

40 produce(void *arg)

41 {

42 for ( : : ) {

43 Pthread niutex 1ock(&shared.mutex);

44 if (shared.nput >= nitems) {

45 Pthread mutex unlock(&shared.mutex):

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 printfCbuffEd] = d\n . i. shared.buff[i]):

62 }

63 return(NULL):

64 }

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

42-53 Критическая область кода производителя состоит из проверки на достижение

конца массива (завершение работы)

if (shared.nput >= nitems)

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

shared.buffEshared.nput] = shared.nval: shared.nput++; shared.nval++:

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

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

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

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

Solaris % prodcons2 1000000 5

countEO] = 167165 countEl] = 249891 countE2] = 194221 countE3] - 191815 countE4] = 196908

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

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