Использование epoll

Мультиплексирование ввода-вывода

Существует множество способов написать сервер, обрабатывающий клиентский запросы. Вот некоторые из них:

• Последовательная обработка каждого запроса;
• Один процесс/поток на клиента;
• Мультиплексирование ввода-вывода.

Каждый подход обладает своими достоинствами и недостатками.

Я опишу мультиплексирование ввода вывода.
Принцип его работы в следующем: дескрипторы ввода/вывода помещается в список ожидания, затем процесс блокируется до наступления одного из событий:

Для читающего сокета:

• Можно выполнить чтение данных. Т.е. количество байт в буфере приёмнике больше или равно значению минимального количества данных (по умолчанию это значение равно 1, но его можно поменять с помощью параметра сокета SO_RCVLOWAT);
• На противоположном конце соединение закрывается;
• Сокет является прослушиваемым и число установленных соединений больше нуля.
• Ошибка сокета;

Для пишущего сокета:

• Можно выполнить запись данных. Т.е. количество байт в буфере передатчике больше или равно значению минимального количества данных (по умолчанию это значение равно 2048, но его можно поменять с помощью параметра сокета SO_SNDLOWAT);
• На противоположном конце соединение закрывается. При этом если записать данные в сокет, сгенерируется сигнал SIGPIPE;
• Ошибка сокета;

Также процесс выходит из состояния блокировки при получении внеполосных данных.

epoll
Системные вызовы группы epoll присущи только Linux. Они появились в ядре версии 2.6 и призваны исправить проблемы системных вызовов poll и select.
Работа с epoll осуществляется в три этапа:
1. Создаём дескриптор epoll, вызвав функцию

int epoll_create(int size);

2. Добавляем дескриптор ввода/вывода в массив ожидания, вызвав функцию

int epoll_ctl(int efd, int op, int fd, struct epool_event* event);

3. Запускаем ожидание готовности, вызвав функцию

int epoll_wait(int efd, struct epoll_event *events, int maxevents, int timeout);

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

Вот небольшой пример кода, использующий epoll:

#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h>
#include <strings.h>
#include <string.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <signal.h>
#include <iostream>

namespace
{

 int setnonblocking(int sock);
 void do_read(int fd);
 void do_write(int fd);
 void process_error(int fd);

}

int main(int argc, char* argv[])
{
 const int MAX_EPOLL_EVENTS = 100;
 const int BACK_LOG = 100;

 if (argc < 2)
 {
  std::cout <<  "Usage: server [port]" << std::endl;
  return 0;
 }

 char* p;
 int serv_port = strtol(argv[1], &p, 10);
 if (*p)
 {
  std::cout << "Invalid port number" << std::endl;
  return -1;
 }

 // Игнорируем сигнал ошибки работы с сокетом
 signal(SIGPIPE, SIG_IGN);

 // Создание дескриптора epoll
 int efd = epoll_create(MAX_EPOLL_EVENTS);

 int listenfd;
 listenfd = socket(AF_INET, SOCK_STREAM, 0);
 if (listenfd < 0)
 {
  perror("Socket creation");
  return -1;
 }

 // Неблокирующий режим
 setnonblocking(listenfd);

 struct sockaddr_in servaddr;
 bzero(&servaddr, sizeof(servaddr));
 servaddr.sin_family = AF_INET;
 servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
 servaddr.sin_port = htons(serv_port);

 if (bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0)
 {
  perror("Socket bind");
  return -1;
 }

 if (listen(listenfd, BACK_LOG) < 0)
 {
  perror("Socket listen");
  return -1;
 }

 // Добавляем дескриптор в массив ожидания
 struct epoll_event listenev;
 listenev.events = EPOLLIN | EPOLLPRI | EPOLLET;
 listenev.data.fd = listenfd;
 if (epoll_ctl(efd, EPOLL_CTL_ADD, listenfd, &listenev) < 0)
 {
  perror("Epoll fd add");
  return -1;
 }

 socklen_t client;
 // Массив готовых дескрипторов
 struct epoll_event events[MAX_EPOLL_EVENTS];
 struct epoll_event connev;
 struct sockaddr_in cliaddr;

 int events_cout = 1;

 for (;;)
 {
  // Блокирование до готовности одно или нескольких дескрипторов
  int nfds = epoll_wait(efd, events, MAX_EPOLL_EVENTS, -1);

  for (int n = 0; n < nfds; ++n)
  {
   // Готов слушающий дескриптор
   if (events[n].data.fd == listenfd)
   {
    client = sizeof(cliaddr);
    int connfd = accept(listenfd, (struct sockaddr*) &cliaddr, &client);
    if (connfd < 0)
    {
     perror("accept");
     continue;
    }

    // Недостаточно места в массиве ожидания
    if (events_cout == MAX_EPOLL_EVENTS-1)
    {
     std::cout << "Event array is full" << std::endl;
     close(connfd);
     continue;
    }

    // Добавление клиентского дескриптора в массив ожидания
    setnonblocking(connfd);
    connev.data.fd = connfd;
    connev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;
    if (!epoll_ctl(efd, EPOLL_CTL_ADD, connfd, &connev) < 0)
    {
     perror("Epoll fd add");
     close(connfd);
     continue;
    }

    ++events_cout;
   }
   // Готов клиентский дескриптор
   else
   {
    // Выполням работу с дескриптором
    int fd = events[n].data.fd;

    if (events[n].events & EPOLLIN)
     do_read(fd);

    if (events[n].events & EPOLLOUT)
     do_write(fd);

    if (events[n].events & EPOLLRDHUP)
     process_error(fd);

    // В даннoм примере дескриптор просто закрывается и удаляется из массива ожидания.
    // В зависимости от логики работы можно не удалять дескриптор и подождать следующую порцию данных
    epoll_ctl(efd, EPOLL_CTL_DEL, fd, &connev);
    --events_cout;
    close(fd);
   }
  }
 }

 return 0;
}

namespace
{

 int setnonblocking(int sock)
 {
  int opts;

  opts = fcntl(sock,F_GETFL);
  if (opts < 0)
  {
   perror("fcntl(F_GETFL)");
   return -1;
  }
  opts = (opts | O_NONBLOCK);
  if (fcntl(sock,F_SETFL,opts) < 0)
  {
   perror("fcntl(F_SETFL)");
   return -1;
  }

  return 0;
 }

 void do_read(int fd)
 {
  std::cout << "do_read" << std::endl;
 }

 void do_write(int fd)
 {
  std::cout << "do_write" << std::endl;
 }

 void process_error(int fd)
 {
  std::cout << "process_error" << std::endl;
 }
}