Средства межпроцессного взаимодействия в реальном времени

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

#include <mqueue.h>
int mq_send (mqd_t mqdes, 
	const char *msg_ptr, 
	size_t msg_len, 
	unsigned msg_prio);

#include <mqueue.h>
#include <time.h>
int mq_timedsend (
    mqd_t mqdes, const char *msg_ptr, 
	size_t msg_len, unsigned msg_prio, 
	const struct timespec *abstime);

Более точно: функции mq_send() и mq_timedsend() помещают сообщение из msg_len байт, на которое указывает аргумент msg_ptr, в очередь, заданную дескриптором mqdes (если она не полна), в соответствии с приоритетом msg_prio (большим значениям msg_prio соответствует более высокий приоритет сообщения ; допустимый диапазон - от 0 до MQ_PRIO_MAX - 1 ).

Если очередь полна, а флаг O_NONBLOCK не установлен, вызов mq_send() блокируется до появления свободного места. Функция mq_timedsend() в таких случаях контролирует длительность ожидания (до заданного аргументом abstime абсолютного момента времени по часам CLOCK_REALTIME ).

Для извлечения (разумеется, с удалением) сообщений из очереди служат функции mq_receive() и mq_timedreceive() (см. листинги 4.7 и 4.8). Извлекается самое старое из сообщений с самым высоким приоритетом и помещается в буфер, на который указывает аргумент msg_ptr. Если размер буфера (значение аргумента msg_len ) меньше атрибута очереди mq_msgsize, вызов завершается неудачей. Если значение msg_prio_ptr отлично от NULL, в указуемый объект помещается приоритет принятого сообщения.

#include <mqueue.h>
ssize_t mq_receive (
    mqd_t mqdes, char *msg_ptr, 
	size_t msg_len, unsigned *msg_prio_ptr);

#include <mqueue.h>
#include <time.h>
ssize_t mq_timedreceive (
    mqd_t mqdes, char *restrict msg_ptr,
	size_t msg_len, 
	unsigned *restrict msg_prio_ptr,
	const struct timespec *restrict abstime);

Если очередь пуста, а флаг O_NONBLOCK не установлен, вызов mq_receive() блокируется до появления сообщения. Функция mq_timedreceive() в таких случаях контролирует длительность ожидания.

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

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

#include <mqueue.h>
int mq_notify (mqd_t mqdes, 
	const struct sigevent *notification);

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

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

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

Поучительно сопоставить два вида очередей сообщений - только что описанные и те, что были представлены в курсе [1]. Сразу видно, что первые устроены проще и, следовательно, могут быть реализованы эффективнее. Во-первых, структура типа mq_attr гораздо компактнее, чем msqid_ds. Нет и речи о хранении времени последних операций и идентификаторов процессов, их выполнивших. Во-вторых, упрощены производимые проверки. Контролируется максимальный размер одного сообщения, а не суммарный размер сообщений в очереди. Права доступа проверяются при открытии, а не при каждой операции. Только наличие приоритетов вносит элемент сложности, сопоставимый с обработкой типов сообщений, но приоритеты так или иначе должны быть реализованы - либо на уровне ОС, либо в приложении, поэтому лучше этот аспект стандартизовать и поручить операционной системе.

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

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

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

/* Число потоков управления, порождающих случайные числа */
#define PT_N 	MQ_OPEN_MAX

/* Число сообщений, генерируемых каждым потоком управления */
#define MSG_N 	128

/* Длина имени очереди сообщений */
#define MQ_NAME_LENGTH 	PATH_MAX

/* Количество целых чисел в одном сообщении */
#define MSG_INT_SIZE 32

/* Максимальное число сообщений в очереди */
#define MQ_MSGS_MAX 16

/* Приоритет порождаемого сообщения */
#define prio_rnd	(rand () % MQ_PRIO_MAX)

/* Номер сигнала, используемого для уведомлений */
#define SIG_MQ_NOTIFY SIGRTMIN

/* Массив идентификаторов очередей сообщений */
static mqd_t mq_des [PT_N];

/* Массив структур для задания уведомлений */
/* о поступлении сообщений в очереди */
static struct sigevent evnt_mq_notify [PT_N];

/* Мьютекс, используемый для синхронизации */
/* доступа к переменной sum */
static pthread_mutex_t sm_mutex = PTHREAD_MUTEX_INITIALIZER;

/* Общий результат суммирования */
static int sum = 0;

/* * * * * * * * * * * * * * * * * * * * * * * */
/* Функция, вызываемая при получении уведомления*/
/* о том, что очередь сообщений стала непустой. */
/* Номер очереди передается как аргумент       */
/* * * * * * * * * * * * * * * * * * * * * * * */
static void msg_arrvd (union sigval pt_nm) {
	int msg_buf [MSG_INT_SIZE];
	mqd_t mqdes;
	unsigned int msg_prio;
	int msg_sum = 0;
	ssize_t msg_size;
	int i;

	mqdes = mq_des [pt_nm.sival_int];

	/* Примем и обработаем имеющиеся сообщения,     */
	/* а затем снова зарегистрируемся на получение  */
	/* такого же уведомления                        */
	while ((msg_size = mq_receive (mqdes, (char *) msg_buf, 
			MSG_INT_SIZE * sizeof (int), msg_prio)) > 0) {
		for (i = 0; i < (msg_size / (signed int) sizeof (int)); 
			i++) {
			msg_sum += msg_buf [i];
		}
		msg_sum *= msg_prio;
	}

	if ((errno = pthread_mutex_lock (sm_mutex)) != 0) {
		perror ("PTHREAD_MUTEX_LOCK");
	}
	sum += msg_sum;
	if ((errno = pthread_mutex_unlock (sm_mutex)) != 0) {
		perror ("PTHREAD_MUTEX_UNLOCK");
	}

	if (mq_notify (mqdes, 
		evnt_mq_notify [pt_nm.sival_int]) != 0) {
		perror ("MQ_NOTIFY");
	}
}


/* * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Стартовая функция потока, генерирующего числа и   */
/* посылающего сообщения.                            */
/* Аргумент - номер очереди сообщений                */
/* * * * * * * * * * * * * * * * * * * * * * * * * * */
void *start_sender (void *pt_nm) {

	int msg_buf [MSG_INT_SIZE];
	int i, j;

	/* Сформируем и пошлем заданное число сообщений */
	/* (проверяя, не переполнилась ли очередь)*/
	for (j = 0; j < MSG_N; j++) {
		for (i = 0; i < MSG_INT_SIZE; i++) {
			msg_buf [i] = rand ();
		}
		if (mq_send (mq_des [(int) pt_nm], (char *) msg_buf, 
			MSG_INT_SIZE * sizeof (int), prio_rnd) != 0) {
			perror ("MQ_SEND");
			return (NULL);
		}
	}

	return (NULL);
}

/* * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Создание очереди сообщений,                       */
/* регистрация на получение уведомлений,             */
/* создание и ожидание завершения потоков управления */
/* * * * * * * * * * * * * * * * * * * * * * * * * * */
int main (void) {
	/* Массив идентификаторов порождаемых */
	/* потоков. Эти потоки будут */
	pthread_t pt_mqs [PT_N]; /* генерировать сообщения*/
	/* Массив для генерации и хранения */
	/* имен очередей сообщений */
	char mq_name [PT_N] [MQ_NAME_LENGTH];
	struct mq_attr mqattrs; /* Атрибуты создаваемых очередей */
	int i;

	for (i = 0; i < PT_N; i++) {
		/* Создадим очереди сообщений */
		/* и зарегистрируемся на получение уведомлений */
		sprintf (mq_name [i], "g%d", i);

		mqattrs.mq_flags = O_NONBLOCK;
		mqattrs.mq_maxmsg = MQ_MSGS_MAX;
		mqattrs.mq_msgsize = MSG_INT_SIZE * sizeof (int);
		mqattrs.mq_curmsgs = 0;
		if ((mq_des [i] = mq_open (mq_name [i], 
			O_RDWR | O_CREAT | O_NONBLOCK, 0777, 
			mqattrs)) == (mqd_t) (-1)) {
			perror ("MQ_OPEN");
			return (-1);
		}

		/* Сформируем структуру evnt_mq_notify */
		evnt_mq_notify [i].sigev_notify = SIGEV_THREAD;
		evnt_mq_notify [i].sigev_signo = SIG_MQ_NOTIFY;
		evnt_mq_notify [i].sigev_value.sival_int = i;
		evnt_mq_notify [i].sigev_notify_function = msg_arrvd;
		evnt_mq_notify [i].sigev_notify_attributes = NULL;
		if (mq_notify (mq_des [i], evnt_mq_notify [i]) != 0) {
			perror ("MQ_NOTIFY_MAIN");
			return (-1);
		}

		/* Создадим потоки управления */
		if ((errno = pthread_create (pt_mqs [i], NULL, 
				start_sender, (void *) i))
			!= 0) {
			perror ("PTHREAD_CREATE");
			return (i);
		}
	} /* for */

	/* Ожидание завершения */
	for (i = 0; i < PT_N; i++) {
		(void) pthread_join (pt_mqs [i], NULL);
	}

	/* Закроем дескрипторы и удалим очереди */
	for (i = 0; i < PT_N; i++) {
		(void) mq_close (mq_des [i]);
		(void) mq_unlink (mq_name [i]);
	}

	printf ("Общая сумма: %d\n", sum);

	if ((errno = pthread_mutex_destroy (sm_mutex)) != 0) {
		perror ("PTHREAD_MUTEX_DESTROY");
		return (errno);
	}

	return 0;
}

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

Семафоры реального времени

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

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

Если при освобождении семафора множество ждущих потоков было непусто, один из них удаляется из этого множества и его выполнение возобновляется; в противном случае значение семафора просто увеличивается.

Семафоры бывают именованными и безымянными ( неименованными ). Первые именуются цепочками символов и создаются функцией sem_open() с флагом O_CREAT, вторые создаются функцией sem_init(). При прочих операциях семафор идентифицируется открытым дескриптором (который может быть унаследован у родительского процесса, вызвавшего fork() ). Дескриптор реализуется как указатель на объект типа sem_t .

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

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

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

#include <semaphore.h>

sem_t *sem_open (const char *name, int oflag, ...);

int sem_init (sem_t *sem, int pshared, unsigned value);

Имена семафоров (аргумент name функции sem_open() ) устроены и трактуются так же, как и описанные выше имена очередей сообщений. Флагов (в аргументе oflag ) может быть установлено два: O_CREAT и/или O_EXCL. Если установлен флаг O_CREAT, то при вызове функции sem_open() необходимо задать два дополнительных аргумента: режим доступа (тип mode_t ) и значение семафора (тип unsigned int ).

В случае ошибки функция sem_open() возвращает значение SEM_FAILED, отличное от любого допустимого указателя на объект типа sem_t, а sem_init() "по старинке" возвращает -1.

Отметим, что нормальный результат для функции sem_init() в POSIX-2001 не стандартизован; вероятно, в будущих версиях им станет нуль. Инициализированный объект типа sem_t помещается по указателю sem. Аргумент value задает начальное значение создаваемого неименованного семафора. Если аргумент pshared отличен от нуля, семафор разделяется между процессами; в противном случае разделение возможно только между потоками управления вызывающего процесса.

Именованный семафор можно закрыть, обратившись к функции sem_close(), и удалить с помощью функции sem_unlink(); для ликвидации неименованных семафоров служит функция sem_destroy() (см. листинг 4.12). Нормальный результат этих функций равен нулю, в случае ошибки возвращается -1.

#include <semaphore.h>

int sem_close (sem_t *sem);

int sem_unlink (const char *name);

int sem_destroy (sem_t *sem);

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

Для захвата семафоров служат функции sem_wait(), sem_trywait() и sem_timedwait() (см. листинги 4.13 и 4.14).

#include <semaphore.h>

int sem_wait (sem_t *sem);

int sem_trywait (sem_t *sem);

#include <semaphore.h>
#include <time.h>
int sem_timedwait (sem_t *restrict sem, 
	const struct timespec *restrict abstime);

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

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

#include <semaphore.h>
int sem_post (sem_t *sem);

Функция sem_getvalue() (см. листинг 4.16) позволяет опросить значение семафора, не меняя его состояния.

#include <semaphore.h>
int sem_getvalue (sem_t *restrict sem, int *restrict sval);

Значение семафора записывается по указателю sval. Если семафор был захвачен, это значение окажется нулевым или отрицательным.

Если сопоставить семафоры реального времени с теми, что были описаны в курсе [1], то можно сделать те же выводы, что и для очередей сообщений. Семафоры реального времени, согласно стандарту POSIX-2001, по сути являются бинарными. Они устроены проще и, следовательно, могут быть реализованы эффективнее. Главное - отсутствуют тяжеловесные, со сложной семантикой, трудные для реализации групповые операции.

Использование семафоров реального времени проиллюстрируем программой, реализующей взаимодействие поставщик/потребитель (см. листинг 4.17).

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

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <semaphore.h>
#include <pthread.h>
#include <errno.h>
/* Буфер для хранения генерируемых данных */
static int my_buf;

/* Семафор, разрешающий записывать в буфер новые данные */
static sem_t w_sem;

/* Семафор, разрешающий читать данные из буфера */
static sem_t r_sem;

/* Общий результат суммирования */
static int sum = 0;

/* * * * * * * * * * * * * * * * * * * * * * * */
/* Стартовая функция потока, генерирующего числа */
/* * * * * * * * * * * * * * * * * * * * * * * */
void *start_writer (void *dummy) {
	while (sem_wait (w_sem) == 0) {
		my_buf = rand ();
		if (sem_post (r_sem) != 0) {
			perror ("SEM_POST-R");
			return (NULL);
		}
	}

	return (NULL);
}

/* * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Стартовая функция потока, читающего и суммирующего числа */
/* * * * * * * * * * * * * * * * * * * * * * * * * * */
void *start_reader (void *dummy) {
	while (sem_wait (r_sem) == 0) {
		sum += my_buf;
		if (sem_post (w_sem) != 0) {
			perror ("SEM_POST-W");
			return (NULL);
		}
	}

	return (NULL);
}
/* * * * * * * * * * * * * * * * * * * * * * * */
/* Инициализация семафоров,                    */
/* создание и терминирование потоков управления*/
/* * * * * * * * * * * * * * * * * * * * * * * */
int main (void) {
	pthread_t w_ptid;/* Идентификатор потока-писателя */
	pthread_t r_ptid;/* Идентификатор потока-читателя */

	/* Инициализируем семафоры.      */
	/* Семафор записи будет свободен,*/
	/* семафор чтения - захвачен     */
	if (sem_init (w_sem, 0, 1) == -1) {
		perror ("SEM_INIT-W");
		return (-1);
	}
	if (sem_init (r_sem, 0, 0) == -1) {
		perror ("SEM_INIT-R");
		return (-1);
	}

	/* Создадим потоки управления - писателя и читателя */
	if ((errno = pthread_create (w_ptid, NULL, 
			start_writer, NULL)) != 0) {
		perror ("PTHREAD_CREATE-W");
		return (errno);
	}
	if ((errno = pthread_create (r_ptid, NULL, 
			start_reader, NULL)) != 0) {
		perror ("PTHREAD_CREATE-R");
		return (errno);
	}

	/* Дадим потокам повыполняться */
	sleep (10);

	/* Терминируем потоки */
	(void) pthread_cancel (w_ptid);
	(void) pthread_cancel (r_ptid);

	/* Дождемся завершения потоков */
	(void) pthread_join (w_ptid, NULL);
	(void) pthread_join (r_ptid, NULL);
	/* Ликвидируем семафоры */
	if (sem_destroy (w_sem) != 0) {
		perror ("SEM_DESTROY-W");
		return (-1);
	}
	if (sem_destroy (r_sem) != 0) {
		perror ("SEM_DESTROY-R");
		return (-1);
	}

	printf ("Сумма сгенерированных чисел: %d\n", sum);

	return 0;
}

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

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

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

Здесь ничего не говорится об уменьшении положительного значения без захвата семафора, зато намеки на это имеются в части, где приводятся детальное разъяснение положений стандарта и обоснование принятых решений. Смущает и то, что в описании функции sem_post() не запрещается освобождать свободный семафор, что, естественно, выливается в увеличение его значения. Далее, в разных местах семафоры называются то бинарными, то целочисленными. Наконец, автору не известно ни одной реализации, где семафоры реального времени не были бы целочисленными. В общем, сложное это дело - толкование внутренне противоречивых стандартов, вступающих в конфликт с реальной жизнью...

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

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

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

/* Буфер для хранения генерируемых данных */
static int my_buf [BUFSIZ];

/* Индекс, по которому можно записать очередной элемент */
static int w_ind = 0;

/* Индекс, по которому можно прочитать очередной элемент */
static int r_ind = 0;

/* Семафор, разрешающий записывать в буфер новые данные */
static sem_t w_sem;

/* Семафор, разрешающий читать данные из буфера */
static sem_t r_sem;

/* Общий результат суммирования */
static int sum = 0;


/* * * * * * * * * * * * * * * * * * * * * * * */
/* Стартовая функция потока, генерирующего числа*/
/* * * * * * * * * * * * * * * * * * * * * * * */
void *start_writer (void *dummy) {
	while (sem_wait (w_sem) == 0) {
		my_buf [w_ind] = rand ();
		w_ind = (w_ind + 1) % BUFSIZ;
		if (sem_post (r_sem) != 0) {
			perror ("SEM_POST-R");
			return (NULL);
		}
	}

	return (NULL);
}

/* * * * * * * * * * * * * * * */
/* Стартовая функция потока,   */
/* читающего и суммирующего числа*/
/* * * * * * * * * * * * * * * */
void *start_reader (void *dummy) {
	while (sem_wait (r_sem) == 0) {
		sum += my_buf [r_ind];
		r_ind = (r_ind + 1) % BUFSIZ;
		if (sem_post (w_sem) != 0) {
			perror ("SEM_POST-W");
			return (NULL);
		}
	}

	return (NULL);
}

/* * * * * * * * * * * * * * * * * * * * * * * */
/* Инициализация семафоров,                    */
/* создание и терминирование потоков управления*/
/* * * * * * * * * * * * * * * * * * * * * * * */
int main (void) {
	pthread_t w_ptid;/* Идентификатор потока-писателя */
	pthread_t r_ptid;/* Идентификатор потока-читателя */


	/* Инициализируем семафоры.      */
	/* Семафор записи будет свободен,*/
	/* разрешая заполнить весь буфер,*/
	/* семафор чтения - захвачен     */
	if (sem_init (w_sem, 0, BUFSIZ) == -1) {
		perror ("SEM_INIT-W");
		return (-1);
	}
	if (sem_init (r_sem, 0, 0) == -1) {
		perror ("SEM_INIT-R");
		return (-1);
	}

	/* Создадим потоки управления - писателя и читателя */
	if ((errno = pthread_create (w_ptid, NULL, 
			start_writer, NULL)) != 0) {
		perror ("PTHREAD_CREATE-W");
		return (errno);
	}
	if ((errno = pthread_create (r_ptid, NULL, 
			start_reader, NULL)) != 0) {
		perror ("PTHREAD_CREATE-R");
		return (errno);
	}

	/* Дадим потокам повыполняться */
	sleep (10);

	/* Терминируем потоки */
	(void) pthread_cancel (w_ptid);
	(void) pthread_cancel (r_ptid);

	/* Дождемся завершения потоков */
	(void) pthread_join (w_ptid, NULL);
	(void) pthread_join (r_ptid, NULL);

	/* Ликвидируем семафоры */
	if (sem_destroy (w_sem) != 0) {
		perror ("SEM_DESTROY-W");
		return (-1);
	}

	if (sem_destroy (r_sem) != 0) {
		perror ("SEM_DESTROY-R");
		return (-1);
	}

	printf ("Сумма сгенерированных чисел: %d\n", sum);

	return 0;
}

Значение семафора w_sem равно числу элементов буфера, доступных для записи, r_sem - для чтения. Многопотоковым инвариантом программы является сумма этих величин, равная размеру буфера. В начальный момент буфер целиком доступен для записи. После этого вызовы sem_wait() и sem_post() уменьшают значение "своего" и увеличивают значение "чужого" семафора. Поток управления приостанавливается в sem_wait(), когда "свое" значение уменьшается до нуля, у нужно ждать, пока другой поток вызовом sem_post() не увеличит его, сделав положительным.

Разумеется, обсуждение темы семафоров было бы неполным без обеда философов. Мы приведем программу, написанную С.В. Самборским (см. листинг 4.19). В ней семафоры используются как бинарные, поэтому ее стандартность и мобильность не вызывают сомнений.

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

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

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

Пример запуска:
	mudrecSem -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: mudrecSem.c,v 1.2 2004/03/18 10:28:38 sambor Exp $";

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <semaphore.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])

/* Массив семафоров для синхронизации доступа к вилкам */
sem_t *semFork;

/* Разные алгоритмы захвата вилок */
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);

			sprintf (buffer, "%s: хочет есть\n", this->name);
			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) {
	sem_post (semFork [this->left_fork - 1]);
	sem_post (semFork [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);

	sem_wait (semFork [first - 1]);
	sem_wait (semFork [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);
	}

	sem_wait (semFork [first - 1]);
	sem_wait (semFork [last - 1]);
}

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

	for (;;) {
		sem_wait (semFork [left - 1]);
		if (0 == sem_trywait (semFork [right - 1])) return;
		sem_post (semFork [left - 1]);
		sem_wait (semFork [right - 1]);
		if (0 == sem_trywait (semFork [left - 1])) return;
		sem_post (semFork [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);

	/* Создадим семафоры для охраны вилок */
	semFork = calloc (sizeof (sem_t), nMudr);
	for (i = 0; i < nMudr; i++) {
		sem_init (semFork [i], 0, 1 /* На каждое место */
		/* по одной вилке */);
	}

	/* Философы входят в столовую */
	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++) {
		sem_destroy (semFork [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 (semFork);
	free (kafedra);

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

	return 0;
}

Разделяемые сегменты памяти

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

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

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

Стандартный программный интерфейс состоит из двух функций: shm_open() и shm_unlink() (см. листинг 4.19).

#include <sys/mman.h>

int shm_open (const char *name, 
    int oflag, mode_t mode);

int shm_unlink (const char *name);

При открытии с помощью функции shm_open() возвращается файловый дескриптор. Имя (аргумент name ) трактуется стандартным для рассматриваемых средств межпроцессного взаимодействия образом. Посредством аргумента oflag могут указываться флаги O_RDONLY, O_RDWR, O_CREAT, O_EXCL и/или O_TRUNC. Если объект создается, то режим доступа к нему формируется в соответствии со значением аргумента mode и маской создания файлов процесса.

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

Представляется естественным, что способ доступа к объектам определяется типом дескриптора, возвращаемого при их открытии. Если это адрес, то доступ сводится к операциям чтения/записи из/в память. Если это файловый дескриптор, то для доступа должны использоваться функции файлового ввода/вывода - read(), write() и т.п. Подобное естественное применение объектов в разделяемой памяти иллюстрируется двухпроцессной программой, копирующей строки со стандартного ввода на стандартный вывод (см. листинги 4.21, 4.22, 4.23). Предполагается, что заголовочный файл называется "g_shm.h", а файл с образом процесса, запускаемого посредством execl(), - "g_r_shm".

#ifndef g_SHM
#define g_SHM

/* Имя объекта в разделяемой памяти */
#define O_SHM_NAME 	"/g_o.shm"

/* Используемый номер сигнала 
   реального времени */
#define SIG_SHM   SIGRTMIN

/* Используемые значения сигнала 
   реального времени */
#define SIGVAL_LINE	0
#define SIGVAL_EOF 	EOF

#endif

/* * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Программа, состоящая из двух процессов, копирует  */
/* строки со стандартного ввода на стандартный вывод,*/
/* "прокачивая" их через разделяемый сегмент памяти. */
/* Для синхронизации доступа к разделяемому сегменту */
/* используются сигналы реального времени            */
/* * * * * * * * * * * * * * * * * * * * * * * * * * */

#include <unistd.h>
#include <stdio.h>
#include <signal.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <limits.h>
#include <sys/wait.h>
#include <assert.h>

#include "g_shm.h"

/* * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Создание разделяемого сегмента памяти,            */
/* чтение со стандартного ввода и запись строк в сегмент*/
/* * * * * * * * * * * * * * * * * * * * * * * * * * */
int main (void) {
	int fd_shm; /* Дескриптор объекта в разделяемой памяти*/
	FILE *fp; /* Поток для записи в объект */
	char line [LINE_MAX]; /* Буфер для копируемых строк */
	struct sigaction sact; /* Структура для обработки сигналов */
	union sigval sg_val;/* Значение сигнала */
	int sg_no; /* Номер принятого сигнала */
	pid_t cpid;/* Идентификатор порожденного процесса */
	/* Создадим разделяемый сегмент памяти */
	if ((fd_shm = shm_open (O_SHM_NAME, O_RDWR | O_CREAT, 
			0777)) < 0) {
		perror ("SHM_CREAT");
		return (1);
	}

	/* Сформируем поток данных по файловому дескриптору */
	/* объекта в разделяемой памяти */
	assert ((fp = fdopen (fd_shm, "w")) != NULL);

	/* Отменим буферизацию вывода */
	setbuf (fp, NULL);

	/* Сформируем маску сигналов (блокируем SIG_SHM) */
	(void) sigemptyset (sact.sa_mask);
	(void) sigaddset (sact.sa_mask, SIG_SHM);
	(void) sigprocmask (SIG_BLOCK, sact.sa_mask, 
		(sigset_t *) NULL);

	/* Установим для сигнала SIG_SHM флаг SA_SIGINFO */
	sact.sa_flags = SA_SIGINFO;
	sact.sa_sigaction = (void (*) (int, siginfo_t *, 
		void *)) SIG_DFL;
	(void) sigaction (SIG_SHM, sact, 
		(struct sigaction *) NULL);


	/* Подготовительная работа закончена */

	switch (cpid = fork ()) {
		case -1:
			perror ("FORK");
			return (2);
		case 0:
			/* Чтение из объекта и выдачу на стандартный */
			/* вывод реализуем в порожденном процессе */
			if (execl ("./g_r_shm", "g_r_shm", 
				(char *) NULL) < 0) {
				perror ("EXECL");
				return (3);
			}
	}

	/* Чтение со стандартного ввода и запись в объект */
	/* возложим на родительский процесс.*/
	/* В начальный момент объект в разделяемой памяти*/
	/* доступен для записи */
	assert (fseek (fp, 0, SEEK_SET) == 0);
	fputs ("Вводите строки\n", fp);
	/* Сообщим порожденному процессу,*/
	/* что объект в разделяемой памяти заполнен*/
	sg_val.sival_int = SIGVAL_LINE;
	assert (sigqueue (cpid, SIG_SHM, sg_val) == 0);

	while (fgets (line, sizeof (line), stdin) != NULL) {
		assert (fseek (fp, 0, SEEK_SET) == 0);
		/* Дождемся, когда в объект можно будет писать */
		if ((sigwait (sact.sa_mask, sg_no) != 0) || 
				(sg_no != SIG_SHM)) {
			return (4);
		}
		assert (fputs ("Вы ввели: ", fp) != EOF);
		assert (fputs (line, fp) != EOF);
		assert (sigqueue (cpid, SIG_SHM, sg_val) == 0);
	}
	/* Сообщим о конце файла */
	sg_val.sival_int = SIGVAL_EOF;
	assert (sigqueue (cpid, SIG_SHM, sg_val) == 0);

	fclose (fp);

	(void) wait (NULL);

	if (shm_unlink (O_SHM_NAME) != 0) {
		perror ("SHM_UNLINK");
		return (7);
	}

	return (0);
}

/* * * * * * * * * * * * * * * * * * * * * * * * * * */
/* Процесс читает строки из объекта в разделяемой памяти */
/* и копирует их на стандартный вывод                */
/* * * * * * * * * * * * * * * * * * * * * * * * * * */

#include <unistd.h>
#include <stdio.h>
#include <signal.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <limits.h>
#include <assert.h>

#include "g_shm.h"

/* * * * * * * * * * * * * * * * * * * * */
/* Открытие разделяемого сегмента памяти,*/
/* чтение из сегмента и выдача строк     */
/* на стандартный вывод                  */
/* * * * * * * * * * * * * * * * * * * * */
int main (void) {
	int fd_shm; /* Дескриптор объекта */
		/* в разделяемой памяти */
	FILE *fp; /* Поток для чтения из объекта */
	char line [LINE_MAX];/* Буфер для копируемых строк */
	sigset_t smask; /* Маска ожидаемых сигналов */
	siginfo_t sinfo; /* Структура для получения */
		/* данных о сигнале */
	pid_t ppid; /* Идентификатор родительского */
		/* процесса */

	/* Откроем разделяемый сегмент памяти */
	if ((fd_shm = shm_open (O_SHM_NAME, O_RDONLY, 
			0777)) < 0) {
		perror ("SHM_OPEN");
		return (1);
	}

	/* Сформируем поток по файловому дескриптору объекта */
	/* в разделяемой памяти */
	assert ((fp = fdopen (fd_shm, "r")) != NULL);

	/* Отменим буферизацию ввода */
	setbuf (fp, NULL);

	/* Запомним идентификатор родительского процесса */
	ppid = getppid ();

	/* Сформируем маску ожидаемых сигналов (SIG_SHM) */
	(void) sigemptyset (smask);
	(void) sigaddset (smask, SIG_SHM);

	/* Подготовительная работа закончена */

	while ((fseek (fp, 0, SEEK_SET) == 0) 
			/* Дождемся, когда из объекта можно будет читать */
		(sigwaitinfo (smask, sinfo) == SIG_SHM) 
			/* И прочитаем строку, а не конец файла */
			(sinfo.si_value.sival_int == SIGVAL_LINE)) {
		(void) fgets (line, sizeof (line), fp);
		/* Сообщим родительскому процессу, */
		/* что данные из объекта извлечены */
		assert (kill (ppid, SIG_SHM) == 0);
		/* Выдадим, наконец, строку на стандартный вывод */
		assert (fputs (line, stdout) != EOF);
	}

	fclose (fp);

	return 0;
}

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

К сожалению, стандарт POSIX-2001 не следует приведенным выше естественным предположениям, касающимся связи между типом дескриптора и способом доступа к объекту, так что приведенная программа, строго говоря, не соответствует стандарту. Дело в том, что файловый дескриптор, возвращаемый функцией shm_open(), является дескриптором "второго сорта" (хотя и первой свежести): результат применения к нему функций fdopen(), read(), write() и т.п. не специфицирован.

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