Use subprocess's timeout feature instead of reinventing the wheel

subprocess's APIs in 3.3+ support passing timeout to calls, such as
.communicate(..), .wait(..), etc. Pass `kill_after_timeout` to those
APIs and remove the watchdog handler code as it's not needed once
`timeout=` is used.

This enables `kill_after_timeout` on Windows platforms by side-effect as
upstream implements `timeout` for all supported platforms.

Signed-off-by: Enji Cooper <yaneurabeya@gmail.com>

Author: ngie-eign

git/cmd.py

@@ -13,7 +13,6 @@
 import logging
 import os
 import re
-import signal
 import subprocess
 from subprocess import DEVNULL, PIPE, Popen
 import sys
@@ -110,7 +109,7 @@ def handle_process_output(
     stderr_handler: Union[None, Callable[[AnyStr], None], Callable[[List[AnyStr]], None]],
     finalizer: Union[None, Callable[[Union[Popen, "Git.AutoInterrupt"]], None]] = None,
     decode_streams: bool = True,
-    kill_after_timeout: Union[None, float] = None,
+    kill_after_timeout: float | None = None,
 ) -> None:
     R"""Register for notifications to learn that process output is ready to read, and
     dispatch lines to the respective line handlers.
@@ -139,7 +138,7 @@ def handle_process_output(
         - decoding must happen later, such as for :class:`~git.diff.Diff`\s.
 
     :param kill_after_timeout:
-        :class:`float` or ``None``, Default = ``None``
+        :class:``int``, ``float``, or ``None`` (block indefinitely), Default = ``None``.
 
         To specify a timeout in seconds for the git command, after which the process
         should be killed.
@@ -326,16 +325,22 @@ class _AutoInterrupt:
     raise.
     """
 
-    __slots__ = ("proc", "args", "status")
+    __slots__ = ("proc", "args", "status", "timeout")
 
     # If this is non-zero it will override any status code during _terminate, used
     # to prevent race conditions in testing.
     _status_code_if_terminate: int = 0
 
-    def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None:
+    def __init__(
+        self,
+        proc: subprocess.Popen | None,
+        args: Any,
+        timeout: float | None = None,
+    ) -> None:
         self.proc = proc
         self.args = args
         self.status: Union[int, None] = None
+        self.timeout = timeout
 
     def _terminate(self) -> None:
         """Terminate the underlying process."""
@@ -365,10 +370,10 @@ def _terminate(self) -> None:
         # Try to kill it.
         try:
             proc.terminate()
-            status = proc.wait()  # Ensure the process goes away.
+            status = proc.wait(timeout=self.timeout)  # Ensure the process goes away.
 
             self.status = self._status_code_if_terminate or status
-        except (OSError, AttributeError) as ex:
+        except (OSError, AttributeError, subprocess.TimeoutExpired) as ex:
             # On interpreter shutdown (notably on Windows), parts of the stdlib used by
             # subprocess can already be torn down (e.g. `subprocess._winapi` becomes None),
             # which can cause AttributeError during terminate(). In that case, we prefer
@@ -400,7 +405,7 @@ def wait(self, stderr: Union[None, str, bytes] = b"") -> int:
         stderr_b = force_bytes(data=stderr, encoding="utf-8")
         status: Union[int, None]
         if self.proc is not None:
-            status = self.proc.wait()
+            status = self.proc.wait(timeout=self.timeout)
             p_stderr = self.proc.stderr
         else:  # Assume the underlying proc was killed earlier or never existed.
             status = self.status
@@ -1106,7 +1111,7 @@ def execute(
         as_process: bool = False,
         output_stream: Union[None, BinaryIO] = None,
         stdout_as_string: bool = True,
-        kill_after_timeout: Union[None, float] = None,
+        kill_after_timeout: float | None = None,
         with_stdout: bool = True,
         universal_newlines: bool = False,
         shell: Union[None, bool] = None,
@@ -1158,13 +1163,12 @@ def execute(
             until the timeout is explicitly specified. Uses of this feature should be
             carefully considered, due to the following limitations:
 
-            1. This feature is not supported at all on Windows.
-            2. Effectiveness may vary by operating system. ``ps --ppid`` is used to
+            1. Effectiveness may vary by operating system. ``ps --ppid`` is used to
                enumerate child processes, which is available on most GNU/Linux systems
                but not most others.
-            3. Deeper descendants do not receive signals, though they may sometimes
+            2. Deeper descendants do not receive signals, though they may sometimes
                terminate as a consequence of their parent processes being killed.
-            4. `kill_after_timeout` uses ``SIGKILL``, which can have negative side
+            3. `kill_after_timeout` uses ``SIGKILL``, which can have negative side
                effects on a repository. For example, stale locks in case of
                :manpage:`git-gc(1)` could render the repository incapable of accepting
                changes until the lock is manually removed.
@@ -1252,15 +1256,7 @@ def execute(
         if inline_env is not None:
             env.update(inline_env)
 
-        if sys.platform == "win32":
-            if kill_after_timeout is not None:
-                raise GitCommandError(
-                    redacted_command,
-                    '"kill_after_timeout" feature is not supported on Windows.',
-                )
-            cmd_not_found_exception = OSError
-        else:
-            cmd_not_found_exception = FileNotFoundError
+        cmd_not_found_exception = OSError if sys.platform == "win32" else FileNotFoundError
         # END handle
 
         stdout_sink = PIPE if with_stdout else getattr(subprocess, "DEVNULL", None) or open(os.devnull, "wb")
@@ -1303,66 +1299,14 @@ def execute(
         if as_process:
             return self.AutoInterrupt(proc, command)
 
-        if sys.platform != "win32" and kill_after_timeout is not None:
-            # Help mypy figure out this is not None even when used inside communicate().
-            timeout = kill_after_timeout
-
-            def kill_process(pid: int) -> None:
-                """Callback to kill a process.
-
-                This callback implementation would be ineffective and unsafe on Windows.
-                """
-                p = Popen(["ps", "--ppid", str(pid)], stdout=PIPE)
-                child_pids = []
-                if p.stdout is not None:
-                    for line in p.stdout:
-                        if len(line.split()) > 0:
-                            local_pid = (line.split())[0]
-                            if local_pid.isdigit():
-                                child_pids.append(int(local_pid))
-                try:
-                    os.kill(pid, signal.SIGKILL)
-                    for child_pid in child_pids:
-                        try:
-                            os.kill(child_pid, signal.SIGKILL)
-                        except OSError:
-                            pass
-                    # Tell the main routine that the process was killed.
-                    kill_check.set()
-                except OSError:
-                    # It is possible that the process gets completed in the duration
-                    # after timeout happens and before we try to kill the process.
-                    pass
-                return
-
-            def communicate() -> Tuple[AnyStr, AnyStr]:
-                watchdog.start()
-                out, err = proc.communicate()
-                watchdog.cancel()
-                if kill_check.is_set():
-                    err = 'Timeout: the command "%s" did not complete in %d secs.' % (
-                        " ".join(redacted_command),
-                        timeout,
-                    )
-                    if not universal_newlines:
-                        err = err.encode(defenc)
-                return out, err
-
-            # END helpers
-
-            kill_check = threading.Event()
-            watchdog = threading.Timer(timeout, kill_process, args=(proc.pid,))
-        else:
-            communicate = proc.communicate
-
         # Wait for the process to return.
         status = 0
         stdout_value: Union[str, bytes] = b""
         stderr_value: Union[str, bytes] = b""
         newline = "\n" if universal_newlines else b"\n"
         try:
             if output_stream is None:
-                stdout_value, stderr_value = communicate()
+                stdout_value, stderr_value = proc.communicate(timeout=kill_after_timeout)
                 # Strip trailing "\n".
                 if stdout_value is not None and stdout_value.endswith(newline) and strip_newline_in_stdout:  # type: ignore[arg-type]
                     stdout_value = stdout_value[:-1]
@@ -1380,8 +1324,15 @@ def communicate() -> Tuple[AnyStr, AnyStr]:
                 # Strip trailing "\n".
                 if stderr_value is not None and stderr_value.endswith(newline):  # type: ignore[arg-type]
                     stderr_value = stderr_value[:-1]
-                status = proc.wait()
+                status = proc.wait(timeout=kill_after_timeout)
             # END stdout handling
+        except subprocess.TimeoutExpired as err:
+            _logger.info(
+                "error: process killed because it timed out. kill_after_timeout=%s seconds",
+                kill_after_timeout,
+            )
+            if with_exceptions:
+                raise GitCommandError(redacted_command, status or 255, err.stderr, err.stdout) from err
         finally:
             if proc.stdout is not None:
                 proc.stdout.close()

git/remote.py

@@ -5,6 +5,8 @@
 
 """Module implementing a remote object allowing easy access to git remotes."""
 
+from __future__ import annotations
+
 __all__ = ["RemoteProgress", "PushInfo", "FetchInfo", "Remote"]
 
 import contextlib
@@ -873,7 +875,7 @@ def _get_fetch_info_from_stderr(
         self,
         proc: "Git.AutoInterrupt",
         progress: Union[Callable[..., Any], RemoteProgress, None],
-        kill_after_timeout: Union[None, float] = None,
+        kill_after_timeout: float | int | None = None,
     ) -> IterableList["FetchInfo"]:
         progress = to_progress_instance(progress)
 
@@ -944,7 +946,7 @@ def _get_push_info(
         self,
         proc: "Git.AutoInterrupt",
         progress: Union[Callable[..., Any], RemoteProgress, None],
-        kill_after_timeout: Union[None, float] = None,
+        kill_after_timeout: float | int | None = None,
     ) -> PushInfoList:
         progress = to_progress_instance(progress)
 
@@ -1002,7 +1004,7 @@ def fetch(
         refspec: Union[str, List[str], None] = None,
         progress: Union[RemoteProgress, None, "UpdateProgress"] = None,
         verbose: bool = True,
-        kill_after_timeout: Union[None, float] = None,
+        kill_after_timeout: float | int | None = None,
         allow_unsafe_protocols: bool = False,
         allow_unsafe_options: bool = False,
         **kwargs: Any,
@@ -1082,7 +1084,7 @@ def pull(
         self,
         refspec: Union[str, List[str], None] = None,
         progress: Union[RemoteProgress, "UpdateProgress", None] = None,
-        kill_after_timeout: Union[None, float] = None,
+        kill_after_timeout: float | int | None = None,
         allow_unsafe_protocols: bool = False,
         allow_unsafe_options: bool = False,
         **kwargs: Any,
@@ -1136,7 +1138,7 @@ def push(
         self,
         refspec: Union[str, List[str], None] = None,
         progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None] = None,
-        kill_after_timeout: Union[None, float] = None,
+        kill_after_timeout: float | int | None = None,
         allow_unsafe_protocols: bool = False,
         allow_unsafe_options: bool = False,
         **kwargs: Any,