Repository: gorakhargosh/watchdog (python)

Issue:
watchdog runs too many times / repeatedly when many files are changed
I am using watchdog to watch a git repository and rerun a build process every time the repo changes.

I have observed that performing a git pull, for example, can cause watchdog to run repeatedly.  I *believe* that the git pull is causing a very large number of files to change, and hence watchdog runs many times because watchdog runs once per file modification.

For me, this is not the desired behavior.  If 5 files are modified at the same time, for example, I only need watchdog to run the build process once.  I am hoping we can add a configuration option to watchdog to allow somebody to opt-in to the following proposed behavior "lazy run":

## Proposed behavior "lazy run":

  * Watchdog maintains a queue of file change events.  This happens concurrently.

  1. Check the queue.
  2. If the queue is nonempty, empty the queue and run the process.
  3. (When the process is finished,) return to step 1.


## Example of "lazy run" in action:

Suppose that the process to run takes a long time, and 5 files are modified in rapid succession.

  1. First file modified.
  2. modification 1 goes in queue
  3. watchdog empties queue and begins running the process
  4. Second file modified
  5. Third file modified
  6. Fourth file modified
  7. Fifth file modified
  8. modifications 2 through 5 are in the queue.
  9. The process completes and watchdog checks the queue again
  10. Modifications are discarded from the queue and watchdog begins running the process again
  11. The process completes again.

As you can see, in this example, even though 5 files were modified, the process only runs 2 times.


Pull request:
Add optional event debouncing for auto-restart
If a debounce interval is specified, upon receiving events, collect events until the interval passes with no incoming events, and only then trigger a restart.

Closes #542.

Note to reviewers: The more involved start and stop mechanisms in AutoRestartTrick are needed to avoid restarting the debouncer when restart the process, and to ensure that the debouncer is properly stopped when stopping the trick.

Patch:
diff --git a/changelog.rst b/changelog.rst
index 2e81f8b..b4410ca 100644
--- a/changelog.rst
+++ b/changelog.rst
@@ -9,7 +9,8 @@ Changelog
 2023-xx-xx • `full history <https://github.com/gorakhargosh/watchdog/compare/v2.2.1...HEAD>`__
 
 - [inotify] Add support for ``IN_OPEN`` events: a ``FileOpenedEvent`` event will be fired. (`#941 <https://github.com/gorakhargosh/watchdog/pull/941>`__)
-- Thanks to our beloved contributors: @BoboTiG, @dstaple
+- [watchmedo] Add optional event debouncing for ``auto-restart``, only restarting once if many events happen in quick succession (`#940 <https://github.com/gorakhargosh/watchdog/pull/940>`__)
+- Thanks to our beloved contributors: @BoboTiG, @dstaple, @taleinat
 
 2.2.1
 ~~~~~
diff --git a/src/watchdog/tricks/__init__.py b/src/watchdog/tricks/__init__.py
index 6e7eaf6..e2ceace 100644
--- a/src/watchdog/tricks/__init__.py
+++ b/src/watchdog/tricks/__init__.py
@@ -46,10 +46,12 @@
 import os
 import signal
 import subprocess
+import threading
 import time
 
 from watchdog.events import PatternMatchingEventHandler
 from watchdog.utils import echo
+from watchdog.utils.event_debouncer import EventDebouncer
 from watchdog.utils.process_watcher import ProcessWatcher
 
 logger = logging.getLogger(__name__)
@@ -177,60 +179,117 @@ class AutoRestartTrick(Trick):
 
     def __init__(self, command, patterns=None, ignore_patterns=None,
                  ignore_directories=False, stop_signal=signal.SIGINT,
-                 kill_after=10):
+                 kill_after=10, debounce_interval_seconds=0):
+        if kill_after < 0:
+            raise ValueError("kill_after must be non-negative.")
+        if debounce_interval_seconds < 0:
+            raise ValueError("debounce_interval_seconds must be non-negative.")
+
         super().__init__(
             patterns=patterns, ignore_patterns=ignore_patterns,
             ignore_directories=ignore_directories)
+
         self.command = command
         self.stop_signal = stop_signal
         self.kill_after = kill_after
+        self.debounce_interval_seconds = debounce_interval_seconds
 
         self.process = None
         self.process_watcher = None
+        self.event_debouncer = None
+        self.restart_count = 0
+
+        self._is_process_stopping = False
+        self._is_trick_stopping = False
+        self._stopping_lock = threading.RLock()
 
     def start(self):
-        # windows doesn't have setsid
-        self.process = subprocess.Popen(self.command, preexec_fn=getattr(os, 'setsid', None))
-        self.process_watcher = ProcessWatcher(self.process, self._restart)
-        self.process_watcher.start()
+        if self.debounce_interval_seconds:
+            self.event_debouncer = EventDebouncer(
+                debounce_interval_seconds=self.debounce_interval_seconds,
+                events_callback=lambda events: self._restart_process(),
+            )
+            self.event_debouncer.start()
+        self._start_process()
 
     def stop(self):
-        if self.process is None:
+        # Ensure the body of the function is only run once.
+        with self._stopping_lock:
+            if self._is_trick_stopping:
+                return
+            self._is_trick_stopping = True
+
+        process_watcher = self.process_watcher
+        if self.event_debouncer is not None:
+            self.event_debouncer.stop()
+        self._stop_process()
+
+        # Don't leak threads: Wait for background threads to stop.
+        if self.event_debouncer is not None:
+            self.event_debouncer.join()
+        process_watcher.join()
+
+    def _start_process(self):
+        if self._is_trick_stopping:
             return
 
-        if self.process_watcher is not None:
-            self.process_watcher.stop()
-            self.process_watcher = None
+        # windows doesn't have setsid
+        self.process = subprocess.Popen(self.command, preexec_fn=getattr(os, 'setsid', None))
+        self.process_watcher = ProcessWatcher(self.process, self._restart_process)
+        self.process_watcher.start()
 
-        def kill_process(stop_signal):
-            if hasattr(os, 'getpgid') and hasattr(os, 'killpg'):
-                os.killpg(os.getpgid(self.process.pid), stop_signal)
-            else:
-                os.kill(self.process.pid, self.stop_signal)
+    def _stop_process(self):
+        # Ensure the body of the function is not run in parallel in different threads.
+        with self._stopping_lock:
+            if self._is_process_stopping:
+                return
+            self._is_process_stopping = True
 
         try:
-            kill_process(self.stop_signal)
-        except OSError:
-            # Process is already gone
-            pass
-        else:
-            kill_time = time.time() + self.kill_after
-            while time.time() < kill_time:
-                if self.process.poll() is not None:
-                    break
-                time.sleep(0.25)
-            else:
+            if self.process_watcher is not None:
+                self.process_watcher.stop()
+                self.process_watcher = None
+
+            if self.process is not None:
                 try:
-                    kill_process(9)
+                    kill_process(self.process.pid, self.stop_signal)
                 except OSError:
                     # Process is already gone
                     pass
-        self.process = None
+                else:
+                    kill_time = time.time() + self.kill_after
+                    while time.time() < kill_time:
+                        if self.process.poll() is not None:
+                            break
+                        time.sleep(0.25)
+                    else:
+                        try:
+                            kill_process(self.process.pid, 9)
+                        except OSError:
+                            # Process is already gone
+                            pass
+                self.process = None
+        finally:
+            self._is_process_stopping = False
 
     @echo_events
     def on_any_event(self, event):
-        self._restart()
+        if self.event_debouncer is not None:
+            self.event_debouncer.handle_event(event)
+        else:
+            self._restart_process()
+
+    def _restart_process(self):
+        if self._is_trick_stopping:
+            return
+        self._stop_process()
+        self._start_process()
+        self.restart_count += 1
+
 
-    def _restart(self):
-        self.stop()
-        self.start()
+if hasattr(os, 'getpgid') and hasattr(os, 'killpg'):
+    def kill_process(pid, stop_signal):
+        os.killpg(os.getpgid(pid), stop_signal)
+else:
+    def kill_process(pid, stop_signal):
+        os.kill(pid, stop_signal)
diff --git a/src/watchdog/utils/event_debouncer.py b/src/watchdog/utils/event_debouncer.py
new file mode 100644
index 0000000..f5e6e6c
--- /dev/null
+++ b/src/watchdog/utils/event_debouncer.py
@@ -0,0 +1,54 @@
+import logging
+import threading
+
+from watchdog.utils import BaseThread
+
+
+logger = logging.getLogger(__name__)
+
+
+class EventDebouncer(BaseThread):
+    """Background thread for debouncing event handling.
+
+    When an event is received, wait until the configured debounce interval
+    passes before calling the callback.  If additional events are received
+    before the interval passes, reset the timer and keep waiting.  When the
+    debouncing interval passes, the callback will be called with a list of
+    events in the order in which they were received.
+    """
+    def __init__(self, debounce_interval_seconds, events_callback):
+        super().__init__()
+        self.debounce_interval_seconds = debounce_interval_seconds
+        self.events_callback = events_callback
+
+        self._events = []
+        self._cond = threading.Condition()
+
+    def handle_event(self, event):
+        with self._cond:
+            self._events.append(event)
+            self._cond.notify()
+
+    def stop(self):
+        with self._cond:
+            super().stop()
+            self._cond.notify()
+
+    def run(self):
+        with self._cond:
+            while True:
+                # Wait for first event (or shutdown).
+                self._cond.wait()
+
+                if self.debounce_interval_seconds:
+                    # Wait for additional events (or shutdown) until the debounce interval passes.
+                    while self.should_keep_running():
+                        if not self._cond.wait(timeout=self.debounce_interval_seconds):
+                            break
+
+                if not self.should_keep_running():
+                    break
+
+                events = self._events
+                self._events = []
+                self.events_callback(events)
diff --git a/src/watchdog/watchmedo.py b/src/watchdog/watchmedo.py
index 976571b..ff14df4 100755
--- a/src/watchdog/watchmedo.py
+++ b/src/watchdog/watchmedo.py
@@ -585,7 +585,13 @@ def shell_command(args):
                    default=10.0,
                    type=float,
                    help='When stopping, kill the subprocess after the specified timeout '
-                   'in seconds (default 10.0).')])
+                   'in seconds (default 10.0).'),
+          argument('--debounce-interval',
+                   dest='debounce_interval',
+                   default=0.0,
+                   type=float,
+                   help='After a file change, Wait until the specified interval (in '
+                   'seconds) passes with no file changes, and only then restart.')])
 def auto_restart(args):
     """
     Command to start a long-running subprocess and restart it on matched events.
@@ -633,7 +639,8 @@ def handler_termination_signal(_signum, _frame):
                                ignore_patterns=ignore_patterns,
                                ignore_directories=args.ignore_directories,
                                stop_signal=stop_signal,
-                               kill_after=args.kill_after)
+                               kill_after=args.kill_after,
+                               debounce_interval_seconds=args.debounce_interval)
     handler.start()
     observer = Observer(timeout=args.timeout)
     try:


Test patch:
diff --git a/tests/test_0_watchmedo.py b/tests/test_0_watchmedo.py
index 0f76e98..224cd55 100644
--- a/tests/test_0_watchmedo.py
+++ b/tests/test_0_watchmedo.py
@@ -106,17 +106,69 @@ def test_shell_command_subprocess_termination_nowait(tmpdir):
     assert not trick.is_process_running()
 
 
+def test_auto_restart_on_file_change(tmpdir, capfd):
+    """Simulate changing 3 files.
+
+    Expect 3 restarts.
+    """
+    from watchdog.tricks import AutoRestartTrick
+    import sys
+    import time
+    script = make_dummy_script(tmpdir, n=2)
+    trick = AutoRestartTrick([sys.executable, script])
+    trick.start()
+    time.sleep(1)
+    trick.on_any_event("foo/bar.baz")
+    trick.on_any_event("foo/bar2.baz")
+    trick.on_any_event("foo/bar3.baz")
+    time.sleep(1)
+    trick.stop()
+    cap = capfd.readouterr()
+    assert cap.out.splitlines(keepends=False).count('+++++ 0') >= 2
+    assert trick.restart_count == 3
+
+
+def test_auto_restart_on_file_change_debounce(tmpdir, capfd):
+    """Simulate changing 3 files quickly and then another change later.
+
+    Expect 2 restarts due to debouncing.
+    """
+    from watchdog.tricks import AutoRestartTrick
+    import sys
+    import time
+    script = make_dummy_script(tmpdir, n=2)
+    trick = AutoRestartTrick([sys.executable, script], debounce_interval_seconds=0.5)
+    trick.start()
+    time.sleep(1)
+    trick.on_any_event("foo/bar.baz")
+    trick.on_any_event("foo/bar2.baz")
+    time.sleep(0.1)
+    trick.on_any_event("foo/bar3.baz")
+    time.sleep(1)
+    trick.on_any_event("foo/bar.baz")
+    time.sleep(1)
+    trick.stop()
+    cap = capfd.readouterr()
+    assert cap.out.splitlines(keepends=False).count('+++++ 0') == 3
+    assert trick.restart_count == 2
+
+
 def test_auto_restart_subprocess_termination(tmpdir, capfd):
+    """Run auto-restart with a script that terminates in about 2 seconds.
+
+    After 5 seconds, expect it to have been restarted at least once.
+    """
     from watchdog.tricks import AutoRestartTrick
     import sys
     import time
     script = make_dummy_script(tmpdir, n=2)
-    a = AutoRestartTrick([sys.executable, script])
-    a.start()
+    trick = AutoRestartTrick([sys.executable, script])
+    trick.start()
     time.sleep(5)
-    a.stop()
+    trick.stop()
     cap = capfd.readouterr()
     assert cap.out.splitlines(keepends=False).count('+++++ 0') > 1
+    assert trick.restart_count >= 1
 
 
 def test_auto_restart_arg_parsing_basic():
@@ -129,11 +181,14 @@ def test_auto_restart_arg_parsing_basic():
 
 
 def test_auto_restart_arg_parsing():
-    args = watchmedo.cli.parse_args(["auto-restart", "-d", ".", "--kill-after", "12.5", "cmd"])
+    args = watchmedo.cli.parse_args(
+        ["auto-restart", "-d", ".", "--kill-after", "12.5", "--debounce-interval=0.2", "cmd"]
+    )
     assert args.func is watchmedo.auto_restart
     assert args.command == "cmd"
     assert args.directories == ["."]
     assert args.kill_after == pytest.approx(12.5)
+    assert args.debounce_interval == pytest.approx(0.2)
 
 
 def test_shell_command_arg_parsing():