_stream.py: Stop using a 'RUNNING' event to know the state of the scheduler

The stream is itself calling the `run` method on the scheduler, we don't
need another indirection

2 files changed, 4 insertions, 11 deletions

diff --git a/src/buildstream/_scheduler/scheduler.py b/src/buildstream/_scheduler/scheduler.py
index 41ff1189f..0c94fd376 100644
--- a/src/buildstream/_scheduler/scheduler.py
+++ b/src/buildstream/_scheduler/scheduler.py
@@ -58,7 +58,6 @@ class NotificationType(FastEnum):
     TICK = "tick"
     TERMINATE = "terminate"
     QUIT = "quit"
-    RUNNING = "running"
     SUSPEND = "suspend"
     UNSUSPEND = "unsuspend"
 
@@ -176,9 +175,6 @@ class Scheduler:
         self.loop = asyncio.new_event_loop()
         asyncio.set_event_loop(self.loop)
 
-        # Notify that the loop has been created
-        self._notify(Notification(NotificationType.RUNNING))
-
         # Add timeouts
         self.loop.call_later(1, self._tick)
 
@@ -214,9 +210,6 @@ class Scheduler:
         failed = any(queue.any_failed_elements() for queue in self.queues)
         self.loop = None
 
-        # Notify that the loop has been reset
-        self._notify(Notification(NotificationType.RUNNING))
-
         if failed:
             status = SchedStatus.ERROR
         elif self.terminated:
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 9a00ecdc0..7c8baf233 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -100,7 +100,7 @@ class Stream:
         self._ticker_callback = ticker_callback
         self._interrupt_callback = interrupt_callback
         self._notifier = self._scheduler._stream_notification_handler  # Assign the schedulers notification handler
-        self._scheduler_running = False
+        self._running = False
         self._terminated = False
         self._suspended = False
 
@@ -1062,7 +1062,7 @@ class Stream:
     #
     @property
     def running(self):
-        return self._scheduler_running
+        return self._running
 
     # suspended
     #
@@ -1377,7 +1377,9 @@ class Stream:
         if self._session_start_callback is not None:
             self._session_start_callback()
 
+        self._running = True
         status = self._scheduler.run(self.queues, self._context.get_cascache().get_casd_process_manager())
+        self._running = False
 
         if status == SchedStatus.ERROR:
             raise StreamError()
@@ -1654,8 +1656,6 @@ class Stream:
             self._interrupt_callback()
         elif notification.notification_type == NotificationType.TICK:
             self._ticker_callback()
-        elif notification.notification_type == NotificationType.RUNNING:
-            self._scheduler_running = not self._scheduler_running
         else:
             raise StreamError("Unrecognised notification type received: {}".format(notification.notification_type))