Add method to stop Select

Add method to stop `Select` instance and cleanup any
pending task.
Select stores async tasks that listen on the receivers.
If select is removed without stopping the receivers tasks,
then we get error: Task was destroyed but it is pending!

We should stop all internal tasks before destoring Select.
Destructor can't be async, so we do that with extra method `stop`.
This method should be called when Select is no longer needed.

Signed-off-by: ela-kotulska-frequenz <[email protected]>

Author: ela-kotulska-frequenz

src/frequenz/channels/util/_select.py

@@ -63,6 +63,9 @@ class Select:
 
     [Receiver][frequenz.channels.Receiver]s also function as `Receiver`.
 
+    When Select is no longer needed, then it should be stopped using
+    `self.stop()` method. This would cleanup any internal pending async tasks.
+
     Example:
         For example, if there are two receivers that you want to
         simultaneously wait on, this can be done with:
@@ -106,6 +109,13 @@ def __del__(self) -> None:
         for task in self._pending:
             task.cancel()
 
+    async def stop(self) -> None:
+        """Stop the `Select` instance and cleanup any pending tasks."""
+        for task in self._pending:
+            task.cancel()
+        await asyncio.gather(*self._pending, return_exceptions=True)
+        self._pending = set()
+
     async def ready(self) -> bool:
         """Wait until there is a message in any of the receivers.