scheduler.py: Remove task-based notifications and use the state

The State is the interface between both, there is no need to do multiple
round-trips to handle such notifications

2 files changed, 5 insertions, 26 deletions

diff --git a/src/buildstream/_scheduler/scheduler.py b/src/buildstream/_scheduler/scheduler.py
index 14d9c53d5..a7d3ff471 100644
--- a/src/buildstream/_scheduler/scheduler.py
+++ b/src/buildstream/_scheduler/scheduler.py
@@ -55,8 +55,6 @@ class SchedStatus(FastEnum):
 #
 class NotificationType(FastEnum):
     INTERRUPT = "interrupt"
-    JOB_START = "job_start"
-    JOB_COMPLETE = "job_complete"
     TICK = "tick"
     TERMINATE = "terminate"
     QUIT = "quit"
@@ -311,7 +309,6 @@ class Scheduler:
         # Remove from the active jobs list
         self._active_jobs.remove(job)
 
-        element_info = None
         if status == JobStatus.FAIL:
             # If it's an elementjob, we want to compare against the failure messages
             # and send the unique_id and display key tuple of the Element. This can then
@@ -322,15 +319,10 @@ class Scheduler:
             else:
                 element_info = None
 
-        # Now check for more jobs
-        notification = Notification(
-            NotificationType.JOB_COMPLETE,
-            full_name=job.name,
-            job_action=job.action_name,
-            job_status=status,
-            element=element_info,
-        )
-        self._notify(notification)
+            self._state.fail_task(job.action_name, job.name, element_info)
+
+        self._state.remove_task(job.action_name, job.name)
+
         self._sched()
 
     #######################################################
@@ -372,13 +364,7 @@ class Scheduler:
         self._active_jobs.append(job)
         job.start()
 
-        notification = Notification(
-            NotificationType.JOB_START,
-            full_name=job.name,
-            job_action=job.action_name,
-            time=self._state.elapsed_time(start_time=self._starttime),
-        )
-        self._notify(notification)
+        self._state.add_task(job.action_name, job.name, self._state.elapsed_time(start_time=self._starttime))
 
     # _sched_queue_jobs()
     #
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 64e4d09ec..277ea7394 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -45,7 +45,6 @@ from ._scheduler import (
     ArtifactPushQueue,
     NotificationType,
     Notification,
-    JobStatus,
 )
 from .element import Element
 from ._pipeline import Pipeline
@@ -1662,12 +1661,6 @@ class Stream:
             self._interrupt_callback()
         elif notification.notification_type == NotificationType.TICK:
             self._ticker_callback()
-        elif notification.notification_type == NotificationType.JOB_START:
-            self._state.add_task(notification.job_action, notification.full_name, notification.time)
-        elif notification.notification_type == NotificationType.JOB_COMPLETE:
-            self._state.remove_task(notification.job_action, notification.full_name)
-            if notification.job_status == JobStatus.FAIL:
-                self._state.fail_task(notification.job_action, notification.full_name, notification.element)
         elif notification.notification_type == NotificationType.SCHED_START_TIME:
             self._starttime = notification.time
         elif notification.notification_type == NotificationType.RUNNING: