Lib/test/test_threading_local.py

   1 import unittest
   2 from doctest import DocTestSuite
   3 from test import test_support
   4 import threading
   5 import weakref
   6 import gc
   7
   8 class Weak(object):
   9     pass
  10
  11 def target(local, weaklist):
  12     weak = Weak()
  13     local.weak = weak
  14     weaklist.append(weakref.ref(weak))
  15
  16 class ThreadingLocalTest(unittest.TestCase):
  17
  18     def test_local_refs(self):
  19         self._local_refs(20)
  20         self._local_refs(50)
  21         self._local_refs(100)
  22
  23     def _local_refs(self, n):
  24         local = threading.local()
  25         weaklist = []
  26         for i in range(n):
  27             t = threading.Thread(target=target, args=(local, weaklist))
  28             t.start()
  29             t.join()
  30         del t
  31
  32         gc.collect()
  33         self.assertEqual(len(weaklist), n)
  34
  35         # XXX threading.local keeps the local of the last stopped thread alive.
  36         deadlist = [weak for weak in weaklist if weak() is None]
  37         self.assertEqual(len(deadlist), n-1)
  38
  39         # Assignment to the same thread local frees it sometimes (!)
  40         local.someothervar = None
  41         gc.collect()
  42         deadlist = [weak for weak in weaklist if weak() is None]
  43         self.assertTrue(len(deadlist) in (n-1, n), (n, len(deadlist)))
  44
  45     def test_derived(self):
  46         # Issue 3088: if there is a threads switch inside the __init__
  47         # of a threading.local derived class, the per-thread dictionary
  48         # is created but not correctly set on the object.
  49         # The first member set may be bogus.
  50         import time
  51         class Local(threading.local):
  52             def __init__(self):
  53                 time.sleep(0.01)
  54         local = Local()
  55
  56         def f(i):
  57             local.x = i
  58             # Simply check that the variable is correctly set
  59             self.assertEqual(local.x, i)
  60
  61         threads= []
  62         for i in range(10):
  63             t = threading.Thread(target=f, args=(i,))
  64             t.start()
  65             threads.append(t)
  66
  67         for t in threads:
  68             t.join()
  69
  70     def test_derived_cycle_dealloc(self):
  71         # http://bugs.python.org/issue6990
  72         class Local(threading.local):
  73             pass
  74         locals = None
  75         passed = [False]
  76         e1 = threading.Event()
  77         e2 = threading.Event()
  78
  79         def f():
  80             # 1) Involve Local in a cycle
  81             cycle = [Local()]
  82             cycle.append(cycle)
  83             cycle[0].foo = 'bar'
  84
  85             # 2) GC the cycle (triggers threadmodule.c::local_clear
  86             # before local_dealloc)
  87             del cycle
  88             gc.collect()
  89             e1.set()
  90             e2.wait()
  91
  92             # 4) New Locals should be empty
  93             passed[0] = all(not hasattr(local, 'foo') for local in locals)
  94
  95         t = threading.Thread(target=f)
  96         t.start()
  97         e1.wait()
  98
  99         # 3) New Locals should recycle the original's address. Creating
 100         # them in the thread overwrites the thread state and avoids the
 101         # bug
 102         locals = [Local() for i in range(10)]
 103         e2.set()
 104         t.join()
 105
 106         self.assertTrue(passed[0])
 107
 108
 109 def test_main():
 110     suite = unittest.TestSuite()
 111     suite.addTest(DocTestSuite('_threading_local'))
 112     suite.addTest(unittest.makeSuite(ThreadingLocalTest))
 113
 114     try:
 115         from thread import _local
 116     except ImportError:
 117         pass
 118     else:
 119         import _threading_local
 120         local_orig = _threading_local.local
 121         def setUp(test):
 122             _threading_local.local = _local
 123         def tearDown(test):
 124             _threading_local.local = local_orig
 125         suite.addTest(DocTestSuite('_threading_local',
 126                                    setUp=setUp, tearDown=tearDown)
 127                       )
 128
 129     test_support.run_unittest(suite)
 130
 131 if __name__ == '__main__':
 132     test_main()