Читать «UNIX: взаимодействие процессов» онлайн - страница 73

30   nfds = Select(pipefd[0] + 1, &rset, NULL, NULL, NULL);
31   if (FD_ISSET(pipefd[0], &rset)) {
32    Read(pipefd[0], &c, 1);
33    Mq_notify(mqd, &sigev); /* перерегистрируемся */
34    while ((n = mq_receive(mqd, buff, attr.mq_msgsize, NULL)) >= 0) {
35     printf("read %ld bytes\n", (long) n);
36    }
37    if (errno != EAGAIN)
38     err_sys("mq_receive error");
39   }
40  }
41  exit(0);
42 }
43 static void
44 sig_usr1(int signo)
45 {
46  Write(pipefd[1], "", 1); /* один байт – 0 */
47  return;
48 }

Создание канала

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

Вызов select

27-40 Мы инициализируем набор дескрипторов rset и при каждом проходе цикла включаем бит, соответствующий дескриптору pipefd[0] (открытый на считывание конец канала). Затем мы вызываем функцию select, ожидая получения единственного дескриптора, хотя в типичном приложении именно здесь осуществлялось бы размножение дескрипторов одного из концов канала. Когда появляется возможность читать из канала, мы перерегистрируемся на уведомление и считываем все доступные сообщения.

Обработчик сигнала

43-48 Единственное, что делает обработчик сигнала, — записывает в канал 1 байт. Как мы уже отмечали, эта операция относится к разрешенным для асинхронных обработчиков.

Пример: запуск нового потока

Альтернативой снятию блокировки сигналом является присваивание sigev_notify значения SIGEV_THREAD, что приводит к созданию нового потока. Функция, указанная в sigev_notify_function, вызывается с параметром sigev_value. Атрибуты нового канала указываются переменной sigev_notify_attributes, которая может быть и нулевым указателем, если нас устраивают устанавливаемые по умолчанию атрибуты. Текст программы приведен в листинге 5.13.

Листинг 5.13. Функция mq_notify, запускающая новый программный поток

//pxmsg/mqnotifythread1.с
1  #include "unpipc.h"
2  mqd_t mqd;
3  struct mq_attr attr;
4  struct sigevent sigev;
5  static void notify_thread(union sigval); /* наш поток */
6  int
7  main(int argc, char **argv)
8  {
9   if (argc != 2)
10   err_quit("usage: mqnotifythread1 <name>");
11  mqd = Mq_open(argv[1], O_RDONLY | O_NONBLOCK);
12  Mq_getattr(mqd, &attr);
13  sigev.sigev_notify = SIGEV_THREAD;
14  sigev.sigev_value.sival_ptr = NULL;
15  sigev.sigev_notify_function = notify_thread;
16  sigev.sigev_notify_attributes = NULL;
17  Mq_notify(mqd, &sigev);
18  for (;;)
19   pause(); /* новый поток делает все */
20  exit(0);
21 }
22 static void
23 notify_thread(union sigval arg)
24 {
25  ssize_t n;
26  void *buff;
27  printf("notify_thread started\n");
28  buff = Malloc(attr.mq_msgsize);
29  Mq_notify(mqd, &sigev); /* перерегистрируемся */
30  while ((n = mq_receive(mqd, buff, attr.mq_msgsize, NULL)) >= 0) {
31   printf("read %ld bytes\n", (long) n);
32  }
33  if (errno != EAGAIN)
34   err_sys("mq_receive error");
35  free(buff);
36  pthread_exit(NULL);