網絡IO-阻塞、非阻塞、IO複用、異步

時間 2019-11-15

原文原文鏈接

　　網絡socket輸入操做分爲兩個階段:等待網絡數據到達和將到達內核的數據複製到應用進程緩衝區。對這兩個階段不一樣的處理方式將網絡IO分爲不一樣的模型:IO阻塞模型、非阻塞模型、多路複用和異步IO。本文可運行代碼連接:https://github.com/killianxu/network_examplehtml

一阻塞模型

　　阻塞模型原理以下圖1.1,當進行系統調用recvfrom時,應用進程進入內核態,內核判斷是否已收到數據報，若沒有則阻塞直到數據報準備好,接着複製數據到應用進程緩衝區,而後函數返回。python

圖1.1 阻塞IO模型linux

　　阻塞模型缺點:若數據報未準備好,則線程阻塞,不能進行其它操做和網絡鏈接請求。
　　利用多進程多線程方案,爲每一個鏈接建立一個進程或線程,這樣一個線程的阻塞不會影響到其它鏈接,但當遇到鏈接請求比較多時,會建立較多的進程或線程,嚴重浪費系統資源,影響進程的響應效率,進程和線程也更容易進入假死狀態。
　　利用線程池或鏈接池,能夠減小資源消耗。線程池利用已有線程,減小線程頻繁建立和銷燬,線程維持在必定數量,當有新的鏈接請求時，重用已有線程。鏈接池儘可能重用已有鏈接,減小鏈接的建立和關閉。線程池和鏈接池必定程度上緩解頻繁IO的資源消耗,但線程池和鏈接池都有必定規模，當鏈接請求數遠超過池上線，池系統構成的響應並不比多線程方案好多少。[1]
　　阻塞模型python實例demo以下:
　　阻塞模型server端git

def start_blocking(self):
        """同步阻塞server"""
        self.ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.ssock.bind(('', 8080))
        self.ssock.listen(5)
        count = 0
        while True:
            conn, addr = self.ssock.accept()
            count += 1
            print 'Connected by', addr
            print 'Accepted clinet count:%d' % count
            data = conn.recv(1024) #若無數據則阻塞
            if data:
                conn.sendall(data)
            conn.close()

　　阻塞模型clientgithub

def start_blocking(self):
        self.host = '123.207.123.108'
        self.port = 8080
        self.csock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.csock.connect((self.host, self.port))
        data = self.csock.recv(1024)
        print data

　　運行server端,並運行兩個client實例去鏈接服務端,運行結果以下圖1.2,能夠看到雖然有兩個客戶端去鏈接,但卻只有一個鏈接上,服務端的socket conn爲阻塞套接字,conn.recv(1024)未收到客戶端發送的數據,處於阻塞狀態,服務端沒法再響應另外一個客戶端的鏈接。
編程

圖1.2 阻塞IO服務端運行結果api

二非阻塞模型

　　因爲阻塞IO沒法知足大規模請求的缺點,所以出現了非阻塞模型。非阻塞IO模型以下圖1.3所示,當數據報未準備好,recvfrom當即返回一個EWOULDBLOCK錯誤,能夠利用輪詢不停調用recvfrom,當數據報準備好,內核則將數據複製到應用進程緩衝區。
數組

圖1.3 非阻塞IO模型網絡

　　非阻塞IO模型須要利用輪詢不斷調用recvfrom,浪費大量CPU時間,且當內核接收到數據時,須要等到下一次輪詢才能複製到應用進程緩衝區,數據得不到馬上處理。
　　非阻塞模型python demo以下:
　　非阻塞服務端多線程

def start_noblocking(self):
        """
        同步非阻塞
        """
        self.ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.ssock.bind(('', 8080))
        self.ssock.listen(5)
        count = 0
        while True:
            conn, addr = self.ssock.accept()
            conn.setblocking(0) #設置爲非阻塞socket
            count += 1
            print 'Connected by', addr
            print 'Accepted clinet count:%d' % count
            try:
                data = conn.recv(1024) #非阻塞,沒有數據會馬上返回
                if data:
                    conn.sendall(data)
            except Exception as e:
               pass
            finally:
                conn.close()

　　運行非阻塞服務端和兩個客戶端實例,結果以下圖1.4所示,服務端接收兩個鏈接請求。因爲conn被設置爲非阻塞socket,即便客戶端並無向服務端發送數據,conn.recv(1024)也會當即返回,不會阻塞,從而進程能夠接收新的鏈接請求。

圖1.4 非阻塞服務端運行結果

三 IO複用

　　IO複用在linux中包括select、poll、epoll模型三種,這三個IO複用模型有各自的API實現,以select模型爲例,調用select函數,進程進入阻塞, 同時監控多個套接字描述符的狀態 ,當有數據報變爲可讀或阻塞超時才返回,接着進程可調用recvfrom接收數據報到應用進程緩衝區。

圖3.1 IO複用模型

　　使用IO複用的優勢是能夠等待多個描述符就緒。^[2]

3.1 select模型

　　select模型api以下:
　　int select(int maxfdp1, fd_set *readset, fd_set *writeset, fd_set *exceptset,struct timeval *timeout);
timout表示內核等待任一描述符就緒可等待的時間,有三種狀況:
　　1) 空指針,表示能夠一直等下去,直到有描述符就緒。
　　2) timeout時間爲0,不等待檢查描述符狀態當即返回。
　　3) 時間不爲0，表示等待必定時間,在有描述符準備好但不超過timeval結構所指定的秒數和微秒數。
readset、writeset、exceptset指定須要內核測試讀、寫和異常條件的描述符。fd_set表示描述符集,在select中用整數數組表示,整數的每一位表示一個描述符, readset、writeset、exceptset這三個參數是值-結果類型。
　　能夠用如下幾個宏設置和測試fd_set。在調用select函數前,用一、二、3設置須要監控的描述符,循環調用4測試調用select函數後的描述符,看是否準備好。
　　1) int FD_ZERO(int fd, fd_set *fdset);
　　2) int FD_CLR(int fd, fd_set *fdset);
　　3) int FD_SET(int fd, fd_set *fd_set);
　　4) int FD_ISSET(int fd, fd_set *fdset);
　　致使select返回某個套接字就緒的條件以下:

圖3.2 就緒條件

　　maxfd1表示指定待測試描述符個數,值爲待測試描述符最大值加1,用這個參數可告訴內核最大隻遍歷到maxfd1-1的描述符。maxfd1最大不能超過常量FD_SETSIZE(值默認爲1024，更改該值需從新編譯內核)。

　　select函數的返回值爲整數,表示跨全部描述符集已就緒的總位數。若是超時則返回0。返回-1表示出錯,好比被中斷^[3]。
　　select實現原理:從用戶空間拷貝fd_set到內核空間,遍歷全部fd,將當前進程掛到各個設備的等待隊列中,掛到隊列的同時會返回是否就緒的掩碼,當全部fd返回的掩碼均未就緒,則當前進程睡眠。當fd對應設備驅動發現可讀寫時，則會喚醒處於睡眠態的進程。若是超過必定時間還未喚醒, 則調用select的進程會從新被喚醒得到CPU，進而從新遍歷fd，判斷有沒有就緒的fd,將fd_set從內核空間拷到用戶空間^[4]。
　　select實現的缺點:
　　1) 每次都須要將fd_set拷貝到內核空間,當fd_set較大時開銷很大
　　2) 每次都須要在內核中遍歷fd加入到等待隊列,fd較多開銷較大
　　3) Select支持的文件描述符過小,默認爲1024。

　　select模型python demo以下:
　　select模型服務端

def start(self):
        # create a socket
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setblocking(False)
        # set option reused
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_address = ('', 8080)
        server.bind(server_address)

        server.listen(10)

        # sockets from which we except to read
        inputs = [server]

        # sockets from which we expect to write
        outputs = []

        # Outgoing message queues (socket:Queue)
        message_queues = {}

        # A optional parameter for select is TIMEOUT
        timeout = 20

        while inputs:
            print "waiting for next event"
            # 每次調用select函數,須要將全部socket從新傳一次
            readable, writable, exceptional = select.select(
                inputs, outputs, inputs, timeout)

            # When timeout reached , select return three empty lists
            if not (readable or writable or exceptional):
                print "Time out ! "
                break
            for s in readable:
                if s is server:  # 監聽套接字
                    # A "readable" socket is ready to accept a connection
                    connection, client_address = s.accept()
                    print "    connection from ", client_address
                    connection.setblocking(0)
                    inputs.append(connection)
                    message_queues[connection] = Queue.Queue()
                else:
                    data = s.recv(1024)  # 接收到數據
                    if data:
                        print " received ", data, "from ", s.getpeername()
                        message_queues[s].put(data)
                        # Add output channel for response
                        if s not in outputs:
                            outputs.append(s)
                    else:  # 讀這端的鏈接關閉
                        # Interpret empty result as closed connection
                        print "  closing", client_address
                        if s in outputs:
                            outputs.remove(s)
                        inputs.remove(s)
                        s.close()
                        # remove message queue
                        del message_queues[s]
            for s in writable:
                try:
                    next_msg = message_queues[s].get_nowait()
                except Queue.Empty:
                    print " ", s.getpeername(), 'queue empty'
                    outputs.remove(s)
                else:
                    print " sending ", next_msg, " to ", s.getpeername()
                    s.send(next_msg)

            for s in exceptional:
                print " exception condition on ", s.getpeername()
                # stop listening for input on the connection
                inputs.remove(s)
                if s in outputs:
                    outputs.remove(s)
                s.close()
                # Remove message queue
                del message_queues[s]

　　select模型客戶端

def start(self):
        messages = ["hello world"]
        print "Connect to the server"
         
        server_address = ("123.207.123.108",8080)
         
        #Create a TCP/IP sock
         
        socks = []
         
        for i in range(3):
            socks.append(socket.socket(socket.AF_INET,socket.SOCK_STREAM))
         
        for s in socks:
            s.connect(server_address)
         
        counter = 0
        for message in messages :
            #Sending message from different sockets
            for s in socks:
                counter+=1
                print "  %s sending %s" % (s.getpeername(),message+" version "+str(counter))
                s.send(message+" version "+str(counter))
            #Read responses on both sockets
            for s in socks:
                data = s.recv(1024)
                print " %s received %s" % (s.getpeername(),data)
                if not data:
                    print "%s closing socket "%s.getpeername()
                    s.close()

　　分別運行服務端和客戶端,結果以下:

圖3.3 select模型服務端運行結果

圖3.4 select模型客戶端運行結果

3.2 poll模型

　　poll模型api以下^[8]:

#include <poll.h>
int poll(struct pollfd fds[], nfds_t nfds, int timeout);

typedef struct pollfd {
        int fd;      // 須要被檢測或選擇的文件描述符
        short events;  // 對文件描述符fd上感興趣的事件
        short revents;   // 文件描述符fd上當前實際發生的事件*/
} pollfd_t;

　　1) poll()函數返回fds集合中就緒的讀、寫，或出錯的描述符數量，返回0表示超時，返回-1表示出錯；
　　2) fds是一個struct pollfd類型的數組，用於存放須要檢測其狀態的socket描述符，而且調用poll函數以後fds數組不會被清空；
　　3) nfds記錄數組fds中描述符的總數量；
　　4) timeout是調用poll函數阻塞的超時時間，單位毫秒；
　　5) 一個pollfd結構體表示一個被監視的文件描述符，經過傳遞fds[]指示 poll() 監視多個文件描述符。其中，結構體的events域是監視該文件描述符的事件掩碼，由用戶來設置這個域，結構體的revents域是文件描述符的操做結果事件掩碼，內核在調用返回時設置這個域。events域中請求的任何事件均可能在revents域中返回。
　　合法的事件以下：
　　1) POLLIN 有數據可讀
　　2) POLLRDNORM 有普通數據可讀
　　3) POLLRDBAND 有優先數據可讀
　　4) POLLPRI 有緊迫數據可讀
　　5) POLLOUT 寫數據不會致使阻塞
　　6) POLLWRNORM 寫普通數據不會致使阻塞
　　7) POLLWRBAND 寫優先數據不會致使阻塞
　　8) POLLERR 發生錯誤
　　9) POLLHUP 發生掛起
　　當須要監聽多個事件時，使用POLLIN | POLLPRI設置 events 域；當poll調用以後檢測某事件是否發生時，fds[i].revents & POLLIN進行判斷
poll模型和select模型類似,poll模型一樣須要將全部監控的描述符從新拷貝到內核,並在內核中對全部描述符進行遍歷,沒有解決select模型的性能問題,可是poll模型沒有最大文件描述符數量的限制。
　　select()和poll()將就緒的文件描述符告訴進程後，若是進程沒有對其進行IO操做，那麼下次調用select()和poll()的時候將再次報告這些文件描述符，因此它們通常不會丟失就緒的消息，這種方式稱爲水平觸發^[5]。
　　poll模型python demo以下:

def start(self)://poll模型服務端
        # Create a TCP/IP socket, and then bind and listen
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setblocking(False)
        server_address = ('', 8080)
        print "Starting up on %s port %s" % server_address
        server.bind(server_address)
        server.listen(5)
        message_queues = {}
        # The timeout value is represented in milliseconds, instead of seconds.
        timeout = 5000
        # Create a limit for the event,POLLIN = POLLRDNORM | POLLRDBAND
        READ_ONLY = (select.POLLIN | select.POLLPRI)
        READ_WRITE = (READ_ONLY | select.POLLOUT) #POLLOUT=POLLWRNORM | POLLWRBAND
        # Set up the poller
        poller = select.poll()
        poller.register(server, READ_ONLY)
        # Map file descriptors to socket objects
        fd_to_socket = {server.fileno(): server, }
        while True:
            print "Waiting for the next event"
            events = poller.poll(timeout)
            if len(events) == 0:
                print 'Time out'
                break
            print "*" * 20
            print len(events)
            print events
            print "*" * 20
            for fd, flag in events:
                s = fd_to_socket[fd]
                if flag & (select.POLLIN | select.POLLPRI):
                    if s is server:
                        # A readable socket is ready to accept a connection
                        connection, client_address = s.accept()
                        print " Connection ", client_address
                        connection.setblocking(False)

                        fd_to_socket[connection.fileno()] = connection
                        poller.register(connection, READ_ONLY)

                        # Give the connection a queue to send data
                        message_queues[connection] = Queue.Queue()
                    else:
                        data = s.recv(1024)
                        if data:
                            # A readable client socket has data
                            print "  received %s from %s " % (data, s.getpeername())
                            message_queues[s].put(data)
                            poller.modify(s, READ_WRITE)
                        else:
                            # Close the connection
                            print "  closing", s.getpeername()
                            # Stop listening for input on the connection
                            poller.unregister(s)
                            s.close()
                            del message_queues[s]
                elif flag & select.POLLHUP:
                    # A client that "hang up" , to be closed.
                    print " Closing ", s.getpeername(), "(HUP)"
                    poller.unregister(s)
                    s.close()
                elif flag & select.POLLOUT:
                    # Socket is ready to send data , if there is any to send
                    try:
                        next_msg = message_queues[s].get_nowait()
                    except Queue.Empty:
                        # No messages waiting so stop checking
                        print s.getpeername(), " queue empty"
                        poller.modify(s, READ_ONLY)
                    else:
                        print " sending %s to %s" % (next_msg, s.getpeername())
                        s.send(next_msg)
                elif flag & select.POLLERR:
                    # Any events with POLLERR cause the server to close the
                    # socket
                    print "  exception on", s.getpeername()
                    poller.unregister(s)
                    s.close()
                    del message_queues[s]

3.3 epoll模型

　　epoll模型api包含三個系統調用^[7]:

#include <sys/epoll.h>
int epoll_create(int size);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

　　1. epoll_create建立epoll句柄epfd。size表示在這個epoll fd上能關注的最大fd數,失敗時返回-1。
　　2. epoll_ctl註冊要監聽的事件。
　　　　1) epfd表示epoll句柄；
　　　　2) op表示fd操做類型：EPOLL_CTL_ADD（註冊新的fd到epfd中），EPOLL_CTL_MOD（修改已註冊的fd的監聽事件），EPOLL_CTL_DEL（從epfd中刪除一個fd）
　　　　3) fd是要監聽的描述符；
　　　　4) event表示要監聽的事件; EPOLLIN表示對應的文件描述符能夠讀（包括對端SOCKET正常關閉）；EPOLLOUT表示對應的文件描述符能夠寫；EPOLLPRI表示對應的文件描述符有緊急的數據可讀（這裏應該表示有帶外數據到來）；EPOLLERR表示對應的文件描述符發生錯誤；EPOLLHUP表示對應的文件描述符被掛斷；EPOLLET將EPOLL設爲邊緣觸發(Edge Triggered)模式，這是相對於水平觸發(Level Triggered)來講的。EPOLLONESHOT只監聽一次事件，當監聽完此次事件以後，若是還須要繼續監聽這個socket的話，須要再次把這個socket加入到EPOLL隊列裏^[8]。
　　3. epoll_wait函數等待事件就緒，成功時返回就緒的事件數目，調用失敗時返回 -1，等待超時返回 0。
　　　　1) epfd是epoll句柄
　　　　2) events表示從內核獲得的就緒事件集合
　　　　3) maxevents告訴內核events的大小
　　　　4) timeout表示等待的超時事件
　　epoll_event結構體定義以下:
　　

struct epoll_event {
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
typedef union epoll_data {
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;

　　epoll模型利用三個函數代替select和poll模型的三個函數,能夠避免select模型的三個缺點。
　　1) 不須要每次都將相同的fd監聽事件從新拷貝到內核。epoll的解決方案在epoll_ctl函數中。每次註冊新的事件到epoll句柄中時（在epoll_ctl中指定EPOLL_CTL_ADD），會把全部的fd拷貝進內核，而不是在epoll_wait的時候重複拷貝。epoll保證了每一個fd在整個過程當中只會拷貝一次。
　　2) 不須要再內核中遍歷全部fd來看事件是否就緒。epoll的解決方案不像select或poll同樣每次都把current進程輪流加入fd對應的設備等待隊列中，而只在epoll_ctl時把current進程掛一遍（這一遍必不可少）併爲每一個fd指定一個回調函數，當設備就緒，喚醒等待隊列上的等待者時，就會調用這個回調函數，而這個回調函數會把就緒的fd加入一個就緒鏈表）。epoll_wait的工做實際上就是在這個就緒鏈表中查看有沒有就緒的fd。
　　3) 所監聽的文件描述符的數目不像select有上限限制, 所支持的FD上限是最大能夠打開文件的數目。

　　epoll對文件描述符的操做有兩種模式：LT(level trigger，水平觸發)和ET(edge trigger)。
　　1) 水平觸發：默認工做模式，即當epoll_wait檢測到某描述符事件就緒並通知應用程序時，應用程序能夠不當即處理該事件；下次調用epoll_wait時，會再次通知此事件。
　　2) 邊緣觸發：當epoll_wait檢測到某描述符事件就緒並通知應用程序時，應用程序必須當即處理該事件。若是不處理，下次調用epoll_wait時，不會再次通知此事件。（直到你作了某些操做致使該描述符變成未就緒狀態了，也就是說邊緣觸發只在狀態由未就緒變爲就緒時通知一次）。

　　ET模式很大程度上減小了epoll事件的觸發次數，所以效率比LT模式高。

　　epoll模型python demo以下:

def start(self):
        # Create a TCP/IP socket, and then bind and listen
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setblocking(False)
        server_address = ('', 8080)
        print "Starting up on %s port %s" % server_address
        server.bind(server_address)
        server.listen(5)
        message_queues = {}
        # The timeout value is represented in milliseconds, instead of seconds.
        timeout = 5000
        # Create a limit for the event
        READ_ONLY = (select.EPOLLIN)
        READ_WRITE = (READ_ONLY | select.EPOLLOUT)
        # Set up the epoll
        epoll = select.epoll()
        epoll.register(server.fileno(), READ_ONLY)
        # Map file descriptors to socket objects
        fd_to_socket = {server.fileno(): server, }
        while True:
            print "Waiting for the next event"
            events = epoll.poll(timeout)
            if len(events) == 0:
                print 'Time out'
                break
            print "*" * 20
            print len(events)
            print events
            print "*" * 20
            for fd, flag in events:
                s = fd_to_socket[fd]
                if flag & (select.EPOLLIN):
                    if s is server:
                        # A readable socket is ready to accept a connection
                        connection, client_address = s.accept()
                        print " Connection ", client_address
                        connection.setblocking(False)

                        fd_to_socket[connection.fileno()] = connection
                        epoll.register(connection, READ_ONLY)

                        # Give the connection a queue to send data
                        message_queues[connection] = Queue.Queue()
                    else:
                        data = s.recv(1024)
                        if data:
                            # A readable client socket has data
                            print "  received %s from %s " % (data, s.getpeername())
                            message_queues[s].put(data)
                            epoll.modify(s, READ_WRITE)
                        else:
                            # Close the connection
                            print "  closing", s.getpeername()
                            # Stop listening for input on the connection
                            epoll.unregister(s)
                            s.close()
                            del message_queues[s]
                elif flag & select.EPOLLHUP:
                    # A client that "hang up" , to be closed.
                    print " Closing ", s.getpeername(), "(HUP)"
                    epoll.unregister(s)
                    s.close()
                elif flag & select.EPOLLOUT:
                    # Socket is ready to send data , if there is any to send
                    try:
                        next_msg = message_queues[s].get_nowait()
                    except Queue.Empty:
                        # No messages waiting so stop checking
                        print s.getpeername(), " queue empty"
                        epoll.modify(s, READ_ONLY)
                    else:
                        print " sending %s to %s" % (next_msg, s.getpeername())
                        s.send(next_msg)
                elif flag & select.epollERR:
                    # Any events with epollR cause the server to close the
                    # socket
                    print "  exception on", s.getpeername()
                    epoll.unregister(s)
                    s.close()
                    del message_queues[s]

3.4 IO複用小結

　　沒有IO複用以前,用阻塞型IO,必須爲每一個創建的鏈接建立線程或線程,當面對大量鏈接時, 嚴重浪費系統資源,影響進程的響應效率,用非阻塞IO,須要輪詢測試socket集合是否已經讀寫就緒,在已經就緒和測試到就緒有必定的時延,數據得不到及時處理。利用IO複用, 同時可監控多個套接字描述符的狀態,而不用像阻塞型IO,每一個套接字須要一個線程或進程處理,也不像非阻塞IO,存在處理時延,IO複用函數是阻塞函數,不用輪詢測試,有socket就緒或超時纔會返回。
IO複用分爲select、poll、epoll模型三種,select模型存在以下三個缺點:
　　1) 每次都須要將fd_set拷貝到內核空間,當fd_set較大時開銷很大
　　2) 每次都須要在內核中遍歷fd加入到等待隊列,fd較多開銷較大
　　3) select支持的文件描述符過小,默認爲1024
　　poll模型不存在同時監聽的描述符大小限制,可是仍然存在缺點1和2。epoll模型克服了這三個缺點,epoll模型對於加入監聽的socket描述符,會將描述符和監聽的事件記在內核,無需像select和poll每次都須要將文件描述符集拷貝到內核。在判斷是否有讀寫就緒時。當有讀寫事件就緒時,內核會調用函數將就緒的fd加入就緒鏈表,所以epoll模型只需讀就緒鏈表,而不須要將全部fd遍歷一遍,性能會比select和poll模型高。

四信號驅動和異步IO

4.1 信號驅動IO

　　信號驅動式IO模型原理以下圖4.1:

圖4.1 信號驅動IO

　　Signal Driven I/O 的工做原理就是用戶進程首先和 kernel 之間創建信號的通知機制，即用戶進程告訴 kernel，若是 kernel 中數據準備好了，就經過 SIGIO 信號通知進程。而後用戶空間的進程就會調用 read 系統調用將準備好的數據從 kernel 拷貝到用戶空間。

　　可是這種 I/O 模型存在一個很是重大的缺陷問題：SIGIO 這種信號對於每一個進程來講只有一個！若是使該信號對進程中的兩個描述符（這兩個文件描述符都等待着 I/O 操做）都起做用，那麼進程在接到此信號後就沒法判別是哪個文件描述符準備好了。因此 Signal Driven I/O 模型在現實中用的很是少。

4.2 異步IO

　　異步IO模型原理以下圖:

圖4.2 異步IO

　　在異步IO中,用戶進程調用aio_read當即返回,直到內核將數據拷貝到進程緩衝區,而後通知進程完成,整個過程徹底沒阻塞,連recvfrom都不用用戶進程調用。其它的IO模型都屬於同步IO。
在異步非阻塞 I/O 中，能夠同時發起多個傳輸操做。這須要每一個傳輸操做都有唯一的上下文，這樣才能在它們完成時區分究竟是哪一個傳輸操做完成了。在 AIO 中，這是一個 aiocb（AIO I/O Control Block）結構。這個結構包含了有關傳輸的全部信息，包括爲數據準備的用戶緩衝區。在產生 I/O （稱爲完成）通知時，aiocb 結構就被用來唯一標識所完成的 I/O 操做。這個 API 的展現顯示瞭如何使用它^[10]。
　　aiocb結構以下:
　　

struct aiocb {
int aio_fildes; // File Descriptor
int aio_lio_opcode; // Valid only for lio_listio (r/w/nop)
volatile void *aio_buf; // Data Buffer
size_t aio_nbytes; // Number of Bytes in Data Buffer
struct sigevent aio_sigevent; // Notification Structure

/* Internal fields */
...

};

　　sigevent 結構告訴 AIO 在 I/O 操做完成時應該執行什麼操做。Aio api以下:
　　1) int aio_read( struct aiocb *aiocbp ) 請求異步讀操做
　　2) aio_error 檢查異步請求的狀態
　　3) aio_return 得到完成的異步請求的返回狀態
　　4) aio_write 請求異步寫操做
　　5) aio_suspend 掛起調用進程，直到一個或多個異步請求已經完成（或失敗）
　　6) aio_cancel 取消異步 I/O 請求
　　7) lio_listio 發起一系列 I/O 操做

　　爲了便於理解,這裏使用c語言,使用 aio_read 進行異步讀操做c實例以下:

//使用aio api讀實例
#include <aio.h>

...

int fd, ret;
struct aiocb my_aiocb;

fd = open( "file.txt", O_RDONLY );
if (fd < 0) perror("open");

/* Zero out the aiocb structure (recommended) */
bzero( (char *)&my_aiocb, sizeof(struct aiocb) );

/* Allocate a data buffer for the aiocb request */
my_aiocb.aio_buf = malloc(BUFSIZE+1);// 清空了 aiocb 結構，分配一個數據緩衝區
if (!my_aiocb.aio_buf) perror("malloc");

/* Initialize the necessary fields in the aiocb */
my_aiocb.aio_fildes = fd; //文件描述符
my_aiocb.aio_nbytes = BUFSIZE;//緩衝區大小
my_aiocb.aio_offset = 0;// // 將 aio_offset 設置成 0（該文件中的第一個偏移量）

ret = aio_read( &my_aiocb );//發起異步讀請求
if (ret < 0) perror("aio_read");

while ( aio_error( &my_aiocb ) == EINPROGRESS ) ;//檢查異步請求是否完成

if ((ret = aio_return( &my_iocb )) > 0) {//所傳輸的字節數，若是發生錯誤，返回值就爲 -1
/* got ret bytes on the read */
} else {
/* read failed, consult errno */
}

　　當異步請求完成時,內核有兩種方式通知進程,一種是經過信號,另外一種是調用回調函數。
　　使用信號做爲AIO通知demo以下,應用程序對指定信號註冊信號處理函數, 在產生指定的信號時就會調用這個處理程序。並指定AIO操做完成時,由內核發出指定信號,將aiocb做爲信號的上下文,用來分辨多個IO請求。
　　

AIO完成通知-信號
void setup_io( ... )
{
int fd;
struct sigaction sig_act;
struct aiocb my_aiocb;

...

/* Set up the signal handler */
sigemptyset(&sig_act.sa_mask);
sig_act.sa_flags = SA_SIGINFO;
sig_act.sa_sigaction = aio_completion_handler;


/* Set up the AIO request */
bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
my_aiocb.aio_fildes = fd;
my_aiocb.aio_buf = malloc(BUF_SIZE+1);
my_aiocb.aio_nbytes = BUF_SIZE;
my_aiocb.aio_offset = next_offset;

/* Link the AIO request with the Signal Handler */
my_aiocb.aio_sigevent.sigev_notify = SIGEV_SIGNAL;//指定信號做爲通知方法
my_aiocb.aio_sigevent.sigev_signo = SIGIO;
my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;

/* Map the Signal to the Signal Handler */
ret = sigaction( SIGIO, &sig_act, NULL );

...

ret = aio_read( &my_aiocb );

}


void aio_completion_handler( int signo, siginfo_t *info, void *context )
{
struct aiocb *req;


/* Ensure it's our signal */
if (info->si_signo == SIGIO) {

req = (struct aiocb *)info->si_value.sival_ptr;

/* Did the request complete? */
if (aio_error( req ) == 0) {

/* Request completed successfully, get the return status */
ret = aio_return( req );

}

}

return;
}

　　使用回調函數做爲異步請求通知demo以下, 這種機制不會爲通知而產生一個信號，而是會調用用戶空間的一個函數來實現通知功能.
　　

//AIO完成通知-回調函數
void setup_io( ... )
{
int fd;
struct aiocb my_aiocb;

...

/* Set up the AIO request */
bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
my_aiocb.aio_fildes = fd;
my_aiocb.aio_buf = malloc(BUF_SIZE+1);
my_aiocb.aio_nbytes = BUF_SIZE;
my_aiocb.aio_offset = next_offset;

/* Link the AIO request with a thread callback */
my_aiocb.aio_sigevent.sigev_notify = SIGEV_THREAD;// SIGEV_THREAD 指定線程回調函數來做爲通知方法
my_aiocb.aio_sigevent.notify_function = aio_completion_handler;
my_aiocb.aio_sigevent.notify_attributes = NULL;
my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;

...

ret = aio_read( &my_aiocb );

}


void aio_completion_handler( sigval_t sigval )
{
struct aiocb *req;

req = (struct aiocb *)sigval.sival_ptr;

/* Did the request complete? */
if (aio_error( req ) == 0) {

/* Request completed successfully, get the return status */
ret = aio_return( req );

}

return;
}

總結

　　網絡IO模型包括阻塞、非阻塞、IO複用、信號驅動IO和異步IO五種類型。阻塞IO沒法應對多個鏈接的情形,單個socket操做阻塞會致使服務端沒法接受其餘鏈接,雖然能夠用多線程、多進程的方式,將不一樣的鏈接放在不一樣的線程中和客戶端交互,並利用線程池和鏈接池進行優化。但建立進程和線程會佔用系統資源,當面對大規模鏈接時,系統資源浪費嚴重,系統響應效率不高。
非阻塞模型當socket讀寫操做未就緒時會當即返回,而不會阻塞等待,能夠利用輪詢的方式來進行讀寫操做,但當內核收到數據報到應用進程感知並處理會有時延。
利用IO複用,將監控socket讀寫操做是否就緒和進行讀寫操做分開,且IO複用可監控socket集合,IO複用包含select、poll、epoll三種模型。
　　select模型存在以下三種缺點:
　　1) 每次都須要將fd_set拷貝到內核空間,當fd_set較大時開銷很大
　　2) 每次都須要在內核中遍歷fd加入到等待隊列,fd較多開銷較大
　　3) select支持的文件描述符過小,默認爲1024。
　　poll模型可同時監控的socket沒有上線限制,取決於系統資源,但poll模型不能避免缺點1和2。epoll模型能夠避免select和poll模型的缺點。select，poll每次調用都要把fd集合從用戶態往內核態拷貝一次，而且要把current進程往設備等待隊列中掛一次，而epoll只要一次拷貝，並且把current進程往等待隊列上掛也只掛一次。這也能節省很多的開銷。select，poll內部實現須要本身不斷輪詢全部fd集合，直到設備就緒，期間可能要睡眠和喚醒屢次交替。而epoll其實也須要調用epoll_wait不斷輪詢就緒鏈表，期間也可能屢次睡眠和喚醒交替，可是它是設備就緒時，調用回調函數，把就緒fd放入就緒鏈表中，並喚醒在epoll_wait中進入睡眠的進程。雖然都要睡眠和交替，可是select和poll在「醒着」的時候要遍歷整個fd集合，而epoll在「醒着」的時候只要判斷一下就緒鏈表是否爲空就好了，這節省了大量的CPU時間。這就是回調機制帶來的性能提高。
　　信號驅動式IO,當內核數據準備好時,發出信號,調用進程提早註冊好的信號處理函數,但當存在多個socket操做時，沒法分清是哪一個socket準備好,所以實際應用中較少。
不管是阻塞IO、非阻塞IO、IO複用仍是信號驅動IO模型,都是同步IO模型。其要麼是監控socket就緒,要麼是從內核拷貝數據到進程緩衝區,至少其中一個是阻塞的,不會當即返回。異步IO模型發起讀寫操做後,當即返回,能夠接着進行其它操做,內核完成將數據拷貝到應用進程後,經過信號或者回調函數通知進程。

參考文獻

[1]. 阻塞IO（blocking IO）. https://www.chenxie.net/archives/1956.html[2]. Unix網絡編程卷1.124~125.[3]. linux select函數詳解. https://blog.csdn.net/lingfengtengfei/article/details/12392449[4]. select，poll，epoll實現分析—結合內核源代碼. https://www.linuxidc.com/Linux/2012-05/59873.htm[5]. Python網絡編程中的select 和 poll I/O複用的簡單使用. https://www.cnblogs.com/coser/archive/2012/01/06/2315216.html[6]. socket選項總結（setsocketopt). https://blog.csdn.net/c1520006273/article/details/50420408[7]. Linux下I/O多路複用系統調用(select, poll, epoll)介紹. https://zhuanlan.zhihu.com/p/22834126[8]. IO多路複用：select、poll、epoll示例. https://blog.csdn.net/lisonglisonglisong/article/details/51328062[9]. Linux I/O 模型. https://woshijpf.github.io/linux/2017/07/10/Linux-IO%E6%A8%A1%E5%9E%8B.html.[10]. 使用異步 I/O 大大提升應用程序的性能. https://www.ibm.com/developerworks/cn/linux/l-async/