码迷,mamicode.com
首页 > 其他好文 > 详细

I/O多路复用-EPOLL探索

时间:2017-10-25 17:11:48      阅读:175      评论:0      收藏:0      [点我收藏+]

标签:sci   存储   threading   listen   lap   i/o   sel   splay   入队   

什么是I/O多路复用

I/O多路复用就是通过一种机制,可以监视多个描述符,一旦某个IO能够读写,通知程序进行相应的读写操作。

I/O多路复用的场合

1、当客户处理多个描述字时(通常是交互式输入和网络套接字),必须使用I/O复用

2、如果一个TCP服务器既要处理监听套接字,又要处理已连接套接字,一般也要用到I/O复用

3、如果一个服务器即要处理TCP,又要处理UDP,一般要使用I/O复用

Linux 下I/O多路复用的方式

SELECT、POLL、EPOLL

对比SELECT、POLL、EPOLL

SELECT缺点

每次调用select,都需要把fd集合从用户态拷贝到内核态,开销和fd的数量成正比

每次调用select都需要在内核遍历传递进来的所有fd,开销和fd的数量成正比

select支持的文件描述符数量太少,32位1024个,64位2048个

POLL缺点

poll和select本质相同,使用链表存储文件描述符,没有最大连接数的限制

下图是select、poll、kqueue(FreeBSD平台)、epoll四种方式连接数和时间关系图

技术分享

上图可以看出,随着fd数量的增大,select、poll的时间消耗非常大,kqueue和epoll基本上没有变化

EPOLL优点

1、它所支持的FD上限是最大可以打开文件的数目,在1GB内存的机器上大约是10万左 右,具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。

2、IO 效率不随FD数目增加而线性下降

3、epoll不像select或poll一样每次都把current轮流加入fd对应的设备等待队列中,而只在epoll_ctl时把current挂一遍(这一遍必不可少)并为每个fd指定一个回调函数,当设备就绪,唤醒等待队列上的等待者时,就会调用这个回调函数,而这个回调函数会把就绪的fd加入一个就绪链表)。epoll_wait的工作实际上就是在这个就绪链表中查看有没有就绪的fd,而不是全部遍历。

 

下面是一个场景,一个服务端Server.py,多个客户端订阅,服务端不断的生成消息,客户端订阅后服务端会不断的把消息发送给客户端:

Server:

技术分享
 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 __author__ = Andy
 5 import socket, select, traceback, time
 6 import threading
 7 import Queue
 8 
 9 gen = Queue.Queue()
10 connections = {}
11 requests = {}
12 responses = {}
13 
14 
15 def run():
16     serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
17     serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
18     serversocket.bind((127.0.0.1, 8080))
19     serversocket.listen(5)
20     serversocket.setblocking(0)
21     epoll = select.epoll()  # 创建一个epoll对象
22     epoll.register(serversocket.fileno(), select.EPOLLIN)  # 给新建的serversocket.fileno注册一个读event
23 
24     try:
25         count = 0
26         while True:
27             events = epoll.poll()  # 激活的fileno举手
28             count += 1
29             for fileno, event in events:
30                 if fileno == serversocket.fileno():  # 当激活的fileno是新建的,给该fileno注册一个读event
31                     connection, address = serversocket.accept()
32                     connection.setblocking(0)
33                     epoll.register(connection.fileno(), select.EPOLLIN)
34                     connections[connection.fileno()] = connection
35                     requests[connection.fileno()] = b‘‘
36                     responses[connection.fileno()] = b""
37                     print "new conn.fileno is %s" % connection.fileno()
38                 elif event & select.EPOLLIN:  # 如果fileno是读event,接收发送来消息,并修改该fileno为写event,下次循环时写数据
39                     print "read event is happing"
40                     requests[fileno] += connections[fileno].recv(1024)
41                     epoll.modify(fileno, select.EPOLLOUT)
42                     print(- * 40 + \n + requests[fileno].decode()[:-2])
43                 elif event & select.EPOLLOUT:  # 如果fileno是写事件,写完后正常的为挂起
44                     if responses[fileno]:
45                         byteswritten = connections[fileno].send(responses[fileno])
46                         responses[fileno] = responses[fileno][byteswritten:]
47                         if len(responses[fileno]) == 0:
48                             epoll.modify(fileno, select.EPOLLOUT)  # 需要向订阅者一直发消息,这里发完后仍为写event
49                             print "change event to write"
50                 elif event & select.EPOLLHUP:
51                     epoll.unregister(fileno)
52                     connections[fileno].close()
53                     del connections[fileno]
54                     print "event is HUP ===%s" % fileno
55         pass
56     except Exception, err:
57         print traceback.print_exc()
58     finally:
59         epoll.unregister(serversocket.fileno())
60         epoll.close()
61         serversocket.close()
62         print "finally"
63 
64 
65 def create_data():
66     count = 1
67     while True:
68         count += 1
69         res = "Message-%s" % count
70         gen.put_nowait(res)  # 把消息放入队列
71         time.sleep(0.5)
72 
73 
74 def update_message():
75     while True:
76         message = gen.get()
77         for res in responses:  # 遍历所有的活跃用户,更新消息
78             responses[res] = message
79         time.sleep(0.05)
80 
81 
82 if __name__ == "__main__":
83     p = threading.Thread(target=create_data)  # 开个线程向队列里放数据
84     p1 = threading.Thread(target=update_message)  # 从队列中取出数据
85     p.start()
86     p1.start()
87     run()
88     p.join()
89     p1.join()
View Code

 

Client:

技术分享
 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 __author__ = Andy
 5 import socket,time
 6 def sim_client(name,i):
 7     connFd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
 8     connFd.connect(("127.0.0.1", 8080))
 9     connFd.send("Process-%s start subscibe\n\n" % name)
10     while True:
11         try:
12             readData = connFd.recv(1024)
13             if readData:
14                 print "*"*40 + "\n" + readData.decode()
15         except:
16             time.sleep(0.2)
17 
18 
19 if __name__=="__main__":
20     sim_client("progressage",1)
View Code

 

来看一下输出结果

服务端接受订阅的消息:

技术分享

 

客户端订阅服务端的消息

 技术分享

 

才疏学浅,目前只研究了EPOLL的水平触发,还有边缘触发需要去探索

I/O多路复用-EPOLL探索

标签:sci   存储   threading   listen   lap   i/o   sel   splay   入队   

原文地址:http://www.cnblogs.com/onepiece-andy/p/epoll_level-triggered.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!