You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@buildstream.apache.org by tv...@apache.org on 2021/02/04 08:12:52 UTC
[buildstream] branch tristan/one-cache-size-job-2 created (now
627a017)
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a change to branch tristan/one-cache-size-job-2
in repository https://gitbox.apache.org/repos/asf/buildstream.git.
at 627a017 Scheduler: Introduced JobStatus instead of simple success boolean
This branch includes the following new commits:
new fa20374 utils.py: Fix stack traces when killing already dead processes
new 9f95650 _scheduler/scheduler.py: Make _schedule_jobs() private
new 8fd2cd9 _scheduler/scheduler.py: Only run one cache size job at a time
new 627a017 Scheduler: Introduced JobStatus instead of simple success boolean
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[buildstream] 04/04: Scheduler: Introduced JobStatus instead of
simple success boolean
Posted by tv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a commit to branch tristan/one-cache-size-job-2
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 627a0179a96ce45babee17e4388d34b95429be9c
Author: Tristan Van Berkom <tr...@codethink.co.uk>
AuthorDate: Mon Jan 7 11:20:11 2019 -0500
Scheduler: Introduced JobStatus instead of simple success boolean
This changes the deepest callback from when a Job completes to
propagate a JobStatus value instead of a simple boolean, and updates
all of the effected code paths which used to receive a boolean
to now handle the JobStatus values.
This further improves the situation for issue #753, as now we avoid
queueing cache size jobs for pull jobs which are skipped.
---
buildstream/_frontend/app.py | 6 ++---
buildstream/_scheduler/__init__.py | 2 +-
buildstream/_scheduler/jobs/__init__.py | 1 +
buildstream/_scheduler/jobs/cachesizejob.py | 6 ++---
buildstream/_scheduler/jobs/cleanupjob.py | 6 ++---
buildstream/_scheduler/jobs/elementjob.py | 6 ++---
buildstream/_scheduler/jobs/job.py | 36 +++++++++++++++++++++++++----
buildstream/_scheduler/queues/buildqueue.py | 6 ++---
buildstream/_scheduler/queues/fetchqueue.py | 5 ++--
buildstream/_scheduler/queues/pullqueue.py | 8 ++++---
buildstream/_scheduler/queues/queue.py | 13 +++++------
buildstream/_scheduler/queues/trackqueue.py | 5 ++--
buildstream/_scheduler/scheduler.py | 6 ++---
13 files changed, 68 insertions(+), 38 deletions(-)
diff --git a/buildstream/_frontend/app.py b/buildstream/_frontend/app.py
index 063ca1e..af38ae9 100644
--- a/buildstream/_frontend/app.py
+++ b/buildstream/_frontend/app.py
@@ -38,7 +38,7 @@ from .._message import Message, MessageType, unconditional_messages
from .._stream import Stream
from .._versions import BST_FORMAT_VERSION
from .. import _yaml
-from .._scheduler import ElementJob
+from .._scheduler import ElementJob, JobStatus
# Import frontend assets
from . import Profile, LogLine, Status
@@ -515,13 +515,13 @@ class App():
self._status.add_job(job)
self._maybe_render_status()
- def _job_completed(self, job, success):
+ def _job_completed(self, job, status):
self._status.remove_job(job)
self._maybe_render_status()
# Dont attempt to handle a failure if the user has already opted to
# terminate
- if not success and not self.stream.terminated:
+ if status == JobStatus.FAIL and not self.stream.terminated:
if isinstance(job, ElementJob):
element = job.element
diff --git a/buildstream/_scheduler/__init__.py b/buildstream/_scheduler/__init__.py
index b6e3eeb..4708598 100644
--- a/buildstream/_scheduler/__init__.py
+++ b/buildstream/_scheduler/__init__.py
@@ -26,4 +26,4 @@ from .queues.pushqueue import PushQueue
from .queues.pullqueue import PullQueue
from .scheduler import Scheduler, SchedStatus
-from .jobs import ElementJob
+from .jobs import ElementJob, JobStatus
diff --git a/buildstream/_scheduler/jobs/__init__.py b/buildstream/_scheduler/jobs/__init__.py
index 4b0b11d..3e21317 100644
--- a/buildstream/_scheduler/jobs/__init__.py
+++ b/buildstream/_scheduler/jobs/__init__.py
@@ -20,3 +20,4 @@
from .elementjob import ElementJob
from .cachesizejob import CacheSizeJob
from .cleanupjob import CleanupJob
+from .job import JobStatus
diff --git a/buildstream/_scheduler/jobs/cachesizejob.py b/buildstream/_scheduler/jobs/cachesizejob.py
index d46fd4c..6e4698a 100644
--- a/buildstream/_scheduler/jobs/cachesizejob.py
+++ b/buildstream/_scheduler/jobs/cachesizejob.py
@@ -16,7 +16,7 @@
# Author:
# Tristan Daniël Maat <tr...@codethink.co.uk>
#
-from .job import Job
+from .job import Job, JobStatus
class CacheSizeJob(Job):
@@ -30,8 +30,8 @@ class CacheSizeJob(Job):
def child_process(self):
return self._artifacts.compute_cache_size()
- def parent_complete(self, success, result):
- if success:
+ def parent_complete(self, status, result):
+ if status == JobStatus.OK:
self._artifacts.set_cache_size(result)
if self._complete_cb:
diff --git a/buildstream/_scheduler/jobs/cleanupjob.py b/buildstream/_scheduler/jobs/cleanupjob.py
index 8bdbba0..e579e97 100644
--- a/buildstream/_scheduler/jobs/cleanupjob.py
+++ b/buildstream/_scheduler/jobs/cleanupjob.py
@@ -16,7 +16,7 @@
# Author:
# Tristan Daniël Maat <tr...@codethink.co.uk>
#
-from .job import Job
+from .job import Job, JobStatus
class CleanupJob(Job):
@@ -29,6 +29,6 @@ class CleanupJob(Job):
def child_process(self):
return self._artifacts.clean()
- def parent_complete(self, success, result):
- if success:
+ def parent_complete(self, status, result):
+ if status == JobStatus.OK:
self._artifacts.set_cache_size(result)
diff --git a/buildstream/_scheduler/jobs/elementjob.py b/buildstream/_scheduler/jobs/elementjob.py
index 8ce5c06..fa0d34f 100644
--- a/buildstream/_scheduler/jobs/elementjob.py
+++ b/buildstream/_scheduler/jobs/elementjob.py
@@ -60,7 +60,7 @@ from .job import Job
# Args:
# job (Job): The job object which completed
# element (Element): The element passed to the Job() constructor
-# success (bool): True if the action_cb did not raise an exception
+# status (JobStatus): The status of whether the workload raised an exception
# result (object): The deserialized object returned by the `action_cb`, or None
# if `success` is False
#
@@ -93,8 +93,8 @@ class ElementJob(Job):
# Run the action
return self._action_cb(self._element)
- def parent_complete(self, success, result):
- self._complete_cb(self, self._element, success, self._result)
+ def parent_complete(self, status, result):
+ self._complete_cb(self, self._element, status, self._result)
def message(self, message_type, message, **kwargs):
args = dict(kwargs)
diff --git a/buildstream/_scheduler/jobs/job.py b/buildstream/_scheduler/jobs/job.py
index 3ab7bfa..2b08096 100644
--- a/buildstream/_scheduler/jobs/job.py
+++ b/buildstream/_scheduler/jobs/job.py
@@ -43,6 +43,22 @@ RC_PERM_FAIL = 2
RC_SKIPPED = 3
+# JobStatus:
+#
+# The job completion status, passed back through the
+# complete callbacks.
+#
+class JobStatus():
+ # Job succeeded
+ OK = 0
+
+ # A temporary BstError was raised
+ FAIL = 1
+
+ # A SkipJob was raised
+ SKIPPED = 3
+
+
# Used to distinguish between status messages and return values
class Envelope():
def __init__(self, message_type, message):
@@ -297,10 +313,10 @@ class Job():
# pass the result to the main thread.
#
# Args:
- # success (bool): Whether the job was successful.
+ # status (JobStatus): The job exit status
# result (any): The result returned by child_process().
#
- def parent_complete(self, success, result):
+ def parent_complete(self, status, result):
raise ImplError("Job '{kind}' does not implement parent_complete()"
.format(kind=type(self).__name__))
@@ -571,9 +587,19 @@ class Job():
self.spawn()
return
- success = returncode in (RC_OK, RC_SKIPPED)
- self.parent_complete(success, self._result)
- self._scheduler.job_completed(self, success)
+ # Resolve the outward facing overall job completion status
+ #
+ if returncode == RC_OK:
+ status = JobStatus.OK
+ elif returncode == RC_SKIPPED:
+ status = JobStatus.SKIPPED
+ elif returncode in (RC_FAIL, RC_PERM_FAIL):
+ status = JobStatus.FAIL
+ else:
+ status = JobStatus.FAIL
+
+ self.parent_complete(status, self._result)
+ self._scheduler.job_completed(self, status)
# Force the deletion of the queue and process objects to try and clean up FDs
self._queue = self._process = None
diff --git a/buildstream/_scheduler/queues/buildqueue.py b/buildstream/_scheduler/queues/buildqueue.py
index d053275..9d8e718 100644
--- a/buildstream/_scheduler/queues/buildqueue.py
+++ b/buildstream/_scheduler/queues/buildqueue.py
@@ -21,7 +21,7 @@
from datetime import timedelta
from . import Queue, QueueStatus
-from ..jobs import ElementJob
+from ..jobs import ElementJob, JobStatus
from ..resources import ResourceType
from ..._message import MessageType
@@ -104,7 +104,7 @@ class BuildQueue(Queue):
if artifacts.has_quota_exceeded():
self._scheduler.check_cache_size()
- def done(self, job, element, result, success):
+ def done(self, job, element, result, status):
# Inform element in main process that assembly is done
element._assemble_done()
@@ -117,5 +117,5 @@ class BuildQueue(Queue):
# artifact cache size for a successful build even though we know a
# failed build also grows the artifact cache size.
#
- if success:
+ if status == JobStatus.OK:
self._check_cache_size(job, element, result)
diff --git a/buildstream/_scheduler/queues/fetchqueue.py b/buildstream/_scheduler/queues/fetchqueue.py
index c58bfdb..fc11fd1 100644
--- a/buildstream/_scheduler/queues/fetchqueue.py
+++ b/buildstream/_scheduler/queues/fetchqueue.py
@@ -24,6 +24,7 @@ from ... import Consistency
# Local imports
from . import Queue, QueueStatus
from ..resources import ResourceType
+from ..jobs import JobStatus
# A queue which fetches element sources
@@ -66,9 +67,9 @@ class FetchQueue(Queue):
return QueueStatus.READY
- def done(self, _, element, result, success):
+ def done(self, _, element, result, status):
- if not success:
+ if status == JobStatus.FAIL:
return
element._update_state()
diff --git a/buildstream/_scheduler/queues/pullqueue.py b/buildstream/_scheduler/queues/pullqueue.py
index b861373..dbeb806 100644
--- a/buildstream/_scheduler/queues/pullqueue.py
+++ b/buildstream/_scheduler/queues/pullqueue.py
@@ -21,6 +21,7 @@
# Local imports
from . import Queue, QueueStatus
from ..resources import ResourceType
+from ..jobs import JobStatus
from ..._exceptions import SkipJob
@@ -54,9 +55,9 @@ class PullQueue(Queue):
else:
return QueueStatus.SKIP
- def done(self, _, element, result, success):
+ def done(self, _, element, result, status):
- if not success:
+ if status == JobStatus.FAIL:
return
element._pull_done()
@@ -64,4 +65,5 @@ class PullQueue(Queue):
# Build jobs will check the "approximate" size first. Since we
# do not get an artifact size from pull jobs, we have to
# actually check the cache size.
- self._scheduler.check_cache_size()
+ if status == JobStatus.OK:
+ self._scheduler.check_cache_size()
diff --git a/buildstream/_scheduler/queues/queue.py b/buildstream/_scheduler/queues/queue.py
index 055e2f8..707fcf5 100644
--- a/buildstream/_scheduler/queues/queue.py
+++ b/buildstream/_scheduler/queues/queue.py
@@ -25,7 +25,7 @@ from enum import Enum
import traceback
# Local imports
-from ..jobs import ElementJob
+from ..jobs import ElementJob, JobStatus
from ..resources import ResourceType
# BuildStream toplevel imports
@@ -133,10 +133,9 @@ class Queue():
# job (Job): The job which completed processing
# element (Element): The element which completed processing
# result (any): The return value of the process() implementation
- # success (bool): True if the process() implementation did not
- # raise any exception
+ # status (JobStatus): The return status of the Job
#
- def done(self, job, element, result, success):
+ def done(self, job, element, result, status):
pass
#####################################################
@@ -291,7 +290,7 @@ class Queue():
#
# See the Job object for an explanation of the call signature
#
- def _job_done(self, job, element, success, result):
+ def _job_done(self, job, element, status, result):
# Update values that need to be synchronized in the main task
# before calling any queue implementation
@@ -301,7 +300,7 @@ class Queue():
# and determine if it should be considered as processed
# or skipped.
try:
- self.done(job, element, result, success)
+ self.done(job, element, result, status)
except BstError as e:
# Report error and mark as failed
@@ -337,7 +336,7 @@ class Queue():
# if they are not skipped.
if job.skipped:
self.skipped_elements.append(element)
- elif success:
+ elif status == JobStatus.OK:
self.processed_elements.append(element)
else:
self.failed_elements.append(element)
diff --git a/buildstream/_scheduler/queues/trackqueue.py b/buildstream/_scheduler/queues/trackqueue.py
index 5144180..245b528 100644
--- a/buildstream/_scheduler/queues/trackqueue.py
+++ b/buildstream/_scheduler/queues/trackqueue.py
@@ -24,6 +24,7 @@ from ...plugin import _plugin_lookup
# Local imports
from . import Queue, QueueStatus
from ..resources import ResourceType
+from ..jobs import JobStatus
# A queue which tracks sources
@@ -47,9 +48,9 @@ class TrackQueue(Queue):
return QueueStatus.READY
- def done(self, _, element, result, success):
+ def done(self, _, element, result, status):
- if not success:
+ if status == JobStatus.FAIL:
return
# Set the new refs in the main process one by one as they complete
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index ecbfef3..97335de 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -237,14 +237,14 @@ class Scheduler():
# Args:
# queue (Queue): The Queue holding a complete job
# job (Job): The completed Job
- # success (bool): Whether the Job completed with a success status
+ # status (JobStatus): The status of the completed job
#
- def job_completed(self, job, success):
+ def job_completed(self, job, status):
self._resources.clear_job_resources(job)
self.active_jobs.remove(job)
if job.action_name in _REDUNDANT_EXCLUSIVE_ACTIONS:
self._exclusive_active.remove(job.action_name)
- self._job_complete_callback(job, success)
+ self._job_complete_callback(job, status)
self._schedule_queue_jobs()
self._sched()
[buildstream] 03/04: _scheduler/scheduler.py: Only run one cache
size job at a time
Posted by tv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a commit to branch tristan/one-cache-size-job-2
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 8fd2cd9ee4e9ef78dde9bdc2b3a13db64076b6cd
Author: Tristan Van Berkom <tr...@codethink.co.uk>
AuthorDate: Sun Jan 6 14:15:46 2019 -0500
_scheduler/scheduler.py: Only run one cache size job at a time
When queuing the special cache management related cleanup and
cache size jobs, we now treat these jobs as special and do the
following:
* Avoid queueing a cleanup/cache_size job if one is already queued
We just drop redundantly queued jobs here.
* Ensure that jobs of this type only run one at a time
This could have been done with the Resources mechanics,
however as these special jobs have the same properties and
are basically owned by the Scheduler, it seemed more straight
forward to handle the behaviors of these special jobs together.
This fixes issue #753
---
buildstream/_scheduler/scheduler.py | 53 +++++++++++++++++++++++++++++++------
1 file changed, 45 insertions(+), 8 deletions(-)
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index 8facb08..ecbfef3 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -38,6 +38,16 @@ class SchedStatus():
TERMINATED = 1
+# Our _REDUNDANT_EXCLUSIVE_ACTIONS jobs are special ones
+# which we launch dynamically, they have the property of being
+# meaningless to queue if one is already queued, and it also
+# doesnt make sense to run them in parallel
+#
+_ACTION_NAME_CLEANUP = 'cleanup'
+_ACTION_NAME_CACHE_SIZE = 'cache_size'
+_REDUNDANT_EXCLUSIVE_ACTIONS = [_ACTION_NAME_CLEANUP, _ACTION_NAME_CACHE_SIZE]
+
+
# Scheduler()
#
# The scheduler operates on a list queues, each of which is meant to accomplish
@@ -94,6 +104,15 @@ class Scheduler():
self._suspendtime = None
self._queue_jobs = True # Whether we should continue to queue jobs
+ # Whether our exclusive jobs, like 'cleanup' are currently already
+ # waiting or active.
+ #
+ # This is just a bit quicker than scanning the wait queue and active
+ # queue and comparing job action names.
+ #
+ self._exclusive_waiting = set()
+ self._exclusive_active = set()
+
self._resources = Resources(context.sched_builders,
context.sched_fetchers,
context.sched_pushers)
@@ -223,6 +242,8 @@ class Scheduler():
def job_completed(self, job, success):
self._resources.clear_job_resources(job)
self.active_jobs.remove(job)
+ if job.action_name in _REDUNDANT_EXCLUSIVE_ACTIONS:
+ self._exclusive_active.remove(job.action_name)
self._job_complete_callback(job, success)
self._schedule_queue_jobs()
self._sched()
@@ -233,14 +254,9 @@ class Scheduler():
# size is calculated, a cleanup job will be run automatically
# if needed.
#
- # FIXME: This should ensure that only one cache size job
- # is ever pending at a given time. If a cache size
- # job is already running, it is correct to queue
- # a new one, it is incorrect to have more than one
- # of these jobs pending at a given time, though.
- #
def check_cache_size(self):
- job = CacheSizeJob(self, 'cache_size', 'cache_size/cache_size',
+ job = CacheSizeJob(self, _ACTION_NAME_CACHE_SIZE,
+ 'cache_size/cache_size',
resources=[ResourceType.CACHE,
ResourceType.PROCESS],
complete_cb=self._run_cleanup)
@@ -263,10 +279,19 @@ class Scheduler():
if not self._resources.reserve_job_resources(job):
continue
+ # Postpone these jobs if one is already running
+ if job.action_name in _REDUNDANT_EXCLUSIVE_ACTIONS and \
+ job.action_name in self._exclusive_active:
+ continue
+
job.spawn()
self.waiting_jobs.remove(job)
self.active_jobs.append(job)
+ if job.action_name in _REDUNDANT_EXCLUSIVE_ACTIONS:
+ self._exclusive_waiting.remove(job.action_name)
+ self._exclusive_active.add(job.action_name)
+
if self._job_start_callback:
self._job_start_callback(job)
@@ -287,6 +312,18 @@ class Scheduler():
#
def _schedule_jobs(self, jobs):
for job in jobs:
+
+ # Special treatment of our redundant exclusive jobs
+ #
+ if job.action_name in _REDUNDANT_EXCLUSIVE_ACTIONS:
+
+ # Drop the job if one is already queued
+ if job.action_name in self._exclusive_waiting:
+ continue
+
+ # Mark this action type as queued
+ self._exclusive_waiting.add(job.action_name)
+
self.waiting_jobs.append(job)
# _schedule_queue_jobs()
@@ -355,7 +392,7 @@ class Scheduler():
if not artifacts.has_quota_exceeded():
return
- job = CleanupJob(self, 'cleanup', 'cleanup/cleanup',
+ job = CleanupJob(self, _ACTION_NAME_CLEANUP, 'cleanup/cleanup',
resources=[ResourceType.CACHE,
ResourceType.PROCESS],
exclusive_resources=[ResourceType.CACHE])
[buildstream] 02/04: _scheduler/scheduler.py: Make _schedule_jobs()
private
Posted by tv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a commit to branch tristan/one-cache-size-job-2
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 9f95650218002cd7abfbe516c142e5be28988883
Author: Tristan Van Berkom <tr...@codethink.co.uk>
AuthorDate: Sun Jan 6 14:21:00 2019 -0500
_scheduler/scheduler.py: Make _schedule_jobs() private
This is not used anywhere outside of the Scheduler, currently
only the Scheduler itself is allowed to queue a job at this level.
If the highlevel business logic for automatic queueing of auxiliary
jobs moves to another location, we can make this public again.
---
buildstream/_scheduler/scheduler.py | 34 ++++++++++++++++++----------------
1 file changed, 18 insertions(+), 16 deletions(-)
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index b76c730..8facb08 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -211,19 +211,6 @@ class Scheduler():
starttime = timenow
return timenow - starttime
- # schedule_jobs()
- #
- # Args:
- # jobs ([Job]): A list of jobs to schedule
- #
- # Schedule 'Job's for the scheduler to run. Jobs scheduled will be
- # run as soon any other queueing jobs finish, provided sufficient
- # resources are available for them to run
- #
- def schedule_jobs(self, jobs):
- for job in jobs:
- self.waiting_jobs.append(job)
-
# job_completed():
#
# Called when a Job completes
@@ -257,7 +244,7 @@ class Scheduler():
resources=[ResourceType.CACHE,
ResourceType.PROCESS],
complete_cb=self._run_cleanup)
- self.schedule_jobs([job])
+ self._schedule_jobs([job])
#######################################################
# Local Private Methods #
@@ -287,6 +274,21 @@ class Scheduler():
if not self.active_jobs and not self.waiting_jobs:
self.loop.stop()
+ # _schedule_jobs()
+ #
+ # The main entry point for jobs to be scheduled.
+ #
+ # This is called either as a result of scanning the queues
+ # in _schedule_queue_jobs(), or directly by the Scheduler
+ # to insert special jobs like cleanups.
+ #
+ # Args:
+ # jobs ([Job]): A list of jobs to schedule
+ #
+ def _schedule_jobs(self, jobs):
+ for job in jobs:
+ self.waiting_jobs.append(job)
+
# _schedule_queue_jobs()
#
# Ask the queues what jobs they want to schedule and schedule
@@ -331,7 +333,7 @@ class Scheduler():
# the next queue and process them.
process_queues = any(q.dequeue_ready() for q in self.queues)
- self.schedule_jobs(ready)
+ self._schedule_jobs(ready)
self._sched()
# _run_cleanup()
@@ -357,7 +359,7 @@ class Scheduler():
resources=[ResourceType.CACHE,
ResourceType.PROCESS],
exclusive_resources=[ResourceType.CACHE])
- self.schedule_jobs([job])
+ self._schedule_jobs([job])
# _suspend_jobs()
#
[buildstream] 01/04: utils.py: Fix stack traces when killing
already dead processes
Posted by tv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a commit to branch tristan/one-cache-size-job-2
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit fa20374fb8ba3af5c01b06254909a33e27cea643
Author: Tristan Van Berkom <tr...@codethink.co.uk>
AuthorDate: Mon Jan 7 11:04:31 2019 -0500
utils.py: Fix stack traces when killing already dead processes
Handle NoSuchProcess errors transparently inside the function, instead
of only from outside of it where Job uses it.
---
buildstream/_scheduler/jobs/job.py | 9 +--------
buildstream/utils.py | 5 +++++
2 files changed, 6 insertions(+), 8 deletions(-)
diff --git a/buildstream/_scheduler/jobs/job.py b/buildstream/_scheduler/jobs/job.py
index 60ae0d0..3ab7bfa 100644
--- a/buildstream/_scheduler/jobs/job.py
+++ b/buildstream/_scheduler/jobs/job.py
@@ -215,17 +215,10 @@ class Job():
# Forcefully kill the process, and any children it might have.
#
def kill(self):
-
# Force kill
self.message(MessageType.WARN,
"{} did not terminate gracefully, killing".format(self.action_name))
-
- try:
- utils._kill_process_tree(self._process.pid)
- # This can happen if the process died of its own accord before
- # we try to kill it
- except psutil.NoSuchProcess:
- return
+ utils._kill_process_tree(self._process.pid)
# suspend()
#
diff --git a/buildstream/utils.py b/buildstream/utils.py
index a460031..4943336 100644
--- a/buildstream/utils.py
+++ b/buildstream/utils.py
@@ -1050,6 +1050,11 @@ def _kill_process_tree(pid):
# Ignore this error, it can happen with
# some setuid bwrap processes.
pass
+ except psutil.NoSuchProcess:
+ # It is certain that this has already been sent
+ # SIGTERM, so there is a window where the process
+ # could have exited already.
+ pass
# Bloody Murder
for child in children: