reuse kill_after_timeout kwarg

Author: sroet

git/cmd.py

@@ -79,7 +79,7 @@ def handle_process_output(process: 'Git.AutoInterrupt' | Popen,
                           finalizer: Union[None,
                                            Callable[[Union[subprocess.Popen, 'Git.AutoInterrupt']], None]] = None,
                           decode_streams: bool = True,
-                          timeout: Union[None, float] = None) -> None:
+                          kill_after_timeout: Union[None, float] = None) -> None:
     """Registers for notifications to learn that process output is ready to read, and dispatches lines to
     the respective line handlers.
     This function returns once the finalizer returns
@@ -94,7 +94,10 @@ def handle_process_output(process: 'Git.AutoInterrupt' | Popen,
         their contents to handlers.
         Set it to False if `universal_newline == True` (then streams are in text-mode)
         or if decoding must happen later (i.e. for Diffs).
-    :param timeout: float, or None timeout to pass to t.join() in case it hangs. Default = None.
+    :param kill_after_timeout:
+        float or None, Default = None
+        To specify a timeout in seconds for the git command, after which the process
+        should be killed.
     """
     # Use 2 "pump" threads and wait for both to finish.
     def pump_stream(cmdline: List[str], name: str, stream: Union[BinaryIO, TextIO], is_decode: bool,
@@ -108,9 +111,12 @@ def pump_stream(cmdline: List[str], name: str, stream: Union[BinaryIO, TextIO],
                         handler(line_str)
                     else:
                         handler(line)
+
         except Exception as ex:
             log.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}")
-            raise CommandError([f'<{name}-pump>'] + remove_password_if_present(cmdline), ex) from ex
+            if "I/O operation on closed file" not in str(ex):
+                # Only reraise if the error was not due to the stream closing
+                raise CommandError([f'<{name}-pump>'] + remove_password_if_present(cmdline), ex) from ex
         finally:
             stream.close()
 
@@ -146,9 +152,16 @@ def pump_stream(cmdline: List[str], name: str, stream: Union[BinaryIO, TextIO],
     ## FIXME: Why Join??  Will block if `stdin` needs feeding...
     #
     for t in threads:
-        t.join(timeout=timeout)
+        t.join(timeout=kill_after_timeout)
         if t.is_alive():
-            raise RuntimeError(f"Thread join() timed out in cmd.handle_process_output(). Timeout={timeout} seconds")
+            if hasattr(process, 'proc'):  # Assume it is a Git.AutoInterrupt:
+                process._terminate()
+            else:  # Don't want to deal with the other case
+                raise RuntimeError(f"Thread join() timed out in cmd.handle_process_output()."
+                                   " kill_after_timeout={kill_after_timeout} seconds")
+            if stderr_handler:
+                stderr_handler("error: process killed because it timed out."
+                               f" kill_after_timeout={kill_after_timeout} seconds")
 
     if finalizer:
         return finalizer(process)
@@ -386,13 +399,15 @@ class AutoInterrupt(object):
         The wait method was overridden to perform automatic status code checking
         and possibly raise."""
 
-        __slots__ = ("proc", "args")
+        __slots__ = ("proc", "args", "status")
 
         def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None:
             self.proc = proc
             self.args = args
+            self.status = None
 
-        def __del__(self) -> None:
+        def _terminate(self) -> None:
+            """Terminate the underlying process"""
             if self.proc is None:
                 return
 
@@ -408,6 +423,7 @@ def __del__(self) -> None:
             # did the process finish already so we have a return code ?
             try:
                 if proc.poll() is not None:
+                    self.status = proc.poll()
                     return None
             except OSError as ex:
                 log.info("Ignored error after process had died: %r", ex)
@@ -419,7 +435,7 @@ def __del__(self) -> None:
             # try to kill it
             try:
                 proc.terminate()
-                proc.wait()    # ensure process goes away
+                self.status = proc.wait()    # ensure process goes away
             except OSError as ex:
                 log.info("Ignored error after process had died: %r", ex)
             except AttributeError:
@@ -431,6 +447,11 @@ def __del__(self) -> None:
                     call(("TASKKILL /F /T /PID %s 2>nul 1>nul" % str(proc.pid)), shell=True)
             # END exception handling
 
+
+
+        def __del__(self) -> None:
+            self._terminate()
+
         def __getattr__(self, attr: str) -> Any:
             return getattr(self.proc, attr)
 
@@ -447,21 +468,26 @@ def wait(self, stderr: Union[None, str, bytes] = b'') -> int:
 
             if self.proc is not None:
                 status = self.proc.wait()
+                p_stderr = self.proc.stderr
+            else: #Assume the underlying proc was killed earlier or never existed
+                status = self.status
+                p_stderr = None
 
-                def read_all_from_possibly_closed_stream(stream: Union[IO[bytes], None]) -> bytes:
-                    if stream:
-                        try:
-                            return stderr_b + force_bytes(stream.read())
-                        except ValueError:
-                            return stderr_b or b''
-                    else:
+            def read_all_from_possibly_closed_stream(stream: Union[IO[bytes], None]) -> bytes:
+                if stream:
+                    try:
+                        return stderr_b + force_bytes(stream.read())
+                    except ValueError:
                         return stderr_b or b''
+                else:
+                    return stderr_b or b''
 
-                if status != 0:
-                    errstr = read_all_from_possibly_closed_stream(self.proc.stderr)
-                    log.debug('AutoInterrupt wait stderr: %r' % (errstr,))
-                    raise GitCommandError(remove_password_if_present(self.args), status, errstr)
             # END status handling
+
+            if status != 0:
+                errstr = read_all_from_possibly_closed_stream(p_stderr)
+                log.debug('AutoInterrupt wait stderr: %r' % (errstr,))
+                raise GitCommandError(remove_password_if_present(self.args), status, errstr)
             return status
 
     # END auto interrupt
@@ -694,7 +720,7 @@ def execute(self,
                 as_process: bool = False,
                 output_stream: Union[None, BinaryIO] = None,
                 stdout_as_string: bool = True,
-                kill_after_timeout: Union[None, int] = None,
+                kill_after_timeout: Union[None, float] = None,
                 with_stdout: bool = True,
                 universal_newlines: bool = False,
                 shell: Union[None, bool] = None,

git/remote.py

@@ -708,7 +708,7 @@ def update(self, **kwargs: Any) -> 'Remote':
 
     def _get_fetch_info_from_stderr(self, proc: 'Git.AutoInterrupt',
                                     progress: Union[Callable[..., Any], RemoteProgress, None],
-                                    timeout: Union[None, float] = None,
+                                    kill_after_timeout: Union[None, float] = None,
                                     ) -> IterableList['FetchInfo']:
 
         progress = to_progress_instance(progress)
@@ -726,7 +726,7 @@ def _get_fetch_info_from_stderr(self, proc: 'Git.AutoInterrupt',
 
         progress_handler = progress.new_message_handler()
         handle_process_output(proc, None, progress_handler, finalizer=None, decode_streams=False,
-                              timeout=timeout)
+                              kill_after_timeout=kill_after_timeout)
 
         stderr_text = progress.error_lines and '\n'.join(progress.error_lines) or ''
         proc.wait(stderr=stderr_text)
@@ -772,7 +772,7 @@ def _get_fetch_info_from_stderr(self, proc: 'Git.AutoInterrupt',
 
     def _get_push_info(self, proc: 'Git.AutoInterrupt',
                        progress: Union[Callable[..., Any], RemoteProgress, None],
-                       timeout: Union[None, float] = None) -> IterableList[PushInfo]:
+                       kill_after_timeout: Union[None, float] = None) -> IterableList[PushInfo]:
         progress = to_progress_instance(progress)
 
         # read progress information from stderr
@@ -790,7 +790,7 @@ def stdout_handler(line: str) -> None:
                 pass
 
         handle_process_output(proc, stdout_handler, progress_handler, finalizer=None, decode_streams=False,
-                              timeout=timeout)
+                              kill_after_timeout=kill_after_timeout)
         stderr_text = progress.error_lines and '\n'.join(progress.error_lines) or ''
         try:
             proc.wait(stderr=stderr_text)
@@ -817,7 +817,8 @@ def _assert_refspec(self) -> None:
 
     def fetch(self, refspec: Union[str, List[str], None] = None,
               progress: Union[RemoteProgress, None, 'UpdateProgress'] = None,
-              verbose: bool = True, timeout: Union[None, float] = None,
+              verbose: bool = True,
+              kill_after_timeout: Union[None, float] = None,
               **kwargs: Any) -> IterableList[FetchInfo]:
         """Fetch the latest changes for this remote
 
@@ -838,6 +839,9 @@ def fetch(self, refspec: Union[str, List[str], None] = None,
             for 'refspec' will make use of this facility.
         :param progress: See 'push' method
         :param verbose: Boolean for verbose output
+        :param kill_after_timeout:
+            To specify a timeout in seconds for the git command, after which the process
+            should be killed. It is set to None by default.
         :param kwargs: Additional arguments to be passed to git-fetch
         :return:
             IterableList(FetchInfo, ...) list of FetchInfo instances providing detailed
@@ -858,20 +862,22 @@ def fetch(self, refspec: Union[str, List[str], None] = None,
 
         proc = self.repo.git.fetch(self, *args, as_process=True, with_stdout=False,
                                    universal_newlines=True, v=verbose, **kwargs)
-        res = self._get_fetch_info_from_stderr(proc, progress, timeout=timeout)
+        res = self._get_fetch_info_from_stderr(proc, progress,
+                                               kill_after_timeout=kill_after_timeout)
         if hasattr(self.repo.odb, 'update_cache'):
             self.repo.odb.update_cache()
         return res
 
     def pull(self, refspec: Union[str, List[str], None] = None,
              progress: Union[RemoteProgress, 'UpdateProgress', None] = None,
-             timeout: Union[None, float] = None,
+             kill_after_timeout: Union[None, float] = None,
              **kwargs: Any) -> IterableList[FetchInfo]:
         """Pull changes from the given branch, being the same as a fetch followed
         by a merge of branch with your local branch.
 
         :param refspec: see 'fetch' method
         :param progress: see 'push' method
+        :param kill_after_timeout: see 'fetch' method
         :param kwargs: Additional arguments to be passed to git-pull
         :return: Please see 'fetch' method """
         if refspec is None:
@@ -880,14 +886,16 @@ def pull(self, refspec: Union[str, List[str], None] = None,
         kwargs = add_progress(kwargs, self.repo.git, progress)
         proc = self.repo.git.pull(self, refspec, with_stdout=False, as_process=True,
                                   universal_newlines=True, v=True, **kwargs)
-        res = self._get_fetch_info_from_stderr(proc, progress, timeout=timeout)
+        res = self._get_fetch_info_from_stderr(proc, progress,
+                                               kill_after_timeout=kill_after_timeout)
         if hasattr(self.repo.odb, 'update_cache'):
             self.repo.odb.update_cache()
         return res
 
     def push(self, refspec: Union[str, List[str], None] = None,
              progress: Union[RemoteProgress, 'UpdateProgress', Callable[..., RemoteProgress], None] = None,
-             timeout: Union[None, float] = None, **kwargs: Any) -> IterableList[PushInfo]:
+             kill_after_timeout: Union[None, float] = None,
+             **kwargs: Any) -> IterableList[PushInfo]:
         """Push changes from source branch in refspec to target branch in refspec.
 
         :param refspec: see 'fetch' method
@@ -903,6 +911,9 @@ def push(self, refspec: Union[str, List[str], None] = None,
               overrides the ``update()`` function.
 
         :note: No further progress information is returned after push returns.
+        :param kill_after_timeout:
+            To specify a timeout in seconds for the git command, after which the process
+            should be killed. It is set to None by default.
         :param kwargs: Additional arguments to be passed to git-push
         :return:
             list(PushInfo, ...) list of PushInfo instances, each
@@ -914,8 +925,11 @@ def push(self, refspec: Union[str, List[str], None] = None,
             be 0."""
         kwargs = add_progress(kwargs, self.repo.git, progress)
         proc = self.repo.git.push(self, refspec, porcelain=True, as_process=True,
-                                  universal_newlines=True, **kwargs)
-        return self._get_push_info(proc, progress, timeout=timeout)
+                                  universal_newlines=True,
+                                  kill_after_timeout=kill_after_timeout,
+                                  **kwargs)
+        return self._get_push_info(proc, progress,
+                                   kill_after_timeout=kill_after_timeout)
 
     @ property
     def config_reader(self) -> SectionConstraint[GitConfigParser]: