Merge branch 'tristan/fix-terminated-jobs' into 'master'

_scheduler: Fix bookkeeping of terminated jobs

Closes #479

See merge request BuildStream/buildstream!850

2 files changed, 22 insertions, 10 deletions

diff --git a/buildstream/_scheduler/jobs/job.py b/buildstream/_scheduler/jobs/job.py
index 1c6b4a582..a1b90a080 100644
--- a/buildstream/_scheduler/jobs/job.py
+++ b/buildstream/_scheduler/jobs/job.py
@@ -119,6 +119,8 @@ class Job():
         self._result = None                    # Return value of child action in the parent
         self._tries = 0                        # Try count, for retryable jobs
         self._skipped_flag = False             # Indicate whether the job was skipped.
+        self._terminated = False               # Whether this job has been explicitly terminated
+
         # If False, a retry will not be attempted regardless of whether _tries is less than _max_retries.
         #
         self._retry_flag = True
@@ -190,6 +192,8 @@ class Job():
         # Terminate the process using multiprocessing API pathway
         self._process.terminate()
 
+        self._terminated = True
+
     # terminate_wait()
     #
     # Wait for terminated jobs to complete
@@ -273,18 +277,22 @@ class Job():
     # running the integration commands).
     #
     # Args:
-    #     (int): The plugin identifier for this task
+    #     task_id (int): The plugin identifier for this task
     #
     def set_task_id(self, task_id):
         self._task_id = task_id
 
     # skipped
     #
+    # This will evaluate to True if the job was skipped
+    # during processing, or if it was forcefully terminated.
+    #
     # Returns:
-    #    bool: True if the job was skipped while processing.
+    #    (bool): Whether the job should appear as skipped
+    #
     @property
     def skipped(self):
-        return self._skipped_flag
+        return self._skipped_flag or self._terminated
 
     #######################################################
     #                  Abstract Methods                   #
diff --git a/buildstream/_scheduler/queues/queue.py b/buildstream/_scheduler/queues/queue.py
index f058663a1..0e07a078f 100644
--- a/buildstream/_scheduler/queues/queue.py
+++ b/buildstream/_scheduler/queues/queue.py
@@ -326,16 +326,20 @@ class Queue():
                           detail=traceback.format_exc())
             self.failed_elements.append(element)
         else:
-
-            # No exception occured, handle the success/failure state in the normal way
             #
+            # No exception occured in post processing
+            #
+
+            # All jobs get placed on the done queue for later processing.
             self._done_queue.append(job)
 
-            if success:
-                if not job.skipped:
-                    self.processed_elements.append(element)
-                else:
-                    self.skipped_elements.append(element)
+            # A Job can be skipped whether or not it has failed,
+            # we want to only bookkeep them as processed or failed
+            # if they are not skipped.
+            if job.skipped:
+                self.skipped_elements.append(element)
+            elif success:
+                self.processed_elements.append(element)
             else:
                 self.failed_elements.append(element)