3. Оборудование

Персональный компьютер типа IBM PC Pentium, опе­рационная система Linux, каталог ovp/lab5 c файлами labos5.doc (методические указания), pipe.c, dup2.c, popen.c и подкаталог pipe c файлами mainpipe.c, client.c, server.c, fduplex.c, mainpopen.c.

4. Задание на работу

4.1. Ознакомиться с различными способами создания и использования неименованных каналов в обычных и клиент-серверных приложениях.

4.2. Перекомпилировать и выполнить тестирование обычных и клиент-серверных программ на языке Си, использующих неименованные каналы.

5. Порядок выполнения работы

5.1. Загрузить операционную систему Linux и после появления приглашения login войти в систему под именем ovp.

5.2. Запустить эмулятор консольного окна. Перейти в каталог lab5 и ознакомиться с содержимым находящихся в нем файлов pipe.c, dup2.c, popen.c, а также файлов mainpipe.c, client.c, server.c, fduplex.c, mainpopen.c в подкаталоге pipe.

5.4. Посредством команд gcc откомпилировать и выполнить данные программы. На различных примерах проверить правильность их работы.

6. Оформление отчета

Отчет должен содержать:

• тексты основных программ;

• примеры тестирования программ.

7. Контрольные вопросы

7.1. Какие объекты IPC используются в различных системах?

7.2. Какое назначение функции fdopen?

7.3. Приведите примеры использования вызова dup2().

7.4. Напишите программу реализации программного канала между двумя дочерними процессами.

7.5. Какое назначение файлового сервера?

7.6. Приведите прототипы функций sprintf() и snprintf() и укажите их назначение.

7.7. В каких случаях процессы превращаются в зомби?

7.8. Как создается двусторонний канал?

7.9. Какими способами можно создать программный канал с утилитой или между двумя утилитами?

7.10. Приведите прототип функции popen().

Лабораторная работа №6

СОЗДАНИЕ И СИНХРОНИЗАЦИЯ ПОТОКОВ

В СИСТЕМЕ LINUX

1. Цель и задачи работы

Ознакомление с потоками системы Linux, предназначенными для параллельного выполнения программного кода процесса, а также получение практических навыков создания многопотоковых программ и синхронизации потоков.

2. Теоретические сведения

2.1. Создание потока

Современные операционные системы предоставляют возможность разбить процесс на ряд подпроцессов, называемых потоками, которые выполняются параллельно. Cистема Linux при создании процесса по умолчанию создает в нем единственный (основной) поток, последовательно выполняющий программный код. Этот поток может создавать дополнительные потоки. Все потоки процесса будут иметь доступ к общему виртуальному пространству, общим дескрипторам файлов и другим системным ресурсам. Если, к примеру, один поток меняет значение переменной, это изменение отразится на других потоках. Точно так же, когда один поток закрывает файл, остальные потоки теряют возможность работать с этим файлом. В связи с тем, что процесс и все его потоки могут выполнять лишь одну программу одновременно, как только один из потоков вызывает функцию семейства exec(), все остальные потоки завершаются (естественно, новая программа может создавать собственные потоки).

В Linux реализована библиотека API-функций работы с потоками, соответствующая стандарту POSIX (она называется Pthreads). Все функции и типы данных библиотеки объявлены в файле <pthread.h>. Эти функции не входят в стандартную библиотеку языка С, поэтому при компоновке программы нужно указывать опцию –lpthread в командной строке.

Для создания программного потока используется функция pthread_create(). При создании поток получает идентификатор – тип данных pthread_t. После создания поток начинает выполнять потоковую функцию - соответствующую часть программы. По завершении этой функции поток уничтожается. В Linux потоковая функция имеет следующий тип:

void* (*) (void*)

Она принимает единственный параметр типа void*, называемый аргументом потока, и возвращает значение аналогичного типа.

Программа, представленная в листинге 1, создает поток, который непрерывно записывает символы 'x' в стандартный поток ошибок. После вызова функции pthread_create() основной поток начинает делать то же самое, но вместо символов 'x' печатаются символы 'o'.

Компиляция и компоновка программы осуществляется следующим образом:

cc –o thread_create thread_create.c –lpthread

Листинг 1. (thread-create.c) Создание дополнительного потока

#include <pthread.h>

#include <stdio.h>

void* print_xs (void* unused)

{

while (1)

fputc ('x', stderr);

return NULL;

}

int main ()

{

pthread_t thread_id;

pthread_create (&thread_id, NULL, &print_xs, NULL);

while (1)

fputc ('o', stderr);

return 0;

}

После запуска программы можно увидеть, что символы 'x' и 'o' чередуются самым непредсказуемым образом.

2.2. Передача данных программному потоку

Потоковый аргумент – удобное средство передачи данных потокам. Но поскольку его тип void*, данные содержатся не в самом аргументе. Он лишь должен указывать на какую-то структуру или массив. Лучше всего создать для каждой потоковой функции собственную структуру, в которой определялись бы "параметры", ожидаемые потоковой функцией.

В программе, приведенной в листинге 2, создаются два дополнительных потока: один отображает символы 'x', а другой – символы 'o'. Чтобы вывод на экран не длился бесконечно, потокам передается дополнительный аргумент, определяющий, сколько раз отобразить символ.

Листинг 2. (thread-create2.c) Создание двух потоков

#include <pthread.h>

#include <stdio.h>

struct char_print_parms

{

char character;

int count;

};

void* char_print (void* parameters)

{

struct char_print_parms* p = (struct char_print_parms*) parameters;

int i;

for (i = 0; i < p->count; ++i)

fputc (p->character, stderr);

return NULL;

}

int main ()

{

pthread_t thread1_id;

pthread_t thread2_id;

struct char_print_parms thread1_args;

struct char_print_parms thread2_args;

/* Создание потока для отображения 30000 символов 'x'. */

thread1_args.character = 'x';

thread1_args.count = 30000;

pthread_create (&thread1_id, NULL, &char_print, &thread1_args);

/* Создание потока для отображения 20000 символов 'o'. */

thread2_args.character = 'o';

thread2_args.count = 20000;

pthread_create (&thread2_id, NULL, &char_print, &thread2_args);

/* Ожидание завершения первого дополнительного потока */

pthread_join (thread1_id, NULL);

/* Ожидание завершения второго дополнительного потока */

pthread_join (thread2_id, NULL);

return 0;

}

Здесь одна и та же функция char_print() эксплуатируется обоими потоками, но каждый из них конфигурируется независимо с помощью структуры char_print_parms.

2.3. Значения, возвращаемые программными потоками

Функция pthread_joint() ожидает завершения потока так же как функция wait() ожидает завершения процесса. Если второй аргумент функции pthread_joint() не равен NULL, то в него помещается значение, возвращаемое потоком. Поскольку тип второго аргумента – указатель void*, то возвращаемое функцией значение (например, типа int) сначала приводится к типу void*, а затем уже вне функции pthread_joint() обратно преобразуется в тип int.

Программа, представленная в листинге 3, в отдельном потоке вычисляет n-е простое число и возвращает его в программу. Тем временем функция main() может продолжать свои собственные вычисления. Отметим, что алгоритм последовательного деления, используемый в потоковой функции compute_prime(), весьма неэффективен. Более мощным алгоритмом является, например, "решето Эратосфена".

Листинг 3. (primes.c) Вычисление простых чисел в дополнительном потоке

#include <pthread.h>

#include <stdio.h>

/* Находим простое число с порядковым номером n, где n - это

значение, на которое указывает параметр arg. */

void* compute_prime (void* arg)

{

int candidate = 2;

int n = *((int*) arg);

while (1) {

int factor;

int is_prime = 1;

/* Проверяем, является ли значение candidate простым числом. */

for (factor = 2; factor < candidate; ++factor)

if (candidate % factor == 0) {

is_prime = 0;

break;

}

if (is_prime) {

if (--n == 0)

/* Возвращаем найденное число в основной поток. */

return (void*) candidate;

}

++candidate;

}

return NULL;

}

int main ()

{

pthread_t thread;

int which_prime = 5000;

int prime;

/* Запускаем поток, вычисляющий 5000-е простое число. */

pthread_create (&thread, NULL, &compute_prime, &which_prime);

/* Выполняем здесь другие действия ... */

pthread_join (thread, (void*) &prime);

printf("The %dth prime number is %d.\n", which_prime, prime);

return 0;

}

Каждый поток имеет набор атрибутов. Потоковые атрибуты – это механизм настройки поведения отдельных потоков. При создании потока через второй аргумент функция pthread_create() принимает указатель на объект атрибутов потока. Если этот указатель равен NULL, поток конфигурируется на основании стандартных атрибутов.

Для большинства Linux-приложений интерес представляет один единственный атрибут (остальные используются в приложениях реального времени): статус отсоединения потока. Поток может быть создан как ожидаемый (по умолчанию) или отсоединенный. Ожидаемый поток, подобно процессу, после своего завершения не удаляется автоматически операционной системой. Код его завершения хранится где-то в системе (как у процесса-зомби), пока какой-нибудь другой поток не вызовет функцию pthread_joint(), чтобы запросить это значение. Только тогда ресурсы потока считаются освобожденными. С другой стороны, отсоединенный поток, завершившись, сразу уничтожается. Другие потоки не могут вызвать по отношению к нему функцию pthread_joint(), чтобы получить возвращаемое значение.

2.4. Отмена потока

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

Во многих случаях поток выполняет код, который нельзя просто взять и прервать. Например, поток может получить какие-то ресурсы, поработать с ними, а затем удалить. Если отмена потока произойдет где-то посередине, освободить занятые ресурсы станет невозможно, вследствие чего они окажутся потерянными для системы. Чтобы учесть эту ситуацию, поток должен определить, где и когда он может быть отменен.

С точки зрения возможности отмены поток находится в одном из трех состояний:

• Асинхронно отменяемый. Такой поток можно отменить в любой точке его выполнения.

• Синхронно отменяемый. Поток можно отменить в определенных местах, называемых точками отмены. Запрос на отмену помещается в очередь, и поток отменяется только при достижении очередной точки отмены.

• Неотменяемый. Попытки отменить поток игнорируются.

Первоначально поток является синхронно отменяемым. Чтобы сделать поток асинхронно отменяемым, используется функция pthread_setcanceltype(). Эта функция влияет на тот поток, в котором она была вызвана. Поток может отказаться удаляться или переключаться, вызвав функцию pthread_setcancelstate().

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

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

В листинге 4 показан пример функции process_transaction(), осуществляющей данную задумку. Функция запрещает отмену потока до тех пор, пока баланс обоих счетов не будет изменен.

Отметим, что по окончании критической секции восстанавливается предыдущее состояние потока, а не режим PTHREAD_CANCEL_ENABLE, разрешающий отмену потока.

Листинг 4. (critical-section.c) Защита транзакции с помощью критической секции

#include <pthread.h>

#include <stdio.h>

#include <string.h>

/* Массив балансов счетов, упорядоченных по номеру счета. */

float* account_balances;

int process_transaction (int from_acct, int to_acct, float dollars)

{

int old_cancel_state;

/* Проверяем баланс на счету from_acct. */

if (account_balances[from_acct] < dollars)

return 1;

/* Начало критической секции. */

pthread_setcancelstate (PTHREAD_CANCEL_DISABLE, &old_cancel_state);

/* Переводим деньги. */

account_balances[to_acct] += dollars;

account_balances[from_acct] -= dollars;

/* Конец критической секции. */

pthread_setcancelstate (old_cancel_state, NULL);

return 0;

}

2.5. Область потоковых данных

Все потоки программы делят общее адресное пространство. Тем не менее у каждого потока – свой собственный стек вызова. Это позволяет всем потокам выполнять разный код, а также вызывать функции традиционным способом. При каждом вызове функции в любом потоке создается отдельный набор локальных переменных, которые сохраняются в стеке этого потока.

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

Потоковые переменные должны иметь тип void*. Ссылка на каждую переменную осуществляется по ключу. Для создания нового ключа, т.е. новой переменной, предназначена функция pthread_key_create(). Первым ее аргументом является указатель на переменную типа pthread_key_t. В нее будет записано значение ключа, посредством которого любой поток сможет обращаться к своей копии данных. Второй аргумент – это указатель на функцию очистки ключа. Она будет автоматически вызываться при уничтожении потока.

После того как ключ создан, каждый поток может назначать ему собственное значение, вызывая функцию pthread_setspecific(). Ее первый аргумент – это ключ, а второй – требуемое значение типа void*. Для чтения потоковых переменных предназначена функция pthread_getspecific(), единственным аргументом которой является ключ.

Предположим, имеется приложение, распределяющее вычисления между несколькими потоками. В целях аудита за каждым потоком закреплен отдельный журнальный файл, куда записываются сообщения о ходе выполнения поставленной задачи. Область потоковых данных – удобное место для хранения указателя на журнальный файл каждого потока.

В листинге 5 показано, как это сделать. Для хранения файлового указателя в функции main() создается ключ, запоминаемый в переменной thread_log_key. Эта переменная является глобальной, поэтому она доступна всем потокам. Когда поток начинает выполнять свою потоковую функцию, он открывает журнальный файл и сохраняет указатель на него в своем ключе. Позднее любой поток может вызвать функцию write_to_thread_log(), чтобы записать сообщение в свой журнальный файл. Эта функция извлекает из области потоковых данных указатель на журнальный файл и помещает в файл требуемое сообщение.

Листинг 5. (tsd.c) Создание журнального файла для каждого потока

#include <malloc.h>

#include <pthread.h>

#include <stdio.h>

/* Ключ, связывающий указатель журнального файла с каждым потоком. */

static pthread_key_t thread_log_key;

/* Запись параметра message в журнальный файл текущего потока. */

void write_to_thread_log (const char* message)

{

FILE* thread_log = (FILE*) pthread_getspecific (thread_log_key);

fprintf (thread_log, "%s\n", message);

}

/* Закрытие журнального файла, на который указывает

параметр thread_log. */

void close_thread_log (void* thread_log)

{

fclose ((FILE*) thread_log);

}

void* thread_function (void* args)

{

char thread_log_filename[20];

FILE* thread_log;

/* Создание имени журнального файла для текущего потока. */

sprintf (thread_log_filename, "thread%d.log", (int) pthread_self ());

/* Открытие журнального файла. */

thread_log = fopen (thread_log_filename, "w");

/* Сохранение указателя файла в области потоковых данных

под ключом thread_log_key. */

pthread_setspecific (thread_log_key, thread_log);

write_to_thread_log ("Thread starting.");

/* Далее идет основное тело потока... */

return NULL;

}

int main ()

{

int i;

pthread_t threads[5];

/* Создание ключа, который будет связывать указатели

журнальных файлов с областью потоковых данных. Функция

close_thread_log() закрывает все журнальные файлы */

pthread_key_create (&thread_log_key, close_thread_log);

/* Создание потоков. */

for (i = 0; i < 5; ++i)

pthread_create (&(threads[i]), NULL, thread_function, NULL);

/* Ожидание завершения всех потоков. */

for (i = 0; i < 5; ++i)

pthread_join (threads[i], NULL);

return 0;

}

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

2.6. Необходимость синхронизации потоков

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

Отладка потоковой программы также затруднена, ведь не всегда можно воссоздать ситуацию, приведшую к проблеме. В одном случае программа работает абсолютно правильно, а в другом – вызывает системный сбой. Нельзя заставить систему распланировать выполнение потоков так, как она сделала при предыдущем запуске программы.

Большинство ошибок, возникающих при работе с потоками, связано с тем, что потоки обращаются к одним и тем же данным. Это достоинство потоков, и оно же является их бедствием. Если один поток заполняет структуру данными в то время, когда второй поток обращается к этой же структуре, возникает хаос. Очень часто неправильно написанные потоковые программы корректно работают только в том случае, когда один поток выполняется с более высоким приоритетом, т.е. чаще или быстрее обращается к процессору, чем другой поток. Говорят, что подобного вида ошибки появляются вследствие возникновения состояния гонки: потоки преследуют друг друга в попытке изменить одни и те же данные.

Предположим, что в программу поступает группа запросов, которые обрабатываются несколькими одновременными потоками. Очередь запросов представлена связанным списком объектов типа struct job.

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

Потоковая функция, работающая с очередью заданий, представлена в листинге 6.

Листинг 6. (job-queue1.c) Потоковая функция, работающая с очередью заданий

#include <malloc.h>

struct job {

/* Ссылка на следующий элемент связанного списка. */

struct job* next;

/* Другие поля структуры job, описывающие задание ... */

};

/* Список отложенных заданий. */

struct job* job_queue;

extern void process_job (struct job*);

/* Обработка заданий до тех пор, пока очередь не опустеет. */

void* thread_function (void* arg)

{

while (job_queue != NULL) {

/* Получаем очередное задание из списка заданий. */

struct job* next_job = job_queue;

/* Удаляем это задание из списка. */

job_queue = job_queue->next;

/* Выполняем задание. */

process_job (next_job);

/* Очистка памяти, занимаемой выполненным заданием. */

free (next_job);

}

return NULL;

}

Теперь предположим, что два потока завершают свои операции примерно в одно и тоже время, а в очереди остается только одно задание. Первый поток проверяет, равен ли указатель job_queue значению NULL, и, обнаружив, что очередь не пуста, входит в цикл, где сохраняет указатель на объект задания в переменной next_job. В этот момент Linux прерывает первый поток и активизирует второй. Он тоже проверяет указатель job_queue, устанавливает, что он не равен NULL, и записывает тот же самый указатель в свою переменную next_job. Таким образом, два потока начнут выполнять одно и то же задание.

Далее ситуация еще ухудшается. Первый поток удаляет последнее задание из очереди, делая переменную job_queue равной NULL. Когда второй поток попытается выполнить операцию job_queue_next, возникает фатальная ошибка сегментации.

Это наглядный пример гонки за ресурсами. Если программе "повезет", система не распланирует потоки именно таким образом и ошибка не проявится. Чтобы исключить здесь возможность гонки, необходимо сделать операции с очередью атомарными. Атомарная операция неделима и непрерывна; если она началась, то уже не может быть приостановлена или прервана, пока, наконец, не завершится. Выполнение других операций в это время становится невозможным. В нашем конкретном примере проверка переменной job_queue и удаление задания из списка должны выполняться как одна атомарная операция.

2.7. Исключающие семафоры

Решение проблемы гонки заключается в том, чтобы позволить только одному потоку обращаться к очереди в конкретный момент времени. Когда поток начинает просматривать очередь, все остальные потоки вынуждены дожидаться, пока он удалит очередное задание из списка.

Реализация такого решения требует поддержки от операционной системы. В Linux имеется специальное средство, называемое исключающим семафором, или мьютексом (MUTual EXclusion – взаимное исключение). Это специальная блокировка, которую в конкретный момент времени может устанавливать только один поток. Если исключающий семафор захвачен каким-либо потоком, другой поток, обращающийся к семафору, оказывается заблокированным или переведенным в режим ожидания. Как только семафор освобождается, задержанный поток продолжает свое выполнение. Система Linux гарантирует, что между потоками, пытающимися захватить исключающий семафор, не возникнет гонки. Такой семафор может принадлежать только одному потоку, а все остальные потоки блокируются.

Чтобы создать исключающий семафор, нужно объявить переменную типа pthread_mutex_t и передать указатель на нее функции pthread_mutex_init(). Вторым аргументом этой функции является указатель на объект атрибутов семафора. Как и в случае функции pthread_create(), если объект атрибутов не создан (аргумент NULL), используются атрибуты по умолчанию. Переменная исключающего семафора инициализируется только один раз. Вот как это делается:

pthread_mutex_t mutex;

pthread_mutex_init (&mutex, NULL);

Более простой способ создания исключающего семафора со стандартными атрибутами – присвоение переменной специального значения PTHREAD_MUTEX_INITIALIZER. Вызывать функцию pthread_mutex_init() в таком случае не требуется. Это особенно удобно для глобальных переменных. Предыдущий фрагмент программы эквивалентен следующей записи:

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

Поток может попытаться захватить исключающий семафор, вызвав функцию pthread_mutex_lock(). Если семафор свободен, он переходит во владение данного потока и функция немедленно завершается. Если же семафор был захвачен другим потоком, выполнение функции pthread_mutex_lock() блокируется и возобновляется только тогда, когда семафор вновь становится свободным. Сразу несколько потоков могут ожидать освобождения исключающего семафора. Когда это событие наступает, только один поток (выбираемый произвольным образом) разблокируется и получает возможность захватить семафор; остальные потоки остаются заблокированными.

Функция pthread_mutex_unlock() освобождает исключающий семафор. Она должна вызываться только из того потока, который захватил семафор.

В листинге 7 представлена обновленная версия программы, работающей с очередью заданий. Теперь очередь "защищена" исключающим семафором. Прежде чем получить доступ к очереди (для чтения или записи), каждый поток сначала захватывает семафор. Только когда вся последовательность операций проверки очереди и извлечения из нее задания (для выполнения) будет закончена, произойдет освобождение семафора. Благодаря этому не возникает описанное выше состояние гонки.

Листинг 7. (job-queue2.c) Работа с очередью, защищенной исключающим семафором

#include <malloc.h>

#include <pthread.h>

struct job {

struct job* next;

/* Другие поля структуры job, описывающие задание ... */

};

/* Список отложенных заданий. */

struct job* job_queue;

extern void process_job (struct job*);

/* Исключающий семафор, защищающий очередь. */

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

/* Обработка заданий до тех пор, пока очередь не опустеет. */

void* thread_function (void* arg)

{

while (1) {

struct job* next_job;

/* Захват семафора, защищающего очередь. */

pthread_mutex_lock (&job_queue_mutex);

/* Теперь можно проверить, является ли очередь пустой. */

if (job_queue == NULL)

next_job = NULL;

else {

/* Получаем очередное задание из списка заданий. */

next_job = job_queue;

/* Удаляем это задание из списка. */

job_queue = job_queue->next;

}

/* Освобождаем семафор, так как работа с очередью окончена. */

pthread_mutex_unlock (&job_queue_mutex);

/* Если очередь пуста, завершаем поток. */

if (next_job == NULL)

break;

/* Выполняем задание. */

process_job (next_job);

/* Очистка памяти, занимаемой выполненным заданием. */

free (next_job);

}

return NULL;

}

Все операции доступа к совместно используемому указателю job_queue происходит между вызовами функций pthread_mutex_lock() и pthread_mutex_unlock(). Обратите внимание на то, что, если очередь пуста (т.е указатель job_queue равен NULL), цикл не завершается немедленно. Это привело бы к тому, что исключающий семафор так и остался бы в захваченном состоянии и не позволил бы ни одному другому потоку получить доступ к очереди заданий. Мы в программе действуем иначе: записываем в переменную next_job значение NULL и выходим из цикла только после освобождения семафора.

В данной программе не показано добавление заданий в очередь. Вот как, например, может выглядеть функция, добавляющая новое задание к очереди:

void enqueue_job (struct job* new_job)

{

pthread_mutex_lock (&job_queue_mutex);

new_job->next = job_queue;

job_queue = new_job;

pthread_mutex_unlock (&job_queue_mutex);

}

Отметим, что исключающий семафор блокирует доступ к участку программы, а не к переменной.

Иногда нужно, не заблокировав программу, проверить, захвачен ли исключающий семафор. Для потока не всегда приемлемо находиться в режиме пассивного ожидания, ведь за это время можно сделать много полезного! Проблему решает функция pthread_mutex_trylock(). Если она обнаруживает, что семафор свободен, то захватывает его также, как и функция pthread_mutex_lock(), возвращая при этом 0. Если же оказывается, что семафор уже захвачен другим потоком, функция pthread_mutex_trylock() не блокирует программу, а немедленно завершается, возвращая код ошибки EBUSY. В этом случае поток попытается обратиться к семафору позже.

2.8. Обычные потоковые семафоры

В предыдущей программе потоки завершаются, если очередь заданий опустела. В результате задание может поступить в очередь, а потоков для обработки заданий уже нет. Следовательно, необходим механизм, который позволит блокировать потоки в случае, когда очередь пуста, а новые задания еще не поступили. Такой механизм называется семафором (обычным потоковым семафором).

Семафор – это счетчик, используемый для синхронизации потоков. Операционная система гарантирует, что проверка и модификация значения семафора могут быть выполнены безопасно и не приведут к возникновению гонки.

Счетчик семафора является неотрицательным целым числом. Семафор поддерживает две базовые операции:

• Операция ожидания (wait) уменьшает значение счетчика на единицу. Если счетчик уже равен нулю, операция блокируется до тех пор, пока значение счетчика не станет положительным (вследствие действий, выполняемых другими потоками). После снятия блокировки значение семафора уменьшается на единицу и операция завершается.

• Операция установки (post) увеличивает значение счетчика на единицу. Если до этого счетчик был равен нулю и существовали потоки, заблокированные в операции ожидания данного семафора, один из них разблокируется и завершает свою операцию (т.е. счетчик семафора снова становится равным нулю).

В Linux имеются две немного отличающиеся реализации семафоров. Та, которую мы опишем ниже, соответствует стандарту POSIX. Такие семафоры применяются для организации взаимодействия потоков. Другая реализация предназначена для межпроцессного взаимодействия и соответствует стандарту System V. В обоих случаях при работе с семафорами необходимо включить в программу файл <semaphore.h>.

Семафор представляется переменной типа sem_t. Семафор следует предварительно инициализировать с помощью функции sem_init(), передав указатель на переменную семафора. Второй параметр этой функции должен быть равен нулю, а третий – это начальное значение счетчика семафора.

Чтобы выполнить операцию ожидания семафора, необходимо вызвать функцию sem_wait(). Функция sem_post() устанавливает семафор. Есть также функция sem_trywait(), реализующая функцию неблокирующего ожидания: если операция ожидания приведет к блокированию потока из-за того, что счетчик семафора равен нулю, функция немедленно завершается, возвращая код ошибки EAGAIN.

В Linux имеется функция sem_getvalue(), позволяющее узнать текущее значение счетчика семафора. Это значение помещается в переменную типа int, на которую ссылается второй аргумент функции. Однако нельзя на основании этой информации входить в критическую секцию, так как это может привести к возникновению гонки: другой поток может успеть изменить это значение.

В листинге 8 представлена еще раз обновленная версия программы, работающей с очередью заданий. Теперь для доступа к очереди используются два семафора - исключающий и обычный.

Листинг 8. (job-queue3.c) Работа с очередью с примененим обычного семафора

#include <malloc.h>

#include <pthread.h>

#include <semaphore.h>

struct job {

struct job* next;

/* Другие поля структуры job, описывающие задание ... */

};

struct job* job_queue;

extern void process_job (struct job*);

pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;

/* Семафор для защиты переменной, сообщающей о наличии

заданий в очереди. */

sem_t job_queue_count;

/* Начальная инициализация очереди. */

void initialize_job_queue ()

{

/* Вначале очередь пуста. */

job_queue = NULL;

/* Начальное значение семафора равно нулю. */

sem_init (&job_queue_count, 0, 0);

}

/* Обработка заданий до тех пор, пока очередь не опустеет. */

void* thread_function (void* arg)

{

while (1) {

struct job* next_job;

/* Ожидаем, если очередь пустая */

sem_wait (&job_queue_count);

/* Захват семафора, защищающего очередь. */

pthread_mutex_lock (&job_queue_mutex);

/* Поскольку мы знаем, что очередь не пуста, без лишних

проверок получаем задание из очереди. */

next_job = job_queue;

/* Удаляем это задание из списка. */

job_queue = job_queue->next;

/* Освобождаем взаимоисключающий семафор, так как

работа с очередью окончена. */

pthread_mutex_unlock (&job_queue_mutex);

/* Выполняем задание. */

process_job (next_job);

/* Очистка памяти, занимаемой выполненным заданием. */

free (next_job);

}

return NULL;

}

/* Добавление нового задания в начало очереди. */

void enqueue_job (/* Передача необходимых данных... */)

{

struct job* new_job;

/* Выделение памяти для нового объекта задания. */

new_job = (struct job*) malloc (sizeof (struct job));

/* Заполнение других полей структуры ... */

/* Захватываем исключающий семафор, прежде чем обратиться к очереди. */

pthread_mutex_lock (&job_queue_mutex);

/* Помещаем новое задание в начало очереди. */

new_job->next = job_queue;

job_queue = new_job;

/* Устанавливаем семафор, сообщая о том, что в очереди

новое задание. Если есть потоки, заблокированные в ожидании

семафора, один из них будет разблокирован для считывания

из очереди этого задания. */

sem_post (&job_queue_count);

/* Освобождаем исключающий семафор. */

pthread_mutex_unlock (&job_queue_mutex);

}