mirror of
https://github.com/python/cpython.git
synced 2025-08-18 15:51:23 +00:00
b15bdf7f7a
When a thread touches such an object for the first time, a new thread-local __dict__ is created, and the __init__ method is run. But a thread switch can occur here; if the other thread touches the same object, it installs another __dict__; when the first thread resumes, it updates the dictionary of the second... This is the deep cause of the failures in test_multiprocessing involving "managers" objects. Backport of r64601.
55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
import unittest
|
|
from doctest import DocTestSuite
|
|
from test import test_support
|
|
|
|
class ThreadingLocalTest(unittest.TestCase):
|
|
def test_derived(self):
|
|
# Issue 3088: if there is a threads switch inside the __init__
|
|
# of a threading.local derived class, the per-thread dictionary
|
|
# is created but not correctly set on the object.
|
|
# The first member set may be bogus.
|
|
import threading
|
|
import time
|
|
class Local(threading.local):
|
|
def __init__(self):
|
|
time.sleep(0.01)
|
|
local = Local()
|
|
|
|
def f(i):
|
|
local.x = i
|
|
# Simply check that the variable is correctly set
|
|
self.assertEqual(local.x, i)
|
|
|
|
threads= []
|
|
for i in range(10):
|
|
t = threading.Thread(target=f, args=(i,))
|
|
t.start()
|
|
threads.append(t)
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
|
|
def test_main():
|
|
suite = DocTestSuite('_threading_local')
|
|
|
|
try:
|
|
from thread import _local
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
import _threading_local
|
|
local_orig = _threading_local.local
|
|
def setUp(test):
|
|
_threading_local.local = _local
|
|
def tearDown(test):
|
|
_threading_local.local = local_orig
|
|
suite.addTest(DocTestSuite('_threading_local',
|
|
setUp=setUp, tearDown=tearDown)
|
|
)
|
|
suite.addTest(unittest.makeSuite(ThreadingLocalTest))
|
|
|
|
test_support.run_suite(suite)
|
|
|
|
if __name__ == '__main__':
|
|
test_main()
|