diff --git a/CHANGELOG.md b/CHANGELOG.md index a36ea95..2b33096 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Add new fields `workflow_input_schema` and `workflow_output_schema` to the spec +- Add new fields `workflow_input_schema` and `workflow_output_schema` to the spec. +- `RedisEwoksEventHandler`: add `disconnect_on_error` and `timeout` argument. +### Fixed + +- `Sqlite3EwoksEventHandler`: implement abstract method `_connected()`. ## [5.0.0] - 2026-07-01 diff --git a/doc/howtoguides/events.rst b/doc/howtoguides/events.rst index affaf95..b2c1e05 100644 --- a/doc/howtoguides/events.rst +++ b/doc/howtoguides/events.rst @@ -33,7 +33,7 @@ Define an Ewoks event handler ----------------------------- You can use any handler derived from `logging.Handler` as an Ewoks event handler. For example if you -have a `Connection` class with a constructor that accepts two arguments, you can define an ewoks +have a `MyConnection` class with a constructor that accepts two arguments, you can define an ewoks event handler as follows: .. code-block:: python @@ -46,22 +46,38 @@ event handler as follows: class MyHandler(EwoksEventHandlerMixIn, ConnectionHandler): - def _connect(self, uri:str, param1:int, timeout=1) -> None: - self._connection = Connection(uri, param1) - self._fields = send_events.FIELDS + def __init__( + self, + uri:str, + param1: int, + disconnect_on_error: bool = False, + ): + super().__init__(disconnect_on_error=disconnect_on_error) + self._uri = uri + self._param1 = param1 + self._connection = None + + def _connect(self) -> None: + self._connection = MyConnection(self._uri, self._param1) def _disconnect(self) -> None: del self._connection self._connection = None - def _send_serialized_record(self, srecord): - self._connection.send(srecord) + def _connected(self) -> bool: + return self._connection is not None def _serialize_record(self, record): + """Convert a record to something that can be given to the connection.""" return { - field: self.get_value(record, field, None) for field in self._fields + field: self.get_value(record, field, None) for field in self.FIELD_TYPES } + def _send_serialized_record(self, serialized_record): + """Send the output from `_serialize_record` to the connection.""" + self._connection.send(serialized_record) + + `ConnectionHandler` is an abstract python logging handler which can be used for convenience. To use this example handler diff --git a/pyproject.toml b/pyproject.toml index 8990a18..bb123ed 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ dependencies = [ "pyyaml >=5.1", "h5py >=2.8", "packaging", - "ewoksutils >=1.10.0", + "ewoksutils >=2.0.0rc1", "importlib_metadata;python_version < '3.9'", "pydantic >= 2", "typing_extensions;python_version < '3.11'", diff --git a/src/ewokscore/events/handlers/sqlite3.py b/src/ewokscore/events/handlers/sqlite3.py index 12c3575..ed07685 100644 --- a/src/ewokscore/events/handlers/sqlite3.py +++ b/src/ewokscore/events/handlers/sqlite3.py @@ -4,5 +4,21 @@ class Sqlite3EwoksEventHandler(EwoksEventHandlerMixIn, Sqlite3Handler): - def __init__(self, uri: str): - super().__init__(uri=uri, table="ewoks_events", field_types=self.FIELD_TYPES) + def __init__( + self, uri: str, timeout: float = 10, disconnect_on_error: bool = False + ): + """ + :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. + An event is dropped when the timeout is reached. + :param disconnect_on_error: disconnect when emitting an event failed. + """ + super().__init__( + uri=uri, + table="ewoks_events", + field_types=self.FIELD_TYPES, + timeout=timeout, + disconnect_on_error=disconnect_on_error, + ) diff --git a/src/ewokscore/tests/test_events.py b/src/ewokscore/tests/test_events.py index b9aa4a5..6088561 100644 --- a/src/ewokscore/tests/test_events.py +++ b/src/ewokscore/tests/test_events.py @@ -1,12 +1,16 @@ import logging +import threading from contextlib import contextmanager from logging.handlers import QueueHandler from queue import Empty from queue import Queue import pytest +from ewoksutils import sqlite3_utils +from ewoksutils.event_utils import FIELD_TYPES from .. import events +from ..events import global_state @contextmanager @@ -130,3 +134,58 @@ def test_workflow_event_error_log_level(caplog): get_event() assert len(caplog.records) == 1 assert caplog.records[0].levelno == logging.ERROR + + +def test_concurrent_send(tmp_path): + """Sending events concurrently is thread-safe, including the first + event which instantiates and registers the event handlers.""" + uri = f"file:{tmp_path / 'ewoks_events.db'}" + handlers = [ + { + "class": "ewokscore.events.handlers.Sqlite3EwoksEventHandler", + "arguments": [{"name": "uri", "value": uri}], + } + ] + + nthreads = 8 + nevents_per_thread = 10 + barrier = threading.Barrier(nthreads, timeout=10) + exceptions = list() + + def send_events(job_id): + execinfo = { + "job_id": job_id, + "host_name": None, + "user_name": None, + "process_id": None, + "workflow_id": None, + "handlers": handlers, + } + try: + # Send the first event from all threads at the same time + barrier.wait() + for _ in range(nevents_per_thread // 2): + events.send_workflow_event(execinfo=execinfo, event="start") + events.send_workflow_event(execinfo=execinfo, event="end") + except BaseException as ex: + exceptions.append(ex) + + threads = [ + threading.Thread(target=send_events, args=(str(job_id),)) + for job_id in range(nthreads) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + try: + assert not exceptions + logger = logging.getLogger(global_state.EWOKS_EVENT_LOGGER_NAME) + assert len(logger.handlers) == 1 + finally: + events.cleanup() + + with sqlite3_utils.connect(uri, uri=True) as conn: + rows = list(sqlite3_utils.select(conn, "ewoks_events", field_types=FIELD_TYPES)) + assert len(rows) == nthreads * nevents_per_thread