В листинге 5.19 приведен конец функции mq_open, в котором осуществляется открытие существующей очереди сообщений.
Листинг 5.19. Третья часть функции mq_open: открытие существующей очереди сообщений
//my_pxmsg_mmap/mq_open.с
109 exists:
110 /* открытие файла и отображение его в память */
111 if ((fd = open(pathname, O_RDWR)) < 0) {
112 if (errno == ENOENT && (oflag & O_CREAT))
113 goto again;
114 goto err;
115 }
116 /* проверяем, что инициализация завершена */
117 for (i = 0; i < MAX TRIES; i++) {
118 if (stat(pathname, &statbuff) == –1) {
119 if (errno == ENOENT && (oflag & O_CREAT)) {
120 close(fd);
121 goto again;
122 }
123 goto err;
124 }
125 if ((statbuff.st_mode & S_IXUSR) == 0)
126 break;
127 sleep(1);
128 }
129 if (i == MAX_TRIES) {
130 errno = ETIMEDOUT;
131 goto err;
132 }
133 filesize = statbuff.st_size;
134 mptr = mmap(NULL, filesize, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
135 if (mptr == MAP_FAILED)
136 goto err;
137 close(fd);
138 /* выделяем одну mymq_info{} для каждого вызова open */
139 if ((mqinfo = malloc(sizeof(struct mymq_info))) == NULL)
140 goto err;
141 rnqinfo->mqi_hdr = (struct mymq_hdr *) mptr;
142 mqinfo->mqi_magic = MQI_MAGIC;
143 mqinfo->mqi_flags = nonblock;
144 return((mymqd_t) mqinfo);
145 pthreaderr:
146 errno = i;
147 err:
148 /* не даем следующим вызовам изменить errno */
149 save_errno = errno;
150 if (created)
151 unlink(pathname);
152 if (mptr != MAP_FAILED)
153 munmap(mptr, filesize);
154 if (mqinfo != NULL)
155 free(mqinfo);
156 close(fd);
157 errno = save_errno;
158 return((mymqd_t) –1);
159 }
Открытие существующей очереди сообщений
109-115 Здесь мы завершаем работу, если флаг O_CREAT не был указан или если он был указан, но очередь уже существовала. В любом случае, мы открываем существующую очередь сообщений. Для этого мы открываем для чтения и записи файл, в котором она содержится, функцией open и отображаем его содержимое в адресное пpocтрaнcтвo процесса (mmap).
ПРИМЕЧАНИЕ
Наша реализация сильно упрощена в том, что касается режима открытия файла. Даже если вызвавший процесс указывает флаг O_RDONLY, мы должны дать возможность доступа для чтения и записи при открытии файла командой open и при отображении его в память командой mmap, поскольку невозможно считать сообщение из очереди, не изменив содержимое файла. Аналогично невозможно записать сообщение в очередь, не имея доступа на чтение. Обойти эту проблему можно, сохранив режим открытия (O_RDONLY, O_WRONLY, O_RDWR) в структуре mq_info и проверяя этот режим в каждой из функций. Например, mq_receive должна возвращать ошибку, если в mq_info хранится значение O_WRONLY.
Проверка готовности очереди
116-132 Нам необходимо дождаться, когда очередь будет проинициализирована (в случае, если несколько потоков сделают попытку открыть ее приблизительно одновременно). Для этого мы вызываем stat и проверяем разрешения доступа к файлу (поле st_mode структуры stat). Если бит user-execute сброшен, очередь уже проинициализирована.
Этот участок кода обрабатывает другую возможную ситуацию гонок. Предположим, что два потока разных процессов попытаются открыть очередь приблизительно одновременно. Первый поток создает файл и блокируется при вызове lseek в листинге 5.18. Второй поток обнаруживает, что файл уже существует, и переходит к метке exists, где он вновь открывает файл функцией open и при этом блокируется. Затем продолжается выполнение первого потока, но его вызов mmap в листинге 5.18 не срабатывает (возможно, он превысил лимит использования памяти), поэтому он переходит на метку err и удаляет созданный файл вызовом unlink. Продолжается выполнение второго потока, но если бы мы вызывали fstat вместо stat, он бы вышел по тайм-ауту в цикле for, ожидая инициализации файла. Вместо этого мы вызываем stat, которая возвращает ошибку, если файл не существует, и, если флаг O_CREAT был указан при вызове mq_open, мы переходим на метку again (листинг 5.17) для повторного создания файла. Эта ситуация гонок заставляет нас также проверять, не возвращается ли при вызове open ошибка ENOENT.
Отображение файла в память; создание и инициализация структуры mq_info
133-144 Файл отображается в память, после чего его дескриптор может быть закрыт. Затем мы выделяем место под структуру mq_infо и инициализируем ее. Возвращаемое значение представляет собой указатель на эту структуру.
Обработка ошибок
145-148 При возникновении ошибок происходит переход к метке err, а переменной errno присваивается значение, которое должно быть возвращено функцией mq_open. Мы аккуратно вызываем функции для очистки памяти от выделенных объектов, чтобы переменная errno не изменила свое значение в случае возникновения ошибки в этих функциях.
Функция mq_close
В листинге 5.20 приведен текст нашей функции mq_close.
Листинг 5.20. Функция mq_close
//my_pxmsg_mmap/mq_close.с
1 #include "unpipc.h"
2 #include "mqueue.h"
3 int
4 mymq_close(mymqd_t mqd)
5 {
6 long msgsize, filesize:
7 struct mymq_hdr *mqhdr;
8 struct mymq_attr *attr;
9 struct mymq_info *mqinfo;
10 mqinfo = mqd;
11 if (mqinfo->mqi_magic != MQI_MAGIC) {
12 errno = EBADF;
13 return(-1);
14 }
15 mqhdr = mqinfo->mqi_hdr;
16 attr = &mqhdr->mqh_attr;
17 if (mymq_notify(mqd, NULL) != 0) /* снятие вызвавшего процесса с регистрации */
18 return(-1);
19 msgsize = MSGSIZE(attr->mq_msgsize);
20 filesize = sizeof(struct mymq_hdr) + (attr->mq_maxmsg *
21 (sizeof(struct mymsg_hdr) + msgsize));
22 if (munmap(mqinfo->mqi_hdr, filesize) == –1)
23 return(-1);
24 mqinfo->mqi_magic = 0; /* на всякий случай */
25 free(mqinfo);
26 return(0);
27 }
Получение указателей на структуры
10-16 Проверяется правильность переданных аргументов, после чего получаются указатели на область, занятую отображенным в память файлом (mqhdr), и атрибуты (в структуре mq_hdr).
Сброс регистрации вызвавшего процесса
17-18 Для сброса регистрации на уведомление вызвавшего процесса мы вызываем mq_notify. Если процесс был зарегистрирован, он будет снят с уведомления, но если нет — ошибка не возвращается.
Отключение отображения файла и освобождение памяти
19-25 Мы вычисляем размер файла для вызова munmap и освобождаем память, используемую структурой mqinfo. На случай, если вызвавший процесс будет продолжать использовать дескриптор очереди сообщений, до того как область памяти будет вновь задействована вызовом malloc, мы устанавливаем значение mq_magiс в ноль, чтобы наши функции для работы с очередью сообщений обнаруживали ошибку.
Обратите внимание, что если процесс завершает работу без вызова mq_close, эти же операции выполняются автоматически: отключается отображение в память, а память освобождается.
Функция mq_unlink
Текст функции mqunlink приведен в листинге 5.21. Она удаляет файл, связанный с очередью сообщений, вызывая функцию unlink.
Листинг 5.21. Функция mq_unlink
//my_pxmsg_mmap/mq_unlink.с
1 #include "unpipc.h"
2 #include "mqueue.h"
3 int
4 mymq_unlink(const char *pathname)
5 {
6 if (unlink(pathname) == –1)
7 return(-1);
8 return(0);
9 }
Функция mq_getattr
В листинге 5.22 приведен текст функции mq_getattr, которая возвращает текущее значение атрибутов очереди.
Листинг 5.22. Функция mq_getattr
//my_pxmsg_mmap/mq_getattr.с
1 #include "unpipc.h"
2 #include "mqueue.h"
3 int
4 mymq_getattr(mymqd_t mqd, struct mymq_attr *mqstat)
5 {
6 int n;
7 struct mymq_hdr *mqhdr;
8 struct mymq_attr *attr;
9 struct mymq_info *mqinfo;
10 mqinfo = mqd;
11 if (mqinfo->mqi_magic != MQI_MAGIC) {
12 errno = EBADF;
13 return(-1);
14 }
15 mqhdr = mqinfo->mqi_hdr;
16 attr = &mqhdr->mqh_attr;
17 if ((n = pthread_mutex_lock(&mqhdr->mqh_lock)) != 0) {
18 errno = n;
19 return (-1);
20 }
21 mqstat->mq_flags = mqinfo->mqi_flags; /* для каждого open */
22 mqstat->mq_maxmsg = attr->mq_maxmsg; /* оставшиеся три – для очереди */
23 mqstat->mq_msgsize = attr->mq_msgsize;
24 mqstat->mq_curmsgs = attr->mq_curmsgs;
25 pthread_mutex_unlock(&mqhdr->mqh_lock);