Linux 异步IO介绍

jopen 10年前

Linux 异步IO介绍

epoll

epoll是Linux对select功能的改进,其性能大大提升,而且和监控的IO个数无关。

  • API:

  • epoll_create:

    ***`int epoll_create(int size);`***    创建一个`epoll`实例,`size`参数是可监控IO的数量大小,但是Linux 2.6.8之后已不再使用。
  • epoll_ctl:

    ***`int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);`***    对`epoll`实例进行操作,由`op`指定操作类型:
  • EPOLL_CTL_ADD:添加一个文件描述符到epoll实例中。

  • EPOLL_CTL_MOD:更新epoll实例中文件描述符。

  • EPOLL_CTL_DEL:删除epoll实例中文件描述符。

  • epoll_wait

    ***```int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);```***    等待已经准备好的IO.`events`参数返回的数据就是在`epoll_ctl`函数第三个参数设置的值。
  • 限制:无

  • 使用范例:

    epoll最多的用途就是socket编程,可以大大提高服务器的性能,此处我们实现一个简单的http服务器。

    #define MAXFDS 128  #define EVENTS 100  #define PORT 8080  #define MAXEPOLLSIZE 1024*10    typedef enum  {      false,      true  }bool;    /***************定义处理socket的回调函数类型***********/  typedef int (*socket_pro)(int fd,void *data);    /***************定义回调函数的用户数据***********/  typedef struct userdata  {      int fd;      socket_pro cb;  }userdata_t;    static int epfd;//epoll句柄    /***************发送一个文件数据的函数***********/  static void cws_client_request (int connfd,void *data)  {      struct epoll_event ev = {0};      char buffer[1024*8] = {0};      int ret;      char *requestPath = NULL;      char tmpPath[512] = {"./www/"};      int pagesize = 0;      ret = recv (connfd, buffer, sizeof (buffer) -1, 0);      if (ret > 0)      {          if (strncmp (buffer, "GET ", 4) != 0)          {              printf("bad request.\n");          }            if (strncmp (buffer, "GET /", 5) == 0)          {                if (strncmp (buffer, "GET / ", 6) == 0)              {                  strcat(tmpPath, "/index.html");                  requestPath = tmpPath;              }              else              {                  requestPath = buffer+5;                  char * pos = strstr(buffer+5, " ");                  strncat(tmpPath, requestPath, pos - requestPath);                  requestPath = tmpPath;              }          }          char * badRequest = (char *)"<b>Error 404 Not Found.</b>";          char * httpStatus200 = (char *)"HTTP/1.0 200 OK\r\nContent-Type:text/html\r\n\r\n";            FILE * fp = fopen(requestPath, "r");          FILE * connfp = fdopen(connfd, "w");          if ( connfp == NULL )          {              perror("fdopen error");//cout <<"bad connfp"<<endl;           }              if (fp == NULL)          {              setlinebuf(connfp);              fwrite(badRequest, strlen(badRequest), 1, connfp);              fclose(connfp);          }          else          {              setlinebuf(connfp);              //fwrite(httpStatus200, strlen(httpStatus200), 1, connfp);              //fflush(connfp);              while ((ret = fread (buffer, 1, sizeof(buffer) -1, fp)) > 0)              {                  fwrite(buffer, 1, ret, connfp);                  pagesize += ret;                  fflush(connfp);              }              printf("pagesize:%d\n",pagesize);//cout <<"pagesize:" << pagesize << endl;              fclose(fp);          }      }        //1      close(connfd);      epoll_ctl (epfd, EPOLL_CTL_DEL, connfd, &ev);  }    /***************处理已经连接socket函数***********/  static int __process_data_fd(int fd,void *data)  {     struct epoll_event *ev = (struct epoll_event *)data;     cws_client_request(fd,ev);     free(data);       return;    }    /***************处理监听socket函数***********/  static int __process_listen_fd(int fd,void *data)  {       struct sockaddr_in caddr = {0};       struct epoll_event ev = {0};       int len = sizeof (caddr);       int cfd = accept (fd, (struct sockaddr *) &caddr, (socklen_t *) & len);       if (-1 == cfd)       {          perror("accpet error");//cout << "server has an error, now accept a socket fd" <<endl;          break;       }       setNonBlock (cfd);         userdata_t *cb_data =  malloc(sizeof(userdata_t));       cb_data->fd = cfd;       cb_data->cb = __process_data_fd;          ev.data.ptr = cfd;       ev.events = EPOLLIN | EPOLLET;       epoll_ctl (epfd, EPOLL_CTL_ADD, cfd, &ev);         return 0;  }    /***************设置描述符为非阻塞***********/  static bool setNonBlock (int fd)  {      int flags = fcntl (fd, F_GETFL, 0);      flags |= O_NONBLOCK;      if (-1 == fcntl (fd, F_SETFL, flags))      {          return false;      }          return true;  }    /***************主函数***********/  int main (int argc, char *argv[])  {      int listen_fd, nfds;      int on = 1;      char buffer[512];      struct sockaddr_in saddr, caddr;      struct epoll_event ev, events[EVENTS];        signal(SIGPIPE, SIG_IGN);        if (fork())      {          exit(0);      }          if (-1 == (listen_fd = socket(AF_INET, SOCK_STREAM, 0)))      {          perror("socket error");//cout << "create socket error!" << endl;          return -1;      }        epfd = epoll_create (MAXFDS);      setsockopt (listen_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on));      bzero (&saddr, sizeof (saddr));      saddr.sin_family = AF_INET;      saddr.sin_port = htons ((short) (PORT));      saddr.sin_addr.s_addr = INADDR_ANY;      if (-1 == bind(listen_fd, (struct sockaddr *) &saddr, sizeof (saddr)))      {          perror("bind error");//cout << " cann't bind socket on server " << endl;          return -1;      }        if (-1 == listen (listen_fd, 32))      {          perror("listen error");//cout << "listen error" << endl;          return -1;      }      userdata_t *cb_data =  (userdata_t *)malloc(sizeof(userdata_t));      cb_data->fd = listen_fd;      cb_data->cb = __process_listen_fd;       ev.data.ptr = cb_data;      ev.events = EPOLLIN|EPOLLET;      epoll_ctl (epfd, EPOLL_CTL_ADD, listen_fd, &ev);      for (;;)      {             int i;          nfds = epoll_wait (epfd, events, MAXFDS, -1);          for (i = 0; i < nfds; ++i)          {              userdata_t *cb_data = (userdata_t *)events[i].data.ptr;              cb_data.cb(cb_data.fd,cb_data);          }      }      if (listen_fd > 0)      {          shutdown (listen_fd, SHUT_RDWR);          close (listen_fd);      }        return 0;  }

linux native aio

Linux native aio 有两种API,一种是libaio提供的API,一种是利用系统调用封装成的API,后者使用的较多,因为不需要额外的库且简单。

  • API

  • io_setup:

    是用来设置一个异步请求的上下文,第一个参数是请求事件的个数,第二个参数唯一标识一个异步请求。
  • io_commit:

    是用来提交一个异步io请求的,在提交之前,需要设置一下结构体`iocb`,该结构体有以下字段需要设置:
    • aio_data::用户设置的数据,到时通过io_getevents函数返回。
    • aio_lio_opcode:异步操作码,有以下几种操作:
      - `IOCB_CMD_PREAD`:读操作,相当于调用`pread`  - `IOCB_CMD_PWRITE`:写操作,相当于`pwrite`  - `IOCB_CMD_FSYNC`:同步操作,相当于调用`fsync`  - `IOCB_CMD_FDSYNC`:同步操作,相当于调用`fdatasync`
    • aio_buf:用户提供的存储数据的buffer
    • aio_offset:文件的中偏移量
    • aio_nbytes:IO操作的数据大小
    • aio_flags:该字段要么不设置,要么设置为IOCB_FLAG_RESFD,表示使用eventfd通知事件的完成。
    • aio_resfd:如果aio_flags字段设置为IOCB_FLAG_RESFD,该字段设置为eventfd的返回值。
    </li>
  • io_getevents:

    用来获取完成的io事件,参数`min_nr`是事件个数的的最小值,`nr`是事件个数的最大值,如果没有足够的事件发生,该函数会阻塞,    `timeout`参数是阻塞的超时时间。该函数会返回一个`io_events`结构体,该结构体有以下字段:    * `data`:这就是在`struct iocb`结构体`aio_data`设置的值。  * `obj`:返回的`struct iocb`结构体,是`io_commit`第三个参数设置数组中的值。  * `res`和`res2`:返回结果。
  • io_destroy:

    在所有时间处理完之后,调用此函数销毁异步io请求。
  • 限制:

    aio只能使用于常规的文件IO,不能使用于socket,管道等IO。

  • 使用范例

    上面已经介绍过了,io_getevents在调用之后会阻塞直到有足够的事件发生,因此要实现真正的异步IO,需要借助eventfdepoll

    达到目的。

  • 首先我们封装一下系统调用来作为我们使用的API:

    int io_setup(unsigned nr, aio_context_t *ctxp)      {          return syscall(__NR_io_setup, nr, ctxp);      }        int io_submit(aio_context_t ctx, long nr, struct iocb **iocbpp)      {          return syscall(__NR_io_submit, ctx, nr, iocbpp);      }        int io_getevents(aio_context_t ctx, long min_nr, long max_nr,                       struct io_event *events, struct timespec *timeout)      {          return syscall(__NR_io_getevents, ctx, min_nr, max_nr, events, timeout);      }        int io_destroy(aio_context_t ctx)      {          return syscall(__NR_io_destroy, ctx);      }        int eventfd2(unsigned int initval, int flags)      {          return syscall(__NR_eventfd2, initval, flags);      }
    1. 定义自己的异步用户数据和回调函数:

      typedef void io_callback_t(aio_context_t ctx, struct iocb *iocb, long res);//回调函数类型    struct userdata//用户数据  {      int64_t offset;      int64_t filesize;      int64_t block_size;  };    struct user_iocb//封装结构体,以便异步返回用户数据。  {      struct iocb iocb;      struct userdata user_cb;  };    void aio_callback(aio_context_t ctx, struct iocb *iocb, long res, long res2)  {      int64_t offset = iocb->aio_offset;      struct custom_iocb *iocbp = (struct custom_iocb *) iocb;      printf("data=%.*s\n",page_size, (char *) iocb->aio_buf);  }
    2. 创建异步请求:

      //异步读取整个文件数据  #define page_size 1024  char *path = "test.txt"    //计算异步请求的个数  static int get_event_num(uint64_t len)  {      return len / page_size + (len % page_size != 0);  }    static int get_filesize(char *path)  {      struct stat buf = {0};      int iret = stat(path, &buf);      return (iret >= 0) ? buf.st_size : iret;  }  int main(void)  {      int event_fd = 0;      int file_size = get_filesize(path);      int event_num = get_event_num(file_size);      struct iocb *iocbs = malloc(event_num * sizeof (struct iocb));      struct iocb * iocbps[event_num] = {0};      aio_context_t ctx = 0;        /****************创建使用的eventfd******************/      event_fd = eventfd2(0, O_NONBLOCK | O_CLOEXEC);        /****************创建异步请求上下文*****************/      if (io_setup(event_num, &ctx))      {          perror("io_setup");          return 4;      }        /*****************设置请求的数据********************/      int fd = open(path, O_RDWR, 0664);      for (int i = 0; i < event_num; i++)      {          struct iocb *_ = iocbs + i;          _->aio_buf = (__u64) ((char *) buf + (i * page_size));//设置存储请求数据的buf          _->aio_fildes = fd;//请求的文件描述符          _->aio_offset = i * page_size;//读文件的偏移量,异步请求文件指针不会移动的,自己设定          _->aio_nbytes = page_size;//每个异步请求请求数据的大小          _->aio_resfd = event_fd;//用来通知有事件完成的eventfd          _->aio_flags = IOCB_FLAG_RESFD;//使用eventfd的标志          _->aio_data = (__u64) aio_callback;//设定用户数据,这里是个回调函数          iocbps[i] = _;      }        /*****************提交异步请求********************/      if (io_submit(ctx, event_num, iocbps) != event_num)      {          perror("io_submit");          return 6;      }        /***************利用epoll等待异步请求事件完成*****/      int epfd = 0;      epfd = epoll_create(1);      if (epfd == -1)      {          perror("epoll_create");          return 7;      }      struct epoll_event epevent = {0};      epevent.events = EPOLLIN | EPOLLET;      epevent.data.ptr = NULL;      if (epoll_ctl(epfd, EPOLL_CTL_ADD, event_fd, &epevent))      {          perror("epoll_ctl");          return 8;      }        /*****************处理异步请求事件********/      int i = 0;      while (i < event_num)      {          int64_t finished_aio = 0;          if (epoll_wait(epfd, &epevent, 1, -1) != 1)//等待事件发生          {              perror("epoll_wait");              return 9;          }            //读取异步事件完成的个数,finished_aio接收          if (read(event_fd, &finished_aio, sizeof (finished_aio)) != sizeof (finished_aio))          {              perror("read");              return 10;          }          printf("finished io number: %"PRIu64"\n", finished_aio);            struct timespec tms;          struct io_event events[event_num];          while (finished_aio > 0)          {              tms.tv_sec = 0;              tms.tv_nsec = 0;              int r = io_getevents(ctx, 1, event_num, events, &tms);//获取发生的事件              if (r > 0)              {                  int j = 0;                  for (j = 0; j < r; ++j)                  {                      ((io_callback_t *) (events[j].data))(ctx, (struct iocb *) events[j].obj, events[j].res,                                                             events[j].res2);//调用用户callback                  }                  i += r;                  finished_aio -= r;              }          }      }        /*****************处理异步事件结束********/      close(epfd);      io_destroy(ctx);      close(fd);      close(event_fd);        return 0;    }
    3. </ol> </li> </ul>

      SIGIO

      在Linux下,每一个文件描述符都可以设置O_ASYNC标识,当文件可读或者可写时可以发送信号通知相关进程,默认信号是SIGIO,可以

      使用fcntl函数改变这个默认信号。

      • 限制:

        1. 在多个文件描述符的情况下,信号驱动模式不能确定哪个描述符是可以读或者写的,所以必须遍历每一个描述符来判断,一个

          好的方法是使用epoll获取哪个描述符可读或者写。

        2. 只能用于socket、管道和终端IO,不能用于普通文件的IO。

      • 使用范例

      • 设置描述符的O_ASYNC标识和描述符的归属进程。

      • 设置文件描述符为非阻塞。

      • 设置信号SIGIO的处理动作。

        void new_op(int signum, siginfo_t *info, void *myact)//信号处理函数的实现      {          int i;          for (i = 0; i < 1; i++)          {              printf("data:%u\n", info->si_int);          }          char buf[10] = {0};          while(read(STDIN_FILENO,buf,10) > 0)          {             printf("input:%s",buf);          }          printf("handle signal %d over;\n", signum);      }          int main(void)      {          int oflags;          struct sigaction act;            sigemptyset(&act.sa_mask);          act.sa_sigaction = new_op; //信号处理函数          if (sigaction(SIGPOLL, &act, NULL) < 0)          {              printf("install sigal error\n");          }            fcntl(STDIN_FILENO, F_SETOWN, getpid());          oflags = fcntl(STDIN_FILENO, F_GETFL);          fcntl(STDIN_FILENO, F_SETFL, oflags | FASYNC|O_NONBLOCK);            while (1){              sleep(1);          };      }