patch threading._DummyThread to avoid leak in threading._active.

Thanks to Wil Tan (#169).

- add gevent.threading monkey-helper module.
- add test__threading.py which also checks that patch_all() does not init the hub.
- update test__all__.py w.r.t gevent.threading
- patched_tests_setup.py: ignore tests in test_threading.py that depend on threading._active not being cleaned up for dummy threads
- update test_threading_2.py

gevent/monkey.py

@@ -101,25 +101,15 @@ def patch_thread(threading=True, _threading_local=True, Event=False):
     If *_threading_local* is true (the default), also patch ``_threading_local.local``.
     """
     patch_module('thread')
-    from gevent.local import local
     if threading:
-        from gevent import thread as green_thread
+        patch_module('threading')
         threading = __import__('threading')
-        # QQQ Note, that importing threading results in get_hub() call.
-        # QQQ Would prefer not to import threading at all if it's not used;
-        # QQQ that should be possible with import hooks
-        threading.local = local
-        threading._start_new_thread = green_thread.start_new_thread
-        threading._allocate_lock = green_thread.allocate_lock
-        threading.Lock = green_thread.allocate_lock
-        threading._get_ident = green_thread.get_ident
-        from gevent.hub import sleep
-        threading._sleep = sleep
         if Event:
             from gevent.event import Event
             threading.Event = Event
     if _threading_local:
         _threading_local = __import__('_threading_local')
+        from gevent.local import local
         _threading_local.local = local
 
 

gevent/threading.py

+from __future__ import absolute_import
+
+
+__implements__ = ['local',
+                  '_start_new_thread',
+                  '_allocate_lock',
+                  'Lock',
+                  '_get_ident',
+                  '_sleep',
+                  '_DummyThread']
+
+
+import threading as __threading__
+_DummyThread_ = __threading__._DummyThread
+from gevent.local import local
+from gevent.thread import start_new_thread as _start_new_thread, allocate_lock as _allocate_lock, get_ident as _get_ident
+from gevent.hub import sleep as _sleep, getcurrent
+Lock = _allocate_lock
+
+
+def _cleanup(g):
+    __threading__._active.pop(id(g))
+
+
+class _DummyThread(_DummyThread_):
+    # instances of this will cleanup its own entry
+    # in ``threading._active``
+
+    def __init__(self):
+        _DummyThread_.__init__(self)
+        g = getcurrent()
+        rawlink = getattr(g, 'rawlink', None)
+        if rawlink is not None:
+            rawlink(_cleanup)

greentest/patched_tests_setup.py

 import sys
-import os
 import re
 
 # By default, test cases are expected to switch and emit warnings if there was none
@@ -132,6 +131,12 @@ disabled_tests = \
     # this uses cookielib which we don't care about
 
     , 'test_thread.ThreadRunningTests.test__count'
+    # XXX
+
+    , 'test_threading.ThreadedTests.test_ident_of_no_threading_threads'
+    , 'test_threading.ThreadedTests.test_foreign_thread'
+    # this assert that dummy thread id is in threading._active
+    # however, we clean that up
 ]
 
 # if 'signalfd' in os.environ.get('GEVENT_BACKEND', ''):

greentest/test__all__.py

@@ -11,7 +11,8 @@ MAPPING = {'gevent.local': '_threading_local',
            'gevent.ssl': 'ssl',
            'gevent.thread': 'thread',
            'gevent.subprocess': 'subprocess',
-           'gevent.os': 'os'}
+           'gevent.os': 'os',
+           'gevent.threading': 'threading'}
 
 
 class ANY(object):
@@ -24,16 +25,22 @@ NOT_IMPLEMENTED = {
     'socket': ['CAPI'],
     'thread': ['allocate', 'exit_thread', 'interrupt_main', 'start_new'],
     'select': ANY,
-    'os': ANY}
+    'os': ANY,
+    'threading': ANY}
 
 COULD_BE_MISSING = {
     'socket': ['create_connection', 'RAND_add', 'RAND_egd', 'RAND_status']}
 
+NO_ALL = ['gevent.threading']
+
 
 class Test(unittest.TestCase):
 
     def check_all(self):
         "Check that __all__ is present and does not contain invalid entries"
+        if not hasattr(self.module, '__all__'):
+            assert self.modname in NO_ALL
+            return
         names = {}
         exec ("from %s import *" % self.modname) in names
         names.pop('__builtins__', None)
@@ -97,7 +104,7 @@ class Test(unittest.TestCase):
         """Check that __all__ (or dir()) of the corresponsing stdlib is a subset of __all__ of this module"""
         missed = []
         for name in self.stdlib_all:
-            if name not in self.module.__all__:
+            if name not in getattr(self.module, '__all__', []):
                 missed.append(name)
 
         # handle stuff like ssl.socket and ssl.socket_error which have no reason to be in gevent.ssl.__all__

greentest/test__threading.py

+from gevent import monkey; monkey.patch_all()
+import gevent.hub
+
+# check that the locks initialized by 'threading' did not init the hub
+assert gevent.hub._get_hub() is None, 'monkey.patch_all() should not init hub'
+
+import gevent
+import greentest
+import threading
+
+
+def helper():
+    threading.currentThread()
+    gevent.sleep(0.2)
+
+
+class Test(greentest.TestCase):
+
+    def test(self):
+        before = len(threading._active)
+        g = gevent.spawn(helper)
+        gevent.sleep(0.1)
+        self.assertEqual(len(threading._active), before + 1)
+        g.join()
+        self.assertEqual(len(threading._active), before)
+
+
+if __name__ == '__main__':
+    greentest.main()

greentest/test_threading_2.py

@@ -144,7 +144,9 @@ class ThreadTests(unittest.TestCase):
             done.wait()
             self.assertFalse(ident[0] is None)
             # Kill the "immortal" _DummyThread
-            del threading._active[ident[0]]
+            # in gevent, we already cleaned that up
+            #del threading._active[ident[0]]
+            assert ident[0] not in threading._active
 
     # run with a small(ish) thread stack size (256kB)
     def test_various_ops_small_stack(self):
@@ -185,10 +187,12 @@ class ThreadTests(unittest.TestCase):
         tid = thread.start_new_thread(f, (mutex,))
         # Wait for the thread to finish.
         mutex.acquire()
-        self.assert_(tid in threading._active)
-        self.assert_(isinstance(threading._active[tid],
-                                threading._DummyThread))
-        del threading._active[tid]
+        # in gevent, we clean up the entry, so the following fails:
+        #self.assert_(tid in threading._active)
+        #self.assert_(isinstance(threading._active[tid],
+        #                        threading._DummyThread))
+        #del threading._active[tid]
+        self.assert_(tid not in threading._active)
 
     # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
     # exposed at the Python level.  This test relies on ctypes to get at it.