mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-30 05:35:08 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-30794: added kill() method to multiprocessing.Process (#2528)

Browse source 
* bpo-30794: added kill() method to multiprocessing.Process

* Added entries to documentation and NEWS

* Refactored test_terminate and test_kill

* Fix SIGTERM and SIGKILL being used on Windows for the tests

* Added "versionadded" marker to the documentation

* Fix trailing whitespace in doc
This commit is contained in:
Vitor Pereira 2017-07-18 16:34:23 +01:00 • committed by Antoine Pitrou
parent f474c5a3f3
commit ba75af7130
6 changed files with 40 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

22
Lib/test/_test_multiprocessing.py
View file

@ -277,18 +277,18 @@ class _TestProcess(BaseTestCase):
self.assertNotIn(p, self.active_children())
@classmethod
def _test_terminate(cls):
def _sleep_some(cls):
time.sleep(100)
@classmethod
def _test_sleep(cls, delay):
time.sleep(delay)
def test_terminate(self):
def _kill_process(self, meth):
if self.TYPE == 'threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))
p = self.Process(target=self._test_terminate)
p = self.Process(target=self._sleep_some)
p.daemon = True
p.start()
@ -309,7 +309,7 @@ class _TestProcess(BaseTestCase):
# XXX maybe terminating too soon causes the problems on Gentoo...
time.sleep(1)
p.terminate()
meth(p)
if hasattr(signal, 'alarm'):
# On the Gentoo buildbot waitpid() often seems to block forever.
@ -333,9 +333,17 @@ class _TestProcess(BaseTestCase):
p.join()
# sometimes get p.exitcode == 0 on Windows ...
return p.exitcode
def test_terminate(self):
exitcode = self._kill_process(multiprocessing.Process.terminate)
if os.name != 'nt':
self.assertEqual(p.exitcode, -signal.SIGTERM)
self.assertEqual(exitcode, -signal.SIGTERM)
def test_kill(self):
exitcode = self._kill_process(multiprocessing.Process.kill)
if os.name != 'nt':
self.assertEqual(exitcode, -signal.SIGKILL)
def test_cpu_count(self):
try:
@ -462,7 +470,7 @@ class _TestProcess(BaseTestCase):
for p in procs:
self.assertEqual(p.exitcode, 0)
procs = [self.Process(target=self._test_terminate)
procs = [self.Process(target=self._sleep_some)
for i in range(N)]
for p in procs:
p.start()