ilusha/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython synced 2026-09-29 12:10:30 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Issue #9260: A finer-grained import lock.

Browse source 
Most of the import sequence now uses per-module locks rather than the
global import lock, eliminating well-known issues with threads and imports.
This commit is contained in:
Antoine Pitrou 2012-05-17 18:55:59 +02:00
parent 5cec9d2ae5
commit ea3eb88bca
12 changed files with 3777 additions and 3041 deletions
Download patch file Download diff file Expand all files Collapse all files

191
Lib/importlib/_bootstrap.py
View file

@ -159,6 +159,145 @@ def new_module(name):
return type(_io)(name)
# Module-level locking ########################################################
# A dict mapping module names to weakrefs of _ModuleLock instances
_module_locks = {}
# A dict mapping thread ids to _ModuleLock instances
_blocking_on = {}
class _DeadlockError(RuntimeError):
pass
class _ModuleLock:
"""A recursive lock implementation which is able to detect deadlocks
(e.g. thread 1 trying to take locks A then B, and thread 2 trying to
take locks B then A).
"""
def __init__(self, name):
self.lock = _thread.allocate_lock()
self.wakeup = _thread.allocate_lock()
self.name = name
self.owner = None
self.count = 0
self.waiters = 0
def has_deadlock(self):
# Deadlock avoidance for concurrent circular imports.
me = _thread.get_ident()
tid = self.owner
while True:
lock = _blocking_on.get(tid)
if lock is None:
return False
tid = lock.owner
if tid == me:
return True
def acquire(self):
"""
Acquire the module lock. If a potential deadlock is detected,
a _DeadlockError is raised.
Otherwise, the lock is always acquired and True is returned.
"""
tid = _thread.get_ident()
_blocking_on[tid] = self
try:
while True:
with self.lock:
if self.count == 0 or self.owner == tid:
self.owner = tid
self.count += 1
return True
if self.has_deadlock():
raise _DeadlockError("deadlock detected by %r" % self)
if self.wakeup.acquire(False):
self.waiters += 1
# Wait for a release() call
self.wakeup.acquire()
self.wakeup.release()
finally:
del _blocking_on[tid]
def release(self):
tid = _thread.get_ident()
with self.lock:
if self.owner != tid:
raise RuntimeError("cannot release un-acquired lock")
assert self.count > 0
self.count -= 1
if self.count == 0:
self.owner = None
if self.waiters:
self.waiters -= 1
self.wakeup.release()
def __repr__(self):
return "_ModuleLock(%r) at %d" % (self.name, id(self))
class _DummyModuleLock:
"""A simple _ModuleLock equivalent for Python builds without
multi-threading support."""
def __init__(self, name):
self.name = name
self.count = 0
def acquire(self):
self.count += 1
return True
def release(self):
if self.count == 0:
raise RuntimeError("cannot release un-acquired lock")
self.count -= 1
def __repr__(self):
return "_DummyModuleLock(%r) at %d" % (self.name, id(self))
# The following two functions are for consumption by Python/import.c.
def _get_module_lock(name):
"""Get or create the module lock for a given module name.
Should only be called with the import lock taken."""
lock = None
if name in _module_locks:
lock = _module_locks[name]()
if lock is None:
if _thread is None:
lock = _DummyModuleLock(name)
else:
lock = _ModuleLock(name)
def cb(_):
del _module_locks[name]
_module_locks[name] = _weakref.ref(lock, cb)
return lock
def _lock_unlock_module(name):
"""Release the global import lock, and acquires then release the
module lock for a given module name.
This is used to ensure a module is completely initialized, in the
event it is being imported by another thread.
Should only be called with the import lock taken."""
lock = _get_module_lock(name)
_imp.release_lock()
try:
lock.acquire()
except _DeadlockError:
# Concurrent circular import, we'll accept a partially initialized
# module object.
pass
else:
lock.release()
# Finder/loader utility code ##################################################
_PYCACHE = '__pycache__'
@ -264,12 +403,15 @@ def module_for_loader(fxn):
else:
module.__package__ = fullname.rpartition('.')[0]
try:
module.__initializing__ = True
# If __package__ was not set above, __import__() will do it later.
return fxn(self, module, *args, **kwargs)
except:
if not is_reload:
del sys.modules[fullname]
raise
finally:
module.__initializing__ = False
_wrap(module_for_loader_wrapper, fxn)
return module_for_loader_wrapper
@ -932,7 +1074,8 @@ def _find_module(name, path):
if not sys.meta_path:
_warnings.warn('sys.meta_path is empty', ImportWarning)
for finder in sys.meta_path:
loader = finder.find_module(name, path)
with _ImportLockContext():
loader = finder.find_module(name, path)
if loader is not None:
# The parent import may have already imported this module.
if name not in sys.modules:
@ -962,8 +1105,7 @@ def _sanity_check(name, package, level):
_ERR_MSG = 'No module named {!r}'
def _find_and_load(name, import_):
"""Find and load the module."""
def _find_and_load_unlocked(name, import_):
path = None
parent = name.rpartition('.')[0]
if parent:
@ -1009,6 +1151,19 @@ def _find_and_load(name, import_):
return module
def _find_and_load(name, import_):
"""Find and load the module, and release the import lock."""
try:
lock = _get_module_lock(name)
finally:
_imp.release_lock()
lock.acquire()
try:
return _find_and_load_unlocked(name, import_)
finally:
lock.release()
def _gcd_import(name, package=None, level=0):
"""Import and return the module based on its name, the package the call is
being made from, and the level adjustment.
@ -1021,17 +1176,17 @@ def _gcd_import(name, package=None, level=0):
_sanity_check(name, package, level)
if level > 0:
name = _resolve_name(name, package, level)
with _ImportLockContext():
try:
module = sys.modules[name]
if module is None:
message = ("import of {} halted; "
"None in sys.modules".format(name))
raise ImportError(message, name=name)
return module
except KeyError:
pass # Don't want to chain the exception
_imp.acquire_lock()
if name not in sys.modules:
return _find_and_load(name, _gcd_import)
module = sys.modules[name]
if module is None:
_imp.release_lock()
message = ("import of {} halted; "
"None in sys.modules".format(name))
raise ImportError(message, name=name)
_lock_unlock_module(name)
return module
def _handle_fromlist(module, fromlist, import_):
@ -1149,7 +1304,17 @@ def _setup(sys_module, _imp_module):
continue
else:
raise ImportError('importlib requires posix or nt')
try:
thread_module = BuiltinImporter.load_module('_thread')
except ImportError:
# Python was built without threads
thread_module = None
weakref_module = BuiltinImporter.load_module('_weakref')
setattr(self_module, '_os', os_module)
setattr(self_module, '_thread', thread_module)
setattr(self_module, '_weakref', weakref_module)
setattr(self_module, 'path_sep', path_sep)
setattr(self_module, 'path_separators', set(path_separators))
# Constants

115
Lib/importlib/test/test_locks.py Normal file
View file

@ -0,0 +1,115 @@
from importlib import _bootstrap
import time
import unittest
import weakref
from test import support
try:
import threading
except ImportError:
threading = None
else:
from test import lock_tests
LockType = _bootstrap._ModuleLock
DeadlockError = _bootstrap._DeadlockError
if threading is not None:
class ModuleLockAsRLockTests(lock_tests.RLockTests):
locktype = staticmethod(lambda: LockType("some_lock"))
# _is_owned() unsupported
test__is_owned = None
# acquire(blocking=False) unsupported
test_try_acquire = None
test_try_acquire_contended = None
# `with` unsupported
test_with = None
# acquire(timeout=...) unsupported
test_timeout = None
# _release_save() unsupported
test_release_save_unacquired = None
else:
class ModuleLockAsRLockTests(unittest.TestCase):
pass
@unittest.skipUnless(threading, "threads needed for this test")
class DeadlockAvoidanceTests(unittest.TestCase):
def run_deadlock_avoidance_test(self, create_deadlock):
NLOCKS = 10
locks = [LockType(str(i)) for i in range(NLOCKS)]
pairs = [(locks[i], locks[(i+1)%NLOCKS]) for i in range(NLOCKS)]
if create_deadlock:
NTHREADS = NLOCKS
else:
NTHREADS = NLOCKS - 1
barrier = threading.Barrier(NTHREADS)
results = []
def _acquire(lock):
"""Try to acquire the lock. Return True on success, False on deadlock."""
try:
lock.acquire()
except DeadlockError:
return False
else:
return True
def f():
a, b = pairs.pop()
ra = _acquire(a)
barrier.wait()
rb = _acquire(b)
results.append((ra, rb))
if rb:
b.release()
if ra:
a.release()
lock_tests.Bunch(f, NTHREADS).wait_for_finished()
self.assertEqual(len(results), NTHREADS)
return results
def test_deadlock(self):
results = self.run_deadlock_avoidance_test(True)
# One of the threads detected a potential deadlock on its second
# acquire() call.
self.assertEqual(results.count((True, False)), 1)
self.assertEqual(results.count((True, True)), len(results) - 1)
def test_no_deadlock(self):
results = self.run_deadlock_avoidance_test(False)
self.assertEqual(results.count((True, False)), 0)
self.assertEqual(results.count((True, True)), len(results))
class LifetimeTests(unittest.TestCase):
def test_lock_lifetime(self):
name = "xyzzy"
self.assertNotIn(name, _bootstrap._module_locks)
lock = _bootstrap._get_module_lock(name)
self.assertIn(name, _bootstrap._module_locks)
wr = weakref.ref(lock)
del lock
support.gc_collect()
self.assertNotIn(name, _bootstrap._module_locks)
self.assertIs(wr(), None)
def test_all_locks(self):
support.gc_collect()
self.assertEqual(0, len(_bootstrap._module_locks))
@support.reap_threads
def test_main():
support.run_unittest(ModuleLockAsRLockTests,
DeadlockAvoidanceTests,
LifetimeTests)
if __name__ == '__main__':
test_main()

2
Lib/pydoc.py
View file

@ -167,7 +167,7 @@ def visiblename(name, all=None, obj=None):
if name in {'__builtins__', '__doc__', '__file__', '__path__',
'__module__', '__name__', '__slots__', '__package__',
'__cached__', '__author__', '__credits__', '__date__',
'__version__', '__qualname__'}:
'__version__', '__qualname__', '__initializing__'}:
return 0
# Private names are hidden, but special names are displayed.
if name.startswith('__') and name.endswith('__'): return 1

12
Lib/test/lock_tests.py
View file

@ -247,7 +247,6 @@ class RLockTests(BaseLockTests):
# Cannot release an unacquired lock
lock = self.locktype()
self.assertRaises(RuntimeError, lock.release)
self.assertRaises(RuntimeError, lock._release_save)
lock.acquire()
lock.acquire()
lock.release()
@ -255,6 +254,17 @@ class RLockTests(BaseLockTests):
lock.release()
lock.release()
self.assertRaises(RuntimeError, lock.release)
def test_release_save_unacquired(self):
# Cannot _release_save an unacquired lock
lock = self.locktype()
self.assertRaises(RuntimeError, lock._release_save)
lock.acquire()
lock.acquire()
lock.release()
lock.acquire()
lock.release()
lock.release()
self.assertRaises(RuntimeError, lock._release_save)
def test_different_thread(self):

2
Lib/test/test_pkg.py
View file

@ -23,6 +23,8 @@ def cleanout(root):
def fixdir(lst):
if "__builtins__" in lst:
lst.remove("__builtins__")
if "__initializing__" in lst:
lst.remove("__initializing__")
return lst

19
Lib/test/test_threaded_import.py
View file

@ -12,7 +12,7 @@ import time
import shutil
import unittest
from test.support import (
verbose, import_module, run_unittest, TESTFN, reap_threads)
verbose, import_module, run_unittest, TESTFN, reap_threads, forget)
threading = import_module('threading')
def task(N, done, done_tasks, errors):
@ -187,7 +187,7 @@ class ThreadedImportTests(unittest.TestCase):
contents = contents % {'delay': delay}
with open(os.path.join(TESTFN, name + ".py"), "wb") as f:
f.write(contents.encode('utf-8'))
self.addCleanup(sys.modules.pop, name, None)
self.addCleanup(forget, name)
results = []
def import_ab():
@ -204,6 +204,21 @@ class ThreadedImportTests(unittest.TestCase):
t2.join()
self.assertEqual(set(results), {'a', 'b'})
def test_side_effect_import(self):
code = """if 1:
import threading
def target():
import random
t = threading.Thread(target=target)
t.start()
t.join()"""
sys.path.insert(0, os.curdir)
self.addCleanup(sys.path.remove, os.curdir)
with open(TESTFN + ".py", "wb") as f:
f.write(code.encode('utf-8'))
self.addCleanup(forget, TESTFN)
__import__(TESTFN)
@reap_threads
def test_main():

2
Lib/token.py
View file

@ -70,7 +70,7 @@ NT_OFFSET = 256
tok_name = {value: name
for name, value in globals().items()
if isinstance(value, int)}
if isinstance(value, int) and not name.startswith('_')}
__all__.extend(tok_name.values())
def ISTERMINAL(x):