首页 > 代码库 > Gevent的协程实现原理
Gevent的协程实现原理
之前之所以看greenlet的代码实现,主要就是想要看看gevent库的实现代码。。。然后知道了gevent的协程是基于greenlet来实现的。。。所以就又先去看了看greenlet的实现。。。
这里就不说greenlet的具体实现了,关键就是栈数据的复制拷贝,栈指针的位移。。。
因为gevent带有自己的I/O以及定时循环,所以它对greenlet又加了一层的扩展。。。
这里我们用如下的代码来举例子,然后再来具体的分析gevent是如何扩展greenlet的吧:
import gevent def hello(fname): print "hello : ", fname gevent.sleep(0) print "12321 : ", fname task1 = gevent.spawn(hello, "fjs1") task2 = gevent.spawn(hello, "fjs2") task1.join()
这段代码的输出如下:
嗯,那么闲来看看spawn方法是如何创建协程的吧:
#类方法,这个说白了gevent提供的一层构造 @classmethod def spawn(cls, *args, **kwargs): """Return a new :class:`Greenlet` object, scheduled to start. The arguments are passed to :meth:`Greenlet.__init__`. """ g = cls(*args, **kwargs) #先构造greenlet对象 g.start() #调用start方法,相当于在hub对象的loop上注册回调,这个回调的作用就是调用当前greenlet的switch切换到这个greenlet的执行 return g
这个方法是一个类方法,用于创建一个Greenlet,不过这个要注意,当前这个greenlet已经不是前面提到的greenlet库中定义的那样了,其做了一层简单的扩展。。。
来看看构造函数:
#继承了greenlet,相当于是过扩展了一些功能 class Greenlet(greenlet): """A light-weight cooperatively-scheduled execution unit.""" def __init__(self, run=None, *args, **kwargs): hub = get_hub() greenlet.__init__(self, parent=hub) #这里将所有创建的greenlet对象的parent都指向了这个唯一的hub对象 if run is not None: self._run = run #记录run信息 self.args = args self.kwargs = kwargs self._links = deque() self.value = http://www.mamicode.com/None>这里直接继承了greenlet库中greenlet的定义,然后在构造函数中有比较重要的地方,可以看到,所有构造出来的协程的parent都将会指向一个名字叫hub的主协程。。。
这个很关键,它就是整个gevent的主循环协程,所有创建的业务协程的运行都要依赖于它的调度和管理。。。
好了,在上面spawn过程中,最后还调用了start方法启动协程,那么我们来看看这个方法的定义吧:
#其实这个主要是在hub对象的loop上面挂起一个要执行的回调,而这个回调的功能就是切换到这个greenlet的运行 def start(self): """Schedule the greenlet to run in this loop iteration""" if self._start_event is None: #其实这个只是在hub的loop上面挂起一个回调,然后在hub的loop里面会执行这个回调 self._start_event = self.parent.loop.run_callback(self.switch) #在hub对象的loop里面调用当前greenlet的switch回调,开始run方法的执行代码还是很简单,其实无非就是在parent也就是hub的loop上面注册了一个回调,而这个回调就是当前这个协程的switch方法,。。那么等到这个回调被执行的时候,那么也就是开始这个协程的执行的时候。。。
这里我们先不去看hub以及它的loop的实现。。。就先将其理解为主循环,管理所有的回调,定时,以及I/O事件。。。
嗯,接下来来看看join方法的实现吧。。熟悉多线程的都知道,在多线程环境下,join方法就是阻塞当前线程,直到join的目的线程返回了为止。。。当然这里就不是线程了,变成了协程。。。。
来看看这个join方法的代码吧:
#将当前运行环境挂起,知道join的greenlet运行完了 def join(self, timeout=None): """Wait until the greenlet finishes or *timeout* expires. Return ``None`` regardless. """ if self.ready(): #如果都已经跑完了,那么直接返回吧 return else: switch = getcurrent().switch #获取当前greenlet的switch self.rawlink(switch) #注册当前环境greenlet的回调,那么以后在这个要等待的greenlet执行完后,将会回调这个 try: t = Timeout.start_new(timeout) #创建一个timer对象 try: result = self.parent.switch() #停止当前环境greenlet的执行,调度hub执行 assert result is self, 'Invalid switch into Greenlet.join(): %r' % (result, ) finally: t.cancel() #取消timeout except Timeout: self.unlink(switch) #在挂起的回调中去除 if sys.exc_info()[1] is not t: raise except: self.unlink(switch) raise首先调用getCurrent方法来获取当前环境的协程,然后获取它的switch方法,将其放置到要join的协程的回调队列里面,当这个要join的协程运行完了之后,将会调用这些回调,这样,就可以恢复当前协程的运行了。。。
在下面我们可以看到,调用了parent也就是hub的switch方法,切换到hub的执行,这个里面将会开始要join的协程的执行,这里并不是直接切换到join的协程的运行。。这点需要注意。。。
另外,gevent自己的greenlet的定义加入了run方法,也就是每次执行都将会从这里开始。。。代码如下:
#当前greenlet的执行部分,其实就是调用传进来的函数,然后运行完了之后再调用那些挂起的回调 def run(self): try: if self._start_event is None: self._start_event = _dummy_event else: self._start_event.stop() try: result = self._run(*self.args, **self.kwargs) #执行传进来的函数 except: self._report_error(sys.exc_info()) return self._report_result(result) #这个主要是用于执行挂起的回调 finally: self.__dict__.pop('_run', None) self.__dict__.pop('args', None) self.__dict__.pop('kwargs', None)
在执行完了之后,将会调用_report_result方法来执行所有挂在这个协程上面的回调函数,这样对于上面join挂起的回调,就会在这里得到执行,从而让join方法返回继续执行,这样join方法的实现也就比较的清楚了。。其实还算是比较简单的。。。另外对于如何运行挂起在这个协程上的回调,例如join的回调,还是比较有讲究的,并不是立即在当前协程中运行,而是在hub的loop上挂起一个回调,嗯,代码如下:#这个主要是为了在hub的loop中挂起回调,用于执行当前这个greenlet所有挂起的回调 #这里也不是立即执行这些在这个greenlet上面挂起的回调,而是执行继续挂到loop的回调上面去,这样可以让当前协程尽快返回 #而且如果就在当前协程运行这些回调会出问题,因为如果回调带有别的协程的switch方法,那么switch之后,就再也回不到这个协程继续运行别的回调了 #而在loop上面执行这些回调,也就是hub上,运行这些回调,即使切换到别的协程,以后也会迟早回到hub上继续执行,所以能保证回调能全部运行完。。 def _report_result(self, result): self._exception = None self.value = http://www.mamicode.com/result>至于为什么这么大费周章,上面的注释应该说的很清楚了吧。。。
好了,接下来再来分析一下sleep的实现,代码如下:
#其实sleep的主要目的就是将当前的执行切换出去,回到hub的主循环 def sleep(seconds=0, ref=True): """Put the current greenlet to sleep for at least *seconds*. *seconds* may be specified as an integer, or a float if fractional seconds are desired. If *ref* is false, the greenlet running sleep() will not prevent gevent.wait() from exiting. """ hub = get_hub() #获取hub对象 loop = hub.loop #获取hub的loop对象 if seconds <= 0: #如果这里并没有时间 waiter = Waiter() #创建waiter对象,主要是为了维护当前greenlet与hub之间的切换 loop.run_callback(waiter.switch) #在loop上面挂起一个回调,其实就是在loop中再恢复当前sleep的greenlet的执行 waiter.get() #在这个里面最主要的功能就是记录当前的greenlet对象,然后将栈切换到hub上面执行 else: hub.wait(loop.timer(seconds, ref=ref)) #带定时的wait其实这里分为了两种种类,就是在sleep的时候传入的超时时间,小于等于0的以及大于0的。。。
对于sleep操作,如果是在多线程的环境里,例如java的sleep,其实就是阻塞当前的线程,这样子jvm会调度别的线程的运行,而对于gevent,其实更多的是可以理解为当前协程主动的放弃CPU资源,等到以后再运行。。。
首先来看看对于超时小于等于零的,其实原理很简单,就是进行switch,切换到hub协程的执行,并且在hub的loop上面注册一个回调,用于切换回到当前协程的执行。。。
这里有一点需要的注意的就是,并没有直接在代码中体现switch的操作,而是多了一个waiter对象。。。然后在loop上面注册的回调是waiter的switch方法,然后调用了waiter对象的get方法。。。
这里看gevent的注释才知道,waiter对象可以理解为gevent封装的协程之间的协作工具,具体的协程之间的切换都由waiter来做,避免让用户自己的代码涉及到switch操作,因为这样子很容易出错。。。我们来看看waiter的定义吧:
#其实这个对象只是为了维护用户greenlet与hub之间的切换关系 #将会在hub里面注册当前waiter对象的switch方法作为回调,然后在hub的loop里面将会执行这个回调 class Waiter(object): """A low level communication utility for greenlets. 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. The :meth:`switch` and :meth:`throw` methods must only be called from the :class:`Hub` greenlet. The :meth:`get` method must be called from a greenlet other than :class:`Hub`. >>> result = Waiter() >>> timer = get_hub().loop.timer(0.1) >>> timer.start(result.switch, 'hello from Waiter') >>> result.get() # blocks for 0.1 seconds 'hello from Waiter' If switch is called before the greenlet gets a chance to call :meth:`get` then :class:`Waiter` stores the value. >>> result = Waiter() >>> timer = get_hub().loop.timer(0.1) >>> timer.start(result.switch, 'hi from Waiter') >>> sleep(0.2) >>> result.get() # returns immediatelly without blocking 'hi from Waiter' .. warning:: This a limited and dangerous way to communicate between greenlets. It can easily leave a greenlet unscheduled forever if used incorrectly. Consider using safer :class:`Event`/:class:`AsyncResult`/:class:`Queue` classes. """ __slots__ = ['hub', 'greenlet', 'value', '_exception'] def __init__(self, hub=None): if hub is None: self.hub = get_hub() #获取顶层hub对象 else: self.hub = hub self.greenlet = None self.value = http://www.mamicode.com/None>这个代码应很好理解,而且注释都说的很清楚。。。比较重要的就是get方法,这个方法将会保存当前执行的协程,然后切换到hub的执行,对于switch方法,将会切换回刚开始的协程的执行。。
好了,上面介绍了sleep不带超时的实现。。
接下来来看看带超时的实现:
hub.wait(loop.timer(seconds, ref=ref)) #带定时的wait这里首先创建了一个timer对象,这个可以理解为在loop上面注册了一个超时,接着看代码:
#用于在loop上面注册watcher并等待 def wait(self, watcher): waiter = Waiter() #首先创建一个waiter对象 unique = object() watcher.start(waiter.switch, unique) #当watcher超时的时候将会调用waiter的switch方法 try: result = waiter.get() #调用waiter的get方法,主要是让将当前调用sleep的greenlet切换出去,然后切换到hub的运行 assert result is unique, 'Invalid switch into %s: %r (expected %r)' % (getcurrent(), result, unique) finally: watcher.stop()依然是创建waiter对象,以及它的get方法,不过这里要注意的是,将waiter的switch回调是注册到刚刚创建的timer对象上的,而不是直接注册到loop上面,这样待会timer超时的时候将会调用回调,恢复sleep的协程的执行。。
好了,这里gevent的大体上协程,以及切换关系都差不多了。。。。
Gevent的协程实现原理