worker: rewrite error handling so os._exit covers all cases
Previously the worker error handling is like:
pid = os.fork() --+
if pid == 0: |
.... | problematic
.... --+
try: --+
.... | worker error handling
--+
If a signal arrives when Python is executing the "problematic" lines, an
external error handling (dispatch.py) will take over the control flow and
it's no longer guaranteed "os._exit" is called (see
86cd09bc13ba for why it
is necessary).
This patch rewrites the error handling so it covers all possible code paths
for a worker even during fork.
Note: "os.getpid() == parentpid" is used to test if the process is parent or
not intentionally, instead of checking "pid", because "pid = os.fork()" may
be not atomic - it's possible that that a signal hits the worker before the
assignment completes [1]. The newly added test replaces "os.fork" to
exercise that extreme case.
[1]: CPython compiles "pid = os.fork()" to 2 byte codes: "CALL_FUNCTION" and
"STORE_FAST", so it's probably not atomic:
def f():
pid = os.fork()
dis.dis(f)
2 0 LOAD_GLOBAL 0 (os)
3 LOAD_ATTR 1 (fork)
6 CALL_FUNCTION 0
9 STORE_FAST 0 (pid)
12 LOAD_CONST 0 (None)
15 RETURN_VALUE
"""test line matching with some failing examples and some which warn
run-test.t only checks positive matches and can not see warnings
(both by design)
"""
from __future__ import absolute_import, print_function
import doctest
import os
import re
# this is hack to make sure no escape characters are inserted into the output
if 'TERM' in os.environ:
del os.environ['TERM']
run_tests = __import__('run-tests')
def prn(ex):
m = ex.args[0]
if isinstance(m, str):
print(m)
else:
print(m.decode('utf-8'))
def lm(expected, output):
r"""check if output matches expected
does it generally work?
>>> lm(b'H*e (glob)\n', b'Here\n')
True
fail on bad test data
>>> try: lm(b'a\n',b'a')
... except AssertionError as ex: print(ex)
missing newline
>>> try: lm(b'single backslash\n', b'single \backslash\n')
... except AssertionError as ex: prn(ex)
single backslash or unknown char
"""
assert (expected.endswith(b'\n')
and output.endswith(b'\n')), 'missing newline'
assert not re.search(br'[^ \w\\/\r\n()*?]', expected + output), \
b'single backslash or unknown char'
match = run_tests.TTest.linematch(expected, output)
if isinstance(match, str):
return 'special: ' + match
elif isinstance(match, bytes):
return 'special: ' + match.decode('utf-8')
else:
return bool(match) # do not return match object
def wintests():
r"""test matching like running on windows
enable windows matching on any os
>>> _osaltsep = os.altsep
>>> os.altsep = True
valid match on windows
>>> lm(b'g/a*/d (glob)\n', b'g\\abc/d\n')
True
direct matching, glob unnecessary
>>> lm(b'g/b (glob)\n', b'g/b\n')
'special: -glob'
missing glob
>>> lm(b'/g/c/d/fg\n', b'\\g\\c\\d/fg\n')
'special: +glob'
restore os.altsep
>>> os.altsep = _osaltsep
"""
pass
def otherostests():
r"""test matching like running on non-windows os
disable windows matching on any os
>>> _osaltsep = os.altsep
>>> os.altsep = False
backslash does not match slash
>>> lm(b'h/a* (glob)\n', b'h\\ab\n')
False
direct matching glob can not be recognized
>>> lm(b'h/b (glob)\n', b'h/b\n')
True
missing glob can not not be recognized
>>> lm(b'/h/c/df/g/\n', b'\\h/c\\df/g\\\n')
False
restore os.altsep
>>> os.altsep = _osaltsep
"""
pass
if __name__ == '__main__':
doctest.testmod()