Merge branch 'gorakhargosh:master' into master

Author: TobiasRzepka

changelog.rst

@@ -3,13 +3,21 @@
 Changelog
 ---------
 
-5.0.3 (dev)
-~~~~~~~~~~~
+5.0.4-dev
+~~~~~~~~~
 
-2024-xx-xx • `full history <https://github.com/gorakhargosh/watchdog/compare/v5.0.2...HEAD>`__
+2024-xx-xx • `full history <https://github.com/gorakhargosh/watchdog/compare/v5.0.3...HEAD>`__
 
-- 
-- Thanks to our beloved contributors: @BoboTiG
+-
+- Thanks to our beloved contributors: @BoboTiG, @
+
+5.0.3
+~~~~~
+
+2024-09-27 • `full history <https://github.com/gorakhargosh/watchdog/compare/v5.0.2...v5.0.3>`__
+
+- [inotify] Improve cleaning up ``Inotify`` threads, and add ``eventlet`` test cases (`#1070 <https://github.com/gorakhargosh/watchdog/pull/1070>`__)
+- Thanks to our beloved contributors: @BoboTiG, @ethan-vanderheijden
 
 5.0.2
 ~~~~~

docs/source/global.rst.inc

@@ -4,7 +4,7 @@
 .. |author_email| replace:: contact@tiger-222.fr
 .. |copyright| replace:: Copyright 2011-2024 Yesudeep Mangalapilly, Mickaël Schoentgen & contributors.
 .. |project_name| replace:: ``watchdog``
-.. |project_version| replace:: 5.0.3
+.. |project_version| replace:: 5.0.4
 
 .. _issue tracker: https://github.com/gorakhargosh/watchdog/issues
 .. _code repository: https://github.com/gorakhargosh/watchdog

src/watchdog/observers/inotify_c.py

@@ -5,6 +5,7 @@
 import ctypes.util
 import errno
 import os
+import select
 import struct
 import threading
 from ctypes import c_char_p, c_int, c_uint32
@@ -148,6 +149,9 @@ def __init__(self, path: bytes, *, recursive: bool = False, event_mask: int | No
             Inotify._raise_error()
         self._inotify_fd = inotify_fd
         self._lock = threading.Lock()
+        self._closed = False
+        self._waiting_to_read = True
+        self._kill_r, self._kill_w = os.pipe()
 
         # Stores the watch descriptor for a given path.
         self._wd_for_path: dict[bytes, int] = {}
@@ -230,13 +234,19 @@ def remove_watch(self, path: bytes) -> None:
     def close(self) -> None:
         """Closes the inotify instance and removes all associated watches."""
         with self._lock:
-            if self._path in self._wd_for_path:
-                wd = self._wd_for_path[self._path]
-                inotify_rm_watch(self._inotify_fd, wd)
+            if not self._closed:
+                self._closed = True
 
-            # descriptor may be invalid because file was deleted
-            with contextlib.suppress(OSError):
-                os.close(self._inotify_fd)
+                if self._path in self._wd_for_path:
+                    wd = self._wd_for_path[self._path]
+                    inotify_rm_watch(self._inotify_fd, wd)
+
+                if self._waiting_to_read:
+                    # inotify_rm_watch() should write data to _inotify_fd and wake
+                    # the thread, but writing to the kill channel will gaurentee this
+                    os.write(self._kill_w, b"!")
+                else:
+                    self._close_resources()
 
     def read_events(self, *, event_buffer_size: int = DEFAULT_EVENT_BUFFER_SIZE) -> list[InotifyEvent]:
         """Reads events from inotify and yields them."""
@@ -276,6 +286,21 @@ def _recursive_simulate(src_path: bytes) -> list[InotifyEvent]:
         event_buffer = None
         while True:
             try:
+                with self._lock:
+                    if self._closed:
+                        return []
+
+                    self._waiting_to_read = True
+
+                select.select([self._inotify_fd, self._kill_r], [], [])
+
+                with self._lock:
+                    self._waiting_to_read = False
+
+                    if self._closed:
+                        self._close_resources()
+                        return []
+
                 event_buffer = os.read(self._inotify_fd, event_buffer_size)
             except OSError as e:
                 if e.errno == errno.EINTR:
@@ -340,6 +365,11 @@ def _recursive_simulate(src_path: bytes) -> list[InotifyEvent]:
 
         return event_list
 
+    def _close_resources(self) -> None:
+        os.close(self._inotify_fd)
+        os.close(self._kill_r)
+        os.close(self._kill_w)
+
     # Non-synchronized methods.
     def _add_dir_watch(self, path: bytes, mask: int, *, recursive: bool) -> None:
         """Adds a watch (optionally recursively) for the given directory path

src/watchdog/utils/bricks.py

@@ -72,14 +72,16 @@ def _init(self, maxsize: int) -> None:
         super()._init(maxsize)
         self._last_item = None
 
-    def _put(self, item: Any) -> None:
+    def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:  # noqa: FBT001,FBT002
+        """This method will be used by `eventlet`, when enabled, so we cannot use force proper keyword-only
+        arguments nor touch the signature. Also, the `timeout` argument will be ignored in that case.
+        """
         if self._last_item is None or item != self._last_item:
-            super()._put(item)
-            self._last_item = item
-        else:
-            # `put` increments `unfinished_tasks` even if we did not put
-            # anything into the queue here
-            self.unfinished_tasks -= 1
+            super().put(item, block, timeout)
+
+    def _put(self, item: Any) -> None:
+        super()._put(item)
+        self._last_item = item
 
     def _get(self) -> Any:
         item = super()._get()

src/watchdog/version.py

@@ -4,7 +4,7 @@
 # ``docs/source/global.rst.inc`` file as well.
 VERSION_MAJOR = 5
 VERSION_MINOR = 0
-VERSION_BUILD = 3
+VERSION_BUILD = 4
 VERSION_INFO = (VERSION_MAJOR, VERSION_MINOR, VERSION_BUILD)
 VERSION_STRING = f"{VERSION_MAJOR}.{VERSION_MINOR}.{VERSION_BUILD}"
 

tests/isolated/__init__.py

@@ -0,0 +1,31 @@
+if __name__ == "__main__":
+    import eventlet
+
+    eventlet.monkey_patch()
+
+    import signal
+    import sys
+    import tempfile
+
+    from watchdog.events import LoggingEventHandler
+    from watchdog.observers import Observer
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+
+        def run_observer():
+            event_handler = LoggingEventHandler()
+            observer = Observer()
+            observer.schedule(event_handler, temp_dir)
+            observer.start()
+            eventlet.sleep(1)
+            observer.stop()
+
+        def on_alarm(signum, frame):
+            print("Observer.stop() never finished!", file=sys.stderr)  # noqa: T201
+            sys.exit(1)
+
+        signal.signal(signal.SIGALRM, on_alarm)
+        signal.alarm(4)
+
+        thread = eventlet.spawn(run_observer)
+        thread.wait()

tests/isolated/eventlet_observer_stops.py

@@ -0,0 +1,33 @@
+if __name__ == "__main__":
+    import eventlet
+
+    eventlet.monkey_patch()
+
+    from watchdog.utils.bricks import SkipRepeatsQueue
+
+    q = SkipRepeatsQueue(10)
+    q.put("A")
+    q.put("A")
+    q.put("A")
+    q.put("A")
+    q.put("B")
+    q.put("A")
+
+    value = q.get()
+    assert value == "A"
+    q.task_done()
+
+    assert q.unfinished_tasks == 2
+
+    value = q.get()
+    assert value == "B"
+    q.task_done()
+
+    assert q.unfinished_tasks == 1
+
+    value = q.get()
+    assert value == "A"
+    q.task_done()
+
+    assert q.empty()
+    assert q.unfinished_tasks == 0

tests/isolated/eventlet_skip_repeat_queue.py

@@ -11,6 +11,7 @@
 import errno
 import logging
 import os
+import select
 import struct
 from typing import TYPE_CHECKING
 from unittest.mock import patch
@@ -56,6 +57,13 @@ def test_late_double_deletion(helper: Helper, p: P, event_queue: TestEventQueue,
         + struct_inotify(wd=3, mask=const.IN_IGNORED)
     )
 
+    select_bkp = select.select
+
+    def fakeselect(read_list, *args, **kwargs):
+        if inotify_fd in read_list:
+            return [inotify_fd], [], []
+        return select_bkp(read_list, *args, **kwargs)
+
     os_read_bkp = os.read
 
     def fakeread(fd, length):
@@ -92,8 +100,9 @@ def inotify_rm_watch(fd, wd):
     mock3 = patch.object(inotify_c, "inotify_init", new=inotify_init)
     mock4 = patch.object(inotify_c, "inotify_add_watch", new=inotify_add_watch)
     mock5 = patch.object(inotify_c, "inotify_rm_watch", new=inotify_rm_watch)
+    mock6 = patch.object(select, "select", new=fakeselect)
 
-    with mock1, mock2, mock3, mock4, mock5:
+    with mock1, mock2, mock3, mock4, mock5, mock6:
         start_watching(path=p(""))
         # Watchdog Events
         for evt_cls in [DirCreatedEvent, DirDeletedEvent] * 2:

tests/markers.py

@@ -0,0 +1,25 @@
+import importlib
+
+import pytest
+
+from watchdog.utils import platform
+
+from .utils import run_isolated_test
+
+
+# Kqueue isn't supported by Eventlet, so BSD is out
+# Current usage ReadDirectoryChangesW on Windows is blocking, though async may be possible
+@pytest.mark.skipif(not platform.is_linux(), reason="Eventlet only supported in Linux")
+def test_observer_stops_in_eventlet():
+    if not importlib.util.find_spec("eventlet"):
+        pytest.skip("eventlet not installed")
+
+    run_isolated_test("eventlet_observer_stops.py")
+
+
+@pytest.mark.skipif(not platform.is_linux(), reason="Eventlet only supported in Linux")
+def test_eventlet_skip_repeat_queue():
+    if not importlib.util.find_spec("eventlet"):
+        pytest.skip("eventlet not installed")
+
+    run_isolated_test("eventlet_skip_repeat_queue.py")

tests/test_inotify_c.py

@@ -1,14 +1,10 @@
 from __future__ import annotations
 
-import pytest
-
 from watchdog import events
 from watchdog.utils.bricks import SkipRepeatsQueue
 
-from .markers import cpython_only
-
 
-def basic_actions():
+def test_basic_queue():
     q = SkipRepeatsQueue()
 
     e1 = (2, "fred")
@@ -25,10 +21,6 @@ def basic_actions():
     assert q.empty()
 
 
-def test_basic_queue():
-    basic_actions()
-
-
 def test_allow_nonconsecutive():
     q = SkipRepeatsQueue()
 
@@ -86,10 +78,3 @@ def test_consecutives_allowed_across_empties():
     q.put(e1)  # this repeat is allowed because 'last' added is now gone from queue
     assert e1 == q.get()
     assert q.empty()
-
-
-@cpython_only
-def test_eventlet_monkey_patching():
-    eventlet = pytest.importorskip("eventlet")
-    eventlet.monkey_patch()
-    basic_actions()

tests/test_isolated.py

@@ -2,6 +2,8 @@
 
 import dataclasses
 import os
+import subprocess
+import sys
 from queue import Queue
 from typing import Protocol
 
@@ -97,3 +99,29 @@ def close(self) -> None:
         alive = [emitter.is_alive() for emitter in self.emitters]
         self.emitters = []
         assert alive == [False] * len(alive)
+
+
+def run_isolated_test(path):
+    isolated_test_prefix = os.path.join("tests", "isolated")
+    path = os.path.abspath(os.path.join(isolated_test_prefix, path))
+
+    src_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src")
+    new_env = os.environ.copy()
+    new_env["PYTHONPATH"] = os.pathsep.join([*sys.path, src_dir])
+
+    new_argv = [sys.executable, path]
+
+    p = subprocess.Popen(
+        new_argv,
+        env=new_env,
+    )
+
+    # in case test goes haywire, don't let it run forever
+    timeout = 10
+    try:
+        p.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        p.kill()
+        raise
+
+    assert p.returncode == 0

tests/test_skip_repeats_queue.py

@@ -33,7 +33,7 @@ extras =
     watchmedo
 commands =
     python -m ruff format docs/source/examples src tests
-    python -m ruff check --fix src docs/source/examples tests
+    python -m ruff check --fix --unsafe-fixes src docs/source/examples tests
 
 [testenv:types]
 usedevelop = true