Linux 异步IO介绍
jopen
11年前
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:用户提供的存储数据的bufferaio_offset:文件的中偏移量aio_nbytes:IO操作的数据大小aio_flags:该字段要么不设置，要么设置为IOCB_FLAG_RESFD,表示使用eventfd通知事件的完成。aio_resfd：如果aio_flags字段设置为IOCB_FLAG_RESFD，该字段设置为eventfd的返回值。
-
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，需要借助
eventfd和epoll达到目的。
-
首先我们封装一下系统调用来作为我们使用的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); }-
定义自己的异步用户数据和回调函数：
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); } -
创建异步请求：
//异步读取整个文件数据 #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; } </ol> </li> </ul> -
限制：
-
在多个文件描述符的情况下，信号驱动模式不能确定哪个描述符是可以读或者写的，所以必须遍历每一个描述符来判断，一个
好的方法是使用
epoll获取哪个描述符可读或者写。 -
只能用于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); }; }
SIGIO
在Linux下，每一个文件描述符都可以设置
O_ASYNC标识，当文件可读或者可写时可以发送信号通知相关进程，默认信号是SIGIO,可以使用
fcntl函数改变这个默认信号。 -