test_multiprocessing.py 文件源码

def test_terminate(self):
        p = self.Pool(4)
        result = p.map_async(
            time.sleep, [0.1 for i in range(10000)], chunksize=1
            )
        p.terminate()
        join = TimingWrapper(p.join)
        join()
        self.assertTrue(join.elapsed < 0.2)