DAY15 (Старые версии Машбука или нечто подобное)

4

Лекция №15

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

Итак, «очередь сообщений». Мы говорили, что «очередь сообщений» является одним из разделяемых ресурсов, предоставляемых IPC. Суть его заключается в следующем. Система создает некую структуру данных, представляющую собой очередь. Очередь содержит сообщения. Каждое сообщение обладает типом и набором данных (набор данных сообщения назовем телом сообщения). В системе может бать создано произвольное число очередей (разумеется, это число ограничивается объемом оперативной памяти). Все сообщения попадают/изымаются в/из очереди по стратегии FIFO. Обычная сквозная очередь может быть представлена в виде нескольких типизированных подочередей:

Для создания/подключения к очереди используется функция msgget:

int msgget(key_t key, int flags);

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

В случае успешного завершения функция возвращает идентификатор созданного/подключенного ресурса, иначе «-1».

Следующая функция - функция отправки сообщения:

int msgsnd( int id, struct msgbuf *buf, int size, int flags);

id - идентификатор очереди сообщения;

struct msgbuf {

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

char mtext[] // указатель на тело сообщения

}

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

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

Симметричная функция - чтение сообщения из очереди:

int msgrcv( int id, struct msgbuf *buf, int size, long type, int flags);

id - идентификатор очереди;

buf - указатель на буфер, куда будет принято сообщение;

size - размер буфера, в котором будет размещено тело сообщения (размер принимаемой информации);

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

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

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

int msgctl( int id, int cmd, struct msgid_dl *buf);

id - идентификатор очереди;

cmd - команда управления, для нас интерес представляет IPC_RMID, которая уничтожит ресурс.

buf - этот параметр будет оставлен без комментария.

С помощью msgctl можно уничтожить/создать разделяемый ресурс, сменить права доступа, посмотреть статус состояния этого разделяемого ресурса.

Мы описали два средства взаимодействия между процессами. Что же мы увидели? Понятно, что названия и описания интерфейсов мало понятны. Прежде всего следует заметить то, что как только мы переходим к вопросу взаимодействия процессов, у нас возникает проблема синхронизации. И здесь мы уже видим проблемы, связанные с тем, что после того, как мы поработали с разделяемой памятью или очередью сообщений, в системе может оставаться “хлам”, например, процессы, которые ожидают сообщений, которые в свою очередь не были посланы. Так, если мы обратились к функции получения сообщений с типом, которое вообще не пришло, и если не стоит ключ IPC_NOWAIT, то процесс будет ждать его появления, пока не исчезнет ресурс. Или мы можем забыть уничтожить ресурс (и система никого не поправит) - этот ресурс останется в виде загрязняющего элемента системы.

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

Сейчас мы напишем следующую программу: первый процесс будет читать некоторую текстовую строку из стандартного ввода и в случае, если строка начинается с буквы '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()

{

key_t key; int msgid; char str[256];

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

// получаем ключ, однозначно определяющий доступ к ресурсу данного типа

msgid=msgget(key, 0666 | IPC_CREAT);

// создаем очередь сообщений , 0666 определяет права доступа

for(;;)

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

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

strcpy(Message.Data, str); // и копируем ее в буфер сообщения

switch(str[0])

{

case 'a':

case 'A': Message.mtype=1;//устанавливаем тип, посылаем сообщение в очередь

msgsnd(msgid, (struct msgbuf*) (&Message), strlen(str)+1, 0);

break;

case 'b':

case 'B': 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); // ждем получения сообщений процессами А и В

msgctl(msgid, IPC_RMID, NULL); // уничтожаем очередь

exit(0);

default: exit(0);

}

}

}

Процесс-приемник А

// процесс В аналогичен с точностью до четвертого параметра в msgrcv

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/message.h>

#include <stdio.h>

struct {long mtype;

char Data[256];

} Message;

int main()

{

key_t key; int msgid;

key=ftok("/usr/mash",'s'); // получаем ключ по тем же параметрам

msgid=msgget(key, 0666 | IPC_CREAT); //создаем очередь сообщений

for(;;)

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

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

// читаем сообщение с типом

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

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

}

exit(0);

}

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