test_multiprocessing.py 文件源码

def test_fs(self):
        with fs_sharing(), leak_checker(self) as lc:
            x = torch.DoubleStorage(4)
            q = mp.Queue()
            self.assertFalse(lc.has_shm_files())
            q.put(x)
            self.assertTrue(lc.has_shm_files(wait=False))
            q.get()
            del x
            del q  # We have to clean up fds for leak_checker