Method test_resource_tracker

Lib/test/_test_multiprocessing.py:6197–6267

(self)

class TestResourceTracker(unittest.TestCase):

    def test_resource_tracker(self):
        #
        # Check that killing process does not leak named semaphores
        #
        cmd = '''if 1:
            import time, os
            import multiprocessing as mp
            from multiprocessing import resource_tracker
            from multiprocessing.shared_memory import SharedMemory

            mp.set_start_method("spawn")


            def create_and_register_resource(rtype):
                if rtype == "semaphore":
                    lock = mp.Lock()
                    return lock, lock._semlock.name
                elif rtype == "shared_memory":
                    sm = SharedMemory(create=True, size=10)
                    return sm, sm._name
                else:
                    raise ValueError(
                        "Resource type {{}} not understood".format(rtype))


            resource1, rname1 = create_and_register_resource("{rtype}")
            resource2, rname2 = create_and_register_resource("{rtype}")

            os.write({w}, rname1.encode("ascii") + b"\\n")
            os.write({w}, rname2.encode("ascii") + b"\\n")

            time.sleep(10)
        '''
        for rtype in resource_tracker._CLEANUP_FUNCS:
            with self.subTest(rtype=rtype):
                if rtype in ("noop", "dummy"):
                    # Artefact resource type used by the resource_tracker
                    # or tests
                    continue
                r, w = os.pipe()
                p = subprocess.Popen([sys.executable,
                                     '-E', '-c', cmd.format(w=w, rtype=rtype)],
                                     pass_fds=[w],
                                     stderr=subprocess.PIPE)
                os.close(w)
                with open(r, 'rb', closefd=True) as f:
                    name1 = f.readline().rstrip().decode('ascii')
                    name2 = f.readline().rstrip().decode('ascii')
                _resource_unlink(name1, rtype)
                p.terminate()
                p.wait()

                err_msg = (f"A {rtype} resource was leaked after a process was "
                           f"abruptly terminated")
                for _ in support.sleeping_retry(support.SHORT_TIMEOUT,
                                                  err_msg):
                    try:
                        _resource_unlink(name2, rtype)