From 55e954d50ab5bc5a05d5da77b27c2b9c90204bc7 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Sun, 2 Aug 2026 14:51:39 +0200 Subject: [PATCH] ensure gevent support --- .github/workflows/test.yml | 13 ++++ pyproject.toml | 1 + src/ewoksutils/logging_utils/cleanup.py | 7 +- src/ewoksutils/logging_utils/sqlite3.py | 70 ++++++++++++++++++-- src/ewoksutils/tests/test_logging_sqlite3.py | 29 ++++++-- 5 files changed, 110 insertions(+), 10 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7b569fc..90cce15 100644 --- a/.github/workflows/test.yml +++ b/.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 diff --git a/pyproject.toml b/pyproject.toml index 6ffac0c..f19a527 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ full = [ test = [ "ewoksutils[full]", "pytest >=7", + "gevent", ] dev = [ "ewoksutils[test]", diff --git a/src/ewoksutils/logging_utils/cleanup.py b/src/ewoksutils/logging_utils/cleanup.py index 2af4689..79b9c25 100644 --- a/src/ewoksutils/logging_utils/cleanup.py +++ b/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() diff --git a/src/ewoksutils/logging_utils/sqlite3.py b/src/ewoksutils/logging_utils/sqlite3.py index ca37490..61b17e1 100644 --- a/src/ewoksutils/logging_utils/sqlite3.py +++ b/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,6 +20,7 @@ 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". @@ -26,14 +28,18 @@ def __init__( (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 diff --git a/src/ewoksutils/tests/test_logging_sqlite3.py b/src/ewoksutils/tests/test_logging_sqlite3.py index 8120c1e..5b438ec 100644 --- a/src/ewoksutils/tests/test_logging_sqlite3.py +++ b/src/ewoksutils/tests/test_logging_sqlite3.py @@ -2,6 +2,7 @@ import sqlite3 import threading import time +from typing import Optional from .. import sqlite3_utils from ..logging_utils.sqlite3 import Sqlite3Handler @@ -9,6 +10,18 @@ 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