Issue #22853: Added regression test for using multiprocessing.Queue at import
time. Patch by Davin Potts.
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index d048f6f..4520c6a 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -713,6 +713,27 @@
for p in workers:
p.join()
+ def test_no_import_lock_contention(self):
+ with test.support.temp_cwd():
+ module_name = 'imported_by_an_imported_module'
+ with open(module_name + '.py', 'w') as f:
+ f.write("""if 1:
+ import multiprocessing
+
+ q = multiprocessing.Queue()
+ q.put('knock knock')
+ q.get(timeout=3)
+ q.close()
+ del q
+ """)
+
+ with test.support.DirsOnSysPath(os.getcwd()):
+ try:
+ __import__(module_name)
+ except pyqueue.Empty:
+ self.fail("Probable regression on import lock contention;"
+ " see Issue #22853")
+
def test_timeout(self):
q = multiprocessing.Queue()
start = time.time()