test_concurrent_futures.py 文件源码

def test_del_shutdown(self):
        executor = futures.ProcessPoolExecutor(max_workers=5)
        list(executor.map(abs, range(-5, 5)))
        queue_management_thread = executor._queue_management_thread
        processes = executor._processes
        del executor
        test.support.gc_collect()

        queue_management_thread.join()
        for p in processes:
            p.join()