test_multiprocessing.py 文件源码

def _test_sharing(self):
        def do_test():
            x = torch.zeros(5, 5)
            q = mp.Queue()
            e = mp.Event()
            data = [x, x[:, 1]]
            q.put(data)
            p = mp.Process(target=simple_fill, args=(q, e))
            lc.check_pid(p.pid)
            p.start()
            e.wait()
            self.assertTrue(data[0].eq(4).all())
            self.assertTrue(data[1].eq(4).all())
            p.join(1)
            self.assertFalse(p.is_alive())

        with leak_checker(self) as lc:
            do_test()