cpython: ff36b8cadfd6 (original) (raw)

--- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -417,6 +417,13 @@ class ThreadTests(BaseTestCase): class ThreadJoinOnShutdown(BaseTestCase):

• Between fork() and exec(), only async-safe functions are allowed (issues

• #12316 and #11870), and fork() from a worker thread is known to trigger

• problems with some operating systems (issue #3863): skip problematic tests

• on platforms known to behave badly.

• platforms_to_skip = ('freebsd4', 'freebsd5', 'freebsd6', 'netbsd5',

•                     'os2emx')[](#l1.12)

+ def _run_and_join(self, script): script = """if 1: import sys, os, time, threading @@ -448,11 +455,10 @@ class ThreadJoinOnShutdown(BaseTestCase) self._run_and_join(script)

• @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
• @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_2_join_in_forked_process(self): # Like the test above, but from a forked interpreter

•    import os[](#l1.25)

•    if not hasattr(os, 'fork'):[](#l1.26)

•        return[](#l1.27)
     script = """if 1:[](#l1.28)
         childpid = os.fork()[](#l1.29)
         if childpid != 0:[](#l1.30)

@@ -466,19 +472,11 @@ class ThreadJoinOnShutdown(BaseTestCase) """ self._run_and_join(script)

• @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
• @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_3_join_in_forked_from_thread(self): # Like the test above, but fork() was called from a worker thread # In the forked process, the main Thread object must be marked as stopped.

•    import os[](#l1.40)

•    if not hasattr(os, 'fork'):[](#l1.41)

•        return[](#l1.42)

•    # Skip platforms with known problems forking from a worker thread.[](#l1.43)

•    # See http://bugs.python.org/issue3863.[](#l1.44)

•    if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'netbsd5',[](#l1.45)

•                       'os2emx'):[](#l1.46)

•        print >>sys.stderr, ('Skipping test_3_join_in_forked_from_thread'[](#l1.47)

•                             ' due to known OS bugs on'), sys.platform[](#l1.48)

•        return[](#l1.49)
     script = """if 1:[](#l1.50)
         main_thread = threading.current_thread()[](#l1.51)
         def worker():[](#l1.52)

@@ -507,15 +505,11 @@ class ThreadJoinOnShutdown(BaseTestCase) self.assertEqual(data, expected_output) @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")

• @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_4_joining_across_fork_in_worker_thread(self): # There used to be a possible deadlock when forking from a child # thread. See http://bugs.python.org/issue6643.[](#l1.60)

•    # Skip platforms with known problems forking from a worker thread.[](#l1.62)

•    # See http://bugs.python.org/issue3863.[](#l1.63)

•    if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'os2emx'):[](#l1.64)

•        raise unittest.SkipTest('due to known OS bugs on ' + sys.platform)[](#l1.65)

- # The script takes the following steps: # - The main thread in the parent process starts a new thread and then # tries to join it. @@ -584,6 +578,7 @@ class ThreadJoinOnShutdown(BaseTestCase) self.assertScriptHasOutput(script, "end of main\n") @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")

• @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_5_clear_waiter_locks_to_avoid_crash(self): # Check that a spawned thread that forks doesn't segfault on certain # platforms, namely OS X. This used to happen if there was a waiter

@@ -596,10 +591,6 @@ class ThreadJoinOnShutdown(BaseTestCase) # lock will be acquired, we can't know if the internal mutex will be # acquired at the time of the fork.

•    # Skip platforms with known problems forking from a worker thread.[](#l1.82)

•    # See http://bugs.python.org/issue3863.[](#l1.83)

•    if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'os2emx'):[](#l1.84)

•        raise unittest.SkipTest('due to known OS bugs on ' + sys.platform)[](#l1.85)
     script = """if True:[](#l1.86)
         import os, time, threading[](#l1.87)