PENDING,RUNNING,CANCELLED,CANCELLED_AND_NOTIFIED,FINISHED,Future,

BrokenExecutor)

fromconcurrent.futures.processimportBrokenProcessPool,_check_system_limits

importmultiprocessing.process

importmultiprocessing.util

ifsupport.check_sanitizer(address=True,memory=True):

self.executor=self.executor_type(

max_workers=self.worker_count,

**self.executor_kwargs)

self._prime_executor()

deftearDown(self):

self.executor.shutdown(wait=True)

super().tearDown()

defget_context(self):

returnget_context(self.ctx)

def_prime_executor(self):

futures= [self.executor.submit(time.sleep,0.1)

for_inrange(self.worker_count)]

forfinfutures:

f.result()

returnmp.get_context(self.ctx)

classThreadPoolMixin(ExecutorMixin):

withself.assertRaises(BrokenExecutor):

self.executor.submit(get_init_status)

def_prime_executor(self):

def_assert_logged(self,msg):

ifself.log_queueisnotNone:

f.result()

deftest_cancel_futures(self):

executor=self.executor_type(max_workers=3)

fs= [executor.submit(time.sleep,.1)for_inrange(50)]

executor.shutdown(cancel_futures=True)

assertself.worker_count<=5,"test needs few workers"

fs= [self.executor.submit(time.sleep,.1)for_inrange(50)]

self.executor.shutdown(cancel_futures=True)

cancelled= [futforfutinfsiffut.cancelled()]

self.assertTrue(len(cancelled)>=35,msg=f"{len(cancelled)=}")

self.assertGreater(len(cancelled),20)

self.assertTrue(len(others)>0,msg=f"{len(others)=}")

self.assertGreater(len(others),0)

deftest_hang_issue39205(self):

deftest_hang_gh83386(self):

ifself.executor_type==futures.ProcessPoolExecutor:

raiseunittest.SkipTest(

"Hangs due tohttps://bugs.python.org/issue39205")

"Hangs, seehttps://github.com/python/cpython/issues/83386")

rc,out,err=assert_python_ok('-c',"""if True:

""".format(executor_type=self.executor_type.__name__))

""".format(executor_type=self.executor_type.__name__,

context=getattr(self,'ctx',None)))

self.assertFalse(err)

self.assertEqual(out.strip(),b"apple")

classThreadPoolShutdownTest(ThreadPoolMixin,ExecutorShutdownTest,BaseTestCase):

def_prime_executor(self):

deftest_threads_terminate(self):

defacquire_lock(lock):

lock.acquire()

classProcessPoolShutdownTest(ExecutorShutdownTest):

def_prime_executor(self):

deftest_processes_terminate(self):

defacquire_lock(lock):

lock.acquire()

mp_context=get_context()

mp_context=self.get_context()

sem=mp_context.Semaphore(0)

for_inrange(3):

self.executor.submit(acquire_lock,sem)

p.join()

deftest_context_manager_shutdown(self):

withfutures.ProcessPoolExecutor(max_workers=5)ase:

withfutures.ProcessPoolExecutor(

max_workers=5,mp_context=self.get_context())ase:

processes=e._processes

self.assertEqual(list(e.map(abs,range(-5,5))),

[5,4,3,2,1,0,1,2,3,4])

p.join()

deftest_del_shutdown(self):

executor=futures.ProcessPoolExecutor(max_workers=5)

executor=futures.ProcessPoolExecutor(

max_workers=5,mp_context=self.get_context())

res=executor.map(abs,range(-5,5))

executor_manager_thread=executor._executor_manager_thread

processes=executor._processes

deftest_shutdown_no_wait(self):

executor=futures.ProcessPoolExecutor(max_workers=5)

executor=futures.ProcessPoolExecutor(

max_workers=5,mp_context=self.get_context())

res=executor.map(abs,range(-5,5))

processes=executor._processes

call_queue=executor._call_queue

pool.submit(submit,pool)

for_inrange(50):

withfutures.ProcessPoolExecutor(1,mp_context=get_context('fork'))asworkers:

withfutures.ProcessPoolExecutor(1,mp_context=mp.get_context('fork'))asworkers:

workers.submit(tuple)

deftest_ressources_gced_in_workers(self):

mgr=get_context(self.ctx).Manager()

mgr=self.get_context().Manager()

obj=EventfulGCObj(mgr)

future=self.executor.submit(id,obj)

future.result()

mgr.join()

deftest_saturation(self):

executor=self.executor_type(4)

mp_context=get_context()

executor=self.executor

mp_context=self.get_context()

sem=mp_context.Semaphore(0)

job_count=15*executor._max_workers

try:

for_inrange(job_count):

executor.submit(sem.acquire)

self.assertEqual(len(executor._processes),executor._max_workers)

for_inrange(job_count):

sem.release()

finally:

executor.shutdown()

for_inrange(job_count):

executor.submit(sem.acquire)

self.assertEqual(len(executor._processes),executor._max_workers)

for_inrange(job_count):

sem.release()

deftest_idle_process_reuse_one(self):

executor=self.executor_type(4)

executor=self.executor

assertexecutor._max_workers>=4

executor.submit(mul,21,2).result()

executor.submit(mul,6,7).result()

executor.submit(mul,3,14).result()

self.assertEqual(len(executor._processes),1)

executor.shutdown()

deftest_idle_process_reuse_multiple(self):

executor=self.executor_type(4)

executor=self.executor

assertexecutor._max_workers<=5

executor.submit(mul,12,7).result()

executor.submit(mul,33,25)

executor.submit(mul,25,26).result()

executor.submit(mul,18,29)

self.assertLessEqual(len(executor._processes),2)

executor.submit(mul,1,2).result()

executor.submit(mul,0,9)

self.assertLessEqual(len(executor._processes),3)

executor.shutdown()

deftest_max_tasks_per_child(self):

executor=self.executor_type(1,max_tasks_per_child=3)

executor=self.executor_type(

1,mp_context=self.get_context(),max_tasks_per_child=3)

f1=executor.submit(os.getpid)

original_pid=f1.result()

executor.shutdown()

deftest_max_tasks_early_shutdown(self):

executor=self.executor_type(3,max_tasks_per_child=1)

executor=self.executor_type(

3,mp_context=self.get_context(),max_tasks_per_child=1)

futures= []

foriinrange(6):

futures.append(executor.submit(mul,i,i))

self.executor.shutdown(wait=True)

executor=self.executor_type(

max_workers=2,mp_context=get_context(self.ctx))

max_workers=2,mp_context=self.get_context())

res=executor.submit(func,*args)

ifignore_stderr:

self.executor.shutdown(wait=True)

withself.executor_type(max_workers=2,

mp_context=get_context(self.ctx))asexecutor:

mp_context=self.get_context())asexecutor:

self.executor=executor# Allow clean up in fail_on_deadlock

f=executor.submit(_crash,delay=.1)

executor.shutdown(wait=True)

self.executor.shutdown(wait=True)

withself.executor_type(max_workers=2,

mp_context=get_context(self.ctx))asexecutor:

mp_context=self.get_context())asexecutor:

self.executor=executor# Allow clean up in fail_on_deadlock

Fix ``test_concurrent_futures`` to test the correct multiprocessing start method context in several cases where the test logic mixed this up.