Add finished

src/tribler/core/utilities/async_group/async_group.py

@@ -7,7 +7,7 @@
 from typing import Coroutine, Iterable, List, Optional, Set
 from weakref import ref
 
-from tribler.core.utilities.async_group.exceptions import CanceledException
+from tribler.core.utilities.async_group.exceptions import CanceledException, FinishedException
 
 
 def done_callback(group_ref):
@@ -54,6 +54,7 @@ def __init__(self):
         self.ref = ref(self)
         self.futures: Set[Future] = set()
         self._canceled = False
+        self._finished = False
 
     def add_task(self, coroutine: Coroutine) -> Task:
         """Add a coroutine to the group.
@@ -64,6 +65,10 @@ def add_task(self, coroutine: Coroutine) -> Task:
             task.cancel()
             raise CanceledException()
 
+        if self._finished:
+            task.cancel()
+            raise FinishedException()
+
         self.futures.add(task)
         self.global_futures.add(task)
 
@@ -73,8 +78,10 @@ def add_task(self, coroutine: Coroutine) -> Task:
     async def wait(self):
         """ Wait for completion of all futures
         """
-        if self.futures:
-            await asyncio.wait(self.futures)
+        while active := set(self._active(self.futures)):
+            await asyncio.wait(active)
+
+        self._finished = True
 
     async def cancel(self) -> List[Future]:
         """Cancel the group.
@@ -99,6 +106,10 @@ async def cancel(self) -> List[Future]:
     def cancelled(self):
         return self._canceled
 
+    @property
+    def finished(self):
+        return self._finished
+
     @staticmethod
     def _active(futures: Iterable[Future]) -> Iterable[Future]:
         return (future for future in futures if not future.done())

src/tribler/core/utilities/async_group/exceptions.py

@@ -1,2 +1,6 @@
 class CanceledException(Exception):
     """A coroutine can not be added to a cancelled AsyncGroup"""
+
+
+class FinishedException(Exception):
+    """A coroutine can not be added to a finished AsyncGroup"""