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

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

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

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

23 Функция set_concurrency (наша собственная) указывает подсистеме потоков количество одновременно выполняемых потоков. В Solaris 2.6 она просто вызывает thr_setconcurrency, причем ее запуск необходим, если мы хотим, чтобы у нескольких процессов-производителей была возможность начать выполняться. Если мы не сделаем этого вызова в системе Solaris, будет запущен только первый поток. В Digital Unix 4.0B наша функция 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. Счетчики инициализируются значением 0, и каждый поток увеличивает значение своего счетчика на 1 при помещении очередного элемента в буфер. Содержимое массива счетчиков затем выводится на экран, так что мы можем узнать, сколько элементов было помещено в буфер каждым из потоков.

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

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

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

//mutex/prodcons2.с
39 void *
40 produce(void *arg)
41 {
42 for (;;) {
43 Pthread_mutex_lock(&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 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;