Средства синхронизации потоков управления

Мьютексы

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

#include <pthread.h>

int pthread_mutex_init (
    pthread_mutex_t *restrict mutex, 
    const pthread_mutexattr_t
        *restrict attr);

int pthread_mutex_destroy (
    pthread_mutex_t *mutex);

pthread_mutex_t mutex =
    PTHREAD_MUTEX_INITIALIZER;

#include <pthread.h>

int pthread_mutex_lock (
    pthread_mutex_t *mutex);

int pthread_mutex_trylock (
    pthread_mutex_t *mutex);

int pthread_mutex_unlock (
    pthread_mutex_t *mutex);

#include <pthread.h>
#include <time.h>
int pthread_mutex_timedlock (
    pthread_mutex_t *restrict mutex,
    const struct
        timespec *restrict abstime);

#include <pthread.h>

int pthread_mutex_getprioceiling (
    const pthread_mutex_t 
    *restrict mutex,
    int *restrict prioceiling);

int pthread_mutex_setprioceiling (
    pthread_mutex_t 
    *restrict mutex, int prioceiling, 
    int *restrict old_ceiling);

#include <pthread.h>

int pthread_mutexattr_init (
    pthread_mutexattr_t *attr);

int pthread_mutexattr_destroy (
    pthread_mutexattr_t *attr);

#include <pthread.h>

int pthread_mutexattr_gettype (
    const pthread_mutexattr_t
        *restrict attr, 
    int *restrict type);

int pthread_mutexattr_settype (
    pthread_mutexattr_t *attr, int type);

int pthread_mutexattr_getpshared (
    const pthread_mutexattr_t
        *restrict attr, 
    int *restrict pshared);

int pthread_mutexattr_setpshared (
    pthread_mutexattr_t *attr,
    int pshared);

int pthread_mutexattr_getprotocol (
    const pthread_mutexattr_t
        *restrict attr, 
    int *restrict protocol);

int pthread_mutexattr_setprotocol (
    *attr, int protocol);

int pthread_mutexattr_getprioceiling (
    const pthread_mutexattr_t
        *restrict attr, 
    int *restrict prioceiling);
int pthread_mutexattr_setprioceiling (
    pthread_mutexattr_t *attr,
    int prioceiling);

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

Сразу после инициализации функцией pthread_mutex_init() мьютекс, разумеется, оказывается свободным. Разрушить функцией pthread_mutex_destroy() можно только инициализированный, свободный мьютекс.

Для инициализации статически описанных мьютексов с подразумеваемыми значениями атрибутов целесообразно воспользоваться макросом PTHREAD_MUTEX_INITIALIZER. Эффект будет тем же, что и после вызова pthread_mutex_init() с пустым указателем в качестве значения аргумента attr, только без накладных расходов на проверку корректности атрибутного объекта.

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

У инициализированного мьютекса имеется четыре атрибута:

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

В стандарте POSIX-2001 определены четыре типа мьютексов.

PTHREAD_MUTEX_NORMAL

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

PTHREAD_MUTEX_ERRORCHECK

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

PTHREAD_MUTEX_RECURSIVE

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

PTHREAD_MUTEX_DEFAULT

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

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

Атрибут " протокол " влияет на планирование потока управления во время владения мьютексом. Согласно стандарту, возможных протоколов три.

PTHREAD_PRIO_NONE

Владение мьютексом не влияет на приоритет и планирование потока управления.

PTHREAD_PRIO_INHERIT

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

PTHREAD_PRIO_PROTECT

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

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

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

Признак использования несколькими процессами мьютекса, в соответствии со стандартом POSIX-2001, может принимать два значения.

PTHREAD_PROCESS_PRIVATE

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

PTHREAD_PROCESS_SHARED

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

На попытки захвата мьютекса, осуществляемые с применением функций pthread_mutex_lock(), pthread_mutex_trylock() или pthread_mutex_timedlock() значения атрибутов влияют так, как описано выше.

Как правило, поток, вызвавший pthread_mutex_lock(), блокируется, если мьютекс уже захвачен, и ждет его освобождения. В аналогичной ситуации функция pthread_mutex_trylock() немедленно завершается, возвращая код ошибки, а функция pthread_mutex_timedlock() блокируется, пока либо мьютекс не освободится, либо не наступит заданный аргументом abstime момент времени, отсчитываемого по часам CLOCK_REALTIME.

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

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

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

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

#ifndef g_MALLOC
#define g_MALLOC

/* Количество размеров (в словах типа size_t), */
/* для которых поддерживаются */
/* разные списки памяти */
#define DIF_SIZES         8

/* Размер пула памяти */
#define POOL_SIZE         65536

/* Указатель на кусок памяти нулевого размера */
#define g_NULL  ((void *) (-1))

/* Первое поле следующей структуры нужно
/* для всех кусков памяти, а второе – */
/* только для провязки свободных.*/
/* При отведении памяти адрес второго */
*/ поля выдается как результат */
typedef struct listi {
    size_t length;
    struct listi *pnext;
} *list_of_mem;

extern void *g_malloc (size_t);
extern void g_free (void *);

#endif

/* * * * * * * * * * * * * * * * * * * * * */
/* Функции выделения и освобождения памяти */
/* * * * * * * * * * * * * * * * * * * * * */

#include <pthread.h>
#include <errno.h>
#include <assert.h>
#include <string.h>
#include "g_malloc.h"

static char mem_pool [POOL_SIZE] = {0, };
/* Размер занятой части  пула памяти */
/* Списки свободного пространства */
/* (по одному на каждый размер */
/* от 1 до DIF_SIZES) */
static size_t cur_pool_size = 0;

static list_of_mem short_lists [DIF_SIZES] =
    {NULL, NULL, NULL, NULL, NULL,
     NULL, NULL, NULL};
/* Список больших */
/* свободных кусков (превосходящих DIF_SIZES) */
/* Разные мьютексы для разных */
/* групп списков свободного пространства */
static list_of_mem big_list = NULL;

static pthread_mutex_t short_lists_mutex = 
        PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t big_list_mutex = 
        PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t pool_mutex = 
        PTHREAD_MUTEX_INITIALIZER;

/* * * * * * * * * * */
/* Выделение памяти     */
/* * * * * * * * * * */
void *g_malloc (size_t size) {
/* Указатель для хождения по списку */
/* больших свободных кусков */
    list_of_mem *p; 
/* Указатель для хождения по спискам */
/* свободных кусков */
    list_of_mem pt;
    /* Индекс в массиве short_lists */
    size_t ls; 
    size_t ts; /* Временная переменная */

    if (size == 0) {
        return (g_NULL);
/* Важно, чтобы результат был */
/* отличен от NULL, поскольку NULL */
/* – признак ненормального завершения */
    }

    /* Округлим запрошенный размер вверх */
    /* до кратного размеру size_t и */
    /* прибавим слово служебной информации */
    size = (size – 1 + 2 * sizeof (size_t)) 
             ~(sizeof (size_t) – 1);

    /* Вычислим индекс в массиве */
    /*  short_lists [], соответствующий */
    /* запрошенному размеру */
    ls = size / sizeof (size_t) – 2;

    if (ls < DIF_SIZES) {
        /* Попробуем выдать кусок */
        /* из списка коротких */
        assert (
            pthread_mutex_lock(
              short_lists_mutex) == 0);
        if ((pt = short_lists [ls]) !=
                  NULL) {
            /* Есть нужный кусок */
          short_lists [ls] =
           (short_lists [ls])->pnext;
          assert (pthread_mutex_unlock(
            short_lists_mutex) == 0);
          return (pt->pnext);
        }
        assert (pthread_mutex_unlock(
          short_lists_mutex) == 0);
    }

    /* Попробуем выдать кусок из */
    /* списка больших */
    assert (pthread_mutex_lock(
      big_list_mutex) == 0);
    for (p = big_list, pt = *p; pt != NULL; 
          p = pt->pnext, pt = *p) {
        if ((signed long) (ts =
          pt->length – size) >= 0) {
          /* Нашли подходящий кусок */
            if (ts < sizeof (*pt)) {
    /* Придется выдать кусок целиком – */
    /* в остатке не помещается */
    /* служебная информация */
    *p = pt->pnext;
        } else {
        /* Отрежем сколько надо и, */
        /* при необходимости,     */
        /* перецепим остаток в */
        /* список коротких     */
        if ((ls = (pt->length = ts) /
             sizeof (size_t) – 2) 
                < DIF_SIZES) {
        *p = pt->pnext;
        assert (pthread_mutex_lock(
          short_lists_mutex) == 0);
        pt->pnext = short_lists [ls];
        short_lists [ls] = pt;
        assert (pthread_mutex_unlock(
          short_lists_mutex) == 0);
            }
        pt = (list_of_mem) ((char *) pt
          + ts);
        pt->length = size;
            }
            assert (
             pthread_mutex_unlock(
              big_list_mutex) == 0);
            return (pt->pnext);
        }
    } /* for */
    assert (pthread_mutex_unlock (
      big_list_mutex) == 0);

    /* Кусок из большого списка */
    /* выдать не удалось. */
    /* Попробуем взять прямо из */
    /* пула памяти */
    assert (pthread_mutex_lock(
      pool_mutex)
      == 0);
    if (cur_pool_size + size <= POOL_SIZE) {
        pt = (list_of_mem) (mem_pool
          + cur_pool_size);
        pt->length = size;
        cur_pool_size += size;
        assert (pthread_mutex_unlock (
          pool_mutex) == 0);
        return (pt->pnext);
    }
    assert (pthread_mutex_unlock(
      pool_mutex) == 0);

    /* Неудача при выделении памяти */
    errno = ENOMEM;
    return (NULL);
}

/* * * * * * * * * * * * * * * * * * */
/* Возврат ранее запрошенной памяти */
/* * * * * * * * * * * * * * * * * * */
void g_free (void *p) {
    list_of_mem pt;
    size_t size, ls;

    if ((p == g_NULL) || (p == NULL)) {
        return;
    }

    /* Установим указатель на */
    /* служебную информацию */
    pt = (list_of_mem) ((char *) p
      – sizeof (size_t));
    size = pt->length;
    ls = size / sizeof (size_t) – 2;
    memset (p, 0, size – sizeof (size_t));

    /* Не из конца ли пула этот кусок? */
    assert (pthread_mutex_lock(
      pool_mutex) == 0);
    if (((char *) pt + size) ==
      (mem_pool + cur_pool_size)) {
        pt->length = 0;
        cur_pool_size -= size;
        assert (pthread_mutex_unlock(
          pool_mutex) == 0);
        return;
    }
    assert (pthread_mutex_unlock(
      pool_mutex) == 0);

/* Добавим освободившийся кусок */
/* к одному из списков */
    if (ls < DIF_SIZES) {
        assert (pthread_mutex_lock(
          short_lists_mutex) == 0);
        pt->pnext = short_lists [ls];
        short_lists [ls] = pt;
        assert (pthread_mutex_unlock(
          short_lists_mutex) == 0);
    } else {
        /* Добавим к большому списку */
        assert (pthread_mutex_lock(
          big_list_mutex) == 0);
        pt->pnext = big_list;
        big_list = pt;
        assert (pthread_mutex_unlock(
          big_list_mutex) == 0);
    }
}

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

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

Для библиотечных функций удобны предоставляемые стандартом POSIX-2001 средства инициализации статически описанных мьютексов.

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

#include <unistd.h>
#include <stdio.h>
#include <pthread.h>
#include "g_malloc.h"

static void *start_func (void *dummy) {
    void *p1, *p2, *p3;

    printf ("g_malloc (65000): %p\n",
      p1 = g_malloc (65000));
    sleep (1);
    printf ("g_malloc (1): %p\n",
      p2 = g_malloc (1));
    sleep (1);
    g_free (p1);
    sleep (1);
    g_free (p2);
    sleep (1);
    printf ("g_malloc (64990): %p\n",
      p1 = g_malloc (64990));
    sleep (1);
    printf ("g_malloc (1): %p\n",
      p2 = g_malloc (1));
    sleep (1);
    printf ("g_malloc (5): %p\n",
      p3 = g_malloc (5));
    sleep (1);
    g_free (p1);
    sleep (1);
    g_free (p2);
    sleep (1);
    g_free (p3);
    sleep (1);
    printf ("g_malloc (100000): %p\n",
      p3 = g_malloc (100000));

    return (NULL);
}

int main (void) {
    pthread_t pt1, pt2;

    pthread_create (pt1, NULL,
      start_func, NULL);
    pthread_create (pt2, NULL,
      start_func, NULL);

    pthread_join (pt1, NULL);
    pthread_join (pt2, NULL);

    return (0);
}

g_malloc (65000): 0x8049024
g_malloc (65000): (nil)
g_malloc (1): 0x8058e10
g_malloc (1): 0x8058e18
g_malloc (64990): 0x804902c
g_malloc (64990): (nil)
g_malloc (1): 0x8049024
g_malloc (1): 0x8058e10
g_malloc (5): 0x8058e18
g_malloc (5): 0x8058e24
g_malloc (100000): (nil)
g_malloc (100000): (nil)

Мьютексы – весьма подходящее средство для реализации обеда философов. Ниже приведена программа, написанная С.В. Самборским (см. листинг 2.11).

/*
Обедающие философы. Многопотоковая
реализация с помощью мьютексов. Запуск:
    mudrecMutex [-a | -p | -I]
      [-t число_секунд] имя_философа ...

Опции:
        -t число_секунд – сколько секунд
         моделируется

        Стратегии захвата вилок:
        -a – сначала захватывается вилка
             с меньшим номером;
        -p – сначала захватывается
             нечетная вилка;
        -I – некорректная (но
             эффективная) интеллигентная 
             стратегия: во время
             ожидания уже захваченная
             вилка кладется.

Пример запуска:
          mudrecMutex -p -t 300 A B C D E F G\
                                H I J K L M N\
                                O P Q R S T U\
                                V W X Y Z
*/

static char rcsid[] __attribute__((unused)) = \
 "$Id: mudrecMutex.c,v 1.14 2003/11/20 16:09:20"
 "sambor Exp $";

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <signal.h>
#include <string.h>
#include <fcntl.h>
#include <limits.h>
#include <errno.h>

#define max(a,b) ((a)>(b)?(a):(b))
#define min(a,b) ((a)>(b)?(b):(a))

struct mudrec {
    char *name;
    int left_fork, right_fork;
    int eat_time, wait_time, think_time,
       max_wait_time;
    int count;
    pthread_t thread;
    int private_pFdIn;
} *kafedra;

/* Глобальные счетчики и логические переменные */
int Stop = 0; /* Признак конца обеда */

/* Различные дескрипторы */
int protokol [2] = {-1, -1};
#define pFdIn (protokol [1])
#define pFdOut (protokol [0])

/* Массив мьютексов для синхронизации доступа */
/* к вилкам */
pthread_mutex_t *mtxFork;

/* Разные алгоритмы захвата вилок */
static void get_forks_simple (
  struct mudrec *this);
static void get_forks_odd (
  struct mudrec *this);
static void get_forks_maybe_infinit_time(
  struct mudrec *this);

/* Используемый метод захвата вилок */
void (*get_forks) (struct mudrec *this) =
  get_forks_simple;

/* Возвращение вилок */
static void put_forks (struct mudrec *this);

/*Потоки-философы */
void *filosof (void *arg) {
    struct mudrec *this = arg;
    char buffer [LINE_MAX];
    int bytes;
    int private_pFdIn = this->private_pFdIn;

    while (!Stop) {
        /* Пора подкрепиться */
        {
            int wait_time, tm =
              time (NULL);

            bytes = write (
              private_pFdIn, buffer, 
              strlen (buffer));

            (*get_forks) (this);

            wait_time =
              time (NULL) – tm;
            this->wait_time +=
              wait_time;
            this->max_wait_time =
              max (wait_time, 
              this->max_wait_time);

        sprintf(
         buffer,
         "%s: ждал вилок %d сек\n", 
         this->name, wait_time);
         bytes = write (private_pFdIn,
                        buffer, 
                        strlen (buffer));
        }

        /* Может, обед уже закончился? */
        if (Stop) {
            put_forks (this);
            break;
        }

        /* Ест */
        {
            int eat_time =
              rand () % 20 + 1;

            sleep (eat_time);

            this->eat_time +=
              eat_time;
            this->count++;
            sprintf (buffer,
              "%s: ел %d сек\n",
              this->name, 
              eat_time);
            bytes = write (
               private_pFdIn, buffer, 
               strlen (buffer));
        }

        /* Отдает вилки */
        put_forks (this);

        if (Stop) break;

        /* Размышляет */
        {
            int think_time =
              rand () % 10 + 1;

            sleep (think_time);
            this->think_time +=
              think_time;
        }
    } /* while (!Stop) */

    sprintf (buffer,"%s: уходит\n",
             this->name);
    bytes = write (private_pFdIn,
                   buffer, strlen (buffer));

    close (private_pFdIn);

    return (NULL);
} /* Поток-философ */

/* Кладет вилки одну за другой */
static void put_forks (struct mudrec *this) {
    pthread_mutex_unlock (
      mtxFork [this->left_fork – 1]);
    pthread_mutex_unlock (
      mtxFork [this->right_fork – 1]);
}

/* Берет вилки по очереди в порядке номеров */
static void get_forks_simple(
     struct mudrec *this) {
    int first = min (
       this->left_fork, this->right_fork);
    int last = max (
       this->left_fork, this->right_fork);
    pthread_mutex_lock (
       mtxFork [first – 1]);
    pthread_mutex_lock (
       mtxFork [last – 1]);
}

/* Берем сначала нечетную вилку */
/* (если обе нечетные – */
/* то с большим номером) */
static void get_forks_odd (struct mudrec *this)
{
    int left = this->left_fork, right =
      this->right_fork;

    int first;
    int last;

    if ((left  1) > (right  1)) {
        first = left;
        last  = right;
    } else if ((left  1) < (right  1)) {
        first = right;
        last  = left;
    } else {
        first = max (left, right);
        last  = min (left, right);
    }

    pthread_mutex_lock (
      mtxFork [first – 1]);
    pthread_mutex_lock (
      mtxFork [last – 1]);
}

/* Берем вилки по очереди, в */
/* произвольном порядке.    */
/* Но если вторая вилка не */
/* берется сразу, то кладем */
/* первую. */
/* То есть философ не расходует */
/* вилочное время впустую.    */
static void get_forks_maybe_infinit_time 
        (struct mudrec *this) {
    int left = this->left_fork, right =
      this->right_fork;

    for (;;) {
        pthread_mutex_lock (
          mtxFork [left – 1]);
        if (0 == pthread_mutex_trylock 
            (mtxFork [right – 1]))
             return;
        pthread_mutex_unlock (
          mtxFork [left – 1]);
        pthread_mutex_lock (
          mtxFork [right – 1]);
        if (0 == pthread_mutex_trylock 
            (mtxFork [left – 1]))
        return;
        pthread_mutex_unlock (
          mtxFork [right – 1]);
    }
}

/* Мелкие служебные функции */
static void stop (int dummy) {
    Stop = 1;
}

static void usage (char name []) {
    fprintf (stderr,
        "Использование: %s"
        " [-a | -p | -I] "
        "[-t число_секунд] "
        "имя_философа ...\n", name);
    exit (1);
}

/* Точка входа демонстрационной программы */
int main (int argc, char *argv []) {
    char buffer [LINE_MAX], *p;
    int i, n, c;
    int open_room_time = 300;
    int nMudr;
    struct sigaction sact;

    while ((c = getopt (argc, argv,
      "apIt:")) != -1) {
        switch (c) {
        case 'a':
          get_forks =
            get_forks_simple;
           break;
        case 'p':
          get_forks =
            get_forks_odd;
           break;
        case 'I':
          get_forks = 
            get_forks_maybe_infinit_time;
          break;
        case 't':
          open_room_time = strtol(
            optarg, p, 0);
        if (optarg [0] == 0 || *p != 0)
          usage (argv [0]);
        break;
        default : usage (argv [0]);
        }
    }

    nMudr = argc – optind;
    /* Меньше двух */
    /* философов неинтересно ... */
    if (nMudr < 2) usage (argv [0]); 

    /* Создание канала для протокола */
    /* обработки событий */
    pipe (protokol);
kafedra = calloc (sizeof (struct mudrec), nMudr);

    /* Зачисление на кафедру */
    for (i = 0; i < nMudr; i++, optind++) {
        kafedra [i].name = argv [optind];
        /* Выдадим телефон */
        kafedra [i].private_pFdIn =
          fcntl (pFdIn, 
          F_DUPFD, 0);

        /* Укажем новичку, */
        /* какими вилками */
        /* пользоваться */
        kafedra [i].left_fork = i + 1;
        kafedra [i].right_fork = i + 2;
    }
    /* Последний */
    /* пользуется вилкой первого */
    kafedra [nMudr – 1].right_fork = 1; 

    /* Зададим реакцию на сигналы */
    /* и установим будильник */
    /* на конец обеда */
    sact.sa_handler = stop;
    (void) sigemptyset (sact.sa_mask);
    sact.sa_flags = 0;
    (void) sigaction (SIGINT, sact, 
        (struct sigaction *) NULL);
    (void) sigaction (SIGALRM, sact, 
        (struct sigaction *) NULL);
    alarm (open_room_time);

    /* Создадим мьютексы для охраны вилок */
    mtxFork = calloc (
      sizeof (pthread_mutex_t),
      nMudr);
    for (i = 0; i < nMudr; i++) {
        pthread_mutex_init (
          mtxFork [i],
          NULL);
    }

    /* Философы входят в столовую */
    for (i = 0; i < nMudr; i++)
        pthread_create (
           kafedra [i].thread,
           NULL, filosof, 
        (void *) kafedra [i]);

/* Выдача сообщений на стандартный */
/* вывод и выход */
/* после окончания всех задач */
    close (pFdIn);
    while (1) {
        n = read (pFdOut, buffer,
          LINE_MAX);
        if (n == 0 || (n < 0
           errno != EINTR)) break;
        for (i = 0; i < n; i++)
          putchar (buffer [i]);
    }
    close (pFdOut);

    /* Уничтожение мьютексов */
    for (i = 0; i < nMudr; i++) {
        pthread_mutex_destroy (
         mtxFork [i]);
    }

    /* Выдача сводной информации */
    {
        int full_eating_time = 0;
        int full_waiting_time = 0;
        int full_thinking_time = 0;
        for (i = 1; i <= nMudr; i++) {
            struct mudrec *this =
              kafedra [i – 1];

            full_eating_time +=
              this->eat_time;
            full_waiting_time +=
              this->wait_time;
            full_thinking_time +=
              this->think_time;

            if (this->count > 0) {
            float count =
              this->count;
            float think_time =
              this->think_time /
                count;
            float eat_time =
              this->eat_time /
                count;
            float wait_time =
              this->wait_time / 
                count;

            printf (
              "%s: ел %d раз в "
              "среднем: думал=%.1f "
              "ел=%.1f ждал=%.1f "
              "(максимум %d)\n", 
              this->name,
              this->count, 
              think_time,eat_time,
              wait_time, 
              this->max_wait_time);
            } else
            printf ("%s: не поел\n",
              this->name);
        } /* for */

        {
        float total_time = (
            full_eating_time + 
            full_waiting_time + 
            full_thinking_time) /
             (float) nMudr;
printf("Среднее число одновременно"
       " едящих = %.3f\n", 
       full_eating_time / total_time);
printf("Среднее число одновременно"
       " ждущих = %.3f\n",
       full_waiting_time / total_time);
        }
        }
        /* Выдача сводной информации */

    free (mtxFork);
    free (kafedra);

    /* Сообщим об окончании работы. */
    printf ("Конец обеда\n");

    return 0;
}

Переменные условия

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

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

while (! предикат) {
    Ожидание на переменной условия
    с освобождением мьютекса.
    После успешного завершения
    ожидания поток вновь 
    оказывается владельцем мьютекса.
}

После нормального выхода из цикла проверяемое условие истинно; можно выполнить требуемые действия и освободить мьютекс.

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

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

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

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

#include <pthread.h>

int pthread_cond_init (
  pthread_cond_t *restrict cond,
  const pthread_condattr_t
    *restrict attr);

int pthread_cond_destroy (
   pthread_cond_t *cond);

pthread_cond_t cond =
  PTHREAD_COND_INITIALIZER;

#include <pthread.h>

int pthread_cond_wait (
  pthread_cond_t *restrict cond,
  pthread_mutex_t *restrict mutex);

int pthread_cond_timedwait (
  pthread_cond_t *restrict cond,
  pthread_mutex_t *restrict mutex, 
  struct timespec *restrict abstime);

#include <pthread.h>

int pthread_cond_broadcast (
  pthread_cond_t *cond);

int pthread_cond_signal (
  pthread_cond_t *cond);

#include <pthread.h>

int pthread_condattr_init (
  pthread_condattr_t *attr);

int pthread_condattr_destroy (
  pthread_condattr_t *attr);

#include <pthread.h>

int pthread_condattr_getpshared (
  const pthread_condattr_t 
  *restrict attr, int *restrict pshared);

int pthread_condattr_setpshared (
  pthread_condattr_t *attr, 
  int pshared);

int pthread_condattr_getclock (
  const pthread_condattr_t 
  *restrict attr, clockid_t
    *restrict clock_id);

int pthread_condattr_setclock (
  pthread_condattr_t *attr, 
  clockid_t clock_id);

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

Функции pthread_cond_wait() и pthread_cond_timedwait() блокируют вызывающий поток управления на переменной условия cond. Этот поток должен являться владельцем мьютекса, заданного аргументом mutex.

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

По завершении ожидания на переменной условия cond поток управления вновь становится владельцем мьютекса mutex. Это дает ему возможность проверить истинность предиката и либо продолжить выполнение, выйдя из цикла, либо возобновить ожидание.

Если значение предиката оказалось ложным, ложным называется и разблокирование потока управления. Для функции pthread_cond_wait() оно может объясняться по крайней мере двумя причинами:

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

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

rc = 0;
while (! predicate  rc == 0) {
    rc = pthread_cond_timedwait (
      cond, mutex, ts);
}

Если бы аргумент ts задавался как интервал, его пришлось бы перевычислять в цикле; кроме того, контроль по времени оказался бы смазанным непредсказуемыми задержками в выполнении потока управления.

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

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

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

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

Не рекомендуется вызывать pthread_cond_signal() (а также освобождать мьютексы ) в (асинхронно выполняемых) функциях обработки сигналов.

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

/* * * * * * * * * * * * * * * * * */
/* Многопотоковый вариант обеда    */
/* философов с использованием      */
/* мьютексов и переменных условия  */
/* * * * * * * * * * * * * * * * * */

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <errno.h>

/* Число обедающих философов */
#define QPH             5

/* Время (в секундах) на обед */
#define FO             15

/* Длительность еды */
#define ernd (rand () % 3 + 1)

/* Длительность разговора */
#define trnd (rand () % 5 + 1)

/* Состояние вилок */
static int fork_busy [QPH] = {0, };

/* Синхронизирующая переменная условия */
static pthread_cond_t forks_cond =
  PTHREAD_COND_INITIALIZER;

/* Синхронизирующий мьютекс */
static pthread_mutex_t forks_mutex =
  PTHREAD_MUTEX_INITIALIZER;

/* * * * * * * * * * * * * * * * * * */
/* Стартовая функция потока-философа.     */
/* Аргумент – номер философа     */
/* * * * * * * * * * * * * * * * * * */
void *start_phil (void *no) {
    /* Время до конца обеда */
    int fo;
    /* Время очередного отрезка */
    /* еды или беседы */     
    int t;         

    fo = FO;
    while (fo > 0) { /* Обед */

        /* Философ говорит */
        printf ("Философ %d беседует\n",
          (int) no);
        t = trnd; sleep (t); fo -= t;

        /* Пытается взять вилки */
        (void) pthread_mutex_lock (
          forks_mutex);
        while (fork_busy [(int) no – 1]
          || 
          fork_busy [(int) no % QPH] ) {
            (void) pthread_cond_wait
            (
            forks_cond, 
            forks_mutex);
        }
        fork_busy [(int) no – 1] = 
            fork_busy [(int) no
              % QPH] = 1;
        (void) pthread_mutex_unlock (
          forks_mutex);

        /* Ест */
        printf ("Философ %d ест\n",
          (int) no);
        t = ernd; sleep (t); fo -= t;

        /* Отдает вилки */
        (void) pthread_mutex_lock (
          forks_mutex);
        fork_busy [(int) no – 1] = 
            fork_busy [(int) no 
            % QPH] = 0;
        (void) pthread_cond_broadcast
          (forks_cond);
        (void) pthread_mutex_unlock
          (forks_mutex);
    } /* while */

    printf ("Философ %d закончил обед\n",
      (int) no);
    return (NULL);
}

/* * * * * * * * * * * * * * * */
/* Создание потоков-философов  */
/*  и ожидание их завершения   */
/* * * * * * * * * * * * * * * */
int main (void) {
/* Массив идентификаторов */
/* потоков-философов */
    pthread_t pt_id [QPH];
    /* Номер философа */    
    int no;
    /* Атрибутный объект для */
    /* создания потоков */                 
    pthread_attr_t attr_obj; 

    if ((errno = pthread_attr_init(
      attr_obj)) != 0) {
        perror ("PTHREAD_ATTR_INIT");
        return (errno);
    }

    /* В очередь, господа */
    /* философы, в очередь! */
    if ((errno = 
            pthread_attr_setschedpolicy(
            attr_obj, 
            SCHED_FIFO)) != 0) {
        perror (
          "PTHREAD_ATTR_SETSCHEDPOLICY");
        return (errno);
    }

    /* Все – к столу */
    for (no = 1;  no <= QPH; no++) {
        if ((errno = pthread_create (
           pt_id [no – 1], 
                attr_obj,
                start_phil,
                (void *) no))
                 != 0)
            {
            perror (
              "PTHREAD_CREATE");
            return (no);
        }
    }

    /* Ожидание завершения обеда */
    for (no = 1; no <= QPH; no++) {
        (void) pthread_join (
         pt_id [no – 1], NULL);
    }

    return 0;
}

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

Философ 1 беседует
Философ 2 беседует
Философ 3 беседует
Философ 4 беседует
Философ 5 беседует
Философ 4 ест
Философ 2 ест
Философ 4 беседует
Философ 5 ест
Философ 2 беседует
Философ 3 ест
Философ 5 беседует
Философ 1 ест
Философ 3 беседует
Философ 4 ест
Философ 1 беседует
Философ 2 ест
Философ 2 беседует
Философ 1 ест
Философ 4 беседует
Философ 3 ест
Философ 1 беседует
Философ 5 ест
Философ 5 беседует
Философ 3 беседует
Философ 4 ест
Философ 2 ест
Философ 4 беседует
Философ 2 беседует
Философ 1 ест
Философ 3 ест
Философ 1 закончил обед
Философ 5 ест
Философ 3 закончил обед
Философ 2 ест
Философ 5 закончил обед
Философ 4 ест
Философ 2 закончил обед
Философ 4 закончил обед

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

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

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

Разумеется, приведенное решение задачи об обедающих философах является нечестным, поскольку во время ожидания на переменной условия, которое длится неопределенно долго, философ "отключается", он не беседует и не ест, что, по условию задачи, запрещено (этим недостатком страдает и программа из курса [1], основанная на групповых операциях с семафорами ). Однако как иллюстрация типичных способов работы с мьютексами и переменными условия данная программа имеет право на существование.

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

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

Типичный пример использования блокировок чтение-запись – синхронизация доступа к буферизованным в памяти фрагментам файловой системы, особенно к каталогам (чем ближе к корню, тем чаще их читают и реже изменяют). Важно отметить в этой связи, что, в отличие от блокировок, реализуемых функцией fcntl() (см. курс [1]), блокировки чтение-запись могут применяться и там, где файловая система отсутствует (например, в минимальных конфигурациях, функционирующих под управлением соответствующей подпрофилю стандарта POSIX операционной системы реального времени).

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

#include <pthread.h>

int pthread_rwlock_init (
  pthread_rwlock_t
    *restrict rwlock, 
  const pthread_rwlockattr_t
    *restrict attr);

int pthread_rwlock_destroy (
  pthread_rwlock_t *rwlock);

#include <pthread.h>

int pthread_rwlock_rdlock (
  pthread_rwlock_t *rwlock);

int pthread_rwlock_tryrdlock (
  pthread_rwlock_t *rwlock);

#include <pthread.h>
#include <time.h>
int pthread_rwlock_timedrdlock (
  pthread_rwlock_t *restrict rwlock, 
  const struct timespec *restrict abstime);

#include <pthread.h>

int pthread_rwlock_wrlock (
  pthread_rwlock_t *rwlock);

int pthread_rwlock_trywrlock (
  pthread_rwlock_t *rwlock);

#include <pthread.h>
#include <time.h>
int pthread_rwlock_timedwrlock (
  pthread_rwlock_t *restrict rwlock,
  const struct timespec *restrict abstime);

#include <pthread.h>
int pthread_rwlock_unlock (
  pthread_rwlock_t *rwlock);

#include <pthread.h>

int pthread_rwlockattr_init (
  pthread_rwlockattr_t *attr);

int pthread_rwlockattr_destroy (
  pthread_rwlockattr_t *attr);

#include <pthread.h>

int pthread_rwlockattr_getpshared (
  const pthread_rwlockattr_t 
  *restrict attr, int *restrict pshared);

int pthread_rwlockattr_setpshared (
  pthread_rwlockattr_t *attr, 
  int pshared);

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

Естественно, если блокировку установил писатель, читателю придется подождать. Если этим писателем оказался тот же поток, ждать придется долго...

Параллельно одним или несколькими потоками управления может быть установлено несколько блокировок на чтение. Сколько раз устанавливали такую блокировку, столько же раз ее необходимо снять. (Установка блокировки писателем возможна только после последнего снятия.) Максимально допустимое число одновременно установленных блокировок на чтение зависит от реализации.

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

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

В качестве иллюстрации применения блокировок чтение-запись приведем набор функций для работы со списками в многопотоковой среде. На листинге 2.28 показан заголовочный файл, содержащий необходимые описания, на листинге 2.29 – C-файл с реализацией функций. Листинг 2.30 содержит текст тестовой программы, листинг 2.31 – результаты ее работы. Обратим внимание на использование блокировок без приостановки выполнения и на проверку статуса завершения попыток их установить.

/* * * * * * * * * * * * * * */
/* Типы, структуры и функции     */
/* для многопотоковой работы     */
/* со списками */
/* * * * * * * * * * * * * * */

#ifndef g_LIST
#define g_LIST

/* Указатель на процедуру */
typedef void (*aproc) (void *, void *);

/* Указатель на целочисленную функцию */
typedef int (*afun) (void *, void *);

/* Элемент списка указателей */
/* на объекты произвольного типа */
typedef struct g_link {
    /* Ссылка на следующий элемент списка */
    struct g_link *pnext;
    /* Ссылка на объект */
    void *pvalue;             
} *g_linkP;

/* Список указателей на объекты */
/* произвольного типа */
typedef struct g_list {
    /* Голова списка */
    g_linkP head; 
  /* Блокировка, ассоциированная */
  pthread_rwlock_t lk;    
  /* со списком */
} *g_listP;

/* Инициализировать список */
extern void g_list_init (g_listP);

/* Завершить работу со списком */
extern void g_list_destroy (g_listP);

/* Вставить новый элемент */
/* в конец списка */
extern void g_list_ins_last (
  g_listP, void *);
/* Вставить новый элемент в */
/* список перед первым,     */
/* удовлетворяющим заданному */
/* свойству (или в конец)     */
extern void g_list_ins_fun (
  g_listP, void *, afun);

/* Удалить элемент из списка */
extern void g_list_del (
  g_listP, void *);

/* Применить процедуру ко всем */
/* элементам списка */
extern void g_list_forall_x (
  g_listP, aproc, void *);

/* Выбрать подходящий элемент списка */
extern void *g_list_suchas_x (
  g_listP, afun, void *);

/* Выдать элемент по номеру */
extern void *g_list_get_bynum (
  g_listP, int);

#endif

/* * * * * * * * * * * * * * * * * * * * */
/* Реализация функций для многопотоковой */
/* работы со списками. */
/* Применяются блокировки чтение-запись */
/* без приостановки выполнения */
/* * * * * * * * * * * * * * * * * * * * */

#define _XOPEN_SOURCE 600

#include <pthread.h>
#include <stdlib.h>
#include <errno.h>
#include "g_list.h"

/* * * * * * * * * * * * * */
/* Инициализировать список     */
/* * * * * * * * * * * * * */
void g_list_init (g_listP plist) {
    plist->head = NULL;
    errno = pthread_rwlock_init (
      plist->lk, NULL);
}
/* * * * * * * * * * * * * * * */
/* Завершить работу со списком     */
/* * * * * * * * * * * * * * * */
void g_list_destroy (g_listP plist) {
    errno = pthread_rwlock_destroy (
      plist->lk);
}

/* * * * * * * * * * * * * * * * * * * * */
/* Вставить новый элемент в конец списка */
/* * * * * * * * * * * * * * * * * * * * */
void g_list_ins_last (g_listP plist,
   void *pval) {
    register g_linkP pt;
    register g_linkP *p;

    if ((errno = pthread_rwlock_trywrlock (
      plist->lk)) != 0) {
        return;
    }

    for (p = plist->head; 
        *p != NULL; p = (*p)->pnext)
        ;
if ((pt = (g_linkP) malloc (sizeof (*pt)))
   != NULL) {
        pt->pnext = NULL;
        pt->pvalue = pval;
        *p = pt;
    }

    (void) pthread_rwlock_unlock (
      plist->lk);
}

/* * * * * * * * * * * * * * * * * */
/* Вставить новый элемент в список */
/* перед первым, */
/* удовлетворяющим заданному свойству */
/* (или в конец) */
/* * * * * * * * * * * * * * * * */
void g_list_ins_fun (g_listP plist,
   void *pval, afun fun) {
    register g_linkP pt;
    register g_linkP *p;

    if ((errno = pthread_rwlock_trywrlock (
      plist->lk)) != 0) {
        return;
    }

    for (p = plist->head;
            (*p != NULL) 
               (fun (pval, 
         (*p)->pnext->pvalue)
                   == 0);
            p = (*p)->pnext)
        ;

    if ((pt = (g_linkP) malloc (
      sizeof (*pt))) != NULL) {
        pt->pnext = NULL;
        pt->pvalue = pval;
        *p = pt;
    }

    (void) pthread_rwlock_unlock (
      plist->lk);
}
    
/* * * * * * * * * * * * * * */
/* Удалить элемент из списка */
/* * * * * * * * * * * * * * */
void g_list_del (g_listP plist,
   void *pval) {
    register g_linkP pt;
    register g_linkP *p;

    if ((errno = pthread_rwlock_trywrlock
      (plist->lk)) != 0) {
        return;
    }

    for (p = plist->head; *p != NULL;
       p = (*p)->pnext) {
        if ((*p)->pvalue == pval) {
            pt = *p;
            *p = pt->pnext;
            free (pt);
            errno = 
              pthread_rwlock_unlock(
              plist->lk);
            return;
        }
    }

    /* Пытаемся удалить */
    /* несуществующий элемент */
    (void) pthread_rwlock_unlock (
      plist->lk);
    errno = EINVAL;
}


/* * * * * * * * * * * * * */
/* Применить процедуру ко */
/* всем элементам списка */
/* * * * * * * * * * * * */
void g_list_forall_x (g_listP plist,
  aproc proc, void *extobj) {
    register g_linkP pt;

    /* Устанавливаем блокировку на запись, */
    /* поскольку, возможно, процедура */
    /* модифицирует     */
    /* объекты, ссылки на которые */
    /* хранятся в списке */
    if ((errno = pthread_rwlock_trywrlock(
      plist->lk)) != 0) {
        return;
    }

    for (pt = plist->head; pt != NULL;
       pt = pt->pnext) {
        proc (extobj, pt->pvalue);
    }

    (void) pthread_rwlock_unlock(
      plist->lk);
}

/* * * * * * * * * * * * * * * * * * */
/* Выбрать подходящий элемент списка     */
/* * * * * * * * * * * * * * * * * * */
void *g_list_suchas_x (g_listP plist,
  afun fun, void *extobj) {
  register g_linkP pt;

    /* Устанавливаем блокировку на */
    /* чтение, так как     */
    /* считаем, что функция не */
    /* модифицирует объекты,     */
    /* ссылки на которые хранятся */
    /* в списке     */
    if ((errno = pthread_rwlock_tryrdlock(
      plist->lk)) != 0) {
        return NULL;
    }

    for (pt = plist->head; pt != NULL;
      pt = pt->pnext) {
        if (fun (extobj, pt->pvalue)) {
            (void) pthread_rwlock_unlock(
              plist->lk);
            return (pt->pvalue);
        }
    }

    (void) pthread_rwlock_unlock(
      plist->lk);
    return NULL;
}

/* * * * * * * * * * * * * * */
/* Выдать элемент по номеру     */
/* * * * * * * * * * * * * * */
void *g_list_get_bynum (g_listP plist,
   int n) {
    register g_linkP pt;

    if ((errno =
       pthread_rwlock_tryrdlock(
      plist->lk)) != 0) {
        return NULL;
    }

    for (pt = plist->head; n--, pt != NULL;
       pt = pt->pnext) {
        if (n == 0) {
           (void) pthread_rwlock_unlock(
              plist->lk);
            return (pt->pvalue);
        }
    }

    /* Пытаемся извлечь несуществующий */
    /* компонент */
    (void) pthread_rwlock_unlock(
      plist->lk);
    errno = EINVAL;

    return NULL;
}

/* * * * * * * * * * * * * * * * * * * */
/* Тест функций многопотоковой работы  */
/* со списками                         */
/* * * * * * * * * * * * * * * * * * * */

#define _XOPEN_SOURCE 600

#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include "g_list.h"

/* Структура для засыпания на */
/* минимальное время */
static struct timespec nslp =
  {0, 1};

/* Структуры для возврата результатов */
/* потоками:    */
/* Сколько операций пытались сделать,      */
/* сколько из них оказались неудачными     */
/* из-за занятости списка         */
static struct pt_res {
    int nops;
    int nbusy;
} ptres [3] = {{0, 0}, {0, 0}, {0, 0}};

/* * * * * * * * * * * * * * * * * * * * */
/* Этот поток добавляет элементы к списку */
/* * * * * * * * * * * * * * * * * * * * */
static void *start_func_1 (void *plist) {
    int *p1;


 while (1) {
    if ((p1 = 
     (int *) malloc (
       sizeof (int))) !=
        NULL) {
        *p1 = rand ();
        errno = 0;
        g_list_ins_last (plist, p1);
        ptres [0].nops++;
        if (errno == EBUSY) {
        ptres [0].nbusy++;
        }
    }
    (void) nanosleep (nslp, NULL);
    }

    return (NULL);
}

/* * * * * * * * * * * * * * * * * * * * * */
/* Процедура для подсчета суммы элементов  */
/* списка                                  */
/* * * * * * * * * * * * * * * * * * * * * */
static void proc_sum (void *sum, void *pval) {
    *((int *) sum) += *((int *) pval);
}
/* * * * * * * * * * * * * * * * * * * * * */
/* Этот поток подсчитывает сумму элементов */
/* списка */
/* * * * * * * * * * * * * * * * * * * * * */
static void *start_func_2 (void *plist) {
    int sum;

    while (1) {
        sum = 0;
        errno = 0;
        g_list_forall_x (plist,
           proc_sum, sum);
        ptres [1].nops++;
        if (errno == EBUSY) {
            ptres [1].nbusy++;
        }
        (void) nanosleep (nslp, NULL);
    }

    return (NULL);
}

/* * * * * * * * * * * * * * * * * */
/* Этот поток удаляет из списка     */
/* элементы со случайными номерами     */
/* * * * * * * * * * * * * * * * * */
static void *start_func_3 (void *plist) {
    int *p1;

    while (1) {
        errno = 0;
        p1 = (int *) g_list_get_bynum(
          plist, rand ());
        ptres [2].nops++;
        if (errno == EBUSY) {
            ptres[2].nbusy++;
        }

        if (p1 != NULL) {
            errno = 0;
            g_list_del (plist, p1);
            ptres [2].nops++;
            if (errno == EBUSY) {
             ptres [2].nbusy++;
            }
            free (p1);
        }
        (void) nanosleep (nslp, NULL);
    }

    return (NULL);
}

/* * * * * * * * * * * * * * * * * * * * * * */
/* Начальный поток всех запустит, поспит, */
/* потом всех терминирует и выдаст статистику */
/* * * * * * * * * * * * * * * * * * * * * * */
int main (void) {
    g_listP plist;
    pthread_t pt1, pt2, pt3;

    g_list_init (plist);

    pthread_create (pt1, NULL,
      start_func_1, plist);
    pthread_create (pt2, NULL,
      start_func_2, plist);
    pthread_create (pt3, NULL,
      start_func_3, plist);

    sleep (10);

    pthread_cancel (pt1);
    pthread_cancel (pt2);
    pthread_cancel (pt3);

    pthread_join (pt1, NULL);
    pthread_join (pt2, NULL);
    pthread_join (pt3, NULL);

    g_list_destroy (plist);

    printf ("Попыток выполнить "
            "операцию со списком: %d\n", 
        ptres [0].nops + ptres [1].nops
          + ptres [2].nops);
    printf ("Число попыток, неудачных"
            " из-за занятости "
                "списка: %d\n", 
        ptres [0].nbusy + ptres [1].nbusy
          + ptres [2].nbusy);

    return (0);
}

Попыток выполнить операцию со списком: 1503 
Число попыток, неудачных из-за
  занятости списка: 0

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

Спин-блокировки

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

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

Для описываемых блокировок стандарт POSIX-2001 не предусматривает установки с ограниченным ожиданием. Это понятно, поскольку накладные расходы по времени на ограничение в типичном случае превысят само время ожидания.

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

Согласно стандарту POSIX-2001, спин-блокировки обслуживаются следующими группами функций:

#include <pthread.h>

int pthread_spin_init (
  pthread_spinlock_t *lock, int pshared);

int pthread_spin_destroy (
  pthread_spinlock_t *lock);

#include <pthread.h>

int pthread_spin_lock (
  pthread_spinlock_t *lock);

int pthread_spin_trylock (
  pthread_spinlock_t *lock);

#include <pthread.h>
int pthread_spin_unlock (
  pthread_spinlock_t *lock);

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

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

pthread_spin_lock (ss->lock);
/* Восстановим старую маску */
ss->blocked = oldmask;
/* Проверим ждущие сигналы */
pending = ss->pending  ~ss->blocked;
pthread_spin_unlock (ss->lock);

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

Барьеры

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

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

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

#include <pthread.h>

int pthread_barrier_init (
  pthread_barrier_t *restrict barrier, 
    const pthread_barrierattr_t
      *restrict attr,
      unsigned count);

int pthread_barrier_destroy (
  pthread_barrier_t *barrier);

#include <pthread.h>
int pthread_barrier_wait (
  pthread_barrier_t *barrier);

#include <pthread.h>

int pthread_barrierattr_init (
  pthread_barrierattr_t *attr);

int pthread_barrierattr_destroy (
  pthread_barrierattr_t *attr);

#include <pthread.h>

int pthread_barrierattr_getpshared 
      (const pthread_barrierattr_t
        *restrict attr, 
      int *restrict pshared);

int pthread_barrierattr_setpshared 
      (pthread_barrierattr_t *attr,
        int pshared);

Обратим внимание на аргумент count в функции инициализации барьера pthread_barrier_init(). Он задает количество синхронизируемых потоков управления. Столько потоков должны вызвать функцию pthread_barrier_wait(), прежде чем каждый из них сможет успешно завершить вызов и продолжить выполнение. (Разумеется, значение count должно быть положительным.)

Когда к функции pthread_barrier_wait() обратилось требуемое число потоков управления, одному из них (стандарт POSIX-2001 не специфицирует, какому именно) в качестве результата возвращается именованная константа PTHREAD_BARRIER_SERIAL_THREAD, а всем другим выдаются нули. После этого барьер возвращается в начальное (инициализированное) состояние, а выделенный поток может выполнить соответствующие объединительные действия.

Описанная схема работы проиллюстрирована листингом 2.40.

if ((status = pthread_barrier_wait(
  barrier)) ==
        PTHREAD_BARRIER_SERIAL_THREAD) {
 /* Выделенные (обычно – объединительные) */
 /* действия.     */
 /* Выполняются каким-то одним потоком */
 /* управления */

} else {
    /* Эта часть выполняется всеми */
    /* прочими потоками */
    /* управления */
    if (status != 0) {
    /* Обработка ошибочной ситуации */
    } else {
        /* Нормальное "невыделенное" */
        /* завершение ожидания */
        /* на барьере */
    }
}

/* Повторная синхронизация – */
/* ожидание завершения выделенных действий */
status = pthread_barrier_wait (barrier);
/* Продолжение параллельной работы */
    . . .

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

Аналогично, не являются точками терминирования функции pthread_mutex_lock() и pthread_spin_lock(). Если бы они были таковыми, то точками терминирования стали бы все функции, в том числе библиотечные, которые их вызывают – malloc(), free() и т.п. Обеспечить в обработчиках завершения корректное состояние объектов, обслуживаемых подобными функциями, довольно сложно; это деятельность, чреватая ошибками, которые трудно не только найти и исправить, но даже воспроизвести. С другой стороны, мьютексы и спин-блокировки предназначены для захвата на короткое время, без длительного ожидания, так что нечувствительность к терминированию в данном случае не составляет большой проблемы.

Работу с барьерами проиллюстрируем коллективными вычислениями, производимыми двумя потоками (см. листинг 2.41).

/* * * * * * * * * * * * * * * * * * * * * * */
/* Программа использует барьеры для слияния  */
/* результатов    */
/* коллективных вычислений ln(2) */
/* * * * * * * * * * * * * * * * * * * * * * */

#define _XOPEN_SOURCE 600

#include <pthread.h>
#include <stdio.h>
#include <errno.h>

static pthread_barrier_t mbrr;

/* Ряд для ln(2) будут суммировать */
/* два потока.    */
/* Один возьмет на себя положительные */
/* слагаемые,     */
/* другой – отрицательные         */
static double sums [2] = {0, 0};

/* * * * * * * * * * * * * * * * * * * * */
/* Стартовая функция потоков управления, */
/* участвующих в вычислениях ln(2).     */
/* Аргумент ns на самом деле     */
/* целочисленный и равен 1 или 2     */
/* * * * * * * * * * * * * * * * * * * * */
static void *start_func (void *ns) {
    double d = 1;
    double s = 0;
    int i;

    /* Вычислим свою часть ряда */
    for (i = (int) ns; i <= 100000000;
      i += 2) {
        s += d / i;
    }

    /* Запомним результат в нужном месте */
    sums [(int) ns – 1] = s;

    /* Синхронизируемся для получения */
    /* общего итога */
    if (pthread_barrier_wait (mbrr) == 
      PTHREAD_BARRIER_SERIAL_THREAD) {
        sums [0] -= sums [1];
    }
/* Указатель на итог возвращают оба потока */
    return (sums);
}

/* * * * * * * * * * * * * * * * * */
/* Инициализация барьера,          */
/* создание и ожидание завершения  */
/* потоков управления     */
/* * * * * * * * * * * * * * * * * */
int main (void) {
    pthread_t pt1, pt2;
    double *pd;

    if ((errno = pthread_barrier_init (
      mbrr, NULL, 2)) != 0) {
        perror ("PTHREAD_BARRIER_INIT");
        return (errno);
    }

    pthread_create (pt1, NULL,
      start_func, (void *) 1);
    pthread_create (pt2, NULL,
      start_func, (void *) 2);

    pthread_join (pt1,
      (void **) pd);
    pthread_join (pt2,
      (void **) pd);

    printf ("Коллективно вычисленное"
            " значение ln(2): %g\n", *pd);

    return (pthread_barrier_destroy(
      mbrr));
}

В данном случае второго ожидания на барьере не понадобилось – вместо этого потоки управления просто завершаются.