6.3.2 Синхронизация потоков

Pthread-ы обеспечивают синхронизацию потоков в форме взаимного исключения (мьютексов) и условных переменных.

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

Общая последовательность для защиты структуры общих данных с помощью мьютекса:

блокируем мьютекс

работаем с общими данными

разблокируем мьютекс

Условная переменная представляет собой механизм синхронизации, который является более полезным для ожидания событий, чем для блокировки ресурса. Условная переменная связана с предикатом (логическим выражением, которое имеет значение ИСТИНА, TRUE, или ЛОЖЬ, FALSE), основанным на некоторых общих данных. Функции отправляются спать на условной переменной и пробуждают один или все потоки, если результат предиката изменяется.

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

Давайте подробно обсудим реализацию мьютекса и условной переменной для pthread-ов.

Мьютекс для pthread-ов

Мьютекс инициализируется во время определения:

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

Он также может быть проинициализирован во время выполнения вызовом функции pthread_mutex_init.

int pthread_mutex_init(pthread_mutex_t *mutex,

               const pthread_mutexattr_t *mutexattr);

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

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

int pthread_mutex_lock(pthread_mutex_t *mutex);

int pthread_mutex_unlock(pthread_mutex_t *mutex);

Общие данные могут быть защищены с помощью функций блокировки и разблокировки мьютекса следующим образом:

pthread_mutex_lock(&lock);

/* работаем с общими данными */

pthread_mutex_unlock(&lock);

Есть три типа мьютекса.

▪Быстрый мьютекс

▪Рекурсивный мьютекс

▪Мьютекс проверки ошибки

Поведения этих трёх типов мьютекса похожи; они отличаются только когда владелец мьютекса вновь вызывает pthread_mutex_lock, что снова захватить его.

▪Для быстрого мьютекса возникает тупиковая ситуация, поскольку поток теперь ждёт самого себя, чтобы разблокировать мьютекс

▪Для рекурсивного мьютекса функция возвращается сразу и счётчик захвата мьютекса увеличивается на единицу. Мьютекс разблокируется только если счетчик дойдёт до нуля; то есть поток должен вызвать pthread_mutex_unlock для каждого вызова pthread_mutex_lock.

▪Для мьютекса проверки ошибки pthread_mutex_lock возвращает ошибку с кодом ошибки EDEADLK.

Быстрый, рекурсивный мьютекс и мьютекс проверки ошибки инициализируются во время определения следующим образом:

/* Быстрый мьютекс */

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

/* Рекурсивный мьютекс */

pthread_mutex_t lock =

   PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;

/* Мьютекс проверки ошибки */

pthread_mutex_t lock =

   PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP;

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

/* Быстрый мьютекс */

pthread_mutex_t lock;

pthread_mutex_init(&lock, NULL);

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

pthread_mutex_t lock;

pthread_mutexattr_t mutex_attr;

pthread_mutexattr_init(&mutex_attr);

pthread_mutexattr_settype(&mutex_attr,

   PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP);

pthread_mutex_init(&lock, &mutex_attr);

Мьютекс проверки ошибки инициализируется во время выполнения аналогично показанному выше; только в вызове pthread_mutexattr_settype изменяется тип мьютекса на PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP.

Условная переменная pthread-ов

Как и мьютекс, условная переменная инициализируется либо во время определения, либо во время выполнения вызовом функции pthread_cond_init.

pthread_cond_t cond_var = PTHREAD_COND_INITIALIZER;

или

pthread_cond_t cond_var;

pthread_cond_init(&cond_var, NULL);

Давайте вернёмся к нашему MP3-плееру, чтобы понять различные операции с условной переменной pthread-ов. В них участвуют три объекта, показанные на Рисунке 6.6.

▪Основной поток: он порождает поток декодера звука. Он также поставляет данные для потока декодера.

▪Поток декодера: он декодирует и воспроизводит звук из данных, предоставленных основным потоком.

▪Очередь: это общая структура данных между основным потоком и потоком декодера. Основной поток читает данные из файла и помещает их в очередь, а поток декодера извлекает данные и работает с ними.

Основной поток порождает поток декодера при запуске приложения.

int main(){

  pthread_t decoder_tid;

    ...

  /* Создаём поток декодера звука */

  if (pthread_create(&decoder_tid, NULL, audio_decoder,

                     NULL ) != 0){

    printf("Audio decoder thread creation failed.\n");

    return FAIL;

  }

  ...

  ...

}

Возникают три вопроса:

▪Как поток декодера узнаёт, что в очереди есть данные? Должен ли он делать опрос? Есть ли более эффективный механизм?

▪Есть ли способ для основного потока проинформировать поток декодера о наличии в очереди данных?

▪Как можно защитить очередь от одновременного доступа со стороны основного потока и потока декодера?

Чтобы ответить на эти вопросы, давайте сначала посмотрим на детали потока декодера звука.

void* audio_decoder(void *unused){

  char *buffer;

  printf("Audio Decoder thread started\n");

  for(;;){

    pthread_mutex_lock(&lock);

    while(is_empty_queue())

      pthread_cond_wait(&cond, &lock);

    buffer = get_queue();

    pthread_mutex_unlock(&lock);

    /* декодируем данные в буфере */

    /* посылаем декодированные данные на выход для воспроизведения */

    free(buffer);

  }

}

Обратите, пожалуйста, внимание на следующий кусок кода в функции audio_decoder.

while(is_empty_queue())

  pthread_cond_wait(&cond, &lock);

Здесь мы ввели условную переменную cond. Предикатом для этой условной переменной является "очередь не пуста". Таким образом, если предикат является ложным (то есть очередь пуста), поток засыпает на условной переменной, вызвав функцию pthread_cond_wait. Поток будет оставаться в состоянии ожидания, пока какой-нибудь другой поток не просигнализирует об изменении условия (то есть изменит предикат путём добавления в очередь данных, что сделает её не пустой). Прототип этой функции:

int pthread_cond_wait(pthread_cond_t *cond,

                      pthread_mutex_t *mutex);

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

while(is_empty_queue())

   <--- О состоянии сигнализирует другой поток --->

  pthread_cond_wait(...);

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

pthread_mutex_lock(&lock);  <-- Получаем мьютекс

while(is_empty_queue())     <-- проверяем предикат

  pthread_cond_wait(&cond,&lock);<-- засыпаем на условной переменной

Чтобы сделать эту работу, необходимо, чтобы все участвующие потоки должны были захватывать мьютекс, изменять/проверять состояние, а затем освобождать мьютекс.

Теперь, что происходит с мьютексом, когда поток переходит в спящий режим в pthread_cond_wait? Если мьютекс остаётся в состоянии блокировки, то никакой другой поток не сможет просигнализировать о состоянии, так как этот поток тоже попытается  захватить тот же мьютекс перед изменением предиката. У нас проблема: один поток удерживает блокировки и спит на условной переменной, а другой поток ожидает блокировку для установки состояния. Для того чтобы избежать такой проблемы, соответствующий мьютекс должен быть разблокирован после засыпания потока на условной переменной. Это делается в функции pthread_cond_wait. Функция помещает поток в состояние сна и автоматически освобождает мьютекс.

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

pthread_mutex_lock(&lock); <-- Получаем мьютекс

while(is_empty_queue()) <-- Проверяем условие

  pthread_cond_wait(&cond,&lock); <-- ждём на условной переменной

<-- Мьютекс повторно захватывается внутри pthread_cond_wait -->

buffer = get_queue(); <-- Обрабатываем условие

pthread_mutex_unlock(&lock);<-- По завершении освобождаем мьютекс

Давайте посмотрим, как поток может просигнализировать о состоянии. Шагами являются:

▪Получение мьютекса.

▪Изменение состояния.

▪Освобождение мьютекса.

▪Пробуждение одного или всех потоков, которые спали на условной переменной.

Основной поток нашего плеера пробуждает поток декодера звука после добавления в очередь данных.

fp = fopen("song.mp3", "r");

while (!feof(fp)){

  char *buffer = (char *)malloc(MAX_SIZE);

  fread(buffer, MAX_SIZE, 1, fp);

  pthread_mutex_lock(&lock); <-- Получаем мьютекс

  add_queue(buffer); <-- изменяем условие. Добавление

                         буфера в очередь делает её

                         не пустой

  pthread_mutex_unlock(&lock); <-- Освобождаем мьютекс

  pthread_cond_signal(&cond); <-- Пробуждаем поток декодера

  usleep(300*1000);

}

pthread_cond_signal пробуждает единственный поток, спящий на условной переменной. Чтобы пробудить все потоки, которые спят на условной переменной, также доступна функция pthread_cond_broadcast.

int pthread_cond_signal(pthread_cond_t *cond);

int pthread_cond_broadcast(pthread_cond_t *cond);