From 5ca212e32347e01f6e5aeea5ce073f57533eabd4 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:18:42 +0200 Subject: [PATCH 1/7] Sqlite3EwoksEventHandler: add constructor arguments --- src/ewokscore/events/handlers/sqlite3.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) 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, + ) From 6f0b46bd779667fdbe27b70551d30fc2e68bcd81 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:18:58 +0200 Subject: [PATCH 2/7] test concurrent ewoks events --- src/ewokscore/tests/test_events.py | 61 ++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/src/ewokscore/tests/test_events.py b/src/ewokscore/tests/test_events.py index b9aa4a5..677b732 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,60 @@ 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 From 60c0debfcf05f55e7808426facfcd35f181d8803 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:19:42 +0200 Subject: [PATCH 3/7] docs: abstract _connected() in ewoks event handler and fix argument passing. --- doc/howtoguides/events.rst | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/doc/howtoguides/events.rst b/doc/howtoguides/events.rst index affaf95..cab0625 100644 --- a/doc/howtoguides/events.rst +++ b/doc/howtoguides/events.rst @@ -46,14 +46,28 @@ event handler as follows: class MyHandler(EwoksEventHandlerMixIn, ConnectionHandler): - def _connect(self, uri:str, param1:int, timeout=1) -> None: - self._connection = Connection(uri, param1) + 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 = Connection(self._uri, self._param1) self._fields = send_events.FIELDS def _disconnect(self) -> None: del self._connection self._connection = None + def _connected(self) -> bool: + return self._connection is not None + def _send_serialized_record(self, srecord): self._connection.send(srecord) From 52aad9b0feac7ffa78fa929f9a799ca1219571f2 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:19:48 +0200 Subject: [PATCH 4/7] modify changelog --- CHANGELOG.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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 From 4d4d088f619b08144376251e1e9a7cef4a4f2412 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Fri, 17 Jul 2026 14:23:58 +0200 Subject: [PATCH 5/7] ewoksutils requirement --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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'", From a04b74449553375e3ae6029875e47944dd1bb72b Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Fri, 17 Jul 2026 14:25:17 +0200 Subject: [PATCH 6/7] fix ruff --- src/ewokscore/tests/test_events.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/ewokscore/tests/test_events.py b/src/ewokscore/tests/test_events.py index 677b732..6088561 100644 --- a/src/ewokscore/tests/test_events.py +++ b/src/ewokscore/tests/test_events.py @@ -187,7 +187,5 @@ def send_events(job_id): events.cleanup() with sqlite3_utils.connect(uri, uri=True) as conn: - rows = list( - sqlite3_utils.select(conn, "ewoks_events", field_types=FIELD_TYPES) - ) + rows = list(sqlite3_utils.select(conn, "ewoks_events", field_types=FIELD_TYPES)) assert len(rows) == nthreads * nevents_per_thread From bc1d86c23c7159e620cac41ebeda37f7984cade1 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Fri, 17 Jul 2026 14:38:10 +0200 Subject: [PATCH 7/7] docs: fix Ewoks event handler docs --- doc/howtoguides/events.rst | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/doc/howtoguides/events.rst b/doc/howtoguides/events.rst index cab0625..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 @@ -58,8 +58,7 @@ event handler as follows: self._connection = None def _connect(self) -> None: - self._connection = Connection(self._uri, self._param1) - self._fields = send_events.FIELDS + self._connection = MyConnection(self._uri, self._param1) def _disconnect(self) -> None: del self._connection @@ -68,14 +67,17 @@ event handler as follows: def _connected(self) -> bool: return self._connection is not None - def _send_serialized_record(self, srecord): - self._connection.send(srecord) - 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