🐛 Fixed GTK race condition in pipelines

src/importer/importer.py

@@ -122,8 +122,7 @@ class Importer:
             pipeline: Pipeline = shared.store.add_game(game)
             if pipeline is not None:
                 logging.info("Imported %s (%s)", game.name, game.game_id)
-                pipeline.connect("manager-done", self.manager_done_callback)
-                pipeline.connect("manager-started", self.manager_started_callback)
+                pipeline.connect("advanced", self.pipeline_advanced_callback)
                 self.game_pipelines.add(pipeline)
 
     def update_progressbar(self):
@@ -136,17 +135,8 @@ class Importer:
         logging.debug("Import done for source %s", source.id)
         self.n_source_tasks_done += 1
 
-    def manager_started_callback(self, pipeline: Pipeline, manager: Manager):
-        """Callback called when a game manager has started"""
-        logging.debug(
-            "Manager %s for %s started", manager.__class__.__name__, pipeline.game.name
-        )
-
-    def manager_done_callback(self, pipeline: Pipeline, manager: Manager):
+    def pipeline_advanced_callback(self, pipeline: Pipeline):
         """Callback called when a pipeline for a game has advanced"""
-        logging.debug(
-            "Manager %s for %s done", manager.__class__.__name__, pipeline.game.name
-        )
         if pipeline.is_done:
             self.n_pipelines_done += 1
             self.update_progressbar()

src/store/managers/async_manager.py

@@ -1,4 +1,4 @@
-from typing import Callable
+from typing import Callable, Any
 
 from gi.repository import Gio
 
@@ -26,8 +26,8 @@ class AsyncManager(Manager):
         Already scheduled Tasks will no longer be cancellable."""
         self.cancellable = Gio.Cancellable()
 
-    def run(self, game: Game, callback: Callable) -> None:
-        task = Task.new(self, self.cancellable, self._task_callback, (callback,))
+    def run(self, game: Game, callback: Callable[["Manager"], Any]) -> None:
+        task = Task.new(None, self.cancellable, self._task_callback, (callback,))
         task.set_task_data((game,))
         task.run_in_thread(self._task_thread_func)
 
@@ -38,5 +38,5 @@ class AsyncManager(Manager):
 
     def _task_callback(self, _source_object, _result, data):
         """Method run after the async task is done"""
-        _game, callback, *_rest = data
+        callback, *_rest = data
         callback(self)

src/store/managers/display_manager.py

@@ -1,16 +1,16 @@
 from src import shared
 from src.game import Game
-from src.store.managers.file_manager import FileManager
+from src.store.managers.sgdb_manager import SGDBManager
+from src.store.managers.steam_api_manager import SteamAPIManager
 from src.store.managers.manager import Manager
 
 
 class DisplayManager(Manager):
     """Manager in charge of adding a game to the UI"""
 
-    run_after = set((FileManager,))
+    run_after = set((SteamAPIManager, SGDBManager))
 
     def final_run(self, game: Game) -> None:
         # TODO decouple a game from its widget
-        # TODO make the display manager async
         shared.win.games[game.game_id] = game
         game.update()

src/store/managers/manager.py

@@ -1,5 +1,5 @@
 from abc import abstractmethod
-from typing import Callable
+from typing import Callable, Any
 
 from src.game import Game
 
@@ -16,6 +16,10 @@ class Manager:
     errors: list[Exception]
     blocking: bool = True
 
+    @property
+    def name(self):
+        return type(self).__name__
+
     def __init__(self) -> None:
         super().__init__()
         self.errors = []
@@ -38,7 +42,7 @@ class Manager:
         * May not raise exceptions, as they will be silently ignored
         """
 
-    def run(self, game: Game, callback: Callable[["Manager"]]) -> None:
+    def run(self, game: Game, callback: Callable[["Manager"], Any]) -> None:
         """Pass the game through the manager.
         In charge of calling the final_run method."""
         self.final_run(game)

src/store/pipeline.py

@@ -1,10 +1,10 @@
+import logging
 from typing import Iterable
 
 from gi.repository import GObject
 
 from src.game import Game
 from src.store.managers.manager import Manager
-from src.utils.task import Task
 
 
 class Pipeline(GObject.Object):
@@ -59,17 +59,18 @@ class Pipeline(GObject.Object):
 
         # Schedule parallel managers, then run the blocking ones
         for manager in (*parallel, *blocking):
-            self.emit("manager-started", manager)
-            manager.run(self.game, self._manager_callback)
+            self.waiting.remove(manager)
+            self.running.add(manager)
+            manager.run(self.game, self.manager_callback)
 
-    def _manager_callback(self, manager: Manager) -> None:
+    def manager_callback(self, manager: Manager) -> None:
         """Method called by a manager when it's done"""
-        self.emit("manager-done", manager)
+        logging.debug("%s done for %s", manager.name, self.game.game_id)
+        self.running.remove(manager)
+        self.done.add(manager)
+        self.emit("advanced")
+        self.advance()
 
-    @GObject.Signal(name="manager-started", arg_types=(object,))
-    def manager_started(self, manager: Manager) -> None:
-        """Signal emitted when a manager is started"""
-
-    @GObject.Signal(name="manager-done", arg_types=(object,))
-    def manager_done(self, manager: Manager) -> None:
-        """Signal emitted when a manager is done"""
+    @GObject.Signal(name="advanced")
+    def advanced(self) -> None:
+        """Signal emitted when the pipeline has advanced"""