62 nready.nready++;
63 Pthread_mutex_unlock(&nready.mutex);
64 *((int *) arg) += 1;
65 }
66 }
67 void*
68 consume(void *arg)
69 {
70 int i;
71 for (i = 0; i < nitems; i++) {
72 Pthread_mutex_lock(&nready.mutex);
73 while (nready.nready == 0)
74 Pthread_cond_wait(&nready.cond, &nready.mutex);
75 nready.nready--;
76 Pthread_mutex_unlock(&nready.mutex);
77 if (buff[i] != i)
78 printf("buff[%d] = *dn", i, buff[i]);
79 }
80 return(NULL);
81 }
Помещение очередного элемента в массив
50-58 Для блокирования критической области в потоке-производителе теперь используется исключение put.mutex.
Уведомление потребителя
59-64 Мы увеличиваем счетчик nready.nready, в котором хранится количество элементов, готовых для обработки потребителем. Перед его увеличением мы проверяем, не было ли значение счетчика нулевым, и если да, то вызывается функция pthread_cond_signal, позволяющая возобновить выполнение всех потоков (в данном случае потребителя), ожидающих установки ненулевого значения этой переменной. Теперь мы видим, как взаимодействуют взаимное исключение и связанная с ним условная переменная. Счетчик используется совместно потребителем и производителями, поэтому доступ к нему осуществляется с блокировкой соответствующего взаимного исключения (nready.mutex). Условная переменная используется для ожидания и передачи сигнала.
Потребитель ждет, пока значение nready.nready не станет отличным от нуля
72-76 Потребитель просто ждет, пока значение счетчика nready. nready не станет отличным от нуля. Поскольку этот счетчик используется совместно с производителями, его значение можно проверять только при блокировке соответствующего взаимного исключения. Если при проверке значение оказывается нулевым, мы вызываем pthread_cond_wait для приостановки процесса. При этом выполняются два атомарных действия:
1. Разблокируется nready.mutex.
2. Выполнение потока приостанавливается, пока какой-нибудь другой поток не вызовет pthread_cond_signal для этой условной переменной.
Перед возвращением управления потоку функция pthread_cond_wait блокирует nready.mutex. Таким образом, если после возвращения из функции мы обнаруживаем, что счетчик имеет ненулевое значение, мы уменьшаем этот счетчик (зная, что взаимное исключение заблокировано) и разблокируем взаимное исключение. Обратите внимание, что после возвращения из pthread_cond_wait мы всегда заново проверяем условие, поскольку может произойти ложное пробуждение при отсутствии выполнения условия. Различные реализации стремятся уменьшить количество ложных пробуждений, но они все равно происходят.
Код, передающий сигнал условной переменной, выглядит следующим образом:
struct {
pthread_mutex_t mutex;
pthread_cond_t cond;
переменные, для которых устанавливается условие
} var = { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, … };
Pthread_mutex_lock(&var.mutex);
установка истинного значения условия
Pthread_cond_signal(&var.cond);
Pthread_mutex_unlock(&var.mutex);
В нашем примере переменная, для которой устанавливалось условие, представляла собой целочисленный счетчик, а установка условия означала просто увеличение счетчика. Мы оптимизировали программу, посылая сигнал только при изменении значения счетчика с 0 на 1.
Код, проверяющий условие и приостанавливающий процесс, если оно не выполняется, обычно выглядит следующим образом:
Pthread_mutex_lock(&var.mutex);
while (условие ложно)
Pthread_cond_wait(&var.cond, &var.mutex);
изменение условия
Pthread_mutex_unlock(&var.mutex);
Исключение конфликтов блокировок
В приведенном выше фрагменте кода, как и в листинге 7.6, функция pthread_cond_signal вызывалась потоком, блокировавшим взаимное исключение, относящееся к условной переменной, для которой отправлялся сигнал. Мы можем представить себе, что в худшем варианте система немедленно передаст управление потоку, которому направляется сигнал, и он начнет выполняться и немедленно остановится, поскольку не сможет заблокировать взаимное исключение. Альтернативный код, помогающий этого избежать, для листинга 7.6 будет иметь следующий вид:
int dosignal;
Pthread_mutex_lock(nready.mutex);
dosignal = (nready.nready == 0);
nready.nready++;
Pthread_mutex_unlock(&nready.mutex);
if (dosignal)
Pthread_cond_signal(&nready.cond);
Здесь мы отправляем сигнал условной переменной только после разблокирования взаимного исключения. Это разрешено стандартом Posix: поток, вызывающий pthread_cond_signal, не обязательно должен в этот момент блокировать связанное с переменной взаимное исключение. Однако Posix говорит, что если требуется предсказуемое поведение при одновременном выполнении потоков, это взаимное исключение должно быть заблокировано процессом, вызывающим pthread_cond_signal.
7.6. Условные переменные: время ожидания и широковещательная передача
В обычной ситуации pthread_cond_signal запускает выполнение одного потока, ожидающего сигнал по соответствующей условной переменной. В некоторых случаях поток знает, что требуется пробудить несколько других процессов. Тогда можно воспользоваться функцией pthread_cond_broadcast для пробуждения всех процессов, заблокированных в ожидании сигнала данной условной переменной.
ПРИМЕЧАНИЕ
Пример ситуации, в которой требуется возобновление выполнения нескольких процессов, появится в главе 8, когда мы будем обсуждать задачу с несколькими считывающими и записывающими процессами. Когда записывающий процесс снимает блокировку, он должен разбудить все ждущие считывающие процессы, поскольку одновременное считывание в данном случае разрешено.
Альтернативным (и более безопасным) методом является использование широковещательной передачи во всех тех случаях, когда требуется использование сигналов. Сигнал является оптимизацией для тех случаев, когда известно, что все ожидающие процессы правильно написаны и требуется разбудить только один из них, и какой именно будет разбужен, значения не имеет. Во всех других ситуациях следует использовать широковещательную передачу.
#include <pthread.h>
int pthread_cond_broadcast(pthread_cond_t *cptr);
int pthread_cond_timedwait(pthread_cond_t *cptr, pthread_mutex_t *mptr, const struct timespec *abstime);
/* Функции возвращают 0 в случае успешного завершения, положительный код Еххх — в случае ошибки */
Функция pthread_cond_timedwait позволяет установить ограничение на время блокирования процесса. Аргумент abstime представляет собой структуру timespec:
struct timespec {
time_t tv_sec; /* секунды */
long tv_nsec; /* наносекунды */
};
Эта структура задает конкретный момент системного времени, в который происходит возврат из функции, даже если сигнал по условной переменной еще не будет получен. В этом случае возвращается ошибка с кодом ETIMEDOUT.
Эта величина представляет собой абсолютное значение времени, а не промежуток. Аргумент abstime задает таким образом количество секунд и наносекунд с 1 января 1970 UTC до того момента времени, в который должен произойти возврат из функции. Это отличает функцию от select, pselect и poll (глава 6 [24]), которые в качестве аргумента принимают некоторое количество долей секунды, спустя которое должен произойти возврат. (Функция select принимает количество микросекунд, pselect — наносекунд, a poll — миллисекунд.) Преимущество использования абсолютного времени заключается в том, что если функция возвратится до ожидаемого момента (например, при перехвате сигнала), ее можно будет вызвать еще раз, не изменяя содержимого структуры timespec.
7.7. Атрибуты взаимных исключений и условных переменных
В наших примерах в этой главе мы хранили взаимные исключения и условные переменные как глобальные данные всего процесса, поскольку они использовались для синхронизации потоков внутри него. Инициализировали мы их с помощью двух констант: PTHREAD_MUTEX_INITIALIZER и PTHREAD_COND_INTIALIZER. Инициализируемые таким образом исключения и условные переменные приобретали значения атрибутов по умолчанию, но мы можем инициализировать их и с другими значениями атрибутов.
Прежде всего инициализировать и удалять взаимное исключение и условную переменную можно с помощью функций
#include <pthread.h>
int pthread_mutex_imt(pthread_mutex_t *mptr, const pthread_mutexattr_t *attr);
int pthread_mutex_destroy(pthread_mutex_t *mptr);
int pthread_cond_init(pthread_cond_t *cрtr, const pthread_condattr_t *attr);
int pthread_cond_destroy(pthread_cond_t *cptr);