Unify how classes/objects that spawn tasks should be managed

[leandro-lucarella-frequenz]

Problem

Right now we have no common approach on how to manage classes/objects that spawn tasks. The most core example are actors, but we also have a lot of other classes that spawn tasks (like the resampler, moving window, etc.).

It should be possible to clearly control the lifespan of these objects, and be able to cleanly and clearly start and stop the tasks they spawn.

For now every class is doing their own thing, sometimes there is no way to stop things, sometimes there is but it is private, etc.

We should have a common way to do this.

Proposal

Add a Worker class to the SDK that provides a simple framework to handle this.

Here is a quick sketch of a possible implementation:

class Worker(ABC):

    def __init__(self):
        self.tasks: set[asyncio.Task] = set

    # async might not be really needed here, but allows sub-classes
    # to do async work as part of starting a task, which can be a
    # good thing since async initializaiton work can't be done in
    # __init__()
    @abstract
    async def start(self) -> None:
        pass

    @property
    def is_stopped(self) -> bool:
        for it in self.tasks:
            if t.done():
                return False
        return True
    def cancel(self) -> None:
        for t in self.tasks:
            t.cancel()

    async def join(self) -> None:
        await asyncio.gather(*[t.join() for t in self.tasks])

    async def stop(self) -> None:
        for t in self.tasks:
            await cancel_and_await(t)

    async __aenter__(self) -> None:
        await self.start()

    async def __aexit__(self, exc_type, exc, tb) -> None:
        await self.stop()

    def __del__(self) -> None:
        self.cancel()
        # Maybe see if there is a loop running and try to run await self.stop() in the loop?

(join() and stop() should probably be implemented using asyncio.wait() in a more resilient way)

Then all other classes can inherit from Worker:

class MovingWindow(Worker):
    def __init__(self, other_params: ..., ...):
        super().__init__()

    async def start(self) -> None:
        self.tasks.add(asyncio.create_task(...))

mw = MovingWindow(...)
await mw.start()
...
await mw.stop()

Advantages

• Having a common, well-known interface (consistency, ease of use, less cognitive burden for users to know what's going on)
• It allows full control over the lifespan of a worker
• It separates initialization from running the task, which could help avoid dependency problems if 2 workers need each other to run.
• It is explicit in terms of when something is running in the background or not.

Disadvantages

• If people forget to call start() it might take a while to figure out what's going on.

Class name

Worker is not a great name, as it is usually used for something that runs in a thread and can run a task, but I could not think of a better name.

I thought of Task but it might be confusing given asyncio.Task, I thought of BackgroundTask as it is something that it is supposed to run in the background, opposed to a general Task that can also be some calculation for which you want some result, but want to run several calculations in parallel.

I thought of Daemon as the functionality is similar to a unix daemon. Or maybe Service in the same sense.

References

• Coming from this discussion: Add a Moving Window Class #190 (comment)

FYI @frequenz-floss/python-sdk-team

[daniel-zullo-frequenz]

Wrt to the class name, I'd suggest something with the prefix Task given the context but the names will be a bit longer than Worker.
This is what comes to my mind:

• TaskManager given that manages tasks such as starting, cancelling, and stopping them
• TaskRunner as it runs the tasks and handles their life cycles
• TaskOrchestrator orchestrates the execution of tasks, it is a suitable name given there is functionality like join() and might fit even better if it will have, for instance, restart(), schedule(), etc functions

[leandro-lucarella-frequenz]

Don't want to dig too deep into the rabbit hole, but this class is not a manager nor a runner, nor an orchestrator, it is really just something running a background service (so maybe I'm leaning more towards Service).

[shsms]

Just want to point out that there are scenarios that this doesn't cover, especially when it comes to short-lived tasks, like [asyncio.create_task(recv.receive()) for recv in list_of_recv].

If there's such a scenario, we won't be able to use self.tasks because we might not want to mix different types of tasks in a single set, for example.

So while the proposal does make sense to simplify most use-cases, a common approach might not work in all cases, and we'll still have to implement custom stop methods.

[leandro-lucarella-frequenz]

About having different types of tasks, I guess you mean you want to stop one type of task and not others? In that case the user can provide extra methods, like stop_receivers(), and stop() would stop all. For me this class is about managing the lifespan of object having tasks in the background, not about fine detail on how to use those objects. You could say is mostly an initialization and finalization thing. If you want to "destroy" some object, you want to make sure all the tasks it runs in the background are stopped too.

But yeah, in any case I can see some sub-classes implementing their own cancel/join/stop methods, I think that's fine.

Side note: This class should also manage automatic finalization (__del__ or whatever, this is also not a trivial topic, there are several options for this, there is also weakref.finalize, which shows it is a hard problem and it should be handled once for all classes needing it). Also this class should probably provide magic methods to make it a context manager so it can be used with the with statement:

async with MovingWindow(...) as mw: # initializes *and** start() mw
    mw.whatever()
# mw is stop()ped and destroyed

or

mw = MovingWindow(...) # initialization
async with mw: # start()
    mw.whatever()
# mw is stop()ped but not destroyed, we can still use it to extract some result for example
result = mw.something()
mw.start() # or maybe even start it again

[leandro-lucarella-frequenz]

I updated the proposal with some additions (context manager, finalization, and boolean testing if it is stopped).

[shsms]

async with is for cases where want to run a few lines with a single instance running. But we are creating a common approach for persistent tasks that need to be accessed from various parts of a program, and that's what most of of use cases are like.

I feel that providing support for async with is confusing and not consistent with out intended use.

[leandro-lucarella-frequenz]

I agree that focusing on the current state of the SDK it is not useful, but for channels for example it could make sense.

Also with actors potentially you could create all actors in the with and then run you application code inside the with, in particular if @matthias-wende-frequenz idea on how to write an app comes to life:

some_actor = SomeActor(...)
other_actor = OtherActor(...)
sdk = Sdk(config)
# more config

async with sdk, some_actor, other_actor:
    while True:
        do_whatever()

This could potentially remove the need for the run() utility function. It should be also familiar for people using standard python's task groups. Just brainstorming here.

[leandro-lucarella-frequenz]

I will give this a go:

• Create a BackgroundService class #240