diff --git a/CHANGELOG.md b/CHANGELOG.md index e042b4d..2655fa0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `RedisEwoksEventHandler`: add `disconnect_on_error` and `timeout` argument. +- `RedisEwoksEventReader`: add `timeout` argument and do not allow unknown arguments. +- `Sqlite3EwoksEventReader`: add `timeout` argument and do not allow unknown arguments. + +### Fixed + +`EwoksEventReader`: log error when last attempt failed when polling for events. + ## [1.6.0rc1] - 2026-07-13 ### Added diff --git a/pyproject.toml b/pyproject.toml index 7d41399..7d4ae39 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ classifiers = [ requires-python = ">=3.8" dependencies = [ "celery[tblib] >= 5.5", - "ewoksutils[full] >=1.10.0rc1", + "ewoksutils[full] >=2.0.0rc2", "pyyaml", ] @@ -30,7 +30,7 @@ Changelog = "https://github.com/ewoks-kit/ewoksjob/-/blob/main/CHANGELOG.md" [project.optional-dependencies] worker = [ - "ewoks >=6.0.0", + "ewoks >=7.0.0rc1", "objgraph", ] redis = [ diff --git a/src/ewoksjob/events/handlers/redis.py b/src/ewoksjob/events/handlers/redis.py index 8db0c82..edfb558 100644 --- a/src/ewoksjob/events/handlers/redis.py +++ b/src/ewoksjob/events/handlers/redis.py @@ -16,22 +16,44 @@ class RedisEwoksEventHandler(EwoksEventHandlerMixIn, ConnectionHandler): # TODO: https://redisql.redbeardlab.com/blog/python/using-redisql-with-python/ - def __init__(self, url: str, ttl=None): - """An example url is "redis://localhost:10003?db=2".""" + def __init__( + self, + url: str, + ttl: Optional[int] = None, + timeout: float = 10, + disconnect_on_error: bool = False, + ): + """ + :param url: for example "redis://localhost:10003?db=2". + :param ttl: time-to-live in seconds of the record keys. + :param timeout: native redis socket timeout: the maximum time to wait + for connecting and for command replies. + A record is dropped when the timeout is reached. + :param disconnect_on_error: disconnect when emitting a record failed. + """ + super().__init__(disconnect_on_error=disconnect_on_error) self._redis_url = url self._ttl = ttl - super().__init__() + self._timeout = timeout + self._connection = None - def _connect(self, timeout=1) -> None: + def _connect(self) -> None: """This is called when no connection exists.""" client_name = f"ewoks:writer:{socket.gethostname()}:{os.getpid()}" self._connection = redis.Redis.from_url( - self._redis_url, client_name=client_name + self._redis_url, + client_name=client_name, + socket_connect_timeout=self._timeout, + socket_timeout=self._timeout, ) def _disconnect(self) -> None: """This is called when a connection exists and is connected.""" self._connection.close() + self._connection = None + + def _connected(self) -> bool: + return self._connection is not None def _serialize_record(self, record: logging.LogRecord) -> Optional[RedisRecordType]: """Convert a record to something that can be given to the connection.""" diff --git a/src/ewoksjob/events/readers/base.py b/src/ewoksjob/events/readers/base.py index d8be945..92999a9 100644 --- a/src/ewoksjob/events/readers/base.py +++ b/src/ewoksjob/events/readers/base.py @@ -1,4 +1,5 @@ import json +import logging import time from datetime import datetime from threading import Event @@ -15,6 +16,7 @@ except ImportError: Variable = VariableContainer = None +logger = logging.getLogger(__name__) EventType = Dict[str, str] @@ -54,20 +56,32 @@ def poll_events( interval = max(interval, 0.001) start = time.time() n = 0 + exception = None while True: try: events = list(self.get_events(**filters)) - except Exception as e: - if "no such table" not in str(e): + except Exception as ex: + if not self._retry_get_events_exception(ex): raise + if exception is None: + logger.warning("Reading events failed (keep retrying): %s", ex) + exception = ex else: + exception = None events = events[n:] n += len(events) yield from events - if timeout is not None and (time.time() - start) > timeout: - return - if stop_event is not None and stop_event.is_set(): + + timed_out = timeout is not None and (time.time() - start) > timeout + stopped = stop_event is not None and stop_event.is_set() + if timed_out or stopped: + if exception is not None: + logger.error( + "Stop polling events (last attempt failed)", + exc_info=exception, + ) return + time.sleep(interval) def get_events(self, **filters) -> Iterator[EventType]: @@ -168,3 +182,7 @@ def split_filter( raise TypeError("starttime needs to be a datetime object") post_filter = {"starttime": starttime, "endtime": endtime} return is_equal_filter, post_filter + + @staticmethod + def _retry_get_events_exception(ex: Exception) -> bool: + return True diff --git a/src/ewoksjob/events/readers/redis.py b/src/ewoksjob/events/readers/redis.py index 11d85cc..ec3db33 100644 --- a/src/ewoksjob/events/readers/redis.py +++ b/src/ewoksjob/events/readers/redis.py @@ -11,10 +11,20 @@ class RedisEwoksEventReader(EwoksEventReader): - def __init__(self, url: str, **_): - client_name = f"ewoks:reader:{socket.gethostname()}:{os.getpid()}" - self._proxy = redis.Redis.from_url(url, client_name=client_name) + def __init__(self, url: str, timeout: float = 10): + """ + :param url: for example "redis://localhost:10003?db=2". + :param timeout: native redis socket timeout: the maximum time to wait + for connecting and for command replies. + """ super().__init__() + client_name = f"ewoks:reader:{socket.gethostname()}:{os.getpid()}" + self._proxy = redis.Redis.from_url( + url, + client_name=client_name, + socket_connect_timeout=timeout, + socket_timeout=timeout, + ) def wait_events(self, **kwargs) -> Iterator[EventType]: yield from self.poll_events(**kwargs) diff --git a/src/ewoksjob/events/readers/sqlite3.py b/src/ewoksjob/events/readers/sqlite3.py index 4fecf55..b4ebca3 100644 --- a/src/ewoksjob/events/readers/sqlite3.py +++ b/src/ewoksjob/events/readers/sqlite3.py @@ -9,9 +9,15 @@ class Sqlite3EwoksEventReader(EwoksEventReader): - def __init__(self, uri: str, **_) -> None: + def __init__(self, uri: str, timeout: float = 10) -> None: + """ + :param uri: for example "file:/path/to/ewoks_events.db" or "file:///path/to/ewoks_events.db". + :param timeout: native sqlite3 busy timeout: the maximum time to wait + for database locks to be released by other connections. + """ super().__init__() self._uri = uri + self._timeout = timeout self.__connection = None self.__sql_types = sqlite3_utils.python_to_sql_types(FIELD_TYPES) @@ -24,7 +30,9 @@ def close(self): @property def _connection(self): if self.__connection is None: - self.__connection = sqlite3.connect(self._uri, uri=True) + self.__connection = sqlite3.connect( + self._uri, uri=True, timeout=self._timeout + ) return self.__connection def wait_events(self, **kwargs) -> Iterator[EventType]: @@ -38,3 +46,16 @@ def get_events(self, **filters) -> Iterator[EventType]: sql_types=self.__sql_types, **filters, ) + + @staticmethod + def _retry_get_events_exception(ex: Exception) -> bool: + if not isinstance(ex, sqlite3.OperationalError): + return False + error_message = str(ex) + if "no such table" in error_message: + # no event was published yet. + return True + if "database is locked" in error_message: + # transient lock by another connection. + return True + return False diff --git a/src/ewoksjob/tests/conftest.py b/src/ewoksjob/tests/conftest.py index f1aa97e..22b8998 100644 --- a/src/ewoksjob/tests/conftest.py +++ b/src/ewoksjob/tests/conftest.py @@ -1,5 +1,6 @@ import gc import os +from typing import Optional import pytest from ewokscore import events @@ -155,7 +156,10 @@ def sqlite3_ewoks_events(tmp_path): handlers = [ { "class": "ewokscore.events.handlers.Sqlite3EwoksEventHandler", - "arguments": [{"name": "uri", "value": uri}], + "arguments": [ + {"name": "uri", "value": uri}, + {"name": "retry_period", "value": _sqlite3_retry_period()}, + ], } ] with read_ewoks_events(uri) as reader: @@ -163,6 +167,18 @@ def sqlite3_ewoks_events(tmp_path): events.cleanup() +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 + + @pytest.fixture() def redis_ewoks_events(redisdb): url = f"unix://{redisdb.connection_pool.connection_kwargs['path']}" diff --git a/src/ewoksjob/tests/test_events_sqlite3.py b/src/ewoksjob/tests/test_events_sqlite3.py new file mode 100644 index 0000000..23f3ad0 --- /dev/null +++ b/src/ewoksjob/tests/test_events_sqlite3.py @@ -0,0 +1,182 @@ +import sqlite3 +import threading +import time + +from ewokscore import events + +from ..events.readers.sqlite3 import Sqlite3EwoksEventReader + + +def test_concurrent_read_write(sqlite3_ewoks_events): + """Events published while a reader is polling must all be received.""" + handlers, reader = sqlite3_ewoks_events + + njobs = 4 + nevents_per_job = 20 + + def publish_events(job_id): + execinfo = _execinfo(handlers, job_id=job_id) + for _ in range(nevents_per_job // 2): + events.send_workflow_event(execinfo=execinfo, event="start") + events.send_workflow_event(execinfo=execinfo, event="end") + + # Publish concurrently (first event instantiates and registers the event handlers) + threads = [ + threading.Thread(target=publish_events, args=(str(job_id),)) + for job_id in range(njobs) + ] + for thread in threads: + thread.start() + + # Poll for events concurrently + try: + evts = list(reader.wait_events(timeout=10)) + finally: + for thread in threads: + thread.join(timeout=10) + + # Check that all events were published + assert len(evts) == njobs * nevents_per_job + + +def test_read_blocked_by_writer(tmp_path, sqlite3_ewoks_events): + """A reader must retry reading while a writer locks the database.""" + lock_seconds = 0.5 + retry_period = lock_seconds / 5 # do not retry long enough + retry_timeout = 5 * lock_seconds # retry long enough + + # Add events to the database + handlers, _ = sqlite3_ewoks_events + execinfo = _execinfo(handlers) + events.send_workflow_event(execinfo=execinfo, event="start") + + # Writer locks the database + write_lock_acquired = threading.Event() + + def write_lock_database(): + conn = sqlite3.connect(str(tmp_path / "ewoks_events.db")) + try: + # Start write transaction + conn.execute("BEGIN EXCLUSIVE") + write_lock_acquired.set() + + # Open write transaction locks the db + time.sleep(lock_seconds) + + # Finish transaction + conn.commit() + finally: + conn.close() + + uri = f"file:{tmp_path / 'ewoks_events.db'}" + + # Read while locked: do not retry long enough + write_lock_acquired.clear() + thread = threading.Thread(target=write_lock_database) + thread.start() + try: + assert write_lock_acquired.wait(timeout=10) + with Sqlite3EwoksEventReader(uri=uri, timeout=retry_period) as reader: + evts = list(reader.wait_events(timeout=retry_period)) + finally: + thread.join(timeout=10) + + # Check that no events were read + assert len(evts) == 0 + + # Read while locked: retry long enough + write_lock_acquired.clear() + thread = threading.Thread(target=write_lock_database) + thread.start() + try: + assert write_lock_acquired.wait(timeout=10) + with Sqlite3EwoksEventReader(uri=uri, timeout=retry_period) as reader: + evts = list(reader.wait_events(timeout=retry_timeout)) + finally: + thread.join(timeout=10) + + # Check that all events were read + assert len(evts) == 1 + + +def test_failed_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys): + """A writer must wait for read locks to be released.""" + _test_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys, succeed=False) + + +def test_succeeded_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys): + """A writer must wait for read locks to be released.""" + _test_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys, succeed=True) + + +def _test_write_blocked_by_reader( + tmp_path, sqlite3_ewoks_events, capsys, succeed: bool +): + """A writer must wait for read locks to be released.""" + lock_seconds = 0.5 + if succeed: + retry_timeout = 5 * lock_seconds # retry long enough + else: + retry_timeout = lock_seconds / 5 # do not retry long enough + + # Add start event to the database + handlers, reader = sqlite3_ewoks_events + timeout_arg = {"name": "timeout", "value": retry_timeout} + handlers[0]["arguments"].append(timeout_arg) + execinfo = _execinfo(handlers) + events.send_workflow_event(execinfo=execinfo, event="start") + + # Reader + read_lock_acquired = threading.Event() + + def read_lock_database(): + conn = sqlite3.connect(str(tmp_path / "ewoks_events.db")) + try: + # Start read transaction + conn.execute("BEGIN") + conn.execute("SELECT COUNT(*) FROM ewoks_events").fetchone() + read_lock_acquired.set() + + # Open read transaction locks the db + time.sleep(lock_seconds) + + # Finish transaction + conn.rollback() + finally: + conn.close() + + # Write while locked + read_lock_acquired.clear() + thread = threading.Thread(target=read_lock_database) + thread.start() + + _ = capsys.readouterr() + try: + assert read_lock_acquired.wait(timeout=10) + events.send_workflow_event(execinfo=execinfo, event="end") + finally: + thread.join(timeout=10) + + # Check the error log of the dropped event + captured = capsys.readouterr() + if succeed: + assert "--- Logging error ---" not in captured.err + else: + assert "--- Logging error ---" in captured.err + assert "sqlite3.OperationalError: database is locked" in captured.err + + # Check that the expected events were published + evts = list(reader.wait_events(timeout=1)) + excepted = 2 if succeed else 1 + assert len(evts) == excepted + + +def _execinfo(handlers, job_id="123") -> dict: + return { + "job_id": job_id, + "workflow_id": "456", + "host_name": None, + "user_name": None, + "process_id": None, + "handlers": handlers, + }