@@ -19,6 +19,7 @@
 import sys
 from textwrap import dedent
 import threading
+import time
 import warnings
 
 from git.compat import defenc, force_bytes, safe_decode
@@ -99,6 +100,41 @@
 ## @{
 
 
+def _kill_process(pid: int) -> bool:
+    """Kill a POSIX process and its descendants, returning whether it was killed."""
+    # Collect descendants before signalling, while their parent PIDs still identify them.
+    pids = [pid]
+    for parent_pid in pids:
+        try:
+            try:
+                p = Popen(["pgrep", "-P", str(parent_pid)], stdout=PIPE)
+            except FileNotFoundError:
+                # POSIX ps does not support selecting by parent PID.
+                with Popen(["ps", "-A", "-o", "pid=", "-o", "ppid="], stdout=PIPE) as p:
+                    if p.stdout is not None:
+                        for line in p.stdout:
+                            fields = line.split()
+                            if len(fields) == 2 and all(field.isdigit() for field in fields):
+                                if int(fields[1]) == parent_pid:
+                                    pids.append(int(fields[0]))
+            else:
+                with p:
+                    if p.stdout is not None:
+                        for line in p.stdout:
+                            if line.strip().isdigit():
+                                pids.append(int(line))
+        except OSError as ex:
+            _logger.info("Unable to enumerate child processes: %r", ex)
+    killed = False
+    for process_pid in pids:
+        try:
+            os.kill(process_pid, signal.SIGKILL)
+            killed = killed or process_pid == pid
+        except OSError:
+            pass
+    return killed
+
+
 def handle_process_output(
     process: Union["Git.AutoInterrupt", Popen],
     stdout_handler: Union[
@@ -165,7 +201,7 @@ def pump_stream(
 
         except Exception as ex:
             _logger.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}")
-            if "I/O operation on closed file" not in str(ex):
+            if not (isinstance(ex, ValueError) and stream.closed):
                 # 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:
@@ -200,10 +236,13 @@ def pump_stream(
         threads.append(t)
 
     # FIXME: Why join? Will block if stdin needs feeding...
+    deadline = None if kill_after_timeout is None else time.monotonic() + kill_after_timeout
     for t in threads:
-        t.join(timeout=kill_after_timeout)
+        t.join(timeout=None if deadline is None else max(0, deadline - time.monotonic()))
         if t.is_alive():
             if isinstance(process, Git.AutoInterrupt):
+                if sys.platform != "win32" and process.proc is not None:
+                    _kill_process(process.proc.pid)
                 process._terminate()
             else:  # Don't want to deal with the other case.
                 raise RuntimeError(
@@ -221,6 +260,7 @@ def pump_stream(
                 # We ignore typing on the next line because mypy does not like the way
                 # we inferred that stderr takes str or bytes.
                 stderr_handler(error_str)  # type: ignore[arg-type]
+            break
 
     if finalizer:
         finalizer(process)
@@ -344,37 +384,45 @@ def _terminate(self) -> None:
 
         proc = self.proc
         self.proc = None
-        if proc.stdin:
-            proc.stdin.close()
-        if proc.stdout:
-            proc.stdout.close()
-        if proc.stderr:
-            proc.stderr.close()
-        # Did the process finish already so we have a return code?
         try:
-            if proc.poll() is not None:
-                self.status = self._status_code_if_terminate or proc.poll()
+            if proc.stdin:
+                # A timed-out process may already have exited before input is flushed.
+                with contextlib.suppress(BrokenPipeError):
+                    proc.stdin.close()
+            # Did the process finish already so we have a return code?
+            try:
+                if proc.poll() is not None:
+                    self.status = self._status_code_if_terminate or proc.poll()
+                    return
+            except OSError as ex:
+                _logger.info("Ignored error after process had died: %r", ex)
+
+            # It can be that nothing really exists anymore...
+            if os is None or getattr(os, "kill", None) is None:
                 return
-        except OSError as ex:
-            _logger.info("Ignored error after process had died: %r", ex)
 
-        # It can be that nothing really exists anymore...
-        if os is None or getattr(os, "kill", None) is None:
-            return
+            # Try to kill it.
+            try:
+                proc.terminate()
+            except (OSError, AttributeError) 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
+                # to silently ignore to avoid noisy "Exception ignored in: __del__" messages.
+                _logger.info("Ignored error while terminating process: %r", ex)
+                return
+            # END exception handling
+        finally:
+            if proc.stdout:
+                proc.stdout.close()
+            if proc.stderr:
+                proc.stderr.close()
 
-        # Try to kill it.
         try:
-            proc.terminate()
-            status = proc.wait()  # Ensure the process goes away.
-
+            status = proc.wait()
             self.status = self._status_code_if_terminate or status
         except (OSError, AttributeError) 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
-            # to silently ignore to avoid noisy "Exception ignored in: __del__" messages.
-            _logger.info("Ignored error while terminating process: %r", ex)
-        # END exception handling
+            _logger.info("Ignored error while waiting for terminated process: %r", ex)
 
     def __del__(self) -> None:
         self._terminate()
@@ -1537,43 +1585,9 @@ def execute(
             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.
-                """
-                child_pids = []
-                try:
-                    p = Popen(["pgrep", "-P", str(pid)], stdout=PIPE)
-                except FileNotFoundError:
-                    # POSIX ps does not support selecting by parent PID.
-                    with Popen(["ps", "-A", "-o", "pid=", "-o", "ppid="], stdout=PIPE) as p:
-                        if p.stdout is not None:
-                            for line in p.stdout:
-                                fields = line.split()
-                                if len(fields) == 2 and all(field.isdigit() for field in fields):
-                                    if int(fields[1]) == pid:
-                                        child_pids.append(int(fields[0]))
-                else:
-                    with p:
-                        if p.stdout is not None:
-                            for line in p.stdout:
-                                if line.strip().isdigit():
-                                    child_pids.append(int(line))
-                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.
+                if _kill_process(pid):
                     assert kill_check is not None
                     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 make_timeout_error() -> Union[str, bytes]:
                 err = f'Timeout: the command "{" ".join(redacted_command)}" did not complete in {timeout:g} secs.'
 
@@ -1,3 +1,9 @@
+from subprocess import PIPE
+import sys
+import threading
+
+import pytest
+
 from git.cmd import Git
 
 
@@ -31,3 +37,36 @@ def test_autointerrupt_terminate_ignores_attributeerror():
 
     # Ensure the reference is cleared to avoid repeated attempts.
     assert ai.proc is None
+
+
+def test_autointerrupt_terminate_closes_buffered_stdin():
+    ai = Git().hash_object("--stdin", as_process=True, istream=PIPE)
+    proc = ai.proc
+    proc.stdin.write(b"buffered input")
+    try:
+        ai._terminate()
+        assert proc.stdin.closed
+        assert proc.stdout.closed
+        assert proc.stderr.closed
+    finally:
+        proc.stdout.close()
+        proc.stderr.close()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX SIGTERM handling")
+def test_autointerrupt_terminate_closes_pipes_before_waiting():
+    script = (
+        "import signal, sys; signal.signal(signal.SIGTERM, signal.SIG_IGN); "
+        "print('ready', flush=True); sys.stdin.read(); sys.stdout.buffer.write(b'x' * 1000000)"
+    )
+    ai = Git().execute([sys.executable, "-c", script], as_process=True, istream=PIPE)
+    proc = ai.proc
+    assert proc.stdout.readline() == b"ready\n"
+    watchdog = threading.Timer(5, proc.kill)
+    watchdog.start()
+    try:
+        ai._terminate()
+        assert proc.returncode != -9, "cleanup waited for exit with undrained output pipes"
+    finally:
+        watchdog.cancel()
+        watchdog.join()
@@ -18,6 +18,7 @@
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 from unittest import mock, skipUnless
 
@@ -367,25 +368,49 @@ def portable_popen(args, **kwargs):
             self.assertFalse(marker.exists(), "the direct child survived the timeout")
 
     @skipUnless(sys.platform != "win32", "kill_after_timeout is not supported on Windows")
-    def test_timeout_ps_fallback_selects_only_direct_children(self):
+    def test_timeout_ps_fallback_selects_only_descendants(self):
         process = mock.MagicMock()
         process.pid = 1234
         process.communicate.return_value = (b"", b"")
         process.returncode = -signal.SIGKILL
         ps = mock.MagicMock()
         ps.__enter__.return_value = ps
-        ps.stdout = io.BytesIO(b"PID PPID\n 321 1\n 5678 1234\n 9012 5678\n\n")
+        ps.stdout = mock.MagicMock()
+        ps.stdout.__iter__.side_effect = lambda: iter([b"PID PPID\n", b" 321 1\n", b" 5678 1234\n", b" 9012 5678\n"])
+
+        def portable_popen(args, **kwargs):
+            if args[0] == "pgrep":
+                raise FileNotFoundError("pgrep is not installed")
+            return ps
 
         with contextlib.ExitStack() as stack:
             stack.enter_context(mock.patch.object(cmd, "safer_popen", return_value=process))
-            stack.enter_context(mock.patch.object(cmd, "Popen", side_effect=[FileNotFoundError, ps]))
+            stack.enter_context(mock.patch.object(cmd, "Popen", side_effect=portable_popen))
             kill = stack.enter_context(mock.patch.object(cmd.os, "kill"))
             timer = stack.enter_context(mock.patch.object(cmd.threading, "Timer"))
             # Run the timeout callback synchronously, with no real processes or signals.
             timer.return_value.start.side_effect = lambda: timer.call_args.args[1](1234)
             self.git.execute(["git", "version"], kill_after_timeout=1, with_exceptions=False)
 
-        self.assertEqual(kill.call_args_list, [mock.call(1234, signal.SIGKILL), mock.call(5678, signal.SIGKILL)])
+        self.assertEqual(
+            kill.call_args_list,
+            [mock.call(1234, signal.SIGKILL), mock.call(5678, signal.SIGKILL), mock.call(9012, signal.SIGKILL)],
+        )
+
+    @skipUnless(sys.platform != "win32", "requires POSIX timeout signalling")
+    def test_timeout_without_process_lookup_tools(self):
+        process = self.git.execute([sys.executable, "-c", "import time; time.sleep(30)"], as_process=True)
+        proc = process.proc
+        assert proc is not None
+        try:
+            with mock.patch.object(cmd, "Popen", side_effect=FileNotFoundError("lookup unavailable")):
+                cmd.handle_process_output(process, None, None, kill_after_timeout=0.1)
+            self.assertIsNotNone(proc.poll())
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait()
+                process._terminate()
 
     def test_it_executes_git_without_stdout_redirect(self):
         returncode, stdout, stderr = self.git.execute(
@@ -1006,6 +1031,29 @@ def stderr_handler(line):
         self.assertEqual(len(actual_lines[1]), expected_line_count, repr(actual_lines[1]))
         self.assertEqual(len(actual_lines[2]), expected_line_count, repr(actual_lines[2]))
 
+    @skipUnless(sys.platform != "win32", "requires POSIX timeout signalling")
+    def test_timeout_does_not_wait_for_blocked_output_handler(self):
+        release = threading.Event()
+        finished = threading.Event()
+
+        def handler(line):
+            try:
+                release.wait(5)
+            finally:
+                finished.set()
+
+        process = self.git.execute(
+            [sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(30)"],
+            as_process=True,
+        )
+        started = time.monotonic()
+        try:
+            cmd.handle_process_output(process, handler, None, kill_after_timeout=0.5)
+            self.assertLess(time.monotonic() - started, 3)
+        finally:
+            release.set()
+            self.assertTrue(finished.wait(5))
+
     def test_execute_kwargs_set_agrees_with_method(self):
         parameter_names = inspect.signature(cmd.Git.execute).parameters.keys()
         self_param, command_param, *most_params, extra_kwargs_param = parameter_names