Merge pull request #593 from gevent/pickled-tracebacks

Avoid potential reference cycles by pickling the tracebacks a Greenlet stores

gevent/_tblib.py

+# -*- coding: utf-8 -*-
+# A vendored version of part of https://github.com/ionelmc/python-tblib
+####
+# Copyright (c) 2013-2014, Ionel Cristian Mărieș
+# All rights reserved.
+
+# Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+# following conditions are met:
+
+# 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+# disclaimer.
+
+# 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following
+# disclaimer in the documentation and/or other materials provided with the distribution.
+
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+# INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+# WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+####
+
+# cpython.py
+
+"""
+Taken verbatim from Jinja2.
+
+https://github.com/mitsuhiko/jinja2/blob/master/jinja2/debug.py#L267
+"""
+#import platform # XXX: gevent cannot import platform at the top level; interferes with monkey patching
+import sys
+
+
+def _init_ugly_crap():
+    """This function implements a few ugly things so that we can patch the
+    traceback objects.  The function returned allows resetting `tb_next` on
+    any python traceback object.  Do not attempt to use this on non cpython
+    interpreters
+    """
+    import ctypes
+    from types import TracebackType
+
+    # figure out side of _Py_ssize_t
+    if hasattr(ctypes.pythonapi, 'Py_InitModule4_64'):
+        _Py_ssize_t = ctypes.c_int64
+    else:
+        _Py_ssize_t = ctypes.c_int
+
+    # regular python
+    class _PyObject(ctypes.Structure):
+        pass
+    _PyObject._fields_ = [
+        ('ob_refcnt', _Py_ssize_t),
+        ('ob_type', ctypes.POINTER(_PyObject))
+    ]
+
+    # python with trace
+    if hasattr(sys, 'getobjects'):
+        class _PyObject(ctypes.Structure):
+            pass
+        _PyObject._fields_ = [
+            ('_ob_next', ctypes.POINTER(_PyObject)),
+            ('_ob_prev', ctypes.POINTER(_PyObject)),
+            ('ob_refcnt', _Py_ssize_t),
+            ('ob_type', ctypes.POINTER(_PyObject))
+        ]
+
+    class _Traceback(_PyObject):
+        pass
+    _Traceback._fields_ = [
+        ('tb_next', ctypes.POINTER(_Traceback)),
+        ('tb_frame', ctypes.POINTER(_PyObject)),
+        ('tb_lasti', ctypes.c_int),
+        ('tb_lineno', ctypes.c_int)
+    ]
+
+    def tb_set_next(tb, next):
+        """Set the tb_next attribute of a traceback object."""
+        if not (isinstance(tb, TracebackType) and
+                (next is None or isinstance(next, TracebackType))):
+            raise TypeError('tb_set_next arguments must be traceback objects')
+        obj = _Traceback.from_address(id(tb))
+        if tb.tb_next is not None:
+            old = _Traceback.from_address(id(tb.tb_next))
+            old.ob_refcnt -= 1
+        if next is None:
+            obj.tb_next = ctypes.POINTER(_Traceback)()
+        else:
+            next = _Traceback.from_address(id(next))
+            next.ob_refcnt += 1
+            obj.tb_next = ctypes.pointer(next)
+
+    return tb_set_next
+
+tb_set_next = None
+# try:
+#     if platform.python_implementation() == 'CPython':
+#         #tb_set_next = _init_ugly_crap()
+#         tb_set_next = None
+# except Exception as exc:
+#     sys.stderr.write("Failed to initialize cpython support: {!r}".format(exc))
+# del _init_ugly_crap
+
+# __init__.py
+try:
+    from __pypy__ import tproxy
+except ImportError:
+    tproxy = None
+
+#if not tb_set_next and not tproxy:
+#    raise ImportError("Cannot use tblib. Runtime not supported.")
+
+
+from types import CodeType
+from types import TracebackType
+
+PY3 = sys.version_info[0] == 3
+
+
+class _AttrDict(dict):
+    def __getattr__(self, attr):
+        return self[attr]
+
+
+class __traceback_maker(Exception):
+    pass
+
+
+class Code(object):
+    def __init__(self, code):
+        self.co_filename = code.co_filename
+        self.co_name = code.co_name
+        self.co_nlocals = code.co_nlocals
+        self.co_stacksize = code.co_stacksize
+        self.co_flags = code.co_flags
+        self.co_firstlineno = code.co_firstlineno
+
+
+class Frame(object):
+    def __init__(self, frame):
+        # gevent: python 2.6 syntax fix
+        self.f_globals = {'__file__': frame.f_globals.get('__file__'),
+                          '__name__': frame.f_globals.get('__name__')}
+        self.f_code = Code(frame.f_code)
+
+
+class Traceback(object):
+    def __init__(self, tb):
+        self.tb_frame = Frame(tb.tb_frame)
+        self.tb_lineno = tb.tb_lineno
+        if tb.tb_next is None:
+            self.tb_next = None
+        else:
+            self.tb_next = Traceback(tb.tb_next)
+
+    def as_traceback(self):
+        if tproxy:
+            return tproxy(TracebackType, self.__tproxy_handler)
+        elif tb_set_next:
+            f_code = self.tb_frame.f_code
+            code = compile('\n' * (self.tb_lineno - 1) + 'raise __traceback_maker', self.tb_frame.f_code.co_filename, 'exec')
+            if PY3:
+                code = CodeType(
+                    0, 0,
+                    f_code.co_nlocals, f_code.co_stacksize, f_code.co_flags,
+                    code.co_code, code.co_consts, code.co_names, code.co_varnames,
+                    f_code.co_filename, f_code.co_name,
+                    code.co_firstlineno, b"",
+                    (), ()
+                )
+            else:
+                code = CodeType(
+                    0,
+                    f_code.co_nlocals, f_code.co_stacksize, f_code.co_flags,
+                    code.co_code, code.co_consts, code.co_names, code.co_varnames,
+                    f_code.co_filename.encode(), f_code.co_name.encode(),
+                    code.co_firstlineno, b"",
+                    (), ()
+                )
+
+            try:
+                exec(code, self.tb_frame.f_globals, {})
+            except:
+                tb = sys.exc_info()[2].tb_next
+                tb_set_next(tb, self.tb_next and self.tb_next.as_traceback())
+                return tb
+        else:
+            raise RuntimeError("Cannot re-create traceback !")
+
+    def __tproxy_handler(self, operation, *args, **kwargs):
+        if operation in ('__getattribute__', '__getattr__'):
+            if args[0] == 'tb_next':
+                return self.tb_next and self.tb_next.as_traceback()
+            else:
+                return getattr(self, args[0])
+        else:
+            return getattr(self, operation)(*args, **kwargs)
+
+
+# pickling_support.py
+
+
+def unpickle_traceback(tb_frame, tb_lineno, tb_next):
+    ret = object.__new__(Traceback)
+    ret.tb_frame = tb_frame
+    ret.tb_lineno = tb_lineno
+    ret.tb_next = tb_next
+    return ret.as_traceback()
+
+
+def pickle_traceback(tb):
+    return unpickle_traceback, (Frame(tb.tb_frame), tb.tb_lineno, tb.tb_next and Traceback(tb.tb_next))
+
+
+def install():
+    try:
+        import copy_reg
+    except ImportError:
+        import copyreg as copy_reg
+
+    copy_reg.pickle(TracebackType, pickle_traceback)
+
+# Added by gevent
+
+# We have to defer the initilization, and especially the import of platform,
+# until runtime.
+
+
+def _import_dump_load():
+    global dumps
+    global loads
+    try:
+        import cPickle as pickle
+    except ImportError:
+        import pickle
+    dumps = pickle.dumps
+    loads = pickle.loads
+
+dumps = loads = None
+
+_installed = False
+
+
+def _init():
+    global _installed
+    global tb_set_next
+    if _installed:
+        return
+
+    _installed = True
+    import platform
+    try:
+        if platform.python_implementation() == 'CPython':
+            tb_set_next = _init_ugly_crap()
+    except Exception as exc:
+        sys.stderr.write("Failed to initialize cpython support: {!r}".format(exc))
+
+    try:
+        from __pypy__ import tproxy
+    except ImportError:
+        tproxy = None
+
+    if not tb_set_next and not tproxy:
+        raise ImportError("Cannot use tblib. Runtime not supported.")
+    _import_dump_load()
+    install()
+
+
+def dump_traceback(tb):
+    _init()
+    return dumps(tb)
+
+
+def load_traceback(s):
+    _init()
+    return loads(s)

gevent/greenlet.py

@@ -12,6 +12,8 @@ from gevent.hub import iwait
 from gevent.hub import reraise
 from gevent.hub import wait
 from gevent.timeout import Timeout
+from gevent._tblib import dump_traceback
+from gevent._tblib import load_traceback
 from collections import deque
 
 
@@ -120,15 +122,7 @@ class Greenlet(greenlet):
         return deque()
 
     def _raise_exception(self):
-        reraise(*self._exc_info)
-
-    def _exc_clear(self):
-        """Throw away the traceback associated with the exception on this object.
-
-        Call this to resolve any reference cycles.
-        """
-        if self._exc_info:
-            self._exc_info = (self._exc_info[0], self._exc_info[1], None)
+        reraise(self._exc_info[0], self._exc_info[1], load_traceback(self._exc_info[2]))
 
     @property
     def loop(self):
@@ -357,7 +351,7 @@ class Greenlet(greenlet):
             self._report_result(exc_info[1])
             return
 
-        self._exc_info = exc_info
+        self._exc_info = exc_info[0], exc_info[1], dump_traceback(exc_info[2])
 
         if self._links and not self._notifier:
             self._notifier = self.parent.loop.run_callback(self._notify_links)
@@ -445,19 +439,11 @@ def _kill(greenlet, exception, waiter):
 def joinall(greenlets, timeout=None, raise_error=False, count=None):
     if not raise_error:
         wait(greenlets, timeout=timeout, count=count)
-        for g in greenlets:
-            if hasattr(g, '_exc_clear'):
-                g._exc_clear()
     else:
         for obj in iwait(greenlets, timeout=timeout, count=count):
             if getattr(obj, 'exception', None) is not None:
                 if hasattr(obj, '_raise_exception'):
-                    try:
                     obj._raise_exception()
-                    finally:
-                        for g in greenlets:
-                            if hasattr(g, '_exc_clear'):
-                                g._exc_clear()
                 else:
                     raise obj.exception
 

gevent/pool.py

@@ -241,7 +241,6 @@ class IMapUnordered(Greenlet):
             self.queue.put(greenlet.value)
         else:
             self.queue.put(Failure(greenlet.exception))
-            greenlet._exc_clear()
         if self.ready() and self.count <= 0 and not self.finished:
             self.queue.put(Failure(StopIteration))
             self.finished = True
@@ -251,7 +250,6 @@ class IMapUnordered(Greenlet):
             return
         if not self.successful():
             self.queue.put(Failure(self.exception))
-            self._exc_clear()
             self.finished = True
             return
         if self.count <= 0:
@@ -317,7 +315,6 @@ class IMap(Greenlet):
             self.queue.put((greenlet.index, greenlet.value))
         else:
             self.queue.put((greenlet.index, Failure(greenlet.exception)))
-            greenlet._exc_clear()
         if self.ready() and self.count <= 0 and not self.finished:
             self.maxindex += 1
             self.queue.put((self.maxindex, Failure(StopIteration)))
@@ -329,7 +326,6 @@ class IMap(Greenlet):
         if not self.successful():
             self.maxindex += 1
             self.queue.put((self.maxindex, Failure(self.exception)))
-            self._exc_clear()
             self.finished = True
             return
         if self.count <= 0:

greentest/test__all__.py

@@ -33,7 +33,8 @@ NOT_IMPLEMENTED = {
 COULD_BE_MISSING = {
     'socket': ['create_connection', 'RAND_add', 'RAND_egd', 'RAND_status']}
 
-NO_ALL = ['gevent.threading', 'gevent._util', 'gevent._socketcommon', 'gevent._fileobjectcommon']
+NO_ALL = ['gevent.threading', 'gevent._util', 'gevent._socketcommon', 'gevent._fileobjectcommon',
+          'gevent._tblib']
 
 
 class Test(unittest.TestCase):

greentest/test__event.py

@@ -96,7 +96,6 @@ class TestAsyncResultAsLinkTarget(greentest.TestCase):
         self.assertRaises(greentest.ExpectedException, s1.get)
         assert gevent.with_timeout(DELAY, s2.get, timeout_value=X) is X
         self.assertRaises(greentest.ExpectedException, s3.get)
-        g._exc_clear()
 
 
 class TestEvent_SetThenClear(greentest.TestCase):

greentest/test__exc_info.py

@@ -46,7 +46,6 @@ class Test(greentest.TestCase):
             except Exception:
                 ex = sys.exc_info()[1]
                 assert ex is error, (ex, error)
-            g._exc_clear()
 
     def test2(self):
         timer = gevent.get_hub().loop.timer(0)

greentest/test__greenlet.py

@@ -67,7 +67,6 @@ class TestLink(greentest.TestCase):
         event = AsyncResult()
         p.link(event)
         self.assertRaises(err, event.get)
-        p._exc_clear()
 
         for i in range(3):
             event2 = AsyncResult()
@@ -239,7 +238,6 @@ class TestRaise_link(LinksTestCase):
         assert not callback_flag, callback_flag
 
         self.check_timed_out(*xxxxx)
-        p._exc_clear()
 
     def test_raise(self):
         p = self.p = gevent.spawn(lambda: getcurrent().throw(ExpectedError('test_raise')))
@@ -277,8 +275,7 @@ class TestStuff(greentest.TestCase):
         self.assertRaises(ExpectedError, gevent.joinall, [x, y], raise_error=True)
         self.assertRaises(ExpectedError, gevent.joinall, [y], raise_error=True)
         x.join()
-        x._exc_clear()
-        y._exc_clear()
+    test_wait_error.ignore_leakcheck = True
 
     def test_joinall_exception_order(self):
         # if there're several exceptions raised, the earliest one must be raised by joinall
@@ -292,6 +289,7 @@ class TestStuff(greentest.TestCase):
         except ExpectedError as ex:
             assert 'second' in str(ex), repr(str(ex))
         gevent.joinall([a, b])
+    test_joinall_exception_order.ignore_leakcheck = True
 
     def test_joinall_count_raise_error(self):
         # When joinall is asked not to raise an error, the 'count' param still
@@ -346,7 +344,6 @@ class TestStuff(greentest.TestCase):
         p.link(listener3)
         sleep(DELAY * 10)
         assert results in [[10, 20], [20, 10]], results
-        p._exc_clear()
 
     class Results(object):
 
@@ -546,7 +543,6 @@ class TestBasic(greentest.TestCase):
         assert g.value is None  # not changed
         assert g.exception.myattr == 5
         assert link_test == [g], link_test
-        g._exc_clear()
 
     def _assertKilled(self, g):
         assert not g
@@ -556,7 +552,6 @@ class TestBasic(greentest.TestCase):
         assert g.successful(), (repr(g), g.value, g.exception)
         assert isinstance(g.value, gevent.GreenletExit), (repr(g), g.value, g.exception)
         assert g.exception is None
-        g._exc_clear()
 
     def assertKilled(self, g):
         self._assertKilled(g)

greentest/test__server.py

@@ -377,11 +377,6 @@ class TestPoolSpawn(TestDefaultSpawn):
 
     test_pool_full.error_fatal = False
 
-    # XXX: Travis CI is reporting a leak test failure with this method
-    # but so far I can't reproduce it locally. Attempting a smash-and-grab
-    # fix
-    test_pool_full.ignore_leakcheck = True
-
 
 class TestNoneSpawn(TestCase):
 

greentest/test__socket_close.py

@@ -36,7 +36,6 @@ class Test(greentest.TestCase):
             self.assertEqual(receiver.exception.errno, socket.EBADF)
         finally:
             receiver.kill()
-            receiver._exc_clear()
 
     def test_recv_twice(self):
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

greentest/test__systemerror.py

@@ -64,7 +64,6 @@ class TestSpawn(Test):
     def tearDown(self):
         gevent.sleep(0.0001)
         assert self.x.dead, self.x
-        self.x._exc_clear()
 
     def start(self, *args):
         self.x = gevent.spawn(*args)