Linux I/O多路復(fù)用

Linux中一切皆文件，不論是我們存儲(chǔ)在磁盤(pán)上的字符文件，可執(zhí)行文件還是我們的接入電腦的I/O設(shè)備等都被VFS抽象成了文件，比如標(biāo)準(zhǔn)輸入設(shè)備默認(rèn)是鍵盤(pán)，我們?cè)诓僮鳂?biāo)準(zhǔn)輸入設(shè)備的時(shí)候，其實(shí)操作的是默認(rèn)打開(kāi)的一個(gè)文件描述符是0的文件，而一切軟件操作硬件都需要通過(guò)OS，而OS操作一切硬件都需要相應(yīng)的驅(qū)動(dòng)程序，這個(gè)驅(qū)動(dòng)程序里配置了這個(gè)硬件的相應(yīng)配置和使用方法。Linux的I/O分為阻塞I/O,非阻塞I/O,I/O多路復(fù)用,信號(hào)驅(qū)動(dòng)I/O四種。對(duì)于I/O設(shè)備的驅(qū)動(dòng)，一般都會(huì)提供關(guān)于阻塞和非阻塞兩種配置。我們最常見(jiàn)的I/O設(shè)備之一--鍵盤(pán)(標(biāo)準(zhǔn)輸入設(shè)備)的驅(qū)動(dòng)程序默認(rèn)是阻塞的。
多路復(fù)用就是為了使進(jìn)程能夠從多個(gè)阻塞I/O中獲得自己想要的數(shù)據(jù)并繼續(xù)執(zhí)行接下來(lái)的任務(wù)。其主要的思路就是同時(shí)監(jiān)視多個(gè)文件描述符，如果有文件描述符的設(shè)定狀態(tài)的被觸發(fā)，就繼續(xù)執(zhí)行進(jìn)程，如果沒(méi)有任何一個(gè)文件描述符的設(shè)定狀態(tài)被觸發(fā)，進(jìn)程進(jìn)入sleep
多路復(fù)用的一個(gè)主要用途就是實(shí)現(xiàn)"I/O多路復(fù)用并發(fā)服務(wù)器"，和多線程并發(fā)或者多進(jìn)程并發(fā)相比，這種服務(wù)器的系統(tǒng)開(kāi)銷(xiāo)更低，更適合做web服務(wù)器，但是由于其并沒(méi)有實(shí)現(xiàn)真正的多任務(wù)，所以當(dāng)壓力大的時(shí)候，部分用戶的請(qǐng)求響應(yīng)會(huì)較慢

阻塞I/O

阻塞I/O，就是當(dāng)進(jìn)程試圖訪問(wèn)這個(gè)I/O設(shè)備而這個(gè)設(shè)備并沒(méi)有準(zhǔn)備好的時(shí)候，設(shè)備的驅(qū)動(dòng)程序會(huì)通過(guò)內(nèi)核讓這個(gè)試圖訪問(wèn)的進(jìn)程進(jìn)入sleep狀態(tài)。阻塞I/O的一個(gè)好處就是可以大大的節(jié)約CPU時(shí)間，因?yàn)橐坏┮粋€(gè)進(jìn)程試圖訪問(wèn)一個(gè)沒(méi)有準(zhǔn)備好的阻塞I/O，就會(huì)進(jìn)入sleep狀態(tài)，而進(jìn)入sleep狀態(tài)的進(jìn)程是不在內(nèi)核的進(jìn)程調(diào)度鏈表中，直到目標(biāo)I/O準(zhǔn)備好了將其喚醒并加入調(diào)度鏈表，這樣就可以節(jié)約CPU時(shí)間。當(dāng)然阻塞I/O也有其固有的缺點(diǎn)，如果進(jìn)程試圖訪問(wèn)一個(gè)阻塞I/O，但是否訪問(wèn)成功并不對(duì)接下來(lái)的任務(wù)有決定性影響，那么直接使其進(jìn)入sleep狀態(tài)顯然會(huì)延誤其任務(wù)的完成。

• 典型的默認(rèn)阻塞IO有標(biāo)準(zhǔn)輸入設(shè)備，socket設(shè)備，管道設(shè)備等，當(dāng)我們使用gets(),scanf(),read()等操作請(qǐng)求這些IO時(shí)而IO并沒(méi)有數(shù)據(jù)流入，就會(huì)造成進(jìn)程的sleep。 進(jìn)程會(huì)一直阻塞下去直到接收緩沖區(qū)中有數(shù)據(jù)可讀，此時(shí)內(nèi)核再去喚醒該進(jìn)程，通過(guò)相應(yīng)的函數(shù)從中獲取數(shù)據(jù)。如果阻塞過(guò)程中對(duì)方發(fā)生故障，那么這個(gè)進(jìn)程將會(huì)永遠(yuǎn)阻塞下去。
• 寫(xiě)操作時(shí)發(fā)生阻塞的情況要比讀操作少，主要發(fā)生在要寫(xiě)入的緩沖區(qū)的大小小于要寫(xiě)入的數(shù)據(jù)量的情況下，這時(shí)寫(xiě)操作將不進(jìn)行任何任何拷貝工作，將發(fā)生阻塞。一旦發(fā)送緩沖區(qū)內(nèi)有足夠的空間，內(nèi)核將喚醒進(jìn)程，將數(shù)據(jù)從用戶緩沖區(qū)中拷貝到相應(yīng)的發(fā)送數(shù)據(jù)緩沖區(qū)。udp不用等待確認(rèn)，沒(méi)有實(shí)際的發(fā)送緩沖區(qū)，所以u(píng)dp協(xié)議中不存在發(fā)送緩沖區(qū)滿的情況，在udp套接字上執(zhí)行的寫(xiě)操作永遠(yuǎn)都不會(huì)阻塞

現(xiàn)假設(shè)一個(gè)進(jìn)程希望通過(guò)三個(gè)管道中任意一個(gè)中讀取數(shù)據(jù)并顯示，偽代碼如下

read(pipe_0,buf,sizeof(buf)); //sleepprint buf;read(pipe_1,buf,sizeof(buf));print buf;read(pipe_2,buf,sizeof(buf));print buf;

由于管道是阻塞I/O，所以如果pipe_0沒(méi)有數(shù)據(jù)流入，進(jìn)程就是在第一個(gè)read()處進(jìn)入sleep狀態(tài)而即使pipe_1和pipe_2有數(shù)據(jù)流入也不會(huì)被讀取。
如果我們使用下述代碼重新設(shè)置管道的阻塞屬性，顯然，如果三個(gè)管道都沒(méi)有數(shù)據(jù)流入，那么進(jìn)程就無(wú)法獲得請(qǐng)求的數(shù)據(jù)而繼續(xù)執(zhí)行，倘若這些數(shù)據(jù)很重要(所以我們才要用阻塞I/O)，那結(jié)果就會(huì)十分的糟糕，改為輪詢卻又大量的占據(jù)CPU時(shí)間。

int fl = fcntl(pipe_fd, F_GETFL);fcntl(pipe_fd, F_SETFL, fl | O_NONBLOCK);

如何讓進(jìn)程同時(shí)監(jiān)視三個(gè)管道，其中一個(gè)有數(shù)據(jù)就繼續(xù)執(zhí)行而不會(huì)sleep，如果全部沒(méi)有數(shù)據(jù)流入再sleep，就是多路復(fù)用技術(shù)需要解決的問(wèn)題。

非阻塞I/O

非阻塞I/O就是當(dāng)一個(gè)進(jìn)程試圖訪問(wèn)一個(gè)I/O設(shè)備的時(shí)候，無(wú)論是否從中獲取了請(qǐng)求的數(shù)據(jù)都會(huì)返回并繼續(xù)執(zhí)行接下來(lái)的任務(wù)。，但非常適合請(qǐng)求是否成功對(duì)接下來(lái)的任務(wù)影響不大的I/O請(qǐng)求。但如果訪問(wèn)一個(gè)非阻塞I/O，但這個(gè)請(qǐng)求如果失敗對(duì)進(jìn)程接下來(lái)的任務(wù)有致命影響，最粗暴的就是使用while(1)｛read()｝輪詢。顯然，這種方式會(huì)占用大量的CPU時(shí)間。

select機(jī)制

select是一種非常"古老"的同步I/O接口，但是提供了一種很好的I/O多路復(fù)用的思路

模型

fd_set //創(chuàng)建fd_set對(duì)象，將來(lái)從中增減需要監(jiān)視的fdFD_ZERO() //清空f(shuō)d_set對(duì)象FD_SET() //將一個(gè)fd加入fd_set對(duì)象中 select() //監(jiān)視fd_set對(duì)象中的文件描述符pselect() //先設(shè)定信號(hào)屏蔽，再監(jiān)視FD_ISSET() //測(cè)試fd是否屬于fd_set對(duì)象FD_CLR() //從fd_set對(duì)象中刪除fd

Note:

• select的第一個(gè)參數(shù)nfds是指集合中的最大的文件描述符+1，因?yàn)閟elect會(huì)無(wú)差別遍歷整個(gè)文件描述符表直到找到目標(biāo)，而文件描述符是從0開(kāi)始的，所以一共是集合中的最大的文件描述符+1次。
• 上一條導(dǎo)致了這種機(jī)制的低效，如果需要監(jiān)視的文件描述符是0和100那么每一次都會(huì)遍歷101次
• select()每次返回都會(huì)修改fd_set，如果要循環(huán)select()，需要先對(duì)初始的fd_set進(jìn)行備

例子_I/O多路復(fù)用并發(fā)服務(wù)器

關(guān)于server本身的編程模型，參見(jiàn)tcp/ip協(xié)議服務(wù)器模型和udp/ip協(xié)議服務(wù)器模型這里僅是使用select實(shí)現(xiàn)偽并行的部分模型

#define BUFSIZE 100#define MAXNFD 1024 int main(){ /***********服務(wù)器的listenfd已經(jīng)準(zhǔn)本好了**************/ fd_set readfds; fd_set writefds; FD_ZERO(&readfds); FD_ZERO(&writefds); FD_SET(listenfd, &readfds); fd_set temprfds = readfds; fd_set tempwfds = writefds; int maxfd = listenfd; int nready; char buf[MAXNFD][BUFSIZE] = {0}; while(1){ temprfds = readfds; tempwfds = writefds; nready = select(maxfd+1, &temprfds, &tempwfds, NULL, NULL) if(FD_ISSET(listenfd, &temprfds))｛ //如果監(jiān)聽(tīng)到的是listenfd就進(jìn)行accept int sockfd = accept(listenfd, (struct sockaddr*)&clientaddr, &len); //將新accept的scokfd加入監(jiān)聽(tīng)集合，并保持maxfd為最大fd FD_SET(sockfd, &readfds); maxfd = maxfd>sockfd?maxfd:sockfd; //如果意見(jiàn)檢查了nready個(gè)fd，就沒(méi)有必要再等了，直接下一個(gè)循環(huán) if(--nready==0) continue; } int fd = 0; //遍歷文件描述符表，處理接收到的消息 for(;fd<=maxfd; fd++)｛ if(fd == listenfd) continue; if(FD_ISSET(fd, &temprfds)){ int ret = read(fd, buf[fd], sizeof buf[0]); if(0 == ret)｛ //客戶端鏈接已經(jīng)斷開(kāi) close(fd); FD_CLR(fd, &readfds); if(maxfd==fd) --maxfd; continue; } //將fd加入監(jiān)聽(tīng)可寫(xiě)的集合 FD_SET(fd, &writefds); } //找到了接收消息的socket的fd，接下來(lái)將其加入到監(jiān)視寫(xiě)的fd_set中 //將在下一次while()循環(huán)開(kāi)始監(jiān)視 if(FD_ISSET(fd, &tempwfds)){ int ret = write(fd, buf[fd], sizeof buf[0]); printf("ret %d: %d\n", fd, ret); FD_CLR(fd, &writefds); } } } close(listenfd);}

poll機(jī)制

poll是一種基于select的改良機(jī)制，其針對(duì)select的一些缺陷進(jìn)行了重新設(shè)計(jì)，包括不需要備份fd_set等等，但是依然是遍歷整個(gè)文件描述符表，效率較低

模型

struct pollfd fds //創(chuàng)建一個(gè)pollfd類型的數(shù)組fds[0].fd //向fds[0]中放入需要監(jiān)視的fdfds[0].events //向fds[0]中放入需要監(jiān)視的fd的觸發(fā)事件 POLLIN //I/O有輸入 POLLPRI //有緊急數(shù)據(jù)需要讀取 POLLOUT //I/O可寫(xiě) POLLRDHUP //流式套接字連接斷開(kāi)或套接字處于半關(guān)閉狀態(tài) POLLERR //錯(cuò)誤條件(僅針對(duì)輸出) POLLHUP //掛起(僅針對(duì)輸出) POLLNVAL //無(wú)效的請(qǐng)求：fd沒(méi)有被打開(kāi)(僅針對(duì)輸出)

例子_I/O多路復(fù)用并發(fā)服務(wù)器

/* ... */int main(){ /* ... */ struct pollfd myfds[MAXNFD] = {0}; myfds[0].fd = listenfd; myfds[0].events = POLLIN; int maxnum = 1; int nready; //準(zhǔn)備二維數(shù)組buf，每個(gè)fd使用buf的一行，數(shù)據(jù)干擾 char buf[MAXNFD][BUFSIZE] = {0}; while(1){ //poll直接返回event被觸發(fā)的fd的個(gè)數(shù) nready = poll(myfds, maxnum, -1) int i = 0; for(;i//poll通過(guò)將相應(yīng)的二進(jìn)制位置一來(lái)表示已經(jīng)設(shè)置 //如果下面的條件成立，表示revent[i]里的POLLIN位已經(jīng)是1了 if(myfds[i].revents & POLLIN){ if(myfds[i].fd == listenfd){ int sockfd = accept(listenfd, (struct sockaddr*)&clientaddr, &len); //將新accept的scokfd加入監(jiān)聽(tīng)集合 myfds[maxnum].fd = sockfd; myfds[maxnum].events = POLLIN; maxnum++; //如果意見(jiàn)檢查了nready個(gè)fd，就直接下一個(gè)循環(huán) if(--nready==0) continue; } else{ int ret = read(myfds[i].fd, buf[myfds[i].fd], sizeof buf[0]); if(0 == ret)｛ //如果連接斷開(kāi)了 close(myfds[i].fd); //初始化將文件描述符表所有的文件描述符標(biāo)記為-1 //close的文件描述符也標(biāo)記為-1 //打開(kāi)新的描述符時(shí)從表中搜索第一個(gè)-1 //open()就是這樣實(shí)現(xiàn)始終使用最小的fd //這里為了演示并沒(méi)有使用這種機(jī)制 myfds[i].fd = -1; continue; } myfds[i].events = POLLOUT; } } else if(myfds[i].revents & POLLOUT){ int ret = write(myfds[i].fd, buf[myfds[i].fd], sizeof buf[0]); myfds[i].events = POLLIN; } } } close(listenfd);}

epoll

epoll在poll基礎(chǔ)上實(shí)現(xiàn)的更為健壯的接口，它每次只會(huì)遍歷我們關(guān)心的文件描述符，也是現(xiàn)在主流的web服務(wù)器使用的多路復(fù)用技術(shù)，epoll一大特色就是支持EPOLLET(邊沿觸發(fā))和EPOLLLT (水平觸發(fā))，前者表示如果讀取之后緩沖區(qū)還有數(shù)據(jù)，那么只要讀取結(jié)束，剩余的數(shù)據(jù)也會(huì)丟棄，而后者表示里面的數(shù)據(jù)不會(huì)丟棄，下次讀的時(shí)候還在，默認(rèn)是EPOLLLT

模型

epoll_create() //創(chuàng)建epoll對(duì)象struct epoll_event //準(zhǔn)備事件結(jié)構(gòu)體和事件結(jié)構(gòu)體數(shù)組 event.events event.data.fd ...epoll_ctl() //配置epoll對(duì)象epoll_wait() //監(jiān)控epoll對(duì)象中的fd及其相應(yīng)的event

例子_I/O多路復(fù)用并發(fā)服務(wù)器

/* ... */int main(){ /* ... */ /* 創(chuàng)建epoll對(duì)象 */ int epoll_fd = epoll_create(1024); //準(zhǔn)備一個(gè)事件結(jié)構(gòu)體 struct epoll_event event = {0}; event.events = EPOLLIN; event.data.fd = listenfd; //data是一個(gè)共用體，除了fd還可以返回其他數(shù)據(jù) //ctl是監(jiān)控listenfd是否有event被觸發(fā) //如果發(fā)生了就把event通過(guò)wait帶出。 //所以，如果event里不標(biāo)明fd，我們將來(lái)獲取就不知道哪個(gè)fd epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listenfd, &event); struct epoll_event revents[MAXNFD] = {0}; int nready; char buf[MAXNFD][BUFSIZE] = {0}; while(1){ //wait返回等待的event發(fā)生的數(shù)目 //并把相應(yīng)的event放到event類型的數(shù)組中 nready = epoll_wait(epoll_fd, revents, MAXNFD, -1) int i = 0; for(;i//wait通過(guò)在events中設(shè)置相應(yīng)的位來(lái)表示相應(yīng)事件的發(fā)生 //如果輸入可用，那么下面的這個(gè)結(jié)果應(yīng)該為真 if(revents[i].events & EPOLLIN){ //如果是listenfd有數(shù)據(jù)輸入 if(revents[i].data.fd == listenfd){ int sockfd = accept(listenfd, (struct sockaddr*)&clientaddr, &len); struct epoll_event event = {0}; event.events = EPOLLIN; event.data.fd = sockfd; epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sockfd, &event); } else{ int ret = read(revents[i].data.fd, buf[revents[i].data.fd], sizeof buf[0]); if(0 == ret){ close(revents[i].data.fd); epoll_ctl(epoll_fd, EPOLL_CTL_DEL, revents[i].data.fd, &revents[i]); } revents[i].events = EPOLLOUT; epoll_ctl(epoll_fd, EPOLL_CTL_MOD, revents[i].data.fd, &revents[i]); } } else if(revents[i].events & EPOLLOUT){ int ret = write(revents[i].data.fd, buf[revents[i].data.fd], sizeof buf[0]); revents[i].events = EPOLLIN; epoll_ctl(epoll_fd, EPOLL_CTL_MOD, revents[i].data.fd, &revents[i]); } } } close(listenfd);}