add ewoksorange.gui.concurrency.executor module

[payno]

This PR; done with Claude support add a gui.concurrency.executor module.

This module include mechanism for execution in the context of ewoks tasks.

The goal is to replace the existing widget executors by this module and add a "demo" for the multi-process. Update of existing will be done in another dedicated PR.

Preparatory PR for #406

AI disclosure: Done using Claude

[codecov]

Codecov Report

❌ Patch coverage is 85.29412% with 35 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...range/gui/concurrency/executor/_ProcessCallable.py	32.35%	23 Missing ⚠️
...ewoksorange/gui/concurrency/executor/TaskFuture.py	75.00%	6 Missing ⚠️
...ksorange/gui/concurrency/executor/EwoksExecutor.py	97.67%	3 Missing ⚠️
...range/gui/concurrency/executor/_EwoksTaskHandle.py	60.00%	2 Missing ⚠️
...nge/gui/concurrency/executor/_EwoksThreadHandle.py	96.96%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

[payno]

I only added one test with the ProcessPoolExecutor as mentioned in the code this is more some code to show the way to go but most likely some time will be needed to check possible bottlenecks.

[payno]

At the time each time we submit a task we will need the creation of those two objects (manager.Queue and manager.Event) that from what I have on mind can take a bit of time.
I think in general as written on the doc the ProcessPool is more a prototype in order to 'show the way'.

[payno]

used to emit the requested signal downstream

[payno]

small hack to get the task future defined downstream and needed at execution time.

[payno]

It makes sure there is no conflict when abort is called (if _task is not created yet in call).

[payno]

This part is needed for submitting process only.

[payno]

This is the implementation proposed by Claude. The combination of thread / process seems a bit heavy.

An alternative could be to say that processes are not cancelled to not use the task.cancel and fill directly the process (using process.kill).

But again I think this submission to process is more a demo at the moment and to be modified by people concerned about this use case as they will probably have specific use cases.

[payno]

An alternative could be to return the finished with a state (succeeded or failed) but this won't make sense to return the results or exception.

[payno]

Maybe this part could be simplified by only emitting finished and the future. From this future we can retrieve the exception or result.

[woutdenolf]

Oef, much more complex than my proposal #403 (comment)

In the spirit of "separation of concerns" could be make the executor Ewoks agnostic?

Can you list the pro's of this approach with respect to #403 (comment) ?

If we can't use the existing

from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor

can we at least derive executors from

from concurrent.futures import Executor

so we don't have to invent things like TaskFuture?

Could you also add a standalone example similar to this one in the scripts folder?

#403 (comment)

[payno]

Well I think the key point is that we want to be able to call the ewoks task cancel function in order to be able to abort it (sorry for misleading names).

And this is why we have this TaskFuture and the EwoksExecutor which is wrapping the a concurrent.futures.Executor.

Behind it we use it as:

        EwoksExecutor(
            ThreadPoolExecutor(max_workers=self._MAX_WORKERS),  # or ProcessPoolExecutor
            self._SUBMIT_POLICY,
        )

Then the Process iteration also add some complexity because it requires the ProcessCallable. Even if on this one (as mentioned in the PR) maybe there is a way to simplify things by calling process.kill directly.

Here is the 'gui' updated with the code you had on your PR.

import logging
import sys
from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor

from AnyQt.QtWidgets import QApplication
from AnyQt.QtWidgets import QGridLayout
from AnyQt.QtWidgets import QPushButton
from AnyQt.QtWidgets import QTextEdit
from AnyQt.QtWidgets import QVBoxLayout
from AnyQt.QtWidgets import QWidget
from ewokscore.tests.examples.tasks.sumtask import SumTask
from ewoksorange.gui.concurrency.executor import EwoksExecutor, SubmitPolicy


class Window(QWidget):
    def __init__(self):
        super().__init__()

        self.setWindowTitle("EwoksExecutor Demo")

        self.log = QTextEdit(readOnly=True)

        self.executors = {
            "T DROP": EwoksExecutor(
                ThreadPoolExecutor(max_workers=1),
                SubmitPolicy.DROP_IF_BUSY,
            ),
            "T QUEUE": EwoksExecutor(
                ThreadPoolExecutor(max_workers=1),
                SubmitPolicy.ALWAYS,
            ),
            "T PARALLEL": EwoksExecutor(
                ThreadPoolExecutor(max_workers=4),
                SubmitPolicy.ALWAYS,
            ),
            "P DROP": EwoksExecutor(
                ProcessPoolExecutor(max_workers=1),
                SubmitPolicy.DROP_IF_BUSY,
            ),
            "P QUEUE": EwoksExecutor(
                ProcessPoolExecutor(max_workers=1),
                SubmitPolicy.ALWAYS,
            ),
            "P PARALLEL": EwoksExecutor(
                ProcessPoolExecutor(max_workers=4),
                SubmitPolicy.ALWAYS,
            ),
        }

        for name, executor in self.executors.items():
            executor.submitted.connect(
                lambda future, n=name: self.on_submitted(n, future)
            )
            executor.ignored.connect(lambda n=name: self.log.append(f"[{n}] ⚠ Ignored"))
            executor.succeeded.connect(
                lambda future, result, n=name: self.log.append(
                    f"[{n}] ✅ {id(future)} -> { {k: v.value for k, v in result.items()} }"
                )
            )
            executor.failed.connect(
                lambda future, exc, n=name: self.log.append(
                    f"[{n}] ❌ {id(future)} -> {exc}"
                )
            )

        layout = QVBoxLayout(self)
        grid = QGridLayout()

        buttons = [
            ("Submit x5 (Thread Drop)", "T DROP"),
            ("Submit x5 (Thread Queue)", "T QUEUE"),
            ("Submit x5 (Thread Parallel)", "T PARALLEL"),
            ("Submit x5 (Process Drop)", "P DROP"),
            ("Submit x5 (Process Queue)", "P QUEUE"),
            ("Submit x5 (Process Parallel)", "P PARALLEL"),
        ]

        for i, (text, key) in enumerate(buttons):
            button = QPushButton(text)
            button.clicked.connect(
                lambda checked=False, k=key: self.submit_many(self.executors[k])
            )
            grid.addWidget(button, i // 2, i % 2)

        layout.addLayout(grid)
        layout.addWidget(self.log)

        self.counter = 1

    def on_submitted(self, name, future):
        if future is None:
            self.log.append(f"[{name}] Submitted -> None")
        else:
            self.log.append(f"[{name}] Submitted -> {id(future)}")

    def submit_many(self, executor):
        for _ in range(5):
            executor.submit_task(
                SumTask, inputs={"a": self.counter, "b": self.counter, "delay": 2}
            )
            self.counter += 1

    def closeEvent(self, event):
        for executor in self.executors.values():
            executor.shutdown()
        super().closeEvent(event)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    app = QApplication(sys.argv)

    window = Window()
    window.resize(900, 500)
    window.show()

    sys.exit(app.exec())

[payno]

We can remove the execution on Process. This is what add the most complexity (see commit e3555ab)
Without it this simplify a lot. But as this was more or less a target to have at one point I wanted to have an idea of how we could do this.

[woutdenolf]

Ok my bad, I saw a raw threading. Thread and queue and thought we were re-implementing the executor logic from python.

[woutdenolf]

Remove try-except. The raises exception is task.exception. We should fix TaskExecutor accordingly. I can do that in a follow-up.

[woutdenolf]

Thanks for the hard work. Perhaps best to discuss in person because there are some tricky things in the current implementation.

Script

Could you make a scripts folder and add the example?

Task execution logic

When executing a task without cancel/abort, this happens:

• submit task to executor
• emit "submitted" Qt signal with future are argument
• emit "started" Qt signal with future are argument
• emit either
  • "failed" Qt signal with future and exception as argument
  • "succeeded" Qt signal with future and task.output_variables as arguments
• emit "finished" with future as argument

The execution state is wrapped with a EwoksWorker. It is intended to be executed once.

The TaskFuture has a reference to the native future and the EwoksWorker.

The EwoksWorker exposes abort() which is exposed in turn by TaskFuture.

When calling TaskFuture.cancel() the task gets dropped from the native executor but we don't has a Qt signal for that. Also the implementation of the submision blocks until the task is actually running. That should be fixed.

Also TaskFuture.abort() returns True when the task is reached and aborted. This is not the case for threads.

Also we need to handle the case where the task gets aborted before it even gets scheduled but the native executor.

Is TaskFuture.abort() supposed to be a blocking call? If not how can it return True or False, unless there is an "_aborted" state?

Harmonization

Both EwoksExecutor._submit_thread and EwoksExecutor._submit_process achieve
the same thing but it is very hard to know from reading the code.

• EwoksThreadWorker -> ProcessCallable + EwoksProcessWorker
• run -> relay_started + ProcessCallable

The logic is valid (except that it blocks until the task is running which needs fixing) but needs refactoring for readability and harmony.

That sounds like an ideal AI job.

If I would do it manually I would start by

• making an abstract EwoksWorker
• rename "worker" to something that reflects what it actually does (this will illustrate what needs to be refactored)
• use that abstract type as a type annotation in TaskFuture.__init__
• then the refactoring

[woutdenolf]

I made a PR to this PR to add a thread cleanup fix script and an example: #411.

I'm still wondering whether the split of _EwoksThreadHandle into _EwoksProcessHandle and _ProcessCallable cannot be done in a cleaner way. Perhaps split _EwoksThreadHandle in the same way for consistency reasons. But this is internal API and can be changed at any time. Functionally everything is sound now. LGTM after #411.

Thanks for the huge effort!