zoukankan      html  css  js  c++  java
  • io多路复用

    #coding=utf-8
    #IO多路复用
    #必须在非阻塞状态下才能使用socket的多路复用
    import select
    import socket
    import sys
    import Queue
    
    sever = socket.socket()
    sever.bind(('localhost',9000))
    sever.listen(1000)
    
    sever.setblocking(False)   #设置非阻塞模式
    inputs = [sever]
    outputs = []
    msg_dict = dict()
    while True:
        readable,writeable,exceptionable = select.select(inputs,outputs,inputs)   #检测到的链接放到inputs里面
        print readable,writeable,exceptionable
        for r in readable:
            if r is sever:
                con,addr=sever.accept()
                inputs.append(con)#因为新建立的链接没有发数据,因为是非阻塞,所以会出错,因此要把con加入select的监测列表
                print '有新的连接接入',con,addr
                msg_dict[con] = Queue.Queue()
            else:
                try:
                    data = r.recv(1024)
                    print '收到数据:',data
                except socket.error:
                    print '客户端断开了:',
                    inputs.remove(r)
                    continue
                r.send(data)
                msg_dict[r].put(data)
                outputs.append(r)#放入返回的链接
    
        for w in writeable:    #要返回给客户端的了链接列表
            data1 = msg_dict[w].get()
            w.send(data1)    #返回给客户端数据
            outputs.remove(w) #确保斜刺循环的时候不返回已经处理的列表
        # sever.accept()
            for e in exceptionable:
                if e in outputs:
                    outputs.remove(e)
                inputs.remove()
                del msg_dict[e]
    
    #selector模块
    import selectors
    import socket
    
    sel = selectors.DefaultSelector()
    
    
    def accept(sock, mask):
        conn, addr = sock.accept()  # Should be ready
        print('accepted', conn, 'from', addr,mask)
        conn.setblocking(False)
        sel.register(conn, selectors.EVENT_READ, read) #新连接注册read回调函数,一调用read说明conn有数据接受了
    
    
    def read(conn, mask):
        data = conn.recv(1024)  # Should be ready
        if data:
            print('echoing', repr(data), 'to', conn)  #repr??
            conn.send(data)  # Hope it won't block
        else:
            print('closing', conn)
            sel.unregister(conn)
            conn.close()
    
    
    sock = socket.socket()
    sock.bind(('localhost', 9999))
    sock.listen(100)
    sock.setblocking(False)
    sel.register(sock, selectors.EVENT_READ, accept)   #让select监听。回调函数accept。有活动就调此函数
    
    while True:
        events = sel.select() #默认阻塞,有活动连接就返回活动的连接列表
        for key, mask in events:
            callback = key.data #accept
            callback(key.fileobj, mask) #key.fileobj=  文件句柄
    

    论事件驱动与异步IO

    通常,我们写服务器处理模型的程序时,有以下几种模型:
    (1)每收到一个请求,创建一个新的进程,来处理该请求;
    (2)每收到一个请求,创建一个新的线程,来处理该请求;
    (3)每收到一个请求,放入一个事件列表,让主进程通过非阻塞I/O方式来处理请求
    上面的几种方式,各有千秋,
    第(1)中方法,由于创建新的进程的开销比较大,所以,会导致服务器性能比较差,但实现比较简单。
    第(2)种方式,由于要涉及到线程的同步,有可能会面临死锁等问题。
    第(3)种方式,在写应用程序代码时,逻辑比前面两种都复杂。
    综合考虑各方面因素,一般普遍认为第(3)种方式是大多数网络服务器采用的方式
     

    看图说话讲事件驱动模型

    在UI编程中,常常要对鼠标点击进行相应,首先如何获得鼠标点击呢?
    方式一:创建一个线程,该线程一直循环检测是否有鼠标点击,那么这个方式有以下几个缺点
    1. CPU资源浪费,可能鼠标点击的频率非常小,但是扫描线程还是会一直循环检测,这会造成很多的CPU资源浪费;如果扫描鼠标点击的接口是阻塞的呢?
    2. 如果是堵塞的,又会出现下面这样的问题,如果我们不但要扫描鼠标点击,还要扫描键盘是否按下,由于扫描鼠标时被堵塞了,那么可能永远不会去扫描键盘;
    3. 如果一个循环需要扫描的设备非常多,这又会引来响应时间的问题;
    所以,该方式是非常不好的。

    方式二:就是事件驱动模型
    目前大部分的UI编程都是事件驱动模型,如很多UI平台都会提供onClick()事件,这个事件就代表鼠标按下事件。事件驱动模型大体思路如下:
    1. 有一个事件(消息)队列;
    2. 鼠标按下时,往这个队列中增加一个点击事件(消息);
    3. 有个循环,不断从队列取出事件,根据不同的事件,调用不同的函数,如onClick()、onKeyDown()等;
    4. 事件(消息)一般都各自保存各自的处理函数指针,这样,每个消息都有独立的处理函数;

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

    让我们用例子来比较和对比一下单线程、多线程以及事件驱动编程模型。下图展示了随着时间的推移,这三种模式下程序所做的工作。这个程序有3个任务需要完成,每个任务都在等待I/O操作时阻塞自身。阻塞在I/O操作上所花费的时间已经用灰色框标示出来了。

     

    在单线程同步模型中,任务按照顺序执行。如果某个任务因为I/O而阻塞,其他所有的任务都必须等待,直到它完成之后它们才能依次执行。这种明确的执行顺序和串行化处理的行为是很容易推断得出的。如果任务之间并没有互相依赖的关系,但仍然需要互相等待的话这就使得程序不必要的降低了运行速度。

    在多线程版本中,这3个任务分别在独立的线程中执行。这些线程由操作系统来管理,在多处理器系统上可以并行处理,或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时其他线程得以继续执行。与完成类似功能的同步程序相比,这种方式更有效率,但程序员必须写代码来保护共享资源,防止其被多个线程同时访问。多线程程序更加难以推断,因为这类程序不得不通过线程同步机制如锁、可重入函数、线程局部存储或者其他机制来处理线程安全问题,如果实现不当就会导致出现微妙且令人痛不欲生的bug。

    在事件驱动版本的程序中,3个任务交错执行,但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时,注册一个回调到事件循环中,然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件,当事件到来时将它们分配给等待处理事件的回调函数。这种方式让程序尽可能的得以执行而不需要用到额外的线程。事件驱动型程序比多线程程序更容易推断出行为,因为程序员不需要关心线程安全问题。

    当我们面对如下的环境时,事件驱动模型通常是一个好的选择:

    1. 程序中有许多任务,而且…
    2. 任务之间高度独立(因此它们不需要互相通信,或者等待彼此)而且…
    3. 在等待事件到来时,某些任务会阻塞。

    当应用程序需要在任务间共享可变的数据时,这也是一个不错的选择,因为这里不需要采用同步处理。

    网络应用程序通常都有上述这些特点,这使得它们能够很好的契合事件驱动编程模型。

    此处要提出一个问题,就是,上面的事件驱动模型中,只要一遇到IO就注册一个事件,然后主程序就可以继续干其它的事情了,只到io处理完毕后,继续恢复之前中断的任务,这本质上是怎么实现的呢?哈哈,下面我们就来一起揭开这神秘的面纱。。。。

    二 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)

    注:由于signal driven IO在实际中并不常用,所以我这只提及剩下的四种IO Model。

    阻塞 I/O(blocking IO)

    在linux中,默认情况下所有的socket都是blocking,一个典型的读操作流程大概是这样:

     

    当用户进程调用了recvfrom这个系统调用,kernel就开始了IO的第一个阶段:准备数据(对于网络IO来说,很多时候数据在一开始还没有到达。比如,还没有收到一个完整的UDP包。这个时候kernel就要等待足够的数据到来)。这个过程需要等待,也就是说数据被拷贝到操作系统内核的缓冲区中是需要一个过程的。而在用户进程这边,整个进程会被阻塞(当然,是进程自己选择的阻塞)。当kernel一直等到数据准备好了,它就会将数据从kernel中拷贝到用户内存,然后kernel返回结果,用户进程才解除block的状态,重新运行起来。

    所以,blocking IO的特点就是在IO执行的两个阶段都被block了。

    非阻塞 I/O(nonblocking IO)

    linux下,可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行读操作时,流程是这个样子:

    当用户进程发出read操作时,如果kernel中的数据还没有准备好,那么它并不会block用户进程,而是立刻返回一个error。从用户进程角度讲 ,它发起一个read操作后,并不需要等待,而是马上就得到了一个结果。用户进程判断结果是一个error时,它就知道数据还没有准备好,于是它可以再次发送read操作。一旦kernel中的数据准备好了,并且又再次收到了用户进程的system call,那么它马上就将数据拷贝到了用户内存,然后返回。

    所以,nonblocking IO的特点是用户进程需要不断的主动询问kernel数据好了没有。

    I/O 多路复用( IO multiplexing)

    IO multiplexing就是我们说的select,poll,epoll,有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select,poll,epoll这个function会不断的轮询所负责的所有socket,当某个socket有数据到达了,就通知用户进程。

    当用户进程调用了select,那么整个进程会被block,而同时,kernel会“监视”所有select负责的socket,当任何一个socket中的数据准备好了,select就会返回。这个时候用户进程再调用read操作,将数据从kernel拷贝到用户进程。

    所以,I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符,而这些文件描述符(套接字描述符)其中的任意一个进入读就绪状态,select()函数就可以返回。

    这个图和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)

    inux下的asynchronous IO其实用得很少。先看一下它的流程:

    用户进程发起read操作之后,立刻就可以开始去做其它的事。而另一方面,从kernel的角度,当它受到一个asynchronous read之后,首先它会立刻返回,所以不会对用户进程产生任何block。然后,kernel会等待数据准备完成,然后将数据拷贝到用户内存,当这一切都完成之后,kernel会给用户进程发送一个signal,告诉它read操作完成了。

    总结

    blocking和non-blocking的区别

    调用blocking IO会一直block住对应的进程直到操作完成,而non-blocking IO在kernel还准备数据的情况下会立刻返回。

    synchronous IO和asynchronous IO的区别

    在说明synchronous IO和asynchronous IO的区别之前,需要先给出两者的定义。POSIX的定义是这样子的:
    - A synchronous I/O operation causes the requesting process to be blocked until that I/O operation completes;
    - An asynchronous I/O operation does not cause the requesting process to be blocked;

    两者的区别就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义,之前所述的blocking IO,non-blocking IO,IO multiplexing都属于synchronous IO。

    有人会说,non-blocking IO并没有被block啊。这里有个非常“狡猾”的地方,定义中所指的”IO operation”是指真实的IO操作,就是例子中的recvfrom这个system call。non-blocking IO在执行recvfrom这个system call的时候,如果kernel的数据没有准备好,这时候不会block进程。但是,当kernel中数据准备好的时候,recvfrom会将数据从kernel拷贝到用户内存中,这个时候进程是被block了,在这段时间内,进程是被block的。

    而asynchronous IO则不一样,当进程发起IO 操作之后,就直接返回再也不理睬了,直到kernel发送一个信号,告诉进程说IO完成。在这整个过程中,进程完全没有被block。

    各个IO Model的比较如图所示:

    通过上面的图片,可以发现non-blocking IO和asynchronous IO的区别还是很明显的。在non-blocking IO中,虽然进程大部分时间都不会被block,但是它仍然要求进程去主动的check,并且当数据准备完成以后,也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人(kernel)完成,然后他人做完后发信号通知。在此期间,用户进程不需要去检查IO操作的状态,也不需要主动的去拷贝数据。

    三 I/O 多路复用之select、poll、epoll详解

    select,poll,epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制,一个进程可以监视多个描述符,一旦某个描述符就绪(一般是读就绪或者写就绪),能够通知程序进行相应的读写操作。但select,poll,epoll本质上都是同步I/O,因为他们都需要在读写事件就绪后自己负责进行读写,也就是说这个读写过程是阻塞的,而异步I/O则无需自己负责进行读写,异步I/O的实现会负责把数据从内核拷贝到用户空间。(这里啰嗦下)

    select

    1
    select(rlist, wlist, xlist, timeout=None)

    select 函数监视的文件描述符分3类,分别是writefds、readfds、和exceptfds。调用后select函数会阻塞,直到有描述副就绪(有数据 可读、可写、或者有except),或者超时(timeout指定等待时间,如果立即返回设为null即可),函数返回。当select函数返回后,可以 通过遍历fdset,来找到就绪的描述符。

    select目前几乎在所有的平台上支持,其良好跨平台支持也是它的一个优点。select的一 个缺点在于单个进程能够监视的文件描述符的数量存在最大限制,在Linux上一般为1024,可以通过修改宏定义甚至重新编译内核的方式提升这一限制,但 是这样也会造成效率的降低。

    poll

    1
    int poll (struct pollfd *fds, unsigned int nfds, int timeout);

    不同与select使用三个位图来表示三个fdset的方式,poll使用一个 pollfd的指针实现。

    struct pollfd {
        int fd; /* file descriptor */
        short events; /* requested events to watch */
        short revents; /* returned events witnessed */
    };
    

    pollfd结构包含了要监视的event和发生的event,不再使用select“参数-值”传递的方式。同时,pollfd并没有最大数量限制(但是数量过大后性能也是会下降)。 和select函数一样,poll返回后,需要轮询pollfd来获取就绪的描述符。

    从上面看,select和poll都需要在返回后,通过遍历文件描述符来获取已经就绪的socket。事实上,同时连接的大量客户端在一时刻可能只有很少的处于就绪状态,因此随着监视的描述符数量的增长,其效率也会线性下降。

      

    epoll

    epoll是在2.6内核中提出的,是之前的select和poll的增强版本。相对于select和poll来说,epoll更加灵活,没有描述符限制。epoll使用一个文件描述符管理多个描述符,将用户关系的文件描述符的事件存放到内核的一个事件表中,这样在用户空间和内核空间的copy只需一次。

    一 epoll操作过程

    epoll操作过程需要三个接口,分别如下:

    1
    2
    3
    int epoll_create(int size);//创建一个epoll的句柄,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. int epoll_create(int size);
    创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大,这个参数不同于select()中的第一个参数,给出最大监听的fd+1的值,参数size并不是限制了epoll所能监听的描述符最大个数,只是对内核初始分配内部数据结构的一个建议
    当创建好epoll句柄后,它就会占用一个fd值,在linux下如果查看/proc/进程id/fd/,是能够看到这个fd的,所以在使用完epoll后,必须调用close()关闭,否则可能导致fd被耗尽。

    2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
    函数是对指定描述符fd执行op操作。
    - epfd:是epoll_create()的返回值。
    - op:表示op操作,用三个宏来表示:添加EPOLL_CTL_ADD,删除EPOLL_CTL_DEL,修改EPOLL_CTL_MOD。分别添加、删除和修改对fd的监听事件。
    - fd:是需要监听的fd(文件描述符)
    - epoll_event:是告诉内核需要监听什么事

    3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
    等待epfd上的io事件,最多返回maxevents个事件。
    参数events用来从内核得到事件的集合,maxevents告之内核这个events有多大,这个maxevents的值不能大于创建epoll_create()时的size,参数timeout是超时时间(毫秒,0会立即返回,-1将不确定,也有说法说是永久阻塞)。该函数返回需要处理的事件数目,如返回0表示已超时。

    #_*_coding:utf-8_*_
    __author__ = 'Alex Li'
    
    import socket, logging
    import select, errno
    
    logger = logging.getLogger("network-server")
    
    def InitLog():
        logger.setLevel(logging.DEBUG)
    
        fh = logging.FileHandler("network-server.log")
        fh.setLevel(logging.DEBUG)
        ch = logging.StreamHandler()
        ch.setLevel(logging.ERROR)
    
        formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
        ch.setFormatter(formatter)
        fh.setFormatter(formatter)
    
        logger.addHandler(fh)
        logger.addHandler(ch)
    
    
    if __name__ == "__main__":
        InitLog()
    
        try:
            # 创建 TCP socket 作为监听 socket
            listen_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        except socket.error as  msg:
            logger.error("create socket failed")
    
        try:
            # 设置 SO_REUSEADDR 选项
            listen_fd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        except socket.error as  msg:
            logger.error("setsocketopt SO_REUSEADDR failed")
    
        try:
            # 进行 bind -- 此处未指定 ip 地址,即 bind 了全部网卡 ip 上
            listen_fd.bind(('', 2003))
        except socket.error as  msg:
            logger.error("bind failed")
    
        try:
            # 设置 listen 的 backlog 数
            listen_fd.listen(10)
        except socket.error as  msg:
            logger.error(msg)
    
        try:
            # 创建 epoll 句柄
            epoll_fd = select.epoll()
            # 向 epoll 句柄中注册 监听 socket 的 可读 事件
            epoll_fd.register(listen_fd.fileno(), select.EPOLLIN)
        except select.error as  msg:
            logger.error(msg)
    
        connections = {}
        addresses = {}
        datalist = {}
        while True:
            # epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
            epoll_list = epoll_fd.poll()
    
            for fd, events in epoll_list:
                # 若为监听 fd 被激活
                if fd == listen_fd.fileno():
                    # 进行 accept -- 获得连接上来 client 的 ip 和 port,以及 socket 句柄
                    conn, addr = listen_fd.accept()
                    logger.debug("accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno()))
                    # 将连接 socket 设置为 非阻塞
                    conn.setblocking(0)
                    # 向 epoll 句柄中注册 连接 socket 的 可读 事件
                    epoll_fd.register(conn.fileno(), select.EPOLLIN | select.EPOLLET)
                    # 将 conn 和 addr 信息分别保存起来
                    connections[conn.fileno()] = conn
                    addresses[conn.fileno()] = addr
                elif select.EPOLLIN & events:
                    # 有 可读 事件激活
                    datas = ''
                    while True:
                        try:
                            # 从激活 fd 上 recv 10 字节数据
                            data = connections[fd].recv(10)
                            # 若当前没有接收到数据,并且之前的累计数据也没有
                            if not data and not datas:
                                # 从 epoll 句柄中移除该 连接 fd
                                epoll_fd.unregister(fd)
                                # server 侧主动关闭该 连接 fd
                                connections[fd].close()
                                logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
                                break
                            else:
                                # 将接收到的数据拼接保存在 datas 中
                                datas += data
                        except socket.error as  msg:
                            # 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
                            # 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
                            if msg.errno == errno.EAGAIN:
                                logger.debug("%s receive %s" % (fd, datas))
                                # 将已接收数据保存起来
                                datalist[fd] = datas
                                # 更新 epoll 句柄中连接d 注册事件为 可写
                                epoll_fd.modify(fd, select.EPOLLET | select.EPOLLOUT)
                                break
                            else:
                                # 出错处理
                                epoll_fd.unregister(fd)
                                connections[fd].close()
                                logger.error(msg)
                                break
                elif select.EPOLLHUP & events:
                    # 有 HUP 事件激活
                    epoll_fd.unregister(fd)
                    connections[fd].close()
                    logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
                elif select.EPOLLOUT & events:
                    # 有 可写 事件激活
                    sendLen = 0
                    # 通过 while 循环确保将 buf 中的数据全部发送出去
                    while True:
                        # 将之前收到的数据发回 client -- 通过 sendLen 来控制发送位置
                        sendLen += connections[fd].send(datalist[fd][sendLen:])
                        # 在全部发送完毕后退出 while 循环
                        if sendLen == len(datalist[fd]):
                            break
                    # 更新 epoll 句柄中连接 fd 注册事件为 可读
                    epoll_fd.modify(fd, select.EPOLLIN | select.EPOLLET)
                else:
                    # 其他 epoll 事件不进行处理
                    continue
    
    epoll socket echo server
    

      

    首先列一下,sellect、poll、epoll三者的区别 
    select 
    select最早于1983年出现在4.2BSD中,它通过一个select()系统调用来监视多个文件描述符的数组,当select()返回后,该数组中就绪的文件描述符便会被内核修改标志位,使得进程可以获得这些文件描述符从而进行后续的读写操作。

    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就绪通知方法。

    epoll可以同时支持水平触发和边缘触发(Edge Triggered,只告诉进程哪些文件描述符刚刚变为就绪状态,它只说一遍,如果我们没有采取行动,那么它将不会再次告知,这种方式称为边缘触发),理论上边缘触发的性能要更高一些,但是代码实现相当复杂。

    epoll同样只告知那些就绪的文件描述符,而且当我们调用epoll_wait()获得就绪文件描述符时,返回的不是实际的描述符,而是一个代表就绪描述符数量的值,你只需要去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, but is not supported under Windows.

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

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    import select
    import socket
    import sys
    import Queue
     
    # Create a TCP/IP socket
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setblocking(0)
     
    # Bind the socket to the port
    server_address = ('localhost'10000)
    print >>sys.stderr, 'starting up on %s port %s' % server_address
    server.bind(server_address)
     
    # Listen for incoming connections
    server.listen(5)

    select()方法接收并监控3个通信列表, 第一个是所有的输入的data,就是指外部发过来的数据,第2个是监控和接收所有要发出去的data(outgoing data),第3个监控错误信息,接下来我们需要创建2个列表来包含输入和输出信息来传给select().

    # Sockets from which we expect to read
    inputs = [ server ]
    
    # Sockets to which we expect to write
    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.

    # Outgoing message queues (socket:Queue)
    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 >>sys.stderr, '
    waiting for the next event'
        readable, writable, exceptional = select.select(inputs, outputs, inputs)
    

     当你把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).

    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.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    # Handle inputs
    for in readable:
     
        if is server:
            # A "readable" server socket is ready to accept a connection
            connection, client_address = s.accept()
            print >>sys.stderr, 'new connection from', client_address
            connection.setblocking(0)
            inputs.append(connection)
     
            # Give the connection a queue for data we want to send
            message_queues[connection] = Queue.Queue()

    第二种情况是这个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.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    else:
         data = s.recv(1024)
         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
             if not in outputs:
                 outputs.append(s)

    第三种情况就是这个客户端已经断开了,所以你再通过recv()接收到的数据就为空了,所以这个时候你就可以把这个跟客户端的连接关闭了。

    A readable socket without data available is from a client that has disconnected, and the stream is ready to be closed.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    else:
        # Interpret empty result as closed connection
        print >>sys.stderr, 'closing', client_address, 'after reading no data'
        # Stop listening for input on the connection
        if in outputs:
            outputs.remove(s)  #既然客户端都断开了,我就不用再给它返回数据了,所以这时候如果这个客户端的连接对象还在outputs列表中,就把它删掉
        inputs.remove(s)    #inputs中也删除掉
        s.close()           #把这个连接关闭掉
     
        # Remove message queue
        del message_queues[s]  

      

    对于writable list中的socket,也有几种状态,如果这个客户端连接在跟它对应的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.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    # Handle outputs
    for in writable:
        try:
            next_msg = message_queues[s].get_nowait()
        except Queue.Empty:
            # No messages waiting so stop checking for writability.
            print >>sys.stderr, 'output queue for', s.getpeername(), 'is empty'
            outputs.remove(s)
        else:
            print >>sys.stderr, 'sending "%s" to %s' % (next_msg, s.getpeername())
            s.send(next_msg)

    最后,如果在跟某个socket连接通信过程中出了错误,就把这个连接对象在inputsoutputsmessage_queue中都删除,再把连接关闭掉

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    # Handle "exceptional conditions"
    for in exceptional:
        print >>sys.stderr, 'handling exceptional condition for', s.getpeername()
        # Stop listening for input on the connection
        inputs.remove(s)
        if in outputs:
            outputs.remove(s)
        s.close()
     
        # Remove message queue
        del message_queues[s]

    客户端

    下面的这个是客户端程序展示了如何通过select()对socket进行管理并与多个连接同时进行交互,

    The example client program uses two sockets to demonstrate how the server with select() manages multiple connections at the same time. The client starts by connecting each TCP/IP socket to the server.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    import socket
    import sys
     
    messages = 'This is the message. ',
                 'It will be sent ',
                 'in parts.',
                 ]
    server_address = ('localhost'10000)
     
    # Create a TCP/IP socket
    socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              ]
     
    # Connect the socket to the port where the server is listening
    print >>sys.stderr, 'connecting to %s port %s' % server_address
    for in socks:
        s.connect(server_address)

    接下来通过循环通过每个socket连接给server发送和接收数据。

    
    

    Then it sends one pieces of the message at a time via each socket, and reads all responses available after writing new data.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    for message in messages:
     
        # Send messages on both sockets
        for in socks:
            print >>sys.stderr, '%s: sending "%s"' % (s.getsockname(), message)
            s.send(message)
     
        # Read responses on both sockets
        for in socks:
            data = s.recv(1024)
            print >>sys.stderr, '%s: received "%s"' % (s.getsockname(), data)
            if not data:
                print >>sys.stderr, 'closing socket', s.getsockname()

    最后服务器端的完整代码如下: 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    #_*_coding:utf-8_*_
    __author__ = 'Alex Li'
     
    import select
    import socket
    import sys
    import queue
     
    # Create a TCP/IP socket
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setblocking(False)
     
    # Bind the socket to the port
    server_address = ('localhost'10000)
    print(sys.stderr, 'starting up on %s port %s' % server_address)
    server.bind(server_address)
     
    # Listen for incoming connections
    server.listen(5)
     
    # Sockets from which we expect to read
    inputs = [ server ]
     
    # Sockets to which we expect to write
    outputs = [ ]
     
    message_queues = {}
    while inputs:
     
        # Wait for at least one of the sockets to be ready for processing
        print' waiting for the next event')
        readable, writable, exceptional = select.select(inputs, outputs, inputs)
        # Handle inputs
        for in readable:
     
            if is server:
                # A "readable" server socket is ready to accept a connection
                connection, client_address = s.accept()
                print('new connection from', client_address)
                connection.setblocking(False)
                inputs.append(connection)
     
                # Give the connection a queue for data we want to send
                message_queues[connection] = queue.Queue()
            else:
                data = s.recv(1024)
                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
                    if not in outputs:
                        outputs.append(s)
                else:
                    # Interpret empty result as closed connection
                    print('closing', client_address, 'after reading no data')
                    # Stop listening for input on the connection
                    if in outputs:
                        outputs.remove(s)  #既然客户端都断开了,我就不用再给它返回数据了,所以这时候如果这个客户端的连接对象还在outputs列表中,就把它删掉
                    inputs.remove(s)    #inputs中也删除掉
                    s.close()           #把这个连接关闭掉
     
                    # Remove message queue
                    del message_queues[s]
        # Handle outputs
        for 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)
            else:
                print'sending "%s" to %s' % (next_msg, s.getpeername()))
                s.send(next_msg)
        # Handle "exceptional conditions"
        for in exceptional:
            print('handling exceptional condition for', s.getpeername() )
            # Stop listening for input on the connection
            inputs.remove(s)
            if in outputs:
                outputs.remove(s)
            s.close()
     
            # Remove message queue
            del message_queues[s]

      

    1
    <strong style="font-family: verdana, Arial, Helvetica, sans-serif; font-size: 14px; line-height: 1.5; ">客户端的完整代码如下:</strong>
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    __author__ = 'jieli'
    import socket
    import sys
     
    messages = 'This is the message. ',
                 'It will be sent ',
                 'in parts.',
                 ]
    server_address = ('localhost'10000)
     
    # Create a TCP/IP socket
    socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              socket.socket(socket.AF_INET, socket.SOCK_STREAM),
              ]
     
    # Connect the socket to the port where the server is listening
    print >>sys.stderr, 'connecting to %s port %s' % server_address
    for in socks:
        s.connect(server_address)
     
    for message in messages:
     
        # Send messages on both sockets
        for in socks:
            print >>sys.stderr, '%s: sending "%s"' % (s.getsockname(), message)
            s.send(message)
     
        # Read responses on both sockets
        for in socks:
            data = s.recv(1024)
            print >>sys.stderr, '%s: received "%s"' % (s.getsockname(), data)
            if not data:
                print >>sys.stderr, 'closing socket', s.getsockname()
                s.close()

    Run the server in one window and the client in another. The output will look like this, with different port numbers.

    $ python ./select_echo_server.py
    starting up on localhost port 10000
    
    waiting for the next event
    new connection from ('127.0.0.1', 55821)
    
    waiting for the next event
    new connection from ('127.0.0.1', 55822)
    received "This is the message. " from ('127.0.0.1', 55821)
    
    waiting for the next event
    sending "This is the message. " to ('127.0.0.1', 55821)
    
    waiting for the next event
    output queue for ('127.0.0.1', 55821) is empty
    
    waiting for the next event
    received "This is the message. " from ('127.0.0.1', 55822)
    
    waiting for the next event
    sending "This is the message. " to ('127.0.0.1', 55822)
    
    waiting for the next event
    output queue for ('127.0.0.1', 55822) is empty
    
    waiting for the next event
    received "It will be sent " from ('127.0.0.1', 55821)
    received "It will be sent " from ('127.0.0.1', 55822)
    
    waiting for the next event
    sending "It will be sent " to ('127.0.0.1', 55821)
    sending "It will be sent " to ('127.0.0.1', 55822)
    
    waiting for the next event
    output queue for ('127.0.0.1', 55821) is empty
    output queue for ('127.0.0.1', 55822) is empty
    
    waiting for the next event
    received "in parts." from ('127.0.0.1', 55821)
    received "in parts." from ('127.0.0.1', 55822)
    
    waiting for the next event
    sending "in parts." to ('127.0.0.1', 55821)
    sending "in parts." to ('127.0.0.1', 55822)
    
    waiting for the next event
    output queue for ('127.0.0.1', 55821) is empty
    output queue for ('127.0.0.1', 55822) is empty
    
    waiting for the next event
    closing ('127.0.0.1', 55822) after reading no data
    closing ('127.0.0.1', 55822) after reading no data
    
    waiting for the next event
    

    The client output shows the data being sent and received using both sockets.

    $ python ./select_echo_multiclient.py
    connecting to localhost port 10000
    ('127.0.0.1', 55821): sending "This is the message. "
    ('127.0.0.1', 55822): sending "This is the message. "
    ('127.0.0.1', 55821): received "This is the message. "
    ('127.0.0.1', 55822): received "This is the message. "
    ('127.0.0.1', 55821): sending "It will be sent "
    ('127.0.0.1', 55822): sending "It will be sent "
    ('127.0.0.1', 55821): received "It will be sent "
    ('127.0.0.1', 55822): received "It will be sent "
    ('127.0.0.1', 55821): sending "in parts."
    ('127.0.0.1', 55822): sending "in parts."
    ('127.0.0.1', 55821): received "in parts."
    ('127.0.0.1', 55822): received "in parts."
  • 相关阅读:
    hdu-2084 数塔
    hdu-2151 Worm
    hdu-1234开门人和关门人
    hdu-1715大菲波数
    linux应用之vsftp服务的安装及配置(centos)
    linux应用之tomcat的安装及配置(centos)
    linux应用之nginx的安装及配置(centos)
    linux启动过程详解
    linux应用之Mongodb的安装及配置(centos)
    linux应用之apache服务的安装及配置(centos)
  • 原文地址:https://www.cnblogs.com/qiangayz/p/8647413.html
Copyright © 2011-2022 走看看