gevent初探

发表于 更新于 分类于 阅读次数：

0x00 前言

有很多Python语言的协程库，如：tornado、asyncio等。这些库在使用时需要使用特定的语法，如：async/await，对于非协程的代码需要改写才能以协程方式运行。

gevent是一个特殊的协程库，它可以将非协程代码以协程方式运行，从而起到提升性能的作用。本文尝试分析一下它的实现原理。

0x01 使用案例

先看以下这段代码：

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
import ctypes
import sys
import threading
import time

def gettid():
    if sys.platform == 'linux2':
        return ctypes.CDLL('libc.so.6').syscall(186)
    else:
        return ctypes.windll.kernel32.GetCurrentThreadId()

def thread_test(index):
    time0 = time.time()
    while time.time() - time0 < 1:
        print('I\'m thread %d: %d %d' % (index, threading.current_thread().ident, gettid()))
        time.sleep(0.1)

thread1 = threading.Thread(target=thread_test, args=(1,))
thread1.start()

thread2 = threading.Thread(target=thread_test, args=(2,))
thread2.start()
print('Main thread sleep')
time.sleep(2)
print('Main thread exit')

输出内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
Main thread sleep
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
I'm thread 1: 140540774946560 32347
I'm thread 2: 140540766553856 32348
Main thread exit

在这段代码前面加上以下代码：

1
2
from gevent import monkey
monkey.patch_thread()

输出如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 1: 21069936 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
I'm thread 2: 14522208 31623
Main thread sleep
Main thread exit

可以看出，在加入gevent后，输出与之前有些不同，最大的区别是：两个线程具有相同的线程ID。也就是说，这两个线程其实是跑在同一个线程里的。

还有一点需要注意的地方：Main thread sleep这句输出在前后两次执行是不同的。

0x02 原理分析

来看下patch_thread函数的实现：

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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
def patch_thread(threading=True, _threading_local=True, Event=True, logging=True,
                 existing_locks=True,
                 _warnings=None):
    if threading:
        threading_mod = __import__('threading')
        # Capture the *real* current thread object before
        # we start returning DummyThread objects, for comparison
        # to the main thread.
        orig_current_thread = threading_mod.current_thread()
    else:
        threading_mod = None
        gevent_threading_mod = None
        orig_current_thread = None

    gevent_thread_mod, thread_mod = _patch_module('thread',
                                                  _warnings=_warnings, _notify_did_subscribers=False)

    if threading:
        gevent_threading_mod, _ = _patch_module('threading',
                                                _warnings=_warnings, _notify_did_subscribers=False)

        if Event:
            from gevent.event import Event
            patch_item(threading_mod, 'Event', Event)
            # Python 2 had `Event` as a function returning
            # the private class `_Event`. Some code may be relying
            # on that.
            if hasattr(threading_mod, '_Event'):
                patch_item(threading_mod, '_Event', Event)

        if existing_locks:
            _patch_existing_locks(threading_mod)

        if logging and 'logging' in sys.modules:
            logging = __import__('logging')
            patch_item(logging, '_lock', threading_mod.RLock())
            for wr in logging._handlerList:
                # In py26, these are actual handlers, not weakrefs
                handler = wr() if callable(wr) else wr
                if handler is None:
                    continue
                if not hasattr(handler, 'lock'):
                    raise TypeError("Unknown/unsupported handler %r" % handler)
                handler.lock = threading_mod.RLock()

    if _threading_local:
        _threading_local = __import__('_threading_local')
        from gevent.local import local
        patch_item(_threading_local, 'local', local)

    def make_join_func(thread, thread_greenlet):
        from gevent.hub import sleep
        from time import time

        def join(timeout=None):
            end = None
            if threading_mod.current_thread() is thread:
                raise RuntimeError("Cannot join current thread")
            if thread_greenlet is not None and thread_greenlet.dead:
                return
            if not thread.is_alive():
                return

            if timeout:
                end = time() + timeout

            while thread.is_alive():
                if end is not None and time() > end:
                    return
                sleep(0.01)
        return join

    if threading:
        from gevent.threading import main_native_thread

        for thread in threading_mod._active.values():
            if thread == main_native_thread():
                continue
            thread.join = make_join_func(thread, None)

    if sys.version_info[:2] >= (3, 4):

        # Issue 18808 changes the nature of Thread.join() to use
        # locks. This means that a greenlet spawned in the main thread
        # (which is already running) cannot wait for the main thread---it
        # hangs forever. We patch around this if possible. See also
        # gevent.threading.
        greenlet = __import__('greenlet')

        if orig_current_thread == threading_mod.main_thread():
            main_thread = threading_mod.main_thread()
            _greenlet = main_thread._greenlet = greenlet.getcurrent()

            main_thread.join = make_join_func(main_thread, _greenlet)

            # Patch up the ident of the main thread to match. This
            # matters if threading was imported before monkey-patching
            # thread
            oldid = main_thread.ident
            main_thread._ident = threading_mod.get_ident()
            if oldid in threading_mod._active:
                threading_mod._active[main_thread.ident] = threading_mod._active[oldid]
            if oldid != main_thread.ident:
                del threading_mod._active[oldid]
        else:
            _queue_warning("Monkey-patching not on the main thread; "
                           "threading.main_thread().join() will hang from a greenlet",
                           _warnings)

    from gevent import events
    _notify_patch(events.GeventDidPatchModuleEvent('thread', gevent_thread_mod, thread_mod))
    _notify_patch(events.GeventDidPatchModuleEvent('threading', gevent_threading_mod, threading_mod))

首先，orig_current_thread变量存储了当前的线程对象，然后调用_patch_module函数patch了threading和thread模块。

_patch_module函数相关实现如下：

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
87
88
89
90
91
92
93
def __call_module_hook(gevent_module, name, module, items, _warnings):
    # This function can raise DoNotPatch on 'will'

    def warn(message):
        _queue_warning(message, _warnings)

    func_name = '_gevent_' + name + '_monkey_patch'
    try:
        func = getattr(gevent_module, func_name)
    except AttributeError:
        func = lambda *args: None


    func(module, items, warn)

def patch_item(module, attr, newitem):
    olditem = getattr(module, attr, _NONE)
    if olditem is not _NONE:
        saved.setdefault(module.__name__, {}).setdefault(attr, olditem)
    setattr(module, attr, newitem)

def patch_module(target_module, source_module, items=None,
                 _warnings=None,
                 _notify_did_subscribers=True):
    """
    patch_module(target_module, source_module, items=None)

    Replace attributes in *target_module* with the attributes of the
    same name in *source_module*.

    The *source_module* can provide some attributes to customize the process:

    * ``__implements__`` is a list of attribute names to copy; if not present,
      the *items* keyword argument is mandatory.
    * ``_gevent_will_monkey_patch(target_module, items, warn, **kwargs)``
    * ``_gevent_did_monkey_patch(target_module, items, warn, **kwargs)``
      These two functions in the *source_module* are called *if* they exist,
      before and after copying attributes, respectively. The "will" function
      may modify *items*. The value of *warn* is a function that should be called
      with a single string argument to issue a warning to the user. If the "will"
      function raises :exc:`gevent.events.DoNotPatch`, no patching will be done. These functions
      are called before any event subscribers or plugins.

    :keyword list items: A list of attribute names to replace. If
       not given, this will be taken from the *source_module* ``__implements__``
       attribute.
    :return: A true value if patching was done, a false value if patching was canceled.

    .. versionadded:: 1.3b1
    """
    from gevent import events

    if items is None:
        items = getattr(source_module, '__implements__', None)
        if items is None:
            raise AttributeError('%r does not have __implements__' % source_module)

    try:
        __call_module_hook(source_module, 'will', target_module, items, _warnings)
        _notify_patch(
            events.GeventWillPatchModuleEvent(target_module.__name__, source_module,
                                              target_module, items),
            _warnings)
    except events.DoNotPatch:
        return False

    for attr in items:
        patch_item(target_module, attr, getattr(source_module, attr))

    __call_module_hook(source_module, 'did', target_module, items, _warnings)

    if _notify_did_subscribers:
        # We allow turning off the broadcast of the 'did' event for the benefit
        # of our internal functions which need to do additional work (besides copying
        # attributes) before their patch can be considered complete.
        _notify_patch(
            events.GeventDidPatchModuleEvent(target_module.__name__, source_module,
                                             target_module)
        )

    return True

def _patch_module(name, items=None, _warnings=None, _notify_did_subscribers=True):

    gevent_module = getattr(__import__('gevent.' + name), name)
    module_name = getattr(gevent_module, '__target__', name)
    target_module = __import__(module_name)

    patch_module(target_module, gevent_module, items=items,
                 _warnings=_warnings,
                 _notify_did_subscribers=_notify_did_subscribers)

    return gevent_module, target_module

这里比较关键的逻辑有两个：

  • 调用__call_module_hook执行目标模块中的_gevent_will_monkey_patch和_gevent_did_monkey_patch方法

  • 调用patch_item将原始模块中的指定方法替换为gevent对应模块中的同名方法

可以看出，patch的关键逻辑就是由patch_item实现的，具体要patch的对象是由gevent模块中的__implements__列表指定的。

0x03 threading模块patch分析

对于threading模块，对应模块就是gevent.threading，里面实现了_gevent_will_monkey_patch方法，在hook前做一些准备工作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import threading as __threading__

def _gevent_will_monkey_patch(native_module, items, warn): # pylint:disable=unused-argument
    # Make sure the MainThread can be found by our current greenlet ID,
    # otherwise we get a new DummyThread, which cannot be joined.
    # Fixes tests in test_threading_2 under PyPy.
    main_thread = main_native_thread()
    if __threading__.current_thread() != main_thread:
        warn("Monkey-patching outside the main native thread. Some APIs "
             "will not be available. Expect a KeyError to be printed at shutdown.")
        return

    if _get_ident() not in __threading__._active:
        main_id = main_thread.ident
        del __threading__._active[main_id]
        main_thread._ident = main_thread._Thread__ident = _get_ident()
        __threading__._active[_get_ident()] = main_thread

在gevent.threading模块中要patch的列表是：

1
2
3
4
5
6
7
8
9
__implements__ = [
    'local',
    '_start_new_thread',
    '_allocate_lock',
    'Lock',
    '_get_ident',
    '_sleep',
    '_DummyThread',
]

而在gevent.thread模块中则是：

1
2
3
4
5
6
7
8
9
__implements__ = [
    'allocate_lock',
    'get_ident',
    'exit',
    'LockType',
    'stack_size',
    'start_new_thread',
    '_local'
]

我们来看下threading模块在启动线程过程中做了哪些操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
  +------------------------+
  |                        |
  | threading.Thread.start |
  |                        |
  +-----------+------------+
              |
              |
              v
+-------------+---------------+
|                             |
| threading._start_new_thread |
|                             |
+-------------+---------------+
              |
              |
              |
              v
  +-----------+-------------+
  |                         |
  | thread.start_new_thread |
  |                         |
  +-------------------------+

上面的threading._start_new_thread和thread.start_new_thread都是在gevent的__implements__列表中的。也就是说：这两个函数都被gevent hook了。

1
from gevent.thread import start_new_thread as _start_new_thread

而由于gevent.threading._start_new_thread其实就是gevent.thread.start_new_thread，所以它们其实是同一个函数。

1
2
3
4
5
6
def start_new_thread(function, args=(), kwargs=None):
    if kwargs is not None:
        greenlet = Greenlet.spawn(function, *args, **kwargs)
    else:
        greenlet = Greenlet.spawn(function, *args)
    return get_ident(greenlet)

从start_new_thread函数可以看出，启动线程逻辑已经被替换成了greenlet的实现，这也是为什么使用了patch_thread后，真实线程id相同的原因。

0x04 总结

gevent使用动态patch的方法，实现了动态将非协程库变成协程库的功能，在极少修改代码的前提下提升了程序的性能。但是，在上面线程的例子中可以看出，patch后程序的行为可能会有一些差异，所以在使用上还是需要小心。