Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ classifiers = [
]
requires-python = ">=3.8"
dependencies = [
"ewokscore >=5",
"ewokscore >=5.1.0rc5",

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could keep version 5 compatibility but this is a bit cumbersome for the inheritance. So I moved forward. Let me know if there is reason to keep compatibly with version 5.0

"orange-canvas-core>=0.1.35",
"orange-widget-base>=4.22.0",
"importlib-resources; python_version<'3.9'",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,4 +32,4 @@ def aborted(self) -> bool:
with self._lock:
task = self._task

return task is not None and task.cancelled
return task is not None and task.stop_requested
16 changes: 6 additions & 10 deletions src/ewoksorange/gui/concurrency/_runners/abstract.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,24 +52,20 @@ def _create_task(self) -> Task:
return self._task_class(**self._task_kwargs)

def _execute(self, task: Task) -> VariableContainer:
"""Execute `task`, cancelling it for as long as `abort_event` is set."""
"""Execute `task`, aborting it for as long as `abort_event` is set."""
done = threading.Event()

def _watch_abort():
"""
Currently `task.cancel()` and `task.cancelled` are ill-defined.

The flag `task.cancelled` could be the request or the state.

`task.execute()` resets the task's `cancelled` flag, which
can silently undo a `cancel()` racing with the start
`task.execute()` resets the task's `stop_requested` flag, which
can silently undo a `request_stop()` racing with the start
of execution.

For this reason this watcher thread keeps re-applying `task.cancel()`
For this reason this watcher thread keeps re-applying `task.request_stop()`
until execution finishes, so a task's `run()` reliably observes it
regardless of the timing.

The output of a cancelled task is undefined. It could be an
The output of an aborted task is undefined. It could be an
exception, undefined outputs, pertially defined outputs or
fully defined outputs.
"""
Expand All @@ -80,7 +76,7 @@ def _watch_abort():
# finished without ever being aborted.
return
while not done.is_set():
task.cancel()
task.request_stop()
done.wait(timeout=0.01)
except self._TRANSIENT_ABORT_ERRORS:
pass
Expand Down
2 changes: 1 addition & 1 deletion src/ewoksorange/gui/concurrency/_runners/process.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ def _announce_started(self, task: Task) -> None:

def _finalize(self, task: Task) -> None:
try:
if task.cancelled:
if task.stop_requested:
self._aborted_event.set()
except self._TRANSIENT_ABORT_ERRORS:
pass
4 changes: 2 additions & 2 deletions src/ewoksorange/gui/concurrency/threaded.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ def stop(self, timeout: Optional[float] = None, wait: bool = False) -> None:
def cancel_running_task(self):
"""
cancel current processing.
The targetted EwoksTask must have implemented the 'cancel' function
The targetted EwoksTask must have implemented the '_on_stop_requested' function
"""
if self.current_task is not None:
self.current_task.cancel()
self.current_task.request_stop()
58 changes: 26 additions & 32 deletions src/ewoksorange/tests/executor/tasks.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import os
import threading
import time

from ewokscore import TaskWithProgress
Expand All @@ -24,7 +25,7 @@ def run(self):
period = self.inputs.delay / 40
for i in range(40):
time.sleep(period)
if self.cancelled:
if self.stop_requested:
raise CustomCancelled(f"cancelled after {(i + 1) * period} seconds")

if self.inputs.fail:
Expand All @@ -37,9 +38,9 @@ def run(self):

self.outputs.result = result

def cancel(self):
"""Interpretation #2: self.cancelled is the request not the state."""
self.cancelled = True
def _on_stop_requested(self):
"""Nothing to do: `run()` polls `self.stop_requested`."""
pass


class RequestCancelTask(
Expand All @@ -53,15 +54,15 @@ def run(self):
step = 0.05
elapsed = 0.0
while elapsed < self.inputs.duration:
if self.cancelled:
if self.stop_requested:
return
time.sleep(step)
elapsed += step
self.outputs.result = f"slept {elapsed:.2f}s"

def cancel(self):
"""Interpretation #2: self.cancelled is the request not the state."""
self.cancelled = True
def _on_stop_requested(self):
"""Nothing to do: `run()` polls `self.stop_requested`."""
pass


class PartialCancelTask(
Expand All @@ -77,60 +78,53 @@ def run(self):
step = 0.05
elapsed = 0.0
while elapsed < self.inputs.duration:
if self.cancelled:
if self.stop_requested:
return
time.sleep(step)
elapsed += step

self.outputs.second = "second done"

def cancel(self):
"""Interpretation #2: self.cancelled is the request not the state."""
self.cancelled = True
def _on_stop_requested(self):
"""Nothing to do: `run()` polls `self.stop_requested`."""
pass


class IgnoreCancelTask(
Task,
input_names=["duration"],
output_names=["result"],
):
"""Never checks `self.cancelled`; always completes fully despite cancel()."""
"""Never checks `self.stop_requested`; always completes fully despite request_stop()."""

def run(self):
time.sleep(self.inputs.duration)
self.outputs.result = "completed despite abort"

def cancel(self):
"""Interpretation #2: self.cancelled is the request not the state."""
self.cancelled = True
def _on_stop_requested(self):
"""Nothing to do: `run()` polls `self.stop_requested`."""
pass


class StateCancelTask(
class InterruptCancelTask(
Comment on lines -108 to +109

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Renamed for clarity. I haven't saw any project using it but still breaking the public API.

Task,
input_names=["duration"],
output_names=["result"],
):
"""cancel() only records a request; run() itself sets `self.cancelled`
once stopped."""
"""Never polls `self.stop_requested`; `_on_stop_requested()` interrupts
the blocking wait of run() instead."""

def __init__(self, **kwargs):
super().__init__(**kwargs)
self.__cancel_requested = False
self.__stop_event = threading.Event()

def run(self):
step = 0.05
elapsed = 0.0
while elapsed < self.inputs.duration:
if self.__cancel_requested:
self.cancelled = True
return
time.sleep(step)
elapsed += step
self.outputs.result = f"slept {elapsed:.2f}s"
if self.__stop_event.wait(timeout=self.inputs.duration):
return
self.outputs.result = f"slept {self.inputs.duration:.2f}s"

def cancel(self):
"""Interpretation #1: self.cancelled is the state not the request."""
self.__cancel_requested = True
def _on_stop_requested(self):
self.__stop_event.set()
Comment on lines -121 to +127

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Simplification of the cancellation.

The origin of the modification is that stop_requested (replacing cancelled) has no more setter. It is set directly when stop_requested is called. Which makes it more robust and clarify the states.

This is more 'strict' this way for the best.



class TimedTask(
Expand Down
10 changes: 5 additions & 5 deletions src/ewoksorange/tests/executor/test_executor_abort.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,9 @@

from .tasks import AddTask
from .tasks import IgnoreCancelTask
from .tasks import InterruptCancelTask
from .tasks import PartialCancelTask
from .tasks import RequestCancelTask
from .tasks import StateCancelTask


def test_abort(ewoksorange_qtapp, executor_context_factory):
Expand Down Expand Up @@ -42,14 +42,14 @@ def test_abort(ewoksorange_qtapp, executor_context_factory):
thread.join(timeout=10)


@pytest.mark.parametrize("task_class", [RequestCancelTask, StateCancelTask])
@pytest.mark.parametrize("task_class", [RequestCancelTask, InterruptCancelTask])
def test_abort_leaves_outputs_undefined(
ewoksorange_qtapp, executor_context_factory, task_class
):
"""Cancellation observed by run(): it returns early, so no output is set.

Covers both interpretations of `Task.cancelled` (RequestCancelTask:
request, StateCancelTask: state).
Covers both ways of reacting to a stop request (RequestCancelTask:
polls `Task.stop_requested`, InterruptCancelTask: `Task._on_stop_requested`).
"""
with executor_context_factory() as (kind, executor, recorder):
inputs = {"duration": 2}
Expand Down Expand Up @@ -118,7 +118,7 @@ def test_abort_leaves_partial_outputs(ewoksorange_qtapp, executor_context_factor
def test_abort_does_not_guarantee_cancellation(
ewoksorange_qtapp, executor_context_factory
):
"""A task that never checks `self.cancelled` always completes normally.
"""A task that never checks `self.stop_requested` always completes normally.

`aborted()` still reports True: it reflects that abort reached the task,
not what the task chose to do about it.
Expand Down
17 changes: 5 additions & 12 deletions src/ewoksorange/tests/owwidget/test_owwidget_parallel.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,6 @@ class Parallel(
input_names=["value", "sleep", "synchronize_run"],
output_names=["value", "time"],
):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.__cancel_requested = False

def run(self):
# Wait for all parallel task to start
self.inputs.synchronize_run.wait(timeout=10)
Expand All @@ -36,18 +32,15 @@ def run(self):
tm = time.perf_counter()

# Return value as output or exception
if self.__cancel_requested:
try:
raise Cancelled(self.inputs.value, tm)
finally:
self.cancelled = True
if self.stop_requested:
raise Cancelled(self.inputs.value, tm)

self.outputs.time = tm
self.outputs.value = self.inputs.value

def cancel(self):
# Interpretation #1: self.cancelled is the state not the request
self.__cancel_requested = True
def _on_stop_requested(self):
# Nothing to do: `run()` checks `self.stop_requested`.
pass


class OWParallel(
Expand Down
Loading