sellect、poll、epoll

时间 2019-11-30

标签 sellect poll epoll 繁體版

原文原文链接

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

select
select最先于1983年出如今4.2BSD中，它经过一个select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程能够得到这些文件描述符从而进行后续的读写操做。python

select目前几乎在全部的平台上支持，其良好跨平台支持也是它的一个优势，事实上从如今看来，这也是它所剩很少的优势之一。数组

select的一个缺点在于单个进程可以监视的文件描述符的数量存在最大限制，在Linux上通常为1024，不过能够经过修改宏定义甚至从新编译内核的方式提高这一限制。缓存

另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增加。同时，因为网络响应时间的延迟使得大量TCP链接处于非活跃状态，但调用select()会对全部socket进行一次线性扫描，因此这也浪费了必定的开销。服务器

poll
poll在1986年诞生于System V Release 3，它和select在本质上没有多大差异，可是poll没有最大文件描述符数量的限制。网络

poll和select一样存在一个缺点就是，包含大量文件描述符的数组被总体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增长而线性增大。数据结构

另外，select()和poll()将就绪的文件描述符告诉进程后，若是进程没有对其进行IO操做，那么下次调用select()和poll()的时候将再次报告这些文件描述符，因此它们通常不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。并发

epoll
直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll，它几乎具有了以前所说的一切优势，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。app

epoll能够同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，若是咱们没有采起行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，可是代码实现至关复杂。socket

epoll一样只告知那些就绪的文件描述符，并且当咱们调用epoll_wait()、create、regester、unregest得到就绪文件描述符时，返回的不是实际的描述符，而是一个表明就绪描述符数量的值，你只须要去epoll指定的一个数组中依次取得相应数量的文件描述符便可，这里也使用了内存映射（mmap）技术，这样便完全省掉了这些文件描述符在系统调用时复制的开销。

另外一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用必定的方法后，内核才对全部监视的文件描述符进行扫描，而epoll事先经过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用相似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便获得通知。

Python select

Python的select()方法直接调用操做系统的IO接口，它监控sockets,open files, and pipes(全部带fileno()方法的文件句柄)什么时候变成readable 和writeable, 或者通讯错误，select()使得同时监控多个链接变的简单，而且这比写一个长循环来等待和监控多客户端链接要高效，由于select直接经过操做系统提供的C的网络接口进行操做，而不是经过Python的解释器。

注意：Using Python’s file objects with select() works for Unix, supported under Windows yet.

接下来经过echo server例子以了解select 是如何经过单进程实现同时处理多个非阻塞的socket链接的

select_server:

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

# 导入模块
import select
import socket
import sys
import queue

# 建立TCP/IP socket套接字实例server
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置 socket为非阻塞
server.setblocking(False)

## 绑定server socket地址(ip、端口)
server_address = ('localhost', 10000)       #设置服务端地址（即IP和端口）
print(sys.stderr, 'starting up on %s port %s' % server_address)     # 打印启动信息(ip、端口)
server.bind(server_address)     # server socket实例绑定地址

## 监听客户端链接,最多容许5个并发链接
server.listen(5)
'''
select()方法接收并监控3个通讯列表， 第一个是全部的输入的data,就是指外部发过来的数据，第2个是监控和接收全部要发出去的data(outgoing data),第3个监控错误信息，接下来咱们须要建立2个列表来包含输入和输出信息来传给select().
'''
'''
select.select()
select(rlist, wlist, xlist, timeout=None)
select(rlist, wlist, xlist[, timeout]) -> (rlist, wlist, xlist)
做用：等到一个或多个文件描述符准备好某种I/O
select() 方法中接收并监控三个通讯列表，
        rlist可读列表，全部输入的数据，监控外部发起的链接及发过来的数据；
        wlist可写列表，监控并接收全部要发出的数据
        xlist错误信息列表，监控错误信息

'''
# Sockets from which we expect to read定义可读列表，保存server socket及已创建的socket链接，首先把server socket放进去
inputs = [server]

# Sockets to which we expect to write定义可写列表，保存须要发送数据的socket链接
outputs = []

'''
全部客户端的进来的链接和数据将会被server的主循环程序放在上面的list中处理，咱们如今的server端须要等待链接可写（writable)以后才能过来，而后接收数据并返回(所以不是在接收到数据以后就马上返回)，由于每一个链接要把输入或输出的数据先缓存到queue里，而后再由select取出来再发出去。

Connections are added to and removed from these lists by the server main loop. Since this version of the server is going to wait for a socket to become writable before sending any data (instead of immediately sending the reply), each output connection needs a queue to act as a buffer for the data to be sent through it.
'''
# 定义发送消息缓冲区（字典类型，每一个链接与数据为其中一个key:value对，如socket_connect:'recv_data'），服务端收到的数据不是即收即发的，收到的数据先放到缓冲区,等到扫描可写时再从缓冲区的队列中读取并发送
message_queues = {}

'''
The main portion of the server program loops, calling select() to block and wait for network activity.

下面是此程序的主循环，调用select()时会阻塞和等待直到新的链接和数据进来
'''


## 程序主循环
while inputs:       # 循环可读列表

    # Wait for at least one of the sockets to be ready for processing
    print( '\nwaiting for the next event')          # 打印提示信息
    '''
    当你把inputs,outputs,exceptional(这里跟inputs共用)传给select()后，它返回3个新的list，咱们上面将他们分别赋值为readable,writable,exceptional, 全部在readable list中的socket链接表明有数据可接收(recv),全部在writable list中的存放着你能够对其进行发送(send)操做的socket链接，当链接通讯出现error时会把error写到exceptional列表中。

    select() returns three new lists, containing subsets of the contents of the lists passed in. All of the sockets in the readable list have incoming data buffered and available to be read. All of the sockets in the writable list have free space in their buffer and can be written to. The sockets returned in exceptional have had an error (the actual definition of “exceptional condition” depends on the platform).
    '''
    '''把 inputs、outputs、exceptional（与inputs共用）列表传给select(),
    用readable, writable, exceptional来承接select()返回的3个新列表,
    readable 列表中的server表示接收新socket链接，其余的socket链接表示有数据可接受
    writable列表中的socket链接表示有数据要发送
    exceptional列表的socket表示在通讯过程当中出现有错误
    '''
    readable, writable, exceptional = select.select(inputs, outputs, inputs, 2)     #设置2秒超时，默认阻塞的

    '''
    Readable list 中的socket 能够有3种可能状态，第一种是若是这个socket是main "server" socket,它负责监听客户端的链接，若是这个main server socket出如今readable里，那表明这是server端已经ready来接收一个新的链接进来了，为了让这个main server能同时处理多个链接，在下面的代码里，咱们把这个main server的socket设置为非阻塞模式。

The “readable” sockets represent three possible cases. If the socket is the main “server” socket, the one being used to listen for connections, then the “readable” condition means it is ready to accept another incoming connection. In addition to adding the new connection to the list of inputs to monitor, this section sets the client socket to not block.
    '''

    # Handle inputs处理可读列表
    '''
    可读列表中有三种状态：
    一、server socket监听 新的socket链接
    二、已创建的socket链接有数据发过来
    三、客户端断开链接（这时候recv()的数据为空）
    '''
    for s in readable:

        if s is server: # socket链接为main "server" socket,表示有新的socket要链接进来
            # A "readable" server socket is ready to accept a connection
            connection, client_address = s.accept()     #接收外部链接
            print('new connection from', client_address)        #打印客户地址(ip,port)
            connection.setblocking(False)       #设置新链接的socket为非阻塞
            inputs.append(connection)       #把新链接的socket追加到可读列表

            # Give the connection a queue for data we want to send 在数据缓存区中为新链接的socket分配列队，以便缓存须要发送给该socket的数据
            message_queues[connection] = queue.Queue()
        else:   #非server socket，即外部已经创建的socket链接，这时候就接收数据
            data = s.recv(1024)
            '''
            第二种状况是这个socket是已经创建了的链接，它把数据发了过来，这个时候你就能够经过recv()来接收它发过来的数据，而后把接收到的数据放到queue里，这样你就能够把接收到的数据再传回给客户端了。

The next case is an established connection with a client that has sent data. The data is read with recv(), then placed on the queue so it can be sent through the socket and back to the client.
            '''
            # 这里作一个简单的处理，就是客户发什么数据过来，服务端就再发什么数据给客户。
            # 若是是WEB服务器则，须要对外部发来的data进行解析，如外部请求是读取一个图片，那么服务端就须要把图片内容放到该socket链接在缓冲区中对应的对象中
            if data:        # 有数据，数据为非空
                # A readable client socket has data
                print(sys.stderr, 'received "%s" from %s' % (data, s.getpeername()) )       #打印接收到的数据及地址信息
                message_queues[s].put(data)     #把接收到的数据放到缓冲区中该链接的队列中
                # Add output channel for response
                # 把该socket链接添加到可写列表中，等到遍历可写列表时再发送上面存放的数据
                if s not in outputs:
                    outputs.append(s)
            else:       # 客户端已经断开链接，因此 recv()接收数据就为空，服务端须要把该客户的链接关闭（这里还须要删除可读、可写列表中的该socket链接，及缓冲区中的数据）
                '''
                第三种状况就是这个客户端已经断开了，因此你再经过recv()接收到的数据就为空了，因此这个时候你就能够把这个跟客户端的链接关闭了
                A readable socket without data available is from a client that has disconnected, and the stream is ready to be closed.
                '''
                # Interpret empty result as closed connection
                print('closing', client_address, 'after reading no data')   #打印客户端地址信息
                # Stop listening for input on the connection
                if s in outputs:        #若是可写列表中有该socket链接，
                    outputs.remove(s)    # 可写列表中删除该socket链接
                inputs.remove(s)    # 可读列表中删除该socket链接
                s.close()           # 关闭该socket链接

                # Remove message queue
                del message_queues[s]       # 消息缓冲区中删除给该socket链接分区的queue队列

    '''
    对于writable list中的socket，也有2种状态，若是这个客户端链接在跟它对应的queue里有数据，就把这个数据取出来再发回给这个客户端，不然就把这个链接从output list中移除，这样下一次循环select()调用时检测到outputs list中没有这个链接，那就会认为这个链接还处于非活动状态

There are fewer cases for the writable connections. If there is data in the queue for a connection, the next message is sent. Otherwise, the connection is removed from the list of output connections so that the next time through the loop select() does not indicate that the socket is ready to send data.
    '''
    # Handle outputs处理可写列表
    '''
    在可写列表中的socke链接有两种状态：
    一、有数据要发送
    二、无数据要发送（客户端为非活跃状态）
    '''
    for s in writable:
        try:
            next_msg = message_queues[s].get_nowait()           # 获取发送数据
        except queue.Empty:     # 无数据要发送
            # No messages waiting so stop checking for writability.
            print('output queue for', s.getpeername(), 'is empty')     #打印客户地址等提示信息
            outputs.remove(s)       # 可写列表中删除该socket链接，这样作也能让可写列表不变得那么大，即不保存非活跃的socket链接
        else:        # 有数据要发送
            print( 'sending "%s" to %s' % (next_msg, s.getpeername()))      #打印发送数据及地址信息
            s.send(next_msg)        # 发送数据

    '''
    最后，若是在跟某个socket链接通讯过程当中出了错误，就把这个链接对象在inputs\outputs\message_queue中都删除，再把链接关闭掉
    '''
    # Handle "exceptional conditions"处理错误列表
    # 中止监听错误列表中的socket链接
    for s in exceptional:
        print('handling exceptional condition for', s.getpeername() )       #打印地址等提示信息
        inputs.remove(s)        # 可读列表中删除该socket链接
        if s in outputs:        # 可写列表中删除该socket链接
            outputs.remove(s)
        s.close()       # 闭关该socket链接

        # 消息缓存区中删除该socket对应的queue队列
        del message_queues[s]

server运行结果：

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> starting up on localhost port 10000
waiting for the next event

waiting for the next event

new connection from ('127.0.0.1', 60486)
waiting for the next event

new connection from ('127.0.0.1', 60487)
waiting for the next event

new connection from ('127.0.0.1', 60488)
waiting for the next event

new connection from ('127.0.0.1', 60489)
waiting for the next event

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60486)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60487)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60488)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60489)
waiting for the next event

sending "b'This is the message.'" to ('127.0.0.1', 60486)
sending "b'This is the message.'" to ('127.0.0.1', 60487)
sending "b'This is the message.'" to ('127.0.0.1', 60488)
sending "b'This is the message.'" to ('127.0.0.1', 60489)
waiting for the next event

output queue for ('127.0.0.1', 60486) is empty
output queue for ('127.0.0.1', 60487) is empty
output queue for ('127.0.0.1', 60488) is empty
output queue for ('127.0.0.1', 60489) is empty
waiting for the next event

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60486)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60487)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60488)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60489)
waiting for the next event

sending "b'It whill be send.'" to ('127.0.0.1', 60486)
sending "b'It whill be send.'" to ('127.0.0.1', 60487)
sending "b'It whill be send.'" to ('127.0.0.1', 60488)
sending "b'It whill be send.'" to ('127.0.0.1', 60489)
waiting for the next event

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60486)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60487)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60488)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60489)
waiting for the next event

sending "b'in parts.'" to ('127.0.0.1', 60486)
sending "b'in parts.'" to ('127.0.0.1', 60487)
sending "b'in parts.'" to ('127.0.0.1', 60488)
sending "b'in parts.'" to ('127.0.0.1', 60489)
waiting for the next event

closing ('127.0.0.1', 60489) after reading no data
waiting for the next event

waiting for the next event

client运行结果：

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> connecting to localhost port 10000
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503) received "b'in parts.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504) received "b'in parts.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505) received "b'in parts.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506) received "b'in parts.'"

epoll_socket_server:

#!/use/local/env python
# -*- coding:utf-8 -*-

'''
http://scotdoyle.com/python-epoll-howto.html
'''

import socket, select

EOL1 = b'\n\n'
EOL2 = b'\n\r\n'
response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
response += b'Hello, world!'

serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind(('0.0.0.0', 8080))
serversocket.listen(1)
serversocket.setblocking(0)
serversocket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

epoll = select.epoll()
epoll.register(serversocket.fileno(), select.EPOLLIN)

try:
    connections = {}; requests = {}; responses = {}
    while True:
        events = epoll.poll(1)
        for fileno, event in events:
            if fileno == serversocket.fileno():
                connection, address = serversocket.accept()
                connection.setblocking(0)
                epoll.register(connection.fileno(), select.EPOLLIN)
                connections[connection.fileno()] = connection
                requests[connection.fileno()] = b''
                responses[connection.fileno()] = response
            elif event & select.EPOLLIN:
                requests[fileno] += connections[fileno].recv(1024)
                if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
                    epoll.modify(fileno, select.EPOLLOUT)
                    print('-'*40 + '\n' + requests[fileno].decode()[:-2])
            elif event & select.EPOLLOUT:
                byteswritten = connections[fileno].send(responses[fileno])
                responses[fileno] = responses[fileno][byteswritten:]
                if len(responses[fileno]) == 0:
                    epoll.modify(fileno, 0)
                    connections[fileno].shutdown(socket.SHUT_RDWR)
            elif event & select.EPOLLHUP:
                epoll.unregister(fileno)
                connections[fileno].close()
                del connections[fileno]
finally:
    epoll.unregister(serversocket.fileno())
    epoll.close()
    serversocket.close()