eventlet 模块搭建 WEB 服务器

eventlet这个强悍的东东，看到我同事的一些整理。故贴出来，你们一块儿分享~

motivation

114.113.199.11服务器上nova服务中基于python eventlet实现的定时任务(periodic_task)和 心跳任务(report_state)都是eventlet的一个greenthread实例.

目前服务器上出现了nova定时任务中某些任务执行时间过长而致使心跳任务不能准时运行的问题.

若是eventlet是一个彻底意义上的相似线程/进程的并发库的话, 不该该出现这个问题, 须要研究 eventlet的并发实现, 了解它的并发实现原理, 避免之后出现相似的问题.

分析

通过阅读eventlet源代码, 能够知道eventlet主要依赖另外2个python package:

• greenlet
• python-epoll (或其余相似的异步IO库, 如poll/select等)

主要作了3个工做:

• 封装greenlet
• 封装epoll
• 改写python标准库中相关的module, 以便支持epoll

epoll

epoll是linux实现的一个基于事件的异步IO库, 在以前相似的异步IO库poll上改进而来.

下面两个例子会演示如何用epoll将阻塞的IO操做用epoll改写为异步非阻塞. (取自官方文档)

blocking IO

    import socket 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) try: while True: connectiontoclient, address = serversocket.accept() request = b'' while EOL1 not in request and EOL2 not in request: request += connectiontoclient.recv(1024) print('-'*40 + '\n' + request.decode()[:-2]) connectiontoclient.send(response) connectiontoclient.close() finally: serversocket.close() 

这个例子实现了一个简单的监听在8080端口的web服务器. 经过一个死循环不停的接收来自8080端口 的链接, 并返回结果.

须要注意的是程序会在

connectiontoclient, address = serversocket.accept()

这一行block住, 直到获取到新的链接, 程序才会继续往下运行.

同时, 这个程序同一个时间内只能处理一个链接, 若是有不少用户同时访问8080端口, 必需要按前后 顺序依次处理这些链接, 前面一个链接成功返回后, 才会处理后面的链接.

下面的例子将用epoll将这个简单的web服务器改写为异步的方式

non-blocking IO by using epoll

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

能够看到, 例子中首先使用serversocket.setblocking(0)将socket设为异步的模式, 而后 用select.epoll()新建了一个epoll, 接着用epoll.register(serversocket.fileno(), select.EPOLLIN) 将该socket上的IO输入事件(select.EPOLLIN)注册到epoll里. 这样作了之后, 就能够将 上面例子中会在socket.accept()这步阻塞的Main Loop改写为基于异步IO事件的epoll循环了.

events = epoll.poll(1)

简单的说, 若是有不少用户同时链接到8080端口, 这个程序会同时accept()全部的socket链接, 而后经过这行代码将发生IO事件socket放到events中, 并在后面循环中处理. 没有发生IO事件的 socket不会在loop中作处理. 这样使用epoll就实现了一个简单的并发web服务器.

注意, 这里提到的并发, 和咱们一般所理解线程/进程的并发并不太同样, 更准确的说, 是 IO多路复用 .

greenlet

greentlet是python中实现咱们所谓的"Coroutine(协程)"的一个基础库.

看了下面的例子就明白了.

    from greenlet import greenlet def test1(): print 12 gr2.switch() print 34 def test2(): print 56 gr1.switch() print 78 gr1 = greenlet(test1) gr2 = greenlet(test2) gr1.switch() 

输出

1.  
   12
2.  
   56
3.  
   34

程序先分别为两个函数定义了2个greenlet: gr1和gr2.

gr1.switch()显式切换到gr1上执行, gr1中输出"12"后gr2.switch()显式切换到gr2上执行 输出56, 又gr1.switch()显式切换到gr1上, 输出34. test1()执行结束, gr1 die. 因而 test2()里的78不会输出.

能够发现greenlet仅仅是实现了一个最简单的"coroutine", 而eventlet中的greenthread是在 greenlet的基础上封装了一些更high-level的功能, 好比greenlet的调度等.

eventlet.green

从epoll的运行机制能够看出, 要使用异步IO, 必需要将相关IO操做改写成non-blocking的方式. 可是咱们用eventlet.spawn()的函数, 并无针对epoll作任何改写, 那eventlet是怎么实现 异步IO的呢?

这也是eventlet这个package最凶残的地方, 它本身重写了python标准库中IO相关的操做, 将它们 改写成支持epoll的模式, 放在eventlet.green中.

好比说, socket.accept()被改为了这样

1.  
   def accept(self):
2.  
   if self.act_non_blocking:
3.  
   return self.fd.accept()
4.  
   fd = self.fd
5.  
   while True:
6.  
   res = socket_accept(fd)
7.  
   if res is not None:
8.  
   client, addr = res
9.  
   set_nonblocking(client)
10.  
    return type(self)(client), addr
11.  
    trampoline(fd, read=True, timeout= self.gettimeout(),
12.  
    timeout_exc=socket.timeout( "timed out"))

而后在eventlet.spawn()的时候, 经过 一些高阶魔法和"huge hack", 将这些改写过得模块"patch"到spawn出的greenthread上, 从而 实现epoll的IO多路复用, 至关凶残.

eventlet并发机制分析

前面说了这么多, 这里能够分析一下eventlet的并发机制了.

eventlet的结构以下图所示

1.  
   ______________________________________ _
2.  
   | python process |
3.  
   | _________________________________ |
4.  
   | | python thread | |
5.  
   | | _____ ___________________ | |
6.  
   | | | hub | | pool | | |
7.  
   | | |_____| | _____________ | | |
8.  
   | | | | greenthread | | | |
9.  
   | | | |_____________| | | |
10.  
    | | | _____________ | | |
11.  
    | | | | greenthread | | | |
12.  
    | | | |_____________| | | |
13.  
    | | | _____________ | | |
14.  
    | | | | greenthread | | | |
15.  
    | | | |_____________| | | |
16.  
    | | | | | |
17.  
    | | | ... | | |
18.  
    | | |___________________| | |
19.  
    | | | |
20.  
    | |_________________________________| |
21.  
    | |
22.  
    | _________________________________ |
23.  
    | | python thread | |
24.  
    | |_________________________________| |
25.  
    | _________________________________ |
26.  
    | | python thread | |
27.  
    | |_________________________________| |
28.  
    | |
29.  
    | ... |
30.  
    |_______________________________________|

其中的hub和greenthread分别对应eventlet.hubs.hub和eventlet.greenthread, 本质都是 一个greenlet的实例.

hub中封装前面提到的epoll, epoll的事件循环是由hub.run()这个方法里实现. 每当用户调用 eventlet.spawn(), 就会在当前python线程的pool里产生一个新的greenthread. 因为greenthread 里的IO相关的python标准库被改写成non-blocking的模式(参考上面的socket.accept()).

每当greenthread里作IO相关的操做时, 最终都会返回到hub中的epoll循环, 而后根据epoll中的 IO事件, 调用响应的函数. 具体以下面所示.

greenthread.sleep(), 实际上也是将CPU控制权交给hub, 而后由hub调度下一个须要运行的 greenthread.

    # in eventlet.hubs.poll.Hub

    def wait(self, seconds=None): readers = self.listeners[READ] writers = self.listeners[WRITE] if not readers and not writers: if seconds: sleep(seconds) return try: presult = self.poll.poll(int(seconds * self.WAIT_MULTIPLIER)) except select.error, e: if get_errno(e) == errno.EINTR: return raise SYSTEM_EXCEPTIONS = self.SYSTEM_EXCEPTIONS for fileno, event in presult: try: if event & READ_MASK: readers.get(fileno, noop).cb(fileno) if event & WRITE_MASK: writers.get(fileno, noop).cb(fileno) if event & select.POLLNVAL: self.remove_descriptor(fileno) continue if event & EXC_MASK: readers.get(fileno, noop).cb(fileno) writers.get(fileno, noop).cb(fileno) except SYSTEM_EXCEPTIONS: raise except: self.squelch_exception(fileno, sys.exc_info()) clear_sys_exc_info() 

总结

eventlet实现的并发和咱们理解的一般意义上相似线程/进程的并发是不一样的, eventlet实现的"并发" 更准确的讲, 是 IO多路复用 . 只有在被eventlet.spawn()的函数中存在能够 支持异步IO 相关的操做, 好比说读写socket/named pipe等时, 才能不用对被调用的函数作任何修改而实现 所谓的"并发".

若是被eventlet.spawn()的函数中存在大量的CPU计算或者读写普通文件, eventlet是没法对其 实现并发操做的. 若是想要在这样的greenthread间实现相似"并发"运行的效果, 须要手动的在函数 中插入greenthread.sleep().