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

Блокировки чтения-записи

8.1. Введение

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

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

1. Любое количество потоков могут заблокировать ресурс для считывания, если ни один процесс не заблокировал его на запись.

2. Блокировка чтения-записи может быть установлена на запись, только если ни один поток не заблокировал ресурс для чтения или для записи.

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

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

Такой вид совместного доступа к ресурсу также носит название совместно-исключающей блокировки (shared-exclusive), поскольку тип используемой блокировки на чтение называется совместной блокировкой (shared lock), а тип используемой блокировки на запись называется исключающей блокировкой (exclusive lock). Существует также специальное название для данной задачи (несколько считывающих процессов и один записывающий): задача читателей и писателей (readers and writers problem), и говорят также о блокировке читателей и писателя (readers-writer lock). В последнем случае слово «читатель» специально употреблено во множественном числе, а «писатель» — в единственном, чтобы подчеркнуть сущность задачи.

ПРИМЕЧАНИЕ

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

Функции, описываемые в этой главе, определены стандартом Unix 98, поскольку блокировки чтения-записи не были частью стандарта Posix.1 1996 года. Эти функции были разработаны группой производителей Unix, известной под названием Aspen Group, в 1995 году вместе с другими расширениями, которые еще не были определены Posix.1. Рабочая группа Posix (1003.1j) в настоящее время разрабатывает набор расширений Pthreads, включающий блокировки чтения-записи, который, хочется верить, совпадет с описываемым в этой главе.

8.2. Получение и сброс блокировки чтения-записи

Блокировка чтения-записи имеет тип pthread_rwlock_t. Если переменная этого типа является статической, она может быть проинициализирована присваиванием значения константы PTHREAD_RWLOCK_INITIALIZER.

Функция pthread_rwlock_rdlock позволяет заблокировать ресурс для чтения, причем вызвавший процесс будет заблокирован, если блокировка чтения-записи уже установлена записывающим процессом. Функция pthread_rwlock_wrlock позволяет заблокировать ресурс для записи, причем вызвавший процесс будет заблокирован, если блокировка чтения-записи уже установлена каким-либо другим процессом (считывающим или записывающим). Функция pthread_rwlock_unlock снимает блокировку любого типа (чтения или записи):

#include

int pthread_rwlock_rdlock(pthread_rwlock_t *rwptr);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwptr);
int pthread_rwlock_unlock(pthread_rwlock_t *rwptr );
/* Все функции возвращают 0 в случае успешного завершения, положительное значение Еххх – в случае ошибки */

Следующие две функции производят попытку заблокировать ресурс для чтения или записи, но если это невозможно, возвращают ошибку с кодом EBUSY, вместо того чтобы приостановить выполнение вызвавшего процесса:

#include 
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwptr);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwptr);
/* Обе функции возвращают 0 в случае успешного завершения, положительное значение Еххх – в случае ошибки */

8.3. Атрибуты блокировки чтения-записи

Мы уже отмечали, что статическая блокировка может быть проинициализирована присваиванием ей значения PTHREAD_RWLOCK_INITIALIZER. Эти переменные могут быть проинициализированы и динамически путем вызова функции pthread_rwlock_init. 

Когда поток перестает нуждаться в блокировке, он может вызвать pthread_rwlock_ destroy:

#include 
int pthread_rwlock_init(pthread_rwlock_t *rwptr, const pthread_rwlockattr_t *attr);
int pthread_rwlock_destroy(pthread_rwlock_t *rwptr);
/* Обе функции возвращают 0 в случае успешного завершения, положительное значение Еххх – в случае ошибки */

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

#include 
int pthread_rwlockattr_init(pthread_rwlockattr_t *attr);
int pthread_rwlockattr_destroy(pthread_rwlockattr_t *attr);
/* Обе функции возвращают 0 в случае успешного завершения, положительное значение Еххх – в случае ошибки */

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

#include 
int pthread_rwlockattr_getpshared(const pthread_rwlockattr_t *attr, int *valptr);
int pthread_rwlockattr_setpshared(pthread_rwlockattr_t *attr, int value );
/* Обе функции возвращают 0 в случае успешного завершения, положительное значение Еххх – в случае ошибки */

Первая функция возвращает текущее значение в целом, на которое указывает аргумент valptr. Вторая функция устанавливает значение этого атрибута равным value, которое может быть либо PTHREAD_PROCESS_PRIVATE, либо PTHREAD_ PROCESS_SHARED.

8.4. Реализация с использованием взаимных исключений и условных переменных

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

ПРИМЕЧАНИЕ

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

Другие реализации блокировок чтения записи заслуживают отдельного изучения. В разделе 7.1.2 книги [3] представлена реализация, в которой приоритет имеют ожидающие записи потоки и предусмотрена обработка отмены выполнения потока (о которой мы вскоре будем говорить подробнее). В разделе В.18.2.3.1 стандарта IEEE 1996 [8] представлена другая реализация, в которой предпочтение имеют ожидающие записи потоки и в которой также предусмотрена обработка отмены. В главе 14 книги [12] также приводится возможная реализация, в которой приоритет имеют ожидающие записи процессы. Реализация, приведенная в этом разделе, взята из пакета АСЕ (http://www.cs.wustl.edu/~schmidt/ACE.html), автором которого является Дуг Шмидт (Doug Schmidt). Аббревиатура АСЕ означает Adaptive Communications Environment. Во всех четырех реализациях используются взаимные исключения и условные переменные. 

Тип данных pthread_rwlock_t

В листинге 8.1[1] приведен текст заголовочного файла pthread_rwlock .h, в котором определен основной тип pthread_rwlock_t и прототипы функций, работающих с блокировками чтения и записи. Обычно все это находится в заголовочном файле .

Листинг 8.1. Определение типа данных pthread_rwlock_t

//my_rwlock/pthread_rwlock.h
1  #ifndef __pthread_rwlock_h
2  #define __pthread_rwlock_h

3  typedef struct {
4   pthread_mutex_t rw_mutex; /* блокировка для структуры */
5   pthread_cond_t rw_condreaders; /* для ждущих читающих потоков */
6   pthread_cond_t rw_condwriters; /* для ждущих пишущих потоков */
7   int rw_magic; /* для проверки ошибок */
8   int rw_nwaitreaders;/* число ожидающих */
9   int rw_nwaitwriters;/* число ожидающих */
10  int rw_refcount;
11  /* –1, если блокировка на запись, иначе – количество блокировок на чтение */
12 } pthread_rwlock_t;

13 #define RW_MAGIC 0x19283746
14 /* порядок должен быть такой же, как у элементов структуры */
15 #define PTHREAD_RWLOCK_INITIALIZER { PTHREAD_MUTEX_INITIALIZER, \
16  PTHREAD_COND_INITIALIZER, PTHREAD_COND_INITIALIZER, \
17  RW_MAGIC, 0, 0, 0 }

18 typedef int pthread_rwlockattr_t; /* не поддерживается */

19 /* прототипы функций */
20 int pthread_rwlock_destroy(pthread_rwlock_t *);
21 int pthread_rwlock_init(pthread_rwlock_t *, pthread_rwlockattr_t *);
22 int pthread_rwlock_rdlock(pthread_rwlock_t *);
23 int pthread_rwlock_tryrdlock(pthread_rwlock_t *);
24 int pthread_rwlock_trywrlock(pthread_rwlock_t *);
25 int pthread_rwlock_unlock(pthread_rwlock_t *);
26 int pthread_rwlock_wrlock(pthread_rwlock_t *);

27 /* и наши функции-обертки */
28 void pthread_rwlock_destroy(pthread_rwlock_t *);
29 void pthread_rwlock_init(pthread_rwlock_t*, pthread_rwlockattr_t *);
30 void Pthread_rwlock_rdlock(pthread_rwlock_t *);
31 int Pthread_rwlock_tryrdlock(pthread_rwlock_t *);
32 int pthread_rwlock_trywrlock(pthread_rwlock_t *);
33 void pthread_rwlock_unlock(pthread_rwlock_t *);
34 void pthread_rwlock_wrlock(pthread_rwlock_t *);

35 #endif __pthread_rwlock_h

3-13 Наш тип pthread_rwlock_t содержит одно взаимное исключение, две условные переменные, один флаг и три счетчика. Мы увидим, для чего все это нужно, когда будем разбираться с работой функций нашей программы. При просмотре или изменении содержимого этой структуры мы должны устанавливать блокировку rw_mutex. После успешной инициализации структуры полю rw_magic присваивается значение RW_MAGIC. Значение этого поля проверяется всеми функциями — таким образом гарантируется, что вызвавший поток передал указатель на проинициализированную блокировку. Оно устанавливается в 0 после уничтожения блокировки.

Обратите внимание, что в счетчике rw_refcount всегда хранится текущий статус блокировки чтения-записи: –1 обозначает блокировку записи (и только одна такая блокировка может существовать в любой момент времени), 0 обозначает, что блокировка доступна и может быть установлена, а любое положительное значение соответствует количеству установленных блокировок на чтение.

14-17 Мы также определяем константу для статической инициализации нашей структуры.

Функция pthread_rwlock_init

Первая функция, pthread_rwlock_init, динамически инициализирует блокировку чтения-записи. Ее текст приведен в листинге 8.2.

7-8 Присваивание атрибутов с помощью этой функции не поддерживается, поэтому мы проверяем, чтобы указатель attr был нулевым.

9-19 Мы инициализируем взаимное исключение и две условные переменные, которые содержатся в нашей структуре. Все три счетчика устанавливаются в 0, а полю rw_magiс присваивается значение, указывающее на то, что структура была проинициализирована.

20-25 Если при инициализации взаимного исключения или условной переменной возникает ошибка, мы аккуратно уничтожаем проинициализированные объекты и возвращаем код ошибки.

Листинг 8.2. Функция pthread_rwlock_init: инициализация блокировки чтения-записи

//my_rwlock/pthread_rwlock_init.с
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_init(pthread_rwlock_t *rw, pthread_rwlockattr_t *attr)
5  {
6   int result;
7   if (attr != NULL)
8    return(EINVAL); /* not supported */
9   if ((result = pthread_mutex_init(&rw->rw_mutex, NULL)) != 0)
10   goto err1;
11  if ((result = pthread_cond_init(&rw->rw_condreaders, NULL)) != 0)
12   goto err2;
13  if ((result = pthread_cond_init(&rw->rw_condwriters, NULL)) != 0)
14   goto err3;
15  rw->rw_nwaitreaders = 0;
16  rw->rw_nwaitwriters = 0;
17  rw->rw_refcount = 0;
18  rw->rw_magic = RW_MAGIC;
19  return(0);
20 err3:
21  pthread_cond_destroy(&rw->rw_condreaders);
22 err2;
23  pthread_mutex_destroy(&rw->rw_mutex);
24 err1:
25  return(result); /* значение errno */
26 }

Функция pthread_rwlock destroy

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

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

Листинг 8.З. Функция pthread_rwlock_destroy: уничтожение блокировки чтения-записи

//my_rwlock/pthread_rwlock_destroy.с
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_destroy(pthread_rwlock_t *rw)
5  {
6   if (rw->rw_magic != RW_MAGIC)
7    return(EINVAL);
8   if (rw->rw_refcount != 0 ||
9    rw->rw_nwaitreaders != 0 || rw->rw_nwaitwriters != 0)
10   return(EBUSY);
11  pthread_mutex_destroy(&rw->rw_mutex);
12  pthread_cond_destroy(&rw->rw_condreaders);
13  pthread_cond_destroy(&rw->rw_condwriters);
14  rw->rw_magic = 0;
15  return(0);
16 }

Функция pthread_rwlock_rdlock

Текст функции pthread_rwlock_rdlock приведен в листинге 8.4.

Листинг 8.4. Функция pthread_rwlock_rdlock: получение блокировки на чтение

//my_rwlock/pthread_rwlock_rdlock.с
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_rdlock(pthread_rwlock_t *rw)
5  {
6   int result;
7   if (rw->rw_magic != RW_MAGIC)
8    return(EINVAL);
9   if ((result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
10   return(result);
11  /* предпочтение отдается ожидающим разрешения на запись процессам */
12  while (rw->rw_refcount < 0 || rw->rw_nwaitwriters > 0) {
13   rw->rw_nwaitreaders++;
14   result = pthread_cond_wait(&rw->rw_condreaders, &rw->rw_mutex);
15   rw->rw_nwaitreaders--;
16   if (result != 0)
17    break;
18  }
19  if (result == 0)
20   rw->rw_refcount++; /* блокировка на чтение уже кем-то установлена */
21  pthread_mutex_unlock(&rw->rw_mutex);
22  return (result);
23 }

9-10 При работе со структурой pthread_rwl ock_t всегда устанавливается блокировка на rw_mutex, являющееся ее полем.

11-18 Нельзя получить блокировку на чтение, если rw_refcount имеет отрицательное значение (блокировка установлена на запись) или имеются потоки, ожидающие возможности получения блокировки на запись (rw_nwaitwriters больше 0). Если одно из этих условий верно, мы увеличиваем значение rw_nwaitreaders и вызываем pthread_cond_wait для условной переменной rw_condreaders. Вскоре мы увидим, что при разблокировании ресурса прежде всего проверяется наличие процессов, ожидающих возможности установить блокировку на запись, и если таковых не существует, проверяется наличие ожидающих возможности считывания. Если они имеются, для условной переменной rw_condreaders передается широковещательный сигнал.

19-20 При получении блокировки на чтение мы увеличиваем значение rw_refcount. Блокировка взаимного исключения после этого снимается.

ПРИМЕЧАНИЕ

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

Функция pthread_rwlock_tryrdlock

В листинге 8.5 показана наша реализация функции pthread_rwlock_tryrdlock, которая не вызывает приостановления вызвавшего ее потока.

Листинг 8.5. Функция pthread_rwlock_tryrdlock: попытка заблокировать ресурс для чтения

//my_rwlock/pthread_rwlock_tryrdlock.с
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_tryrdlock(pthread_rwlock_t *rw)
5  {
6   int result;
7   if (rw->rwjnagic != RW_MAGIC)
8    return(EINVAL);
9   if ((result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
10   return(result);
11  if (rw->rw_refcount < 0 || rw->rw_nwaitwriters > 0)
12   result = EBUSY; /* блокировка установлена пишущим потоком или есть пишущие потоки, ожидающие освобождения ресурса */
13  else
14   rw->rw_refcount++; /* увеличение количества блокировок на чтение */
15  pthread_mutex_unlock(&rw->rw_mutex);
16  return(result);
17 }

11-14 Если блокировка в данный момент установлена на запись или есть процессы, ожидающие возможности установить ее на запись, возвращается ошибка с кодом EBUSY. В противном случае мы устанавливаем блокировку, увеличивая значение счетчика rw_refcount.

Функция pthread_rwlock_wrlock

Текст функции pthread_rwlock_wrlock приведен в листинге 8.6.

11-17 Если ресурс заблокирован на считывание или запись (значение rw_refcount отлично от 0), мы приостанавливаем выполнение потока. Для этого мы увеличиваем rw_nwaitwriters и вызываем pthread_cond_wait с условной переменной rw_condwriters. Для этой переменной посылается сигнал при снятии блокировки чтения-записи, если имеются ожидающие разрешения на запись процессы.

18-19 После получения блокировки на запись мы устанавливаем значение rw_refcount в –1.

Листинг 8.6. Функция pthread_rwlock_wrlock: получение блокировки на запись

//my_rwlock/pthread_rwlock_wrlock.c
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_wrlock(pthread_rwlock_t *rw)
5  {
6   int result;
7   if (rw->rw_magic != RW_MAGIC)
8    return(EINVAL);
9   if ((result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
10   return(result);
11  while (rw->rw_refcount != 0) {
12   rw->rw_nwaitwriters++;
13   result = pthread_cond_wait(&rw->rw_condwriters, &rw->rw_mutex);
14   rw->rw_nwaitwriters--;
15   if (result != 0)
16    break;
17  }
18  if (result == 0)
19   rw->rw_refcount = –1;
20  pthread_mutex_unlock(&rw->rw_mutex);
21  return(result);
22 }

Функция pthread_rwlock_trywrlock

Неблокируемая функция pthread_rwlock_trywrlock показана в листинге 8.7.

11-14 Если значение счетчика rw_refcount отлично от нуля, блокировка в данный момент уже установлена считывающим или записывающим процессом (это безразлично) и мы возвращаем ошибку с кодом EBUSY. В противном случае мы устанавливаем блокировку на запись, присвоив переменной rw_refcount значение –1.

Листинг 8.7. Функция pthread_rwlock_trywrlock: попытка получения блокировки на запись

//my_rwlock/pthread_rwlock_trywrlock.c
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_trywrlock(pthread_rwlock_t *rw)
5  {
6   int result;
7   if (rw->rw_magic != RW_MAGIC)
8    return(EINVAL);
9   if ((result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
10   return(result);
11  if (rw->rw_refcount != 0)
12   result = EBUSY; /* заблокирован пишущим потоком или ожидающим возможности записи */
13  else
14   rw->rw_refcount = –1; /* доступна */
15  pthread_mutex_unlock(&rw->rw_mutex);
16  return(result);
17 }

Функция pthread_rwlock_unlock

Последняя функция, pthread_rwlock_unlock, приведена в листинге 8.8.

Листинг 8.8. Функция pthread_rwlock_unlock: разблокирование ресурса

//my_rwlock/pthread_rwlock_unlock.c
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  int
4  pthread_rwlock_unlock(pthread_rwlock_t *rw)
5  {
6   int result;
7   if (rw->rw_magic != RW_MAGIC)
8    return(EINVAL);
9   if ((result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
10   return(result);
11  if (rw->rw_refcount > 0)
12   rw->rw_refcount--; /* снятие блокировки на чтение */
13  else if (rw->rw_refcount == –1)
14   rw->rw_refcount = 0; /* снятие блокировки на запись */
15  else
16   err_dump("rw_refcount = %d", rw->rw_refcount);
17  /* преимущество отдается ожидающим возможности записи потокам */
18  if (rw->rw_nwaitwriters > 0) {
19   if (rw->rw_refcount == 0)
20    result = pthread_cond_signal(&rw->rw_condwriters);
21  } else if (rw->rw_nwaitreaders > 0)
22   result = pthread_cond_broadcast(&rw->rw_condreaders);
23  pthread_mutex_unlock(&rw->rw_mutex);
24  return(result);
25 }

11-16 Если rw_refcount больше 0, считывающий поток снимает блокировку на чтение. Если rw_refcount равно –1, записывающий поток снимает блокировку на запись.

17-22 Если имеются ожидающие разрешения на запись потоки, по условной переменной rw_condwriters передается сигнал (если блокировка свободна, то есть значение счетчика rw_refcount равно 0). Мы знаем, что только один поток может осуществлять запись, поэтому используем функцию pthread_cond_signal. Если нет потоков, ожидающих возможности записи, но есть потоки, ожидающие возможности чтения, мы вызываем pthread_cond_broadcast для переменной rw_condreaders, поскольку возможно одновременное считывание несколькими потоками. Обратите внимание, что мы перестаем устанавливать блокировку для считывающих потоков, если появляются потоки, ожидающие возможности записи. В противном случае постоянно появляющиеся потоки с запросами на чтение могли бы заставить поток, ожидающий возможности записи, ждать целую вечность. По этой причине мы используем два отдельных оператора if и не можем написать просто:

/* предпочтение отдается записывающим процессам */
if (rw->rw_nwaitreaders > 0 && rw->rw_refcount == 0)
 result = pthread_cond_signal(&rw->rw_condwriters);
else if (rw->rw_nwaitreaders > 0)
 result = pthread_cond_broadcast(&rw->rw_condreaders);

Мы могли бы исключить и проверку rw->rw_refcount, но это может привести к вызовам pthread_cond_signal даже при наличии блокировок на чтение, что приведет к потере эффективности.

8.5. Отмена выполнения потоков

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

#include 
int pthread_cancel(pthread_t tid);
/* Возвращает 0 в случае успешного завершения, положительное значение Еххх –в случае ошибки */

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

Для обработки отмены выполнения поток может установить (push) или снять (pop) обработчик-очиститель (cleanup handler):

#include 
void pthread_cleanup_push(void (*function) (void *) void *arg);
void pthread_cleanup_pop(int execute);

Эти обработчики представляют собой обычные функции, которые вызываются:

■ в случае отмены выполнения потока (другим потоком, вызвавшим pthread_ cancel);

■ в случае добровольного завершения работы (вызовом pthread_exit или выходом из начальной функции потока).

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

Аргумент function представляет собой адрес вызываемой функции, а arg — ее единственный аргумент. Функция pthread_cleanup_pop всегда удаляет обработчик из верхушки стека и вызывает эту функцию, если значение execute отлично от 0.

ПРИМЕЧАНИЕ

Мы снова встретимся с проблемой отмены выполнения потоков в связи с листингом 15.26, где может произойти отмена выполнения сервера с дверьми при завершении работы клиента в процессе обработки вызванной им процедуры. 

Пример

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

Рис. 8.1. Временная диаграмма выполнения программы из листинга 8.9

Создание двух потоков

10-13 Создаются два потока, первый из которых выполняет функцию thread1, а второй — thread2. После создания первого делается пауза длительностью в одну секунду, чтобы он успел заблокировать ресурс на чтение. 

Ожидание завершения потоков

14-23 Мы ожидаем завершения работы второго потока и проверяем, что его статус имеет значение PTHREAD_CANCEL. Затем мы ждем завершения работы первого потока и проверяем, что его статус представляет собой нулевой указатель. Затем мы выводим значение трех счетчиков в структуре pthread_rwlock_t и уничтожаем блокировку.

Листинг 8.9. Тестовая программа, иллюстрирующая отмену выполнения потока

//my_rwlock_cancel/testcancel.с
1  #include "unpipc.h"
2  #include "pthread_rwlock.h"

3  pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;
4  pthread_t tid1, tid2;
5  void *thread1(void *), *thread2(void *);

6  int
7  main(int argc, char **argv)
8  {
9   void *status;
10  Set_concurrency(2);
11  Pthread_create(&tid1, NULL, thread1, NULL);
12  sleep(1); /* даем первому потоку возможность получить блокировку */
13  Pthread_create(&tid2, NULL, thread2, NULL);
14  Pthread_join(tid2, &status);
15  if (status != PTHREAD_CANCELED)
16   printf("thread2 status = %p\n", status);
17  Pthread_join(tid1, &status);
18  if (status != NULL)
19   printf("thread1 status = %p\n", status);
20  printf("rw_refcount = %d, rw_nwaitreaders = %d, rw_nwaitwriters = %d\n",
21   rwlock.rw_refcount, rwlock.rw_nwaitreaders,
22   rwlock.rw_nwaitwriters);
23  Pthread_rwlock_destroy(&rwlock);
24  exit(0);
25 }

26 void *
27 thread1(void *arg)
28 {
29  Pthread_rwlock_rdlock(&rwlock);
30  printf("thread1() got a read lock\n");
31  sleep(3); /* даем второму потоку возможность заблокироваться при вызове pthread_rwlock_wrlock() */
32  pthread_cancel(tid2);
33  sleep(3);
34  Pthread_rwlock_unlock(&rwlock);
35  return(NULL);
36 }

37 void *
38 thread2(void *arg)
39 {
40  printf("thread2() trying to obtain a write lock\n"):
41  Pthread_rwlock_wrlock(&rwlock);
42  printf("thread2() got a write lock\n"); /* не будет выполнено */
43  sleep(1);
44  Pthread_rwlock_unlock(&rwlock);
45  return(NULL);
46 }

Функция thread1

26-36 Поток получает блокировку на чтение и ждет 3 секунды. Эта пауза дает возможность другому потоку вызвать pthread_rwlock_wrlock и заблокироваться при вызове pthread_cond_wait, поскольку блокировка на запись не может быть установлена из-за наличия блокировки на чтение. Затем первый поток вызывает pthread_cancel для отмены выполнения второго потока, ждет 3 секунды, освобождает блокировку на чтение и завершает работу.

Функция thread2

37-46 Второй поток делает попытку получить блокировку на запись (которую он получить не может, поскольку первый поток получил блокировку на чтение). Оставшаяся часть функции никогда не будет выполнена.

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

solaris % testcancel
thread1() got a read lock
thread2() trying to obtain a write lock

и мы никогда не вернемся к приглашению интерпретатора. Программа зависнет. Произошло вот что:

1. Второй поток вызвал pthread_rwlock_wrlock (листинг 8.6), которая была заблокирована в вызове pthread_cond_wait.

2. Первый поток вернулся из вызова slеер(3) и вызвал pthread_cancel.

3. Второй поток был отменен и завершил работу. При отмене потока, заблокированного в ожидании сигнала по условной переменной, взаимное исключение блокируется до вызова первого обработчика-очистителя. (Мы не устанавливали обработчик, но взаимное исключение все равно блокируется до завершения потока.) Следовательно, при отмене выполнения второго потока взаимное исключение осталось заблокированным и значение rw_nwaitwriters в листинге 8.6 было увеличено.

4. Первый поток вызывает pthread_rwlock_unlock и блокируется навсегда при вызове pthread_mutex_lock (листинг 8.8), потому что взаимное исключение все еще заблокировано отмененным потоком.

Если мы уберем вызов pthread_rwlock_unlock в функции thread1, функция main выведет вот что:

rw_refcount = 1, rw_nwaitreaders = 0, rw_nwaitwriters = 1
pthread_rwlock_destroy error: Device busy

Первый счетчик имеет значение 1, поскольку мы удалили вызов pthread_rwlock_ unlock, а последний счетчик имеет значение 1, поскольку он был увеличен вторым потоком до того, как тот был отменен.

Исправить эту проблему просто. Сначала добавим две строки к функции pthread_rwlock_rdlock в листинге 8.4. Строки отмечены знаком +:

  rw->rw_nwaitreaders++;
+ pthread_cleanup_push(rwlock_cancelrdwait, (void *) rw);
  result = pthread_cond_wait(&rw->rw_condreaders, &rw->rw_mutex);
+ pthread_cleanup_pop(0);
  rw->rw_nwaitreaders++;

Первая новая строка устанавливает обработчик-очиститель (функцию rwlock_cancelrdwait), а его единственным аргументом является указатель rw. После возвращения из pthread_cond_wait вторая новая строка удаляет обработчик. Аргумент функции pthread_cleanup_pop означает, что функцию-обработчик при этом вызывать не следует. Если этот аргумент имеет ненулевое значение, обработчик будет сначала вызван, а затем удален.

Если поток будет отменен при вызове pthread_cond_wait, возврата из нее не произойдет. Вместо этого будут запущены обработчики (после блокирования соответствующего взаимного исключения, как мы отметили в пункте 3 чуть выше).

В листинге 8.10 приведен текст функции rwlock_cancelrdwait, являющейся обработчиком-очистителем для phtread_rwlock_rdlock.

Листинг 8.10. Функция rwlock_cancelrdwait: обработчик для блокировки чтения

//my_rwlock_cancel/pthread_rwlock_rdlock.с
3  static void
4  rwlock_cancelrdwait(void *arg)
5  {
6   pthread_rwlock_t *rw;
7   rw = arg;
8   rw->rw_nwaitreaders--;
9   pthread_mutex_unlock(&rw->rw_mutex);
10 }

8-9 Счетчик rw_nwaitreaders уменьшается, а затем разблокируется взаимное исключение. Это состояние, которое должно быть восстановлено при отмене потока.

Аналогично мы исправим текст функции pthread_rwlock_wrlock из листинга 8.6. Сначала добавим две новые строки рядом с вызовом pthread_cond_wait:

  rw->rw_nwaitreaders++;
+ pthread_cleanup_push(rwlock_cancelrwrwait, (void*) rw);
  result = pthread_cond_wait(&rw->rw_condwriters, &rw->rw_mutex);
+ pthread_cleanup_pop(0);
  rw->rw_nwaitreaders--;

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

Листинг 8.11. Функция rwlock_cancelwrwait: обработчик для блокировки записи

//my_rwlock_cancel/pthread_rwlock_wrlock.с
3  static void
4  rwlock_cancelwrwait(void *arg)
5  {
6   pthread_rwlock_t *rw;
7   rw = arg;
8   rw->rw_nwaitwriters––;
9   pthread_mutex_unlock(&rw->rw_mutex);
10 }

8-9 Счетчик rw_nwaitwriters уменьшается, и взаимное исключение разблокируется. При запуске нашей тестовой программы из листинга 8.9 с этими новыми функциями мы получим правильные результаты:

solaris %testcancel
thread1() got a read lock
thread2() trying to obtain a write lock
rw_refcount = 0, rw_nwaitreaders = 0, rw_nwaitwriters = 0

Теперь три счетчика имеют правильные значения, первый поток возвращается из вызова pthread_rwlock_unlock, а функция pthread_rwlock_destroy не возвращает ошибку EBUSY.

ПРИМЕЧАНИЕ

Этот раздел представляет собой обзор вопросов, связанных с отменой выполнения потоков. Для более детального изучения этих проблем можно обратиться, например, к разделу 5.3 книги [3].

8.6. Резюме

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

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

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

Упражнения

1. Измените реализацию в разделе 8.4 таким образом, чтобы приоритет имели считывающие, а не записывающие потоки.

2. Сравните скорость работы нашей реализации из раздела 8.4 с предоставленной производителем. 

ГЛАВА 9