Очередь сообщений IPC

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

Рис. 90. Очередь сообщений IPC.

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

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/message.h>

int msgget(key_t key, int msgflag);

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

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

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

int msgsnd(int msqid, const void *msgp, size_t msgsz,

int msgflg);

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

- long msgtype — тип сообщения (только положительное длинное целое);

- char msgtext[] — данные (тело сообщения).

Последний аргумент функции — это флаги. Среди разнообразных флагов можно выделить те, которые определяют режим блокировки при отправке сообщения. Если флаг равен 0, то вызов будет блокироваться, если для отправки недостаточно системных ресурсов. Можно установить флаг IPC_NOWAIT, который позволяет работать без блокировки: тогда в случае возникновения ошибки при отправке сообщения, вызов вернет -1, а переменная errno будет содержать соответствующий код ошибки.

Для получения сообщений используется функция msgrcv().

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

int msgrcv(int msqid, void *msgp, size_t msgsz,

long msgtyp, int msgflg);

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

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

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

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

В случае успешного завершения функция возвращает количество успешно прочитанных байтов в теле сообщения.

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

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

int msgctl(int msqid, int cmd, struct msgid_ds *buf);

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

Возможные значения аргумента cmd:

- IPC_STAT — скопировать структуру, описывающую управляющие параметры очереди по адресу, указанному в параметре buf;

- IPC_SET — заменить структуру, описывающую управляющие параметры очереди, на структуру, находящуюся по адресу, указанному в параметре buf;

- IPC_RMID — удалить очередь.

Пример. Использование очереди сообщений. В приведенном ниже примере участвуют три процесса: основной процесс и процессы A и B. Основной процесс считывает из стандартного ввода текстовую строку. Если она начинается на букву A, то эта строка посылается процессу A, если на B — то процессу B, если на Q — то обоим процессам (в этом случае основной процесс ждет некоторое время, затем удаляет очередь сообщений и завершается). Процессы A и B считывают из очереди адресуемые им сообщения и распечатывают их на экране. Если пришедшее сообщение начинается с буквы Q, то процесс завершается.

/* ОСНОВНОЙ ПРОЦЕСС */

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/message.h>

#include <stdio.h>

/* декларация структуры сообщения */

struct

{

long mtype; /* тип сообщения */

char Data[256]; /* сообщение */

} Message;

int main(int argc, char **argv)

{

key_t key;

int msgid;

char str[256];

/*получаем уникальный ключ, однозначно определяющий

доступ к ресурсу данного типа */

key = ftok("/usr/mash",'s');

/* создаем новую очередь сообщений,

0666 определяет права доступа */

msgid = msgget(key, 0666 | IPC_CREAT | IPC_EXCL);

/* запускаем вечный цикл */

for(;;)

{

gets(str); /* читаем из стандартного ввода строку */

/* и копируем ее в буфер сообщения */

strcpy(Message.Data, str);

/* анализируем первый символ прочитанной строки */

switch(str[0])

{

case 'a':

case 'A':

/* устанавливаем тип 1 для ПРОЦЕССА A*/

Message.mtype = 1;

/* посылаем сообщение в очередь */

msgsnd(msgid, (struct msgbuf*) (&Message),

strlen(str)+1, 0);

break;

case 'b':

case 'B':

/* устанавливаем тип 2 для ПРОЦЕССА A*/

Message.mtype = 2;

msgsnd(msgid, (struct msgbuf*) (&Message),

strlen(str)+1, 0);

break;

case 'q':

case 'Q':

Message.mtype = 1;

msgsnd(msgid, (struct msgbuf*) (&Message),

strlen(str)+1, 0);

Message.mtype = 2;

msgsnd(msgid, (struct msgbuf*) (&Message),

strlen(str)+1, 0);

/* ждем получения сообщений

процессами А и В */

sleep(10); [2]

/* уничтожаем очередь */

msgctl(msgid, IPC_RMID, NULL);

exit(0);

default:

/* игнорируем остальные случаи */

break;

}

}

}

/* ПРИНИМАЮЩИЙ ПРОЦЕСС A (процесс B будет аналогичным) */

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/message.h>

#include <stdio.h>

struct

{

long mtype; /* тип сообщения */

char Data[256]; /* сообщение */

} Message;

int main(int argc, char **argv)

{

key_t key;

int msgid;

/* получаем ключ по тем же параметрам */

key = ftok("/usr/mash",'s');

/*подключаемся к очереди сообщений */

msgid = msgget(key, 0666);

/* запускаем вечный цикл */

for(;;)

{

/* читаем сообщение с типом 1 для ПРОЦЕССА A */

msgrcv(msgid, (struct msgbuf*) (&Message), 256, 1, 0); [3]

printf("%s", Message.Data);

if(Message.Data[0] == 'q' || Message.Data[0] == 'Q')

break;

}

return 0;

}

Пример. Очередь сообщений. Модель «клиент-сервер». В приведенном ниже примере имеется совокупность взаимодействующих процессов. Эта модель несимметричная: один из процессов назначается сервером, и его задачей становится обслуживание запросов остальных процессов-клиентов. В данном примере сервер принимает запросы от клиентов в виде сообщений (из очереди сообщений) с типом 1. Тело сообщения-запроса содержит идентификатор клиентского процесса, который выслал данный запрос. Для каждого запроса сервер генерирует ответ, которое также посылает через очередь сообщений, но посылаемое сообщение будет иметь тип, равный идентификатору процесса-адресата. В свою очередь, клиентский процесс будет брать из очереди сообщений сообщения с типом, равным его идентификатору.

/* СЕРВЕР */

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

#include <stdlib.h>

#include <string.h>

int main(int argc, char **argv)

{

struct

{

long mestype;

char mes[100];

} messageto;

struct

{

long mestype;

long mes;

} messagefrom;

key_t key;

int mesid;

key = ftok("example", 'r');

mesid = msgget (key, 0666 | IPC_CREAT | IPC_EXCL);

while(1)

{

msgrcv(mesid, &messagefrom, sizeof(messagefrom) –

sizeof(long), 1, 0);

messageto.mestype = messagefrom.mes;

strcpy(messageto.mes, "Message for client");

msgsnd (mesid, &messageto, sizeof(messageto) –

sizeof(long),0);

}

}

/* КЛИЕНТ */

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

int main(int argc, char **argv)

{

struct

{

long mestype;

long mes;

} messageto;

struct

{

long mestype;

char mes[100];

} messagefrom;

key_t key;

int mesid;

long pid = getpid();

key = ftok("example", 'r');

mesid = msgget (key, 0666);

messageto.mestype = 1;

messageto.mes = pid;

msgsnd(mesid, &messageto, sizeof(messageto) –

sizeof(long),0);

msgrcv(mesid,&messagefrom, sizeof(messagefrom) –

sizeof(long),pid,0);

printf("%s", messagefrom.mes);

return 0;

}

В серверном процессе декларируются две структуры для принимаемого (meassagefrom) и посылаемого (messageto) сообщений, а также ключ key и дескриптор очереди сообщений mesid. Затем сервер предпринимает традиционные действия: получает ключ, а по нему — дескриптор очереди сообщений. Затем он входит в бесконечный цикл, в котором и обрабатывает клиентские запросы. Каждая итерация цикла выглядит следующим образом. Из очереди выбирается сообщение с типом 1 (это сообщения с запросами от клиентов). Из тела этого сообщения считывается информация об идентификаторе клиента, и этот идентификатор сразу заносится в поле типа посылаемого сообщения. Затем сервер генерирует тело посылаемого сообщения, после чего отправляет созданное сообщение в очередь. На этом итерация цикла завершается.

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