Gevent 调度流程解析,gevent调度流程,gevent的调度流程,
Gevent 调度流程解析,gevent调度流程,gevent的调度流程,
gevent是目前应用非常广泛的网络库,高效的轮询IO库libev加上greenlet实现的协程(coroutine),使得gevent的性能非常出色,尤其是在web应用中。本文介绍gevent的调度流程,主要包括gevent对greenlet的封装和使用,以及greenlet与libev的协作。阅读本文需要对greenlet有一定的认识,可以参考这篇文章,另外,本文分析的gevent版本为1.2,可以通过gevent.version_info查看版本号。
gevent简介:
gevent是基于协程(greenlet)的网络库,底层的事件轮询基于libev(早期是libevent),gevent的API概念和Python标准库一致(如事件,队列)。gevent有一个很有意思的东西-monkey-patch,能够使python标准库中的阻塞操作变成异步,如socket的读写。
gevent来源于eventlet,自称比后者实现更简单、API更方便且性能更好,许多开源的web服务器也使用了gevent,如gunicorn、paste,当然gevent本生也可以作为一个python web服务器使用。这篇文章对常见的wsgi server进行性能对比,gevent不管在http1.0还是http1.1都表现非常出色。下图是目前常用的http1.1标准下的表现:
gevent高效的秘诀就是greenlet和libev啦,greenlet在之前的博文有介绍,gevent对greenlet的使用比较限制,只能在两层协程之间切换,简单也不容易出错。libev使用轮训非阻塞的方式进行事件处理,比如unix下的epoll。早期gevent使用libevent,后来替换成libev,因为libev“提供更少的核心功能以求更改的效率”,这里有libev和libevent的性能对比:
greenlet回顾:
如果想了解gevent的调度流程,最重要的是对greenlet有基本的了解。下面总结一些个人认为比较重要的点:
- 每一个greenlet.greenlet实例都有一个parent(可指定,默认为创生新的greenlet.greenlet所在环境),当greenlet.greenlet实例执行完逻辑正常结束、或者抛出异常结束时,执行逻辑切回到其parent。
- 可以继承greenlet.greenlet,子类需要实现run方法,当调用greenlet.switch方法时会调用到这个run方法
在gevent中,有两个类继承了greenlet.greenlet,分别是gevent.hub.Hub和gevent.greenlet.Greenlet。后文中,如果是greenlet.greenlet这种写法,那么指的是原生的类库greentlet,如果是greenlet(或者Greenlet)那么指gevent封装后的greenlet。
greenlet调度流程:
首先,给出总结性的结论,后面再结合实例和源码一步步分析。
每个gevent线程都有一个hub,前面提到hub是greenlet.greenlet的实例。hub实例在需要的时候创生(Lazy Created),那么其parent是main greenlet。之后任何的Greenlet(注意是greenlet.greenlet的子类)实例的parent都设置成hub。hub调用libev提供的事件循环来处理Greenlet代表的任务,当Greenlet实例结束(正常或者异常)之后,执行逻辑又切换到hub。
gevent调度示例1:
我们看下面最简单的代码:
Python>>> import gevent >>> gevent.sleep(1)
上面的代码很简单,但事实上gevent的核心都包含在其中,接下来结合源码进行分析。
首先看sleep函数(gevent.hub.sleep):
def sleep(seconds=0, ref=True): hub = get_hub() loop = hub.loop if seconds <= 0: waiter = Waiter() loop.run_callback(waiter.switch) waiter.get() else: hub.wait(loop.timer(seconds, ref=ref))
首先是获取hub(第2行),然后在hub上wait这个定时器事件(第9行)。get_hub源码如下(gevent.hub.get_hub):
def get_hub(*args, **kwargs): """ Return the hub for the current thread. """ hub = _threadlocal.hub if hub is None: hubtype = get_hub_class() hub = _threadlocal.hub = hubtype(*args, **kwargs) return hub
可以看到,hub是线程内唯一的,之前也提到过greenlet是线程独立的,每个线程有各自的greenlet栈。hubtype默认就是gevent.hub.Hub,在hub的初始化函数(__init__)中,会创建loop属性,默认也就是libev的python封装。
回到sleep函数定义,hub.wait(loop.timer(seconds, ref=ref))。hub.wait函数非常关键,对于任何阻塞性操作,比如timer、io都会调用这个函数,其作用一句话概括:从当前协程切换到hub,直到watcher对应的事件就绪再从hub切换回来。wait函数源码如下(gevent.hub.Hub.wait):
def wait(self, watcher): """ Wait until the *watcher* (which should not be started) is ready. """ waiter = Waiter() unique = object() watcher.start(waiter.switch, unique) try: result = waiter.get() if result is not unique: raise InvalidSwitchError('Invalid switch into %s: %r (expected %r)' % (getcurrent(), result, unique)) finally: watcher.stop()
形参watcher就是loop.timer实例,其cython描述在corecext.pyx,我们简单理解成是一个定时器事件就行了。上面的代码中,创建了一个Waiter(gevent.hub.Waiter)对象,这个对象起什么作用呢,这个类的doc写得非常清楚:
PythonWaiter.__doc__ A low level communication utility for greenlets. Waiter is a wrapper around greenlet's ``switch()`` and ``throw()`` calls that makes them somewhat safer: * switching will occur only if the waiting greenlet is executing :meth:`get` method currently; * any error raised in the greenlet is handled inside :meth:`switch` and :meth:`throw` * if :meth:`switch`/:meth:`throw` is called before the receiver calls :meth:`get`, then :class:`Waiter` will store the value/exception. The following :meth:`get` will return the value/raise the exception
简而言之,是对greenlet.greenlet类switch 和 throw函数的分装,用来存储返回值greenlet的返回值或者捕获在greenlet中抛出的异常。我们知道,在原生的greenlet中,如果一个greenlet抛出了异常,那么该异常将会展开至其parent greenlet。
回到Hub.wait函数,第8行 watcher.start(waiter.switch, unique) 注册了一个回调,在一定时间(1s)之后调用回调函数waiter.switch。注意,waiter.switch此时并没有执行。然后第10行调用waiter.get。看看这个get函数(gevent.hub.Waiter.get):
def get(self): """If a value/an exception is stored, return/raise it. Otherwise until switch() or throw() is called.""" if self._exception is not _NONE: if self._exception is None: return self.value else: getcurrent().throw(*self._exception) else: if self.greenlet is not None: raise ConcurrentObjectUseError('This Waiter is already used by %r' % (self.greenlet, )) self.greenlet = getcurrent() # 存储当前协程,之后从hub switch回来的时候使用 try: return self.hub.switch() # switch到hub finally: self.greenlet = None
核心的逻辑在第11到15行,11行中,getcurrent获取当前的greenlet(在这个测试代码中,是main greenlet,即最原始的greenlet),将其复制给waiter.greenlet。然后13行switch到hub,在greenlet回顾章节的第二条提到,greenlet.greenlet的子类需要重写run方法,当调用子类的switch时会调用到该run方法。Hub的run方法实现如下:
def run(self): """ Entry-point to running the loop. This method is called automatically when the hub greenlet is scheduled; do not call it directly. :raises LoopExit: If the loop finishes running. This means that there are no other scheduled greenlets, and no active watchers or servers. In some situations, this indicates a programming error. """ assert self is getcurrent(), 'Do not call Hub.run() directly' while True: loop = self.loop loop.error_handler = self try: loop.run() finally: loop.error_handler = None # break the refcount cycle self.parent.throw(LoopExit('This operation would block forever', self))
loop自然是libev的事件循环。doc中提到,这个loop理论上会一直循环,如果结束,那么表明没有任何监听的事件(包括IO 定时等)。之前在Hub.wait函数中注册了定时器,那么在这个run中,如果时间到了,那么会调用定时器的callback,也就是之前的waiter.switch, 我们再来看看这个函数(gevent.hub.Waiter.switch):
def switch(self, value=None): """Switch to the greenlet if one's available. Otherwise store the value.""" greenlet = self.greenlet if greenlet is None: self.value = value self._exception = None else: assert getcurrent() is self.hub, "Can only use Waiter.switch method from the Hub greenlet" switch = greenlet.switch try: switch(value) except: self.hub.handle_error(switch, *sys.exc_info())
核心代码在第8到13行,第8行保证调用到该函数的时候一定在hub这个协程中,这是很自然的,因为这个函数一定是在Hub.run中被调用。第11行switch到waiter.greenlet这个协程,在讲解waiter.get的时候就提到了waiter.greenlet是main greenlet。注意,这里得switch会回到main greenlet被切出的地方(也就是main greenlet挂起的地方),那就是在waiter.get的第10行,整个逻辑也就恢复到main greenlet继续执行。
总结:sleep的作用很简单,触发一个阻塞的操作,导致调用hub.wait,从当前greenlet.greenlet切换至Hub,超时之后再从hub切换到之前的greenlet继续执行。通过这个例子可以知道,gevent将任何阻塞性的操作封装成一个Watcher,然后从调用阻塞操作的协程切换到Hub,等到阻塞操作完成之后,再从Hub切换到之前的协程。
gevent调度示例2:
上面这个例子,虽然能够理顺gevent的调度流程,但事实上并没有体现出gevent 协作的优势。接下来看看gevent tutorial的例子:
import gevent def foo(): print('Running in foo') gevent.sleep(0) print('Explicit context switch to foo again') def bar(): print('Explicit context to bar') gevent.sleep(0) print('Implicit context switch back to bar') gevent.joinall([ gevent.spawn(foo), gevent.spawn(bar), ]) # output Running in foo Explicit context to bar Explicit context switch to foo again Implicit context switch back to bar
从输出可以看到, foo和bar依次输出,显然是在gevent.sleep的时候发生了执行流程切换,gevent.sleep再前面已经介绍了,那么这里主要关注spawn和joinall函数。
gevent.spawn本质调用了gevent.greenlet.Greenlet的类方法spawn:
@classmethod def spawn(cls, *args, **kwargs): g = cls(*args, **kwargs) g.start() return g
这个类方法调用了Greenlet的两个函数,__init__ 和 start. init函数中最为关键的是这段代码:
def __init__(self, run=None, *args, **kwargs): greenlet.__init__(self, None, get_hub()) # 将新创生的greenlet实例的parent一律设置成hub if run is not None: self._run = run
start函数的定义也很简单(gevent.greenlet.Greenlet.start):
def start(self): """Schedule the greenlet to run in this loop iteration""" if self._start_event is None: self._start_event = self.parent.loop.run_callback(self.switch)
注册回调事件self.switch到hub.loop,注意Greenlet.switch最终会调用到Greenlet._run, 也就是spawn函数传入的callable对象(foo、bar)。这里仅仅是注册,但还没有开始事件轮询,gevent.joinall就是用来启动事件轮询并等待运行结果的。
joinall函数会一路调用到gevent.hub.iwait函数:
def iwait(objects, timeout=None, count=None): """ Iteratively yield *objects* as they are ready, until all (or *count*) are ready or *timeout* expired. """ # QQQ would be nice to support iterable here that can be generated slowly (why?) if objects is None: yield get_hub().join(timeout=timeout) return count = len(objects) if count is None else min(count, len(objects)) waiter = _MultipleWaiter() # _MultipleWaiter是Waiter的子类 switch = waiter.switch if timeout is not None: timer = get_hub().loop.timer(timeout, priority=-1) timer.start(switch, _NONE) try: for obj in objects: obj.rawlink(switch) # 这里往hub.loop注册了回调 for idx in xrange(count): print 'for in iwait', idx item = waiter.get() # 这里会切换到hub print 'come here ', item, getcurrent() waiter.clear() if item is _NONE: return yield item finally: if timeout is not None: timer.stop() for obj in objects: unlink = getattr(obj, 'unlink', None) if unlink: try: unlink(switch) except: traceback.print_exc()
然后iwait函数第23行开始的循环,逐个调用waiter.get。这里的waiter是_MultipleWaiter(Waiter)的实例,其get函数最终调用到Waiter.get。前面已经详细介绍了Waiter.get,简而言之,就是switch到hub。我们利用greenlet的tracing功能可以看到整个greenlet.greenlet的switch流程,修改后的代码如下:
import gevent import greenlet def callback(event, args): print event, args[0], '===:>>>>', args[1] def foo(): print('Running in foo') gevent.sleep(0) print('Explicit context switch to foo again') def bar(): print('Explicit context to bar') gevent.sleep(0) print('Implicit context switch back to bar') print 'main greenlet info: ', greenlet.greenlet.getcurrent() print 'hub info', gevent.get_hub() oldtrace = greenlet.settrace(callback) gevent.joinall([ gevent.spawn(foo), gevent.spawn(bar), ]) greenlet.settrace(oldtrace)
切换流程及原因见下图:
总结:gevent.spawn创建一个新的Greenlet,并注册到hub的loop上,调用gevent.joinall或者Greenlet.join的时候开始切换到hub。
本文通过两个简单的例子并结合源码分析了gevent的协程调度流程。gevent的使用非常方便,尤其是在web server中,基本上应用App什么都不用做就能享受gevent带来的好处。笔者阅读gevent源码最重要的原因在于想了解gevent对greenlet的封装和使用,greenlet很强大,强大到容易出错,而gevent保证在两层协程之间切换,值得借鉴!
参考
- http://www.cnblogs.com/xybaby/p/6337944.html
- http://www.gevent.org/
- https://pypi.python.org/pypi/greenlet
- http://software.schmorp.de/pkg/libev.html
- http://libevent.org/
- http://eventlet.net/
- http://nichol.as/benchmark-of-python-web-servers
- http://libev.schmorp.de/bench.html
- http://sdiehl.github.io/gevent-tutorial/
相关内容
- Python 中的异步编程:Asyncio,pythonasyncio, 注:哪怕连异
- Greenlet 详解,greenlet详解,主要包括以下内容:什么是
- Python协程,python协, ——蒙田《蒙田随笔全集
- python中协程,python协程, 进程: 进程是正在执行
- Python并发编程之协程/异步IO,pythonio,Python社区虽然对于
- 玩转 Python 3.5 的 await/async,awaitasync,新功能中很大一部
- Python之线程、进程和协程,python线程和协程,一般我们在
- Python 3.5 协程究竟是个啥,python3.5协程,我发现总有一些
- 玩转 Python 3.5 的 await/async,awaitasync,新功能中很大一部
- Python协程的用法和例子详解,python协程详解
评论关闭