summaryrefslogtreecommitdiffstats
path: root/Lib/test/_test_multiprocessing.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/_test_multiprocessing.py')
-rw-r--r--Lib/test/_test_multiprocessing.py53
1 files changed, 53 insertions, 0 deletions
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index ec003d8..a94eb6c 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -4455,6 +4455,59 @@ class _TestSharedMemory(BaseTestCase):
"resource_tracker: There appear to be 1 leaked "
"shared_memory objects to clean up at shutdown", err)
+ @unittest.skipIf(os.name != "posix", "resource_tracker is posix only")
+ def test_shared_memory_untracking(self):
+ # gh-82300: When a separate Python process accesses shared memory
+ # with track=False, it must not cause the memory to be deleted
+ # when terminating.
+ cmd = '''if 1:
+ import sys
+ from multiprocessing.shared_memory import SharedMemory
+ mem = SharedMemory(create=False, name=sys.argv[1], track=False)
+ mem.close()
+ '''
+ mem = shared_memory.SharedMemory(create=True, size=10)
+ # The resource tracker shares pipes with the subprocess, and so
+ # err existing means that the tracker process has terminated now.
+ try:
+ rc, out, err = script_helper.assert_python_ok("-c", cmd, mem.name)
+ self.assertNotIn(b"resource_tracker", err)
+ self.assertEqual(rc, 0)
+ mem2 = shared_memory.SharedMemory(create=False, name=mem.name)
+ mem2.close()
+ finally:
+ try:
+ mem.unlink()
+ except OSError:
+ pass
+ mem.close()
+
+ @unittest.skipIf(os.name != "posix", "resource_tracker is posix only")
+ def test_shared_memory_tracking(self):
+ # gh-82300: When a separate Python process accesses shared memory
+ # with track=True, it must cause the memory to be deleted when
+ # terminating.
+ cmd = '''if 1:
+ import sys
+ from multiprocessing.shared_memory import SharedMemory
+ mem = SharedMemory(create=False, name=sys.argv[1], track=True)
+ mem.close()
+ '''
+ mem = shared_memory.SharedMemory(create=True, size=10)
+ try:
+ rc, out, err = script_helper.assert_python_ok("-c", cmd, mem.name)
+ self.assertEqual(rc, 0)
+ self.assertIn(
+ b"resource_tracker: There appear to be 1 leaked "
+ b"shared_memory objects to clean up at shutdown", err)
+ finally:
+ try:
+ mem.unlink()
+ except OSError:
+ pass
+ resource_tracker.unregister(mem._name, "shared_memory")
+ mem.close()
+
#
# Test to verify that `Finalize` works.
#