| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent b5d6a79 commit a720bd8
4 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -19,6 +19,7 @@ | |||
| 19 | 19 | import sys | |
| 20 | 20 | from textwrap import dedent | |
| 21 | 21 | import threading | |
| 22 | + import time | ||
| 22 | 23 | import warnings | |
| 23 | 24 | ||
| 24 | 25 | from git.compat import defenc, force_bytes, safe_decode | |
@@ -99,6 +100,42 @@ | |||
| 99 | 100 | ## @{ | |
| 100 | 101 | ||
| 101 | 102 | ||
| 103 | + def _kill_process(pid: int) -> bool: | ||
| 104 | + """Kill a POSIX process and its descendants, returning whether it was killed.""" | ||
| 105 | + # Collect descendants before signalling, while their parent PIDs still identify them. | ||
| 106 | + pids = [pid] | ||
| 107 | + for parent_pid in pids: | ||
| 108 | + try: | ||
| 109 | + try: | ||
| 110 | + p = Popen(["pgrep", "-P", str(parent_pid)], stdout=PIPE) | ||
| 111 | + except FileNotFoundError: | ||
| 112 | + # POSIX ps does not support selecting by parent PID. | ||
| 113 | + with Popen(["ps", "-A", "-o", "pid=", "-o", "ppid="], stdout=PIPE) as p: | ||
| 114 | + if p.stdout is not None: | ||
| 115 | + for line in p.stdout: | ||
| 116 | + fields = line.split() | ||
| 117 | + if len(fields) == 2 and all(field.isdigit() for field in fields): | ||
| 118 | + if int(fields[1]) == parent_pid: | ||
| 119 | + pids.append(int(fields[0])) | ||
| 120 | + else: | ||
| 121 | + with p: | ||
| 122 | + if p.stdout is not None: | ||
| 123 | + for line in p.stdout: | ||
| 124 | + if line.strip().isdigit(): | ||
| 125 | + pids.append(int(line)) | ||
| 126 | + except OSError as ex: | ||
| 127 | + _logger.info("Unable to enumerate child processes: %r", ex) | ||
| 128 | + killed = False | ||
| 129 | + if sys.platform != "win32": | ||
| 130 | + for process_pid in pids: | ||
| 131 | + try: | ||
| 132 | + os.kill(process_pid, signal.SIGKILL) | ||
| 133 | + killed = killed or process_pid == pid | ||
| 134 | + except OSError: | ||
| 135 | + pass | ||
| 136 | + return killed | ||
| 137 | + | ||
| 138 | + | ||
| 102 | 139 | def handle_process_output( | |
| 103 | 140 | process: Union["Git.AutoInterrupt", Popen], | |
| 104 | 141 | stdout_handler: Union[ | |
@@ -165,7 +202,7 @@ def pump_stream( | |||
| 165 | 202 | ||
| 166 | 203 | except Exception as ex: | |
| 167 | 204 | _logger.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}") | |
| 168 | - if "I/O operation on closed file" not in str(ex): | ||
| 205 | + if not (isinstance(ex, ValueError) and stream.closed): | ||
| 169 | 206 | # Only reraise if the error was not due to the stream closing. | |
| 170 | 207 | raise CommandError([f"<{name}-pump>"] + remove_password_if_present(cmdline), ex) from ex | |
| 171 | 208 | finally: | |
@@ -200,27 +237,23 @@ def pump_stream( | |||
| 200 | 237 | threads.append(t) | |
| 201 | 238 | ||
| 202 | 239 | # FIXME: Why join? Will block if stdin needs feeding... | |
| 240 | + deadline = None if kill_after_timeout is None else time.monotonic() + kill_after_timeout | ||
| 203 | 241 | for t in threads: | |
| 204 | - t.join(timeout=kill_after_timeout) | ||
| 242 | + t.join(timeout=None if deadline is None else max(0, deadline - time.monotonic())) | ||
| 205 | 243 | if t.is_alive(): | |
| 206 | 244 | if isinstance(process, Git.AutoInterrupt): | |
| 245 | + if sys.platform != "win32" and process.proc is not None: | ||
| 246 | + _kill_process(process.proc.pid) | ||
| 207 | 247 | process._terminate() | |
| 208 | 248 | else: # Don't want to deal with the other case. | |
| 209 | 249 | raise RuntimeError( | |
| 210 | 250 | "Thread join() timed out in cmd.handle_process_output()." | |
| 211 | 251 | f" kill_after_timeout={kill_after_timeout} seconds" | |
| 212 | 252 | ) | |
| 213 | - if stderr_handler: | ||
| 214 | - error_str: Union[str, bytes] = ( | ||
| 215 | - f"error: process killed because it timed out. kill_after_timeout={kill_after_timeout} seconds" | ||
| 216 | - ) | ||
| 217 | - if not decode_streams and isinstance(p_stderr, BinaryIO): | ||
| 218 | - # Assume stderr_handler needs binary input. | ||
| 219 | - error_str = cast(str, error_str) | ||
| 220 | - error_str = error_str.encode() | ||
| 221 | - # We ignore typing on the next line because mypy does not like the way | ||
| 222 | - # we inferred that stderr takes str or bytes. | ||
| 223 | - stderr_handler(error_str) # type: ignore[arg-type] | ||
| 253 | + process._timeout_error = ( | ||
| 254 | + f"error: process killed because it timed out. kill_after_timeout={kill_after_timeout} seconds" | ||
| 255 | + ) | ||
| 256 | + break | ||
| 224 | 257 | ||
| 225 | 258 | if finalizer: | |
| 226 | 259 | finalizer(process) | |
@@ -326,7 +359,7 @@ class _AutoInterrupt: | |||
| 326 | 359 | raise. | |
| 327 | 360 | """ | |
| 328 | 361 | ||
| 329 | - __slots__ = ("proc", "args", "status") | ||
| 362 | + __slots__ = ("proc", "args", "status", "_timeout_error") | ||
| 330 | 363 | ||
| 331 | 364 | # If this is non-zero it will override any status code during _terminate, used | |
| 332 | 365 | # to prevent race conditions in testing. | |
@@ -336,6 +369,7 @@ def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None: | |||
| 336 | 369 | self.proc = proc | |
| 337 | 370 | self.args = args | |
| 338 | 371 | self.status: Union[int, None] = None | |
| 372 | + self._timeout_error: Optional[str] = None | ||
| 339 | 373 | ||
| 340 | 374 | def _terminate(self) -> None: | |
| 341 | 375 | """Terminate the underlying process.""" | |
@@ -344,37 +378,45 @@ def _terminate(self) -> None: | |||
| 344 | 378 | ||
| 345 | 379 | proc = self.proc | |
| 346 | 380 | self.proc = None | |
| 347 | - if proc.stdin: | ||
| 348 | - proc.stdin.close() | ||
| 349 | - if proc.stdout: | ||
| 350 | - proc.stdout.close() | ||
| 351 | - if proc.stderr: | ||
| 352 | - proc.stderr.close() | ||
| 353 | - # Did the process finish already so we have a return code? | ||
| 354 | 381 | try: | |
| 355 | - if proc.poll() is not None: | ||
| 356 | - self.status = self._status_code_if_terminate or proc.poll() | ||
| 382 | + if proc.stdin: | ||
| 383 | + # A timed-out process may already have exited before input is flushed. | ||
| 384 | + with contextlib.suppress(BrokenPipeError): | ||
| 385 | + proc.stdin.close() | ||
| 386 | + # Did the process finish already so we have a return code? | ||
| 387 | + try: | ||
| 388 | + if proc.poll() is not None: | ||
| 389 | + self.status = self._status_code_if_terminate or proc.poll() | ||
| 390 | + return | ||
| 391 | + except OSError as ex: | ||
| 392 | + _logger.info("Ignored error after process had died: %r", ex) | ||
| 393 | + | ||
| 394 | + # It can be that nothing really exists anymore... | ||
| 395 | + if os is None or getattr(os, "kill", None) is None: | ||
| 357 | 396 | return | |
| 358 | - except OSError as ex: | ||
| 359 | - _logger.info("Ignored error after process had died: %r", ex) | ||
| 360 | 397 | ||
| 361 | - # It can be that nothing really exists anymore... | ||
| 362 | - if os is None or getattr(os, "kill", None) is None: | ||
| 363 | - return | ||
| 398 | + # Try to kill it. | ||
| 399 | + try: | ||
| 400 | + proc.terminate() | ||
| 401 | + except (OSError, AttributeError) as ex: | ||
| 402 | + # On interpreter shutdown (notably on Windows), parts of the stdlib used by | ||
| 403 | + # subprocess can already be torn down (e.g. `subprocess._winapi` becomes None), | ||
| 404 | + # which can cause AttributeError during terminate(). In that case, we prefer | ||
| 405 | + # to silently ignore to avoid noisy "Exception ignored in: __del__" messages. | ||
| 406 | + _logger.info("Ignored error while terminating process: %r", ex) | ||
| 407 | + return | ||
| 408 | + # END exception handling | ||
| 409 | + finally: | ||
| 410 | + if proc.stdout: | ||
| 411 | + proc.stdout.close() | ||
| 412 | + if proc.stderr: | ||
| 413 | + proc.stderr.close() | ||
| 364 | 414 | ||
| 365 | - # Try to kill it. | ||
| 366 | 415 | try: | |
| 367 | - proc.terminate() | ||
| 368 | - status = proc.wait() # Ensure the process goes away. | ||
| 369 | - | ||
| 416 | + status = proc.wait() | ||
| 370 | 417 | self.status = self._status_code_if_terminate or status | |
| 371 | 418 | except (OSError, AttributeError) as ex: | |
| 372 | - # On interpreter shutdown (notably on Windows), parts of the stdlib used by | ||
| 373 | - # subprocess can already be torn down (e.g. `subprocess._winapi` becomes None), | ||
| 374 | - # which can cause AttributeError during terminate(). In that case, we prefer | ||
| 375 | - # to silently ignore to avoid noisy "Exception ignored in: __del__" messages. | ||
| 376 | - _logger.info("Ignored error while terminating process: %r", ex) | ||
| 377 | - # END exception handling | ||
| 419 | + _logger.info("Ignored error while waiting for terminated process: %r", ex) | ||
| 378 | 420 | ||
| 379 | 421 | def __del__(self) -> None: | |
| 380 | 422 | self._terminate() | |
@@ -395,6 +437,8 @@ def wait(self, stderr: Union[None, str, bytes] = b"") -> int: | |||
| 395 | 437 | :raise git.exc.GitCommandError: | |
| 396 | 438 | If the return status is not 0. | |
| 397 | 439 | """ | |
| 440 | + if self._timeout_error is not None: | ||
| 441 | + stderr = self._timeout_error | ||
| 398 | 442 | stderr_b = force_bytes(data=stderr, encoding="utf-8") or b"" | |
| 399 | 443 | status: Union[int, None] | |
| 400 | 444 | if self.proc is not None: | |
@@ -1382,8 +1426,8 @@ def execute( | |||
| 1382 | 1426 | 2. Enumerating child processes requires ``pgrep -P``, or a ``ps`` command | |
| 1383 | 1427 | supporting the POSIX ``-A`` and ``-o`` options if ``pgrep`` is not | |
| 1384 | 1428 | installed. Effectiveness may vary on systems without these commands. | |
| 1385 | - 3. Deeper descendants do not receive signals, though they may sometimes | ||
| 1386 | - terminate as a consequence of their parent processes being killed. | ||
| 1429 | + 3. Descendants are enumerated before signalling. Processes that detach | ||
| 1430 | + or spawn after enumeration may not receive signals. | ||
| 1387 | 1431 | 4. `kill_after_timeout` uses ``SIGKILL``, which can have negative side | |
| 1388 | 1432 | effects on a repository. For example, stale locks in case of | |
| 1389 | 1433 | :manpage:`git-gc(1)` could render the repository incapable of accepting | |
@@ -1537,43 +1581,9 @@ def execute( | |||
| 1537 | 1581 | timeout = kill_after_timeout | |
| 1538 | 1582 | ||
| 1539 | 1583 | def kill_process(pid: int) -> None: | |
| 1540 | - """Callback to kill a process. | ||
| 1541 | - | ||
| 1542 | - This callback implementation would be ineffective and unsafe on Windows. | ||
| 1543 | - """ | ||
| 1544 | - child_pids = [] | ||
| 1545 | - try: | ||
| 1546 | - p = Popen(["pgrep", "-P", str(pid)], stdout=PIPE) | ||
| 1547 | - except FileNotFoundError: | ||
| 1548 | - # POSIX ps does not support selecting by parent PID. | ||
| 1549 | - with Popen(["ps", "-A", "-o", "pid=", "-o", "ppid="], stdout=PIPE) as p: | ||
| 1550 | - if p.stdout is not None: | ||
| 1551 | - for line in p.stdout: | ||
| 1552 | - fields = line.split() | ||
| 1553 | - if len(fields) == 2 and all(field.isdigit() for field in fields): | ||
| 1554 | - if int(fields[1]) == pid: | ||
| 1555 | - child_pids.append(int(fields[0])) | ||
| 1556 | - else: | ||
| 1557 | - with p: | ||
| 1558 | - if p.stdout is not None: | ||
| 1559 | - for line in p.stdout: | ||
| 1560 | - if line.strip().isdigit(): | ||
| 1561 | - child_pids.append(int(line)) | ||
| 1562 | - try: | ||
| 1563 | - os.kill(pid, signal.SIGKILL) | ||
| 1564 | - for child_pid in child_pids: | ||
| 1565 | - try: | ||
| 1566 | - os.kill(child_pid, signal.SIGKILL) | ||
| 1567 | - except OSError: | ||
| 1568 | - pass | ||
| 1569 | - # Tell the main routine that the process was killed. | ||
| 1584 | + if _kill_process(pid): | ||
| 1570 | 1585 | assert kill_check is not None | |
| 1571 | 1586 | kill_check.set() | |
| 1572 | - except OSError: | ||
| 1573 | - # It is possible that the process gets completed in the duration | ||
| 1574 | - # after timeout happens and before we try to kill the process. | ||
| 1575 | - pass | ||
| 1576 | - return | ||
| 1577 | 1587 | ||
| 1578 | 1588 | def make_timeout_error() -> Union[str, bytes]: | |
| 1579 | 1589 | err = f'Timeout: the command "{" ".join(redacted_command)}" did not complete in {timeout:g} secs.' | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1,3 +1,9 @@ | |||
| 1 | + from subprocess import PIPE | ||
| 2 | + import sys | ||
| 3 | + import threading | ||
| 4 | + | ||
| 5 | + import pytest | ||
| 6 | + | ||
| 1 | 7 | from git.cmd import Git | |
| 2 | 8 | ||
| 3 | 9 | ||
@@ -31,3 +37,36 @@ def test_autointerrupt_terminate_ignores_attributeerror(): | |||
| 31 | 37 | ||
| 32 | 38 | # Ensure the reference is cleared to avoid repeated attempts. | |
| 33 | 39 | assert ai.proc is None | |
| 40 | + | ||
| 41 | + | ||
| 42 | + def test_autointerrupt_terminate_closes_buffered_stdin(): | ||
| 43 | + ai = Git().hash_object("--stdin", as_process=True, istream=PIPE) | ||
| 44 | + proc = ai.proc | ||
| 45 | + proc.stdin.write(b"buffered input") | ||
| 46 | + try: | ||
| 47 | + ai._terminate() | ||
| 48 | + assert proc.stdin.closed | ||
| 49 | + assert proc.stdout.closed | ||
| 50 | + assert proc.stderr.closed | ||
| 51 | + finally: | ||
| 52 | + proc.stdout.close() | ||
| 53 | + proc.stderr.close() | ||
| 54 | + | ||
| 55 | + | ||
| 56 | + @pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX SIGTERM handling") | ||
| 57 | + def test_autointerrupt_terminate_closes_pipes_before_waiting(): | ||
| 58 | + script = ( | ||
| 59 | + "import signal, sys; signal.signal(signal.SIGTERM, signal.SIG_IGN); " | ||
| 60 | + "print('ready', flush=True); sys.stdin.read(); sys.stdout.buffer.write(b'x' * 1000000)" | ||
| 61 | + ) | ||
| 62 | + ai = Git().execute([sys.executable, "-c", script], as_process=True, istream=PIPE) | ||
| 63 | + proc = ai.proc | ||
| 64 | + assert proc.stdout.readline() == b"ready\n" | ||
| 65 | + watchdog = threading.Timer(5, proc.kill) | ||
| 66 | + watchdog.start() | ||
| 67 | + try: | ||
| 68 | + ai._terminate() | ||
| 69 | + assert proc.returncode != -9, "cleanup waited for exit with undrained output pipes" | ||
| 70 | + finally: | ||
| 71 | + watchdog.cancel() | ||
| 72 | + watchdog.join() | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -18,6 +18,7 @@ | |||
| 18 | 18 | import subprocess | |
| 19 | 19 | import sys | |
| 20 | 20 | import tempfile | |
| 21 | + import threading | ||
| 21 | 22 | import time | |
| 22 | 23 | from unittest import mock, skipUnless | |
| 23 | 24 | ||
@@ -367,25 +368,49 @@ def portable_popen(args, **kwargs): | |||
| 367 | 368 | self.assertFalse(marker.exists(), "the direct child survived the timeout") | |
| 368 | 369 | ||
| 369 | 370 | @skipUnless(sys.platform != "win32", "kill_after_timeout is not supported on Windows") | |
| 370 | - def test_timeout_ps_fallback_selects_only_direct_children(self): | ||
| 371 | + def test_timeout_ps_fallback_selects_only_descendants(self): | ||
| 371 | 372 | process = mock.MagicMock() | |
| 372 | 373 | process.pid = 1234 | |
| 373 | 374 | process.communicate.return_value = (b"", b"") | |
| 374 | 375 | process.returncode = -signal.SIGKILL | |
| 375 | 376 | ps = mock.MagicMock() | |
| 376 | 377 | ps.__enter__.return_value = ps | |
| 377 | - ps.stdout = io.BytesIO(b"PID PPID\n 321 1\n 5678 1234\n 9012 5678\n\n") | ||
| 378 | + ps.stdout = mock.MagicMock() | ||
| 379 | + ps.stdout.__iter__.side_effect = lambda: iter([b"PID PPID\n", b" 321 1\n", b" 5678 1234\n", b" 9012 5678\n"]) | ||
| 380 | + | ||
| 381 | + def portable_popen(args, **kwargs): | ||
| 382 | + if args[0] == "pgrep": | ||
| 383 | + raise FileNotFoundError("pgrep is not installed") | ||
| 384 | + return ps | ||
| 378 | 385 | ||
| 379 | 386 | with contextlib.ExitStack() as stack: | |
| 380 | 387 | stack.enter_context(mock.patch.object(cmd, "safer_popen", return_value=process)) | |
| 381 | - stack.enter_context(mock.patch.object(cmd, "Popen", side_effect=[FileNotFoundError, ps])) | ||
| 388 | + stack.enter_context(mock.patch.object(cmd, "Popen", side_effect=portable_popen)) | ||
| 382 | 389 | kill = stack.enter_context(mock.patch.object(cmd.os, "kill")) | |
| 383 | 390 | timer = stack.enter_context(mock.patch.object(cmd.threading, "Timer")) | |
| 384 | 391 | # Run the timeout callback synchronously, with no real processes or signals. | |
| 385 | 392 | timer.return_value.start.side_effect = lambda: timer.call_args.args[1](1234) | |
| 386 | 393 | self.git.execute(["git", "version"], kill_after_timeout=1, with_exceptions=False) | |
| 387 | 394 | ||
| 388 | - self.assertEqual(kill.call_args_list, [mock.call(1234, signal.SIGKILL), mock.call(5678, signal.SIGKILL)]) | ||
| 395 | + self.assertEqual( | ||
| 396 | + kill.call_args_list, | ||
| 397 | + [mock.call(1234, signal.SIGKILL), mock.call(5678, signal.SIGKILL), mock.call(9012, signal.SIGKILL)], | ||
| 398 | + ) | ||
| 399 | + | ||
| 400 | + @skipUnless(sys.platform != "win32", "requires POSIX timeout signalling") | ||
| 401 | + def test_timeout_without_process_lookup_tools(self): | ||
| 402 | + process = self.git.execute([sys.executable, "-c", "import time; time.sleep(30)"], as_process=True) | ||
| 403 | + proc = process.proc | ||
| 404 | + assert proc is not None | ||
| 405 | + try: | ||
| 406 | + with mock.patch.object(cmd, "Popen", side_effect=FileNotFoundError("lookup unavailable")): | ||
| 407 | + cmd.handle_process_output(process, None, None, kill_after_timeout=0.1) | ||
| 408 | + self.assertIsNotNone(proc.poll()) | ||
| 409 | + finally: | ||
| 410 | + if proc.poll() is None: | ||
| 411 | + proc.kill() | ||
| 412 | + proc.wait() | ||
| 413 | + process._terminate() | ||
| 389 | 414 | ||
| 390 | 415 | def test_it_executes_git_without_stdout_redirect(self): | |
| 391 | 416 | returncode, stdout, stderr = self.git.execute( | |
@@ -1006,6 +1031,41 @@ def stderr_handler(line): | |||
| 1006 | 1031 | self.assertEqual(len(actual_lines[1]), expected_line_count, repr(actual_lines[1])) | |
| 1007 | 1032 | self.assertEqual(len(actual_lines[2]), expected_line_count, repr(actual_lines[2])) | |
| 1008 | 1033 | ||
| 1034 | + @skipUnless(sys.platform != "win32", "requires POSIX timeout signalling") | ||
| 1035 | + @ddt.data(False, True) | ||
| 1036 | + def test_timeout_does_not_wait_for_blocked_output_handler(self, stderr): | ||
| 1037 | + release = threading.Event() | ||
| 1038 | + finished = threading.Event() | ||
| 1039 | + | ||
| 1040 | + def handler(line): | ||
| 1041 | + try: | ||
| 1042 | + release.wait(5) | ||
| 1043 | + finally: | ||
| 1044 | + finished.set() | ||
| 1045 | + | ||
| 1046 | + process = self.git.execute( | ||
| 1047 | + [ | ||
| 1048 | + sys.executable, | ||
| 1049 | + "-c", | ||
| 1050 | + "import sys, time; " | ||
| 1051 | + "print('ready', file=sys.stderr if sys.argv[1] == 'True' else sys.stdout, flush=True); " | ||
| 1052 | + "time.sleep(30)", | ||
| 1053 | + str(stderr), | ||
| 1054 | + ], | ||
| 1055 | + as_process=True, | ||
| 1056 | + ) | ||
| 1057 | + started = time.monotonic() | ||
| 1058 | + try: | ||
| 1059 | + cmd.handle_process_output( | ||
| 1060 | + process, None if stderr else handler, handler if stderr else None, kill_after_timeout=0.5 | ||
| 1061 | + ) | ||
| 1062 | + self.assertLess(time.monotonic() - started, 3) | ||
| 1063 | + with self.assertRaisesRegex(GitCommandError, "process killed because it timed out"): | ||
| 1064 | + process.wait() | ||
| 1065 | + finally: | ||
| 1066 | + release.set() | ||
| 1067 | + self.assertTrue(finished.wait(5)) | ||
| 1068 | + | ||
| 1009 | 1069 | def test_execute_kwargs_set_agrees_with_method(self): | |
| 1010 | 1070 | parameter_names = inspect.signature(cmd.Git.execute).parameters.keys() | |
| 1011 | 1071 | self_param, command_param, *most_params, extra_kwargs_param = parameter_names | |
| Back | FazBrowse Home | New Git URL |
0 commit comments