Issue #15646: Prevent equivalent of a fork bomb when using multiprocessing

on Windows without the "if __name__ == '__main__'" idiom.
This commit is contained in:
Richard Oudkerk 2012-08-14 11:41:19 +01:00
parent fe9efc5732
commit faee75c33a
4 changed files with 43 additions and 2 deletions

View File

@ -336,7 +336,7 @@ else:
''' '''
Returns prefix of command line used for spawning a child process Returns prefix of command line used for spawning a child process
''' '''
if process.current_process()._identity==() and is_forking(sys.argv): if getattr(process.current_process(), '_inheriting', False):
raise RuntimeError(''' raise RuntimeError('''
Attempt to start a new process before the current process Attempt to start a new process before the current process
has finished its bootstrapping phase. has finished its bootstrapping phase.

16
Lib/test/mp_fork_bomb.py Normal file
View File

@ -0,0 +1,16 @@
import multiprocessing
def foo(conn):
conn.send("123")
# Because "if __name__ == '__main__'" is missing this will not work
# correctly on Windows. However, we should get a RuntimeError rather
# than the Windows equivalent of a fork bomb.
r, w = multiprocessing.Pipe(False)
p = multiprocessing.Process(target=foo, args=(w,))
p.start()
w.close()
print(r.recv())
r.close()
p.join()

View File

@ -16,6 +16,7 @@ import socket
import random import random
import logging import logging
import errno import errno
import test.script_helper
from test import test_support from test import test_support
from StringIO import StringIO from StringIO import StringIO
_multiprocessing = test_support.import_module('_multiprocessing') _multiprocessing = test_support.import_module('_multiprocessing')
@ -2349,8 +2350,28 @@ class TestTimeouts(unittest.TestCase):
finally: finally:
socket.setdefaulttimeout(old_timeout) socket.setdefaulttimeout(old_timeout)
#
# Test what happens with no "if __name__ == '__main__'"
#
class TestNoForkBomb(unittest.TestCase):
def test_noforkbomb(self):
name = os.path.join(os.path.dirname(__file__), 'mp_fork_bomb.py')
if WIN32:
rc, out, err = test.script_helper.assert_python_failure(name)
self.assertEqual('', out.decode('ascii'))
self.assertIn('RuntimeError', err.decode('ascii'))
else:
rc, out, err = test.script_helper.assert_python_ok(name)
self.assertEqual('123', out.decode('ascii').rstrip())
self.assertEqual('', err.decode('ascii'))
#
#
#
testcases_other = [OtherTest, TestInvalidHandle, TestInitializers, testcases_other = [OtherTest, TestInvalidHandle, TestInitializers,
TestStdinBadfiledescriptor, TestTimeouts] TestStdinBadfiledescriptor, TestTimeouts, TestNoForkBomb]
# #
# #

View File

@ -92,6 +92,10 @@ Core and Builtins
Library Library
------- -------
- Issue #15646: Prevent equivalent of a fork bomb when using
multiprocessing on Windows without the "if __name__ == '__main__'"
idiom.
- Issue #15567: Fix NameError when running threading._test - Issue #15567: Fix NameError when running threading._test
- Issue #15424: Add a __sizeof__ implementation for array objects. - Issue #15424: Add a __sizeof__ implementation for array objects.