Chiebot-Mirror
/
meson
mirror of https://github.com/mesonbuild/meson.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

mtest: refactor _run_cmd

Browse Source 
A large part of _run_cmd is devoted to setting up and killing the
test subprocess.  Move that to a separate function to make the
test runner logic easier to understand.
pull/7836/head
Paolo Bonzini 4 years ago
parent 8cf90e6370
commit 57b918f281
1 changed files with 101 additions and 85 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 186
      mesonbuild/mtest.py

186
mesonbuild/mtest.py
Unescape View File

@ -665,6 +665,97 @@ class SingleTestRunner:
self.test.timeout = None
return self._run_cmd(wrap + cmd + self.test.cmd_args + self.options.test_args)
def _run_subprocess(self, args: T.List[str], *, timeout: T.Optional[int],
stdout: T.IO, stderr: T.IO,
env: T.Dict[str, str], cwd: T.Optional[str]) -> T.Tuple[T.Optional[int], bool, T.Optional[str]]:
def kill_process(p: subprocess.Popen) -> T.Optional[str]:
# Python does not provide multiplatform support for
# killing a process and all its children so we need
# to roll our own.
if is_windows():
subprocess.run(['taskkill', '/F', '/T', '/PID', str(p.pid)])
else:
def _send_signal_to_process_group(pgid: int, signum: int) -> None:
""" sends a signal to a process group """
try:
os.killpg(pgid, signum)
except ProcessLookupError:
# Sometimes (e.g. with Wine) this happens.
# There's nothing we can do (maybe the process
# already died) so carry on.
pass
# Send a termination signal to the process group that setsid()
# created - giving it a chance to perform any cleanup.
_send_signal_to_process_group(p.pid, signal.SIGTERM)
# Make sure the termination signal actually kills the process
# group, otherwise retry with a SIGKILL.
try:
p.communicate(timeout=0.5)
except subprocess.TimeoutExpired:
_send_signal_to_process_group(p.pid, signal.SIGKILL)
try:
p.communicate(timeout=1)
except subprocess.TimeoutExpired:
# An earlier kill attempt has not worked for whatever reason.
# Try to kill it one last time with a direct call.
# If the process has spawned children, they will remain around.
p.kill()
try:
p.communicate(timeout=1)
except subprocess.TimeoutExpired:
additional_error = 'Test process could not be killed.'
except ValueError:
additional_error = 'Could not read output. Maybe the process has redirected its stdout/stderr?'
return additional_error
# Let gdb handle ^C instead of us
if self.options.gdb:
previous_sigint_handler = signal.getsignal(signal.SIGINT)
# Make the meson executable ignore SIGINT while gdb is running.
signal.signal(signal.SIGINT, signal.SIG_IGN)
def preexec_fn() -> None:
if self.options.gdb:
# Restore the SIGINT handler for the child process to
# ensure it can handle it.
signal.signal(signal.SIGINT, signal.SIG_DFL)
else:
# We don't want setsid() in gdb because gdb needs the
# terminal in order to handle ^C and not show tcsetpgrp()
# errors avoid not being able to use the terminal.
os.setsid()
p = subprocess.Popen(args,
stdout=stdout,
stderr=stderr,
env=env,
cwd=cwd,
preexec_fn=preexec_fn if not is_windows() else None)
timed_out = False
kill_test = False
try:
p.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
if self.options.verbose:
print('{} time out (After {} seconds)'.format(self.test.name, timeout))
timed_out = True
except KeyboardInterrupt:
mlog.warning('CTRL-C detected while running {}'.format(self.test.name))
kill_test = True
finally:
if self.options.gdb:
# Let us accept ^C again
signal.signal(signal.SIGINT, previous_sigint_handler)
if kill_test or timed_out:
additional_error = kill_process(p)
return p.returncode, timed_out, additional_error
else:
return p.returncode, False, None
def _run_cmd(self, cmd: T.List[str]) -> TestRun:
starttime = time.time()
@ -696,23 +787,6 @@ class SingleTestRunner:
if self.test.protocol is TestProtocol.TAP and stderr is stdout:
stdout = tempfile.TemporaryFile("wb+")
# Let gdb handle ^C instead of us
if self.options.gdb:
previous_sigint_handler = signal.getsignal(signal.SIGINT)
# Make the meson executable ignore SIGINT while gdb is running.
signal.signal(signal.SIGINT, signal.SIG_IGN)
def preexec_fn() -> None:
if self.options.gdb:
# Restore the SIGINT handler for the child process to
# ensure it can handle it.
signal.signal(signal.SIGINT, signal.SIG_DFL)
else:
# We don't want setsid() in gdb because gdb needs the
# terminal in order to handle ^C and not show tcsetpgrp()
# errors avoid not being able to use the terminal.
os.setsid()
extra_cmd = [] # type: T.List[str]
if self.test.protocol is TestProtocol.GTEST:
gtestname = self.test.name
@ -720,77 +794,19 @@ class SingleTestRunner:
gtestname = os.path.join(self.test.workdir, self.test.name)
extra_cmd.append('--gtest_output=xml:{}.xml'.format(gtestname))
p = subprocess.Popen(cmd + extra_cmd,
stdout=stdout,
stderr=stderr,
env=self.env,
cwd=self.test.workdir,
preexec_fn=preexec_fn if not is_windows() else None)
timed_out = False
kill_test = False
if self.test.timeout is None:
timeout = None
elif self.options.timeout_multiplier is not None:
timeout = self.test.timeout * self.options.timeout_multiplier
else:
timeout = self.test.timeout
try:
p.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
if self.options.verbose:
print('{} time out (After {} seconds)'.format(self.test.name, timeout))
timed_out = True
except KeyboardInterrupt:
mlog.warning('CTRL-C detected while running {}'.format(self.test.name))
kill_test = True
finally:
if self.options.gdb:
# Let us accept ^C again
signal.signal(signal.SIGINT, previous_sigint_handler)
additional_error = None
if kill_test or timed_out:
# Python does not provide multiplatform support for
# killing a process and all its children so we need
# to roll our own.
if is_windows():
subprocess.run(['taskkill', '/F', '/T', '/PID', str(p.pid)])
else:
def _send_signal_to_process_group(pgid: int, signum: int) -> None:
""" sends a signal to a process group """
try:
os.killpg(pgid, signum)
except ProcessLookupError:
# Sometimes (e.g. with Wine) this happens.
# There's nothing we can do (maybe the process
# already died) so carry on.
pass
# Send a termination signal to the process group that setsid()
# created - giving it a chance to perform any cleanup.
_send_signal_to_process_group(p.pid, signal.SIGTERM)
# Make sure the termination signal actually kills the process
# group, otherwise retry with a SIGKILL.
try:
p.communicate(timeout=0.5)
except subprocess.TimeoutExpired:
_send_signal_to_process_group(p.pid, signal.SIGKILL)
try:
p.communicate(timeout=1)
except subprocess.TimeoutExpired:
# An earlier kill attempt has not worked for whatever reason.
# Try to kill it one last time with a direct call.
# If the process has spawned children, they will remain around.
p.kill()
try:
p.communicate(timeout=1)
except subprocess.TimeoutExpired:
additional_error = 'Test process could not be killed.'
except ValueError:
additional_error = 'Could not read output. Maybe the process has redirected its stdout/stderr?'
returncode, timed_out, additional_error = self._run_subprocess(cmd + extra_cmd,
timeout=timeout,
stdout=stdout,
stderr=stderr,
env=self.env,
cwd=self.test.workdir)
endtime = time.time()
duration = endtime - starttime
if additional_error is None:
@ -808,16 +824,16 @@ class SingleTestRunner:
stdo = ""
stde = additional_error
if timed_out:
return TestRun(self.test, self.test_env, TestResult.TIMEOUT, [], p.returncode, starttime, duration, stdo, stde, cmd)
return TestRun(self.test, self.test_env, TestResult.TIMEOUT, [], returncode, starttime, duration, stdo, stde, cmd)
else:
if self.test.protocol is TestProtocol.EXITCODE:
return TestRun.make_exitcode(self.test, self.test_env, p.returncode, starttime, duration, stdo, stde, cmd)
return TestRun.make_exitcode(self.test, self.test_env, returncode, starttime, duration, stdo, stde, cmd)
elif self.test.protocol is TestProtocol.GTEST:
return TestRun.make_gtest(self.test, self.test_env, p.returncode, starttime, duration, stdo, stde, cmd)
return TestRun.make_gtest(self.test, self.test_env, returncode, starttime, duration, stdo, stde, cmd)
else:
if self.options.verbose:
print(stdo, end='')
return TestRun.make_tap(self.test, self.test_env, p.returncode, starttime, duration, stdo, stde, cmd)
return TestRun.make_tap(self.test, self.test_env, returncode, starttime, duration, stdo, stde, cmd)
class TestHarness:

Write Preview
Loading…
Cancel
Save