Part9 - 异步IO

本节内容

1. 事件驱动
2. IO多路复用
   1. 用户空间和内存空间
   2. 进程切换
   3. 进程的阻塞
   4. 文件描述符fd
   5. 缓存IO
3. Select\Poll\Epoll异步IO
4. Twsited异步网络框架

1、事件驱动

事件驱动编程是一种编程范式，这里程序的执行流由外部事件来决定。它的特色是包含一个事件循环，当外部事件发生时使用回调机制来触发相应的处理。另外两种常见的编程范式是（单线程）同步以及多线程编程。

 

当咱们面对以下的环境时，事件驱动模型一般是一个好的选择：

• 程序中有许多任务
• 任务之间高度独立（所以它们不须要互相通讯，或者等待彼此）
• 在等待事件到来时，某些任务会阻塞。
• 当应用程序须要在任务间共享可变的数据时，这也是一个不错的选择，由于这里不须要采用同步处理。

网络应用程序一般都有上述这些特色，这使得它们可以很好的契合事件驱动编程模型。

2、IO多路复用

具体介绍见：IO多路复用(番外篇)

讨论的背景是Linux环境下的network IO

IO多路复用（epool）和gevent模块（协程）区别与联系：

都是遇到IO就切换，在linux下底层都是经过libevent.so实现。能够认为gevent是对IO多路复用更上层的封装，IO多路复用是其默认设置，其更专一于任务之间的切换。

1.用户空间和内核空间

操做系统的核心是内核，能够访问受保护的内存空间，也有访问底层硬件设备的全部权限。为了保证内核的安全，用户进程不能直接操做内核（kernel），系统将虚拟空间划分为两部分，一部分为内核空间，一部分为用户空间。

2.进程切换

为了控制进程的执行，内核必须有能力挂起正在CPU上运行的进程，并恢复之前挂起的某个进程的执行。这种行为被称为进程切换。

3.进程的阻塞

正在执行的进程，因为期待的某些事件未发生，如请求系统资源失败、等待某种操做的完成、新数据还没有到达或无新工做作等，则由系统自动执行阻塞原语(Block)，使本身由运行状态变为阻塞状态。可见，进程的阻塞是进程自身的一种主动行为，也所以只有处于运行态的进程（得到CPU），才可能将其转为阻塞状态。当进程进入阻塞状态，是不占用CPU资源的。

4.文件描述符fd

文件描述符（File descriptor）是计算机科学中的一个术语，是一个用于表述指向文件的引用的抽象化概念。文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每个进程所维护的该进程打开文件的记录表。（文件句柄是真实的文件对象）

注：文件描述符这一律念每每只适用于UNIX、Linux这样的操做系统。

5.缓存 I/O

又被称做标准 I/O，大多数文件系统的默认 I/O 操做都是缓存 I/O。在 Linux 的缓存 I/O 机制中，操做系统会将 I/O 的数据缓存在文件系统的页缓存（ page cache ）中，也就是说，数据会先被拷贝到操做系统内核的缓冲区中，而后才会从操做系统内核的缓冲区拷贝到应用程序的地址空间。

缺点：
数据在传输过程当中须要在应用程序地址空间和内核进行屡次数据拷贝操做，这些数据拷贝操做所带来的 CPU 以及内存开销是很是大的。

6.IO模式

对于一次IO访问（以read举例），数据会先被拷贝到操做系统内核的缓冲区中，而后才会从操做系统内核的缓冲区拷贝到应用程序的地址空间。（内核态到用户态数据拷贝）

因此，当一个read操做发生时，它会经历两个阶段：
1. 等待数据准备 (Waiting for the data to be ready)
2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正是由于这两个阶段，linux系统产生了下面五种网络模式的方案。

• 阻塞 I/O（blocking IO）
• 非阻塞 I/O（nonblocking IO）
• I/O 多路复用（ IO multiplexing）
• 信号驱动 I/O（ signal driven IO，不经常使用）
• 异步 I/O（asynchronous IO，实现复杂用得少）

阻塞IO（bloking IO）

linux下，默认状况下全部的socket都是blocking。其特色就是在IO执行的两个阶段都是阻塞的 。

非阻塞 I/O（nonblocking IO）

linux下，能够经过设置socket使其变为non-blocking。其特色是IO执行的第一个阶段不阻塞，用户进程不断的主动询问kernel数据有没有准备好，kernel作出相应的回应，但IO执行的第二个阶段仍然是阻塞的。

I/O 多路复用（ IO multiplexing）

就是咱们说的select，poll，epoll，也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就能够同时处理多个网络链接的IO。它的基本原理就是select，poll，epoll这个function会不断的轮询所负责的全部socket，当某个socket有数据到达了，就通知用户进程数据准备好了，但IO执行的第二个阶段仍然是阻塞的。

这个图和blocking IO的图其实并无太大的不一样，事实上，其性能还更差一些。由于这里须要使用两个system call (select 和 recvfrom)，而blocking IO只调用了一个system call (recvfrom)。可是，用select的优点在于它能够同时处理多个connection。

因此，若是处理的链接数不是很高的话，使用select/epoll的web server不必定比使用multi-threading + blocking IO的web server性能更好，可能延迟还更大。select/epoll的优点并非对于单个链接能处理得更快，而是在于能处理更多的链接。

在IO multiplexing Model中，实际中，对于每个socket，一般都设置成为non-blocking，可是，如上图所示，整个用户的process实际上是一直被block的。只不过process是被select这个函数block，而不是被socket IO给block。

异步 I/O（asynchronous IO）

linux下的asynchronous IO其实用得不多。

用户进程发起read操做以后，就能够开始去作其它的事。从kernel的角度，当它收到一个asynchronous read以后，首先它会马上返回信号给用户进程（确认收到请求的信号？），因此不会对用户进程产生任何block。而后，kernel等待数据准备完成后，将数据拷贝到用户内存，拷贝完成以后，kernel会给用户进程发送一个signal，告诉它read操做完成了。

各个IO Model的比较

3、异步IO（Select\Poll\Epoll）

具体介绍见：Python Select 解析

select目前几乎在全部的平台上支持，其良好跨平台支持也是它的一个优势，可是使用select单个进程可以监视的文件描述符的数量存在最大限制，在Linux上通常为1024，不过能够经过修改宏定义甚至从新编译内核的方式提高这一限制。select监视到文件描述符有活跃，只向用户进程返回有活跃信号，没有返回具体活跃的文件描述符，用户进程还需再次循环检查浪费时间和资源。

poll和select在本质上没有多大差异，可是poll没有最大文件描述符数量的限制，能够看做是一个过渡阶段。

epoll直到Linux2.6才出现，它几乎具有了以前所说的一切优势，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。epoll能够同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，若是咱们没有采起行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，可是代码实现至关复杂。

注：epoll在linux上的文件描述符数量，可能会受到OS用户最大链接数限制，注意调整。

1.用select实现socket服务端

服务端

1. import select  
2. import socket  
3. import queue  
4.     
5. server = socket.socket()  
6. server.bind(('localhost',9000))  
7. server.listen(1000)  
8.     
9. server.setblocking(False) #必需要先设置不阻塞  
10.     
11. msg_dic = {}  
12.     
13. inputs = [server,]  
14. #inputs = [server,conn] #[conn,]  
15. #inputs = [server,conn,conn2] #[conn2,]  
16. outputs = [] #  
17. #outputs = [r1,] #  
18. while True:  
19.     readable ,writeable,exceptional= select.select(inputs, outputs, inputs )  
20.     print(readable,writeable,exceptional)  
21.     for r in readable:  
22.         if r is server: #表明来了一个新链接  
23.             conn,addr = server.accept()  
24.             print("来了个新链接",addr)  
25.             inputs.append(conn) #是由于这个新创建的链接还没发数据过来，如今就接收的话程序就报错了，  
26.             #所以要想实现这个客户端发数据来时server端能知道，就须要让select再监测这个conn  
27.             msg_dic[conn] = queue.Queue() #初始化一个队列，后面存要返回给这个客户端的数据  
28.         else: #conn2  
29.             data = r.recv(1024)  
30.             print("收到数据",data)  
31.             msg_dic[r].put(data)  
32.     
33.             outputs.append(r) #放入返回的链接队列里  
34.             # r.send(data)  
35.             # print("send done....")  
36.     
37.     for w in writeable: #要返回给客户端的链接列表  
38.         data_to_client = msg_dic[w].get()  
39.         w.send(data_to_client) #返回给客户端源数据  
40.     
41.         outputs.remove(w) #确保下次循环的时候writeable,不返回这个已经处理完的链接了  
42.     
43.     for e in exceptional:  
44.         if e in outputs:  
45.             outputs.remove(e)  
46.     
47.         inputs.remove(e)  
48.     
49.         del msg_dic[e]  

客户端

1. import socket  
2.     
3. HOST = 'localhost'  # The remote host  
4. PORT = 9999  # The same port as used by the server  
5. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
6. s.connect((HOST, PORT))  
7. while True:  
8.     msg = bytes(input(">>:"), encoding="utf8")  
9.     s.sendall(msg)  
10.     data = s.recv(1024)  
11.     
12.     #  
13.     print('Received', data)  
14. s.close() 

2.selectors模块

封装好的IO多路复用模块，默认使用epool，当系统不支持时使用select。

服务端

1. import selectors  
2. import socket  
3.     
4. sel = selectors.DefaultSelector()  
5.     
6.     
7. def accept(sock, mask):  
8.     conn, addr = sock.accept()  # Should be ready  
9.     print('accepted', conn, 'from', addr,mask)  
10.     conn.setblocking(False)  
11.     sel.register(conn, selectors.EVENT_READ, read) #新链接注册read回调函数  
12.     
13.     
14. def read(conn, mask):  
15.     data = conn.recv(1024)  # Should be ready  
16.     if data:  
17.         print('echoing', repr(data), 'to', conn)  
18.         conn.send(data)  # Hope it won't block  
19.     else:  
20.         print('closing', conn)  
21.         sel.unregister(conn)  
22.         conn.close()  
23.     
24.     
25. sock = socket.socket()  
26. sock.bind(('localhost', 9999))  
27. sock.listen(100)  
28. sock.setblocking(False)  
29. sel.register(sock, selectors.EVENT_READ, accept)  
30.     
31. while True:  
32.     events = sel.select() #默认阻塞，有活动链接就返回活动的链接列表  
33.     for key, mask in events:  
34.         callback = key.data #accept  
35.         callback(key.fileobj, mask) #key.fileobj=  文件句柄 

多并发客户端

1. import socket  
2. import sys  
3.     
4. messages = [ b'This is the message. ',  
5.              b'It will be sent ',  
6.              b'in parts.',  
7.              ]  
8. server_address = ('192.168.16.130', 9998)  
9.     
10. # Create a TCP/IP socket  
11. socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM) for i in range(11000)]  
12. print(socks)  
13. # Connect the socket to the port where the server is listening  
14. print('connecting to %s port %s' % server_address)  
15. for s in socks:  
16.     s.connect(server_address)  
17.     
18. for message in messages:  
19.     
20.     # Send messages on both sockets  
21.     for s in socks:  
22.         print('%s: sending "%s"' % (s.getsockname(), message) )  
23.         s.send(message)  
24.     
25.     # Read responses on both sockets  
26.     for s in socks:  
27.         data = s.recv(1024)  
28.         print( '%s: received "%s"' % (s.getsockname(), data) )  
29.         if not data:  
30.             print( 'closing socket', s.getsockname() )  

4、Twsited异步网络框架

Twisted本身用异步形式重写了SSH、DNS、FTP、HTTP等，代码比较复杂，游戏开发会用到。

http://www.cnblogs.com/alex3714/articles/5248247.html

参考：

http://www.cnblogs.com/alex3714/articles/5248247.html

http://www.cnblogs.com/alex3714/articles/5876749.html

http://www.cnblogs.com/alex3714/p/4372426.html