ensure gevent support

.github/workflows/test.yml

@@ -64,6 +64,19 @@ jobs:
             python-version: "3.9"
             enable-coverage: "false"
 
+  test-3-12-gevent:
+    needs: build
+    runs-on: ubuntu-latest
+    env:
+      JUPYTER_PLATFORM_DIRS: 1
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ewoks-kit/.github/.github/actions/setup-python-package@main
+        with:
+          python-version: "3.12"
+          install-extras: "test"
+      - run: python -m gevent.monkey --module pytest . -v -ra -W error
+
   # Run linter / checks
   checks:
     uses: ewoks-kit/.github/.github/workflows/python-check.yml@main

pyproject.toml

@@ -30,6 +30,7 @@ full = [
 test = [
     "ewoksutils[full]",
     "pytest >=7",
+    "gevent",
 ]
 dev = [
     "ewoksutils[test]",

src/ewoksutils/logging_utils/cleanup.py

@@ -49,8 +49,13 @@ def cleanup_handler(handler: logging.Handler) -> None:
         try:
             q = handler.queue
             if isinstance(q, queue.Queue):
-                with q.mutex:
+                # gevent's monkey-patched `queue.Queue` has no `mutex` attribute.
+                mutex = getattr(q, "mutex", None)
+                if mutex is None:
                     q.queue.clear()
+                else:
+                    with mutex:
+                        q.queue.clear()
         finally:
             handler.release()
     handler.close()

src/ewoksutils/logging_utils/sqlite3.py

@@ -1,4 +1,5 @@
 import sqlite3
+import time
 from typing import Any
 from typing import Dict
 from typing import List
@@ -19,21 +20,26 @@ def __init__(
         field_types: Dict,
         timeout: float = 10,
         disconnect_on_error: bool = False,
+        retry_period: Optional[float] = None,
     ):
         """
         :param uri: for example "file:/path/to/test.db" or "file:///path/to/test.db".
         :param table: name of the database table in which records are inserted
                       (the table is created when missing).
         :param field_types: mapping from record attribute names (table columns)
                             to python types.
-        :param timeout: native sqlite3 busy timeout: the maximum time to wait
-                        for database locks to be released by other connections.
-                        A record is dropped when the timeout is reached.
+        :param timeout: maximum time to wait for database locks to be released
+                        by other connections. A record is dropped when the
+                        timeout is reached.
         :param disconnect_on_error: disconnect when emitting a record failed.
+        :param retry_period: when `None` (default), `timeout` is used as sqlite3's
+                        native busy timeout. When set, retrying is done at the
+                        python level instead.
         """
         super().__init__(disconnect_on_error=disconnect_on_error)
         self._uri = uri
         self._timeout = timeout
+        self._retry_period = retry_period
         self._field_sql_types = sqlite3_utils.python_to_sql_types(field_types)
 
         self._ensure_table_query = sqlite3_utils.ensure_table_query(
@@ -47,8 +53,15 @@ def __init__(
         self._connection_context = None
 
     def _connect(self) -> None:
+        if self._retry_period is None:
+            native_timeout = self._timeout
+        else:
+            native_timeout = min(self._timeout, self._retry_period)
         ctx = sqlite3_utils.connect(
-            self._uri, timeout=self._timeout, uri=True, check_same_thread=False
+            self._uri,
+            timeout=native_timeout,
+            uri=True,
+            check_same_thread=False,
         )
         try:
             conn = ctx.__enter__()
@@ -88,6 +101,17 @@ def _sql_query(
     ) -> None:
         if conn is None:
             conn = self._connection
+        if conn is None:
+            raise RuntimeError("Sqlite3Handler is not connected")
+        if self._retry_period is None:
+            self._sql_query_single_attempt(sql, parameters, conn)
+        else:
+            self._sql_query_with_retries(sql, parameters, conn, self._retry_period)
+
+    @staticmethod
+    def _sql_query_single_attempt(
+        sql: str, parameters: Sequence, conn: sqlite3.Connection
+    ) -> None:
         try:
             conn.execute(sql, parameters)
             conn.commit()
@@ -98,3 +122,41 @@ def _sql_query(
             except sqlite3.OperationalError:
                 pass
             raise
+
+    def _sql_query_with_retries(
+        self,
+        sql: str,
+        parameters: Sequence,
+        conn: sqlite3.Connection,
+        retry_period: float,
+    ) -> None:
+        start = time.time()
+        while True:
+            try:
+                conn.execute(sql, parameters)
+                conn.commit()
+                return
+            except BaseException as ex:
+                # Do not leave the failed query pending in an open transaction.
+                try:
+                    conn.rollback()
+                except sqlite3.OperationalError:
+                    pass
+
+                # retry when certain exceptions
+                do_retry = self._retry_sqlite3_exception(ex)
+                if do_retry and time.time() - start < self._timeout:
+                    time.sleep(retry_period)
+                    continue
+
+                raise
+
+    @staticmethod
+    def _retry_sqlite3_exception(ex: BaseException) -> bool:
+        if not isinstance(ex, sqlite3.OperationalError):
+            return False
+        error_message = str(ex)
+        if "database is locked" in error_message:
+            # transient lock by another connection.
+            return True
+        return False

src/ewoksutils/tests/test_logging_sqlite3.py

@@ -2,13 +2,26 @@
 import sqlite3
 import threading
 import time
+from typing import Optional
 
 from .. import sqlite3_utils
 from ..logging_utils.sqlite3 import Sqlite3Handler
 
 FIELD_TYPES = {"field1": 0, "field2": ""}
 
 
+def _sqlite3_retry_period() -> Optional[float]:
+    """Sqlite3Handler's native busy timeout blocks without yielding, which
+    can deadlock a cooperative event loop (e.g. gevent) when the lock holder
+    runs in another greenlet. Retry at the python level instead in that case.
+    """
+    try:
+        from gevent.monkey import is_module_patched
+    except ImportError:
+        return None
+    return 0.1 if is_module_patched("threading") else None
+
+
 def test_concurrent_read_write(tmp_path):
     """Records emitted from several handlers while a reader is polling
     must all be inserted."""
@@ -66,14 +79,16 @@ def read_records():
 
 def test_write_blocked_by_reader(tmp_path):
     """A writer must wait for read locks to be released."""
+    retry_period = _sqlite3_retry_period()
     lock_seconds = 0.5
-    retry_period = lock_seconds / 5  # do not retry long enough
-    retry_timeout = 5 * lock_seconds  # retry long enough
+    short_timeout = lock_seconds / 5  # do not retry long enough
+    long_timeout = 5 * lock_seconds  # retry long enough
 
     # Add record to the database
     uri = str(tmp_path / "test.db")
-    handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES)
+    handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES, retry_period=retry_period)
     handler.handle(_make_record(field1=1))
+    handler.close()
 
     # Reader
     read_lock_acquired = threading.Event()
@@ -99,7 +114,9 @@ def read_lock_database():
     thread = threading.Thread(target=read_lock_database)
     thread.start()
 
-    handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES, timeout=retry_period)
+    handler = Sqlite3Handler(
+        uri, "mytable", FIELD_TYPES, timeout=short_timeout, retry_period=retry_period
+    )
     try:
         assert read_lock_acquired.wait(timeout=10)
         handler.handle(_make_record(field1=2))
@@ -112,7 +129,9 @@ def read_lock_database():
     thread = threading.Thread(target=read_lock_database)
     thread.start()
 
-    handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES, timeout=retry_timeout)
+    handler = Sqlite3Handler(
+        uri, "mytable", FIELD_TYPES, timeout=long_timeout, retry_period=retry_period
+    )
     try:
         assert read_lock_acquired.wait(timeout=10)
         # Write while the reader locks