diff options
Diffstat (limited to 'Lib/test/_test_multiprocessing.py')
| -rw-r--r-- | Lib/test/_test_multiprocessing.py | 53 |
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. # |
