I/O多路复用-EPOLL探索

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：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

__author__ = ‘Andy‘
import socket, select, traceback, time
import threading
import Queue

gen = Queue.Queue()
connections = {}
requests = {}
responses = {}


def run():
    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    serversocket.bind((‘127.0.0.1‘, 8080))
    serversocket.listen(5)
    serversocket.setblocking(0)
    epoll = select.epoll()  # 创建一个epoll对象
    epoll.register(serversocket.fileno(), select.EPOLLIN)  # 给新建的serversocket.fileno注册一个读event

    try:
        count = 0
        while True:
            events = epoll.poll()  # 激活的fileno举手
            count += 1
            for fileno, event in events:
                if fileno == serversocket.fileno():  # 当激活的fileno是新建的,给该fileno注册一个读event
                    connection, address = serversocket.accept()
                    connection.setblocking(0)
                    epoll.register(connection.fileno(), select.EPOLLIN)
                    connections[connection.fileno()] = connection
                    requests[connection.fileno()] = b‘‘
                    responses[connection.fileno()] = b""
                    print "new conn.fileno is %s" % connection.fileno()
                elif event & select.EPOLLIN:  # 如果fileno是读event,接收发送来消息,并修改该fileno为写event,下次循环时写数据
                    print "read event is happing"
                    requests[fileno] += connections[fileno].recv(1024)
                    epoll.modify(fileno, select.EPOLLOUT)
                    print(‘-‘ * 40 + ‘\n‘ + requests[fileno].decode()[:-2])
                elif event & select.EPOLLOUT:  # 如果fileno是写事件,写完后正常的为挂起
                    if responses[fileno]:
                        byteswritten = connections[fileno].send(responses[fileno])
                        responses[fileno] = responses[fileno][byteswritten:]
                        if len(responses[fileno]) == 0:
                            epoll.modify(fileno, select.EPOLLOUT)  # 需要向订阅者一直发消息,这里发完后仍为写event
                            print "change event to write"
                elif event & select.EPOLLHUP:
                    epoll.unregister(fileno)
                    connections[fileno].close()
                    del connections[fileno]
                    print "event is HUP ===%s" % fileno
        pass
    except Exception, err:
        print traceback.print_exc()
    finally:
        epoll.unregister(serversocket.fileno())
        epoll.close()
        serversocket.close()
        print "finally"


def create_data():
    count = 1
    while True:
        count += 1
        res = "Message-%s" % count
        gen.put_nowait(res)  # 把消息放入队列
        time.sleep(0.5)


def update_message():
    while True:
        message = gen.get()
        for res in responses:  # 遍历所有的活跃用户,更新消息
            responses[res] = message
        time.sleep(0.05)


if __name__ == "__main__":
    p = threading.Thread(target=create_data)  # 开个线程向队列里放数据
    p1 = threading.Thread(target=update_message)  # 从队列中取出数据
    p.start()
    p1.start()
    run()
    p.join()
    p1.join()

 

Client:

#!/usr/bin/env python
# -*- coding:utf-8 -*-

__author__ = ‘Andy‘
import socket,time
def sim_client(name,i):
    connFd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    connFd.connect(("127.0.0.1", 8080))
    connFd.send("Process-%s start subscibe\n\n" % name)
    while True:
        try:
            readData = connFd.recv(1024)
            if readData:
                print "*"*40 + "\n" + readData.decode()
        except:
            time.sleep(0.2)


if __name__=="__main__":
    sim_client("progressage",1)

 

来看一下输出结果

服务端接受订阅的消息：

 

客户端订阅服务端的消息

 

 

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