You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@buildstream.apache.org by ro...@apache.org on 2020/12/29 13:32:52 UTC
[buildstream] branch phil/712 created (now 3b63a8f)
This is an automated email from the ASF dual-hosted git repository.
root pushed a change to branch phil/712
in repository https://gitbox.apache.org/repos/asf/buildstream.git.
at 3b63a8f fixup! WIP: _schedular: introduce second 'high priority' waiting jobs queue
This branch includes the following new commits:
new 9e60c9b Revert "WIP: use priority queue in schedular"
new 7666f16 WIP: Prepend fetch jobs to waiting jobs
new 391bda3 WIP: _schedular: introduce second 'high priority' waiting jobs queue
new 3b63a8f fixup! WIP: _schedular: introduce second 'high priority' waiting jobs queue
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: fixup! WIP: _schedular: introduce second 'high
priority' waiting jobs queue
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch phil/712
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 3b63a8fc0cab6d94510e8bab5679348a3e644e93
Author: Phil Dawson <ph...@codethink.co.uk>
AuthorDate: Tue Nov 27 15:20:28 2018 +0000
fixup! WIP: _schedular: introduce second 'high priority' waiting jobs queue
---
buildstream/_scheduler/queues/fetchqueue.py | 1 +
buildstream/_scheduler/queues/queue.py | 1 +
buildstream/_scheduler/scheduler.py | 58 +++++++++++++++++------------
3 files changed, 37 insertions(+), 23 deletions(-)
diff --git a/buildstream/_scheduler/queues/fetchqueue.py b/buildstream/_scheduler/queues/fetchqueue.py
index 446dbbd..5c441ba 100644
--- a/buildstream/_scheduler/queues/fetchqueue.py
+++ b/buildstream/_scheduler/queues/fetchqueue.py
@@ -33,6 +33,7 @@ class FetchQueue(Queue):
action_name = "Fetch"
complete_name = "Fetched"
resources = [ResourceType.DOWNLOAD]
+ high_priority = True
def __init__(self, scheduler, skip_cached=False):
super().__init__(scheduler)
diff --git a/buildstream/_scheduler/queues/queue.py b/buildstream/_scheduler/queues/queue.py
index 909cebb..3c04140 100644
--- a/buildstream/_scheduler/queues/queue.py
+++ b/buildstream/_scheduler/queues/queue.py
@@ -58,6 +58,7 @@ class Queue():
action_name = None
complete_name = None
resources = [] # Resources this queues' jobs want
+ high_priority = False # If jobs from this queue should be prioritised by the scheduler
def __init__(self, scheduler):
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index b76c730..1b05415 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -71,12 +71,13 @@ class Scheduler():
#
# Public members
#
- self.active_jobs = [] # Jobs currently being run in the scheduler
- self.waiting_jobs = [] # Jobs waiting for resources
- self.queues = None # Exposed for the frontend to print summaries
- self.context = context # The Context object shared with Queues
- self.terminated = False # Whether the scheduler was asked to terminate or has terminated
- self.suspended = False # Whether the scheduler is currently suspended
+ self.active_jobs = [] # Jobs currently being run in the scheduler
+ self.waiting_jobs = [] # Jobs waiting for resources
+ self.waiting_priority_jobs = [] # High priority jobs waiting for resources
+ self.queues = None # Exposed for the frontend to print summaries
+ self.context = context # The Context object shared with Queues
+ self.terminated = False # Whether the scheduler was asked to terminate or has terminated
+ self.suspended = False # Whether the scheduler is currently suspended
# These are shared with the Job, but should probably be removed or made private in some way.
self.loop = None # Shared for Job access to observe the message queue
@@ -220,7 +221,9 @@ class Scheduler():
# run as soon any other queueing jobs finish, provided sufficient
# resources are available for them to run
#
- def schedule_jobs(self, jobs):
+ def schedule_jobs(self, jobs, priority_jobs):
+ for job in priority_jobs:
+ self.waiting_priority_jobs.append(job)
for job in jobs:
self.waiting_jobs.append(job)
@@ -257,7 +260,7 @@ class Scheduler():
resources=[ResourceType.CACHE,
ResourceType.PROCESS],
complete_cb=self._run_cleanup)
- self.schedule_jobs([job])
+ self.schedule_jobs([job], [])
#######################################################
# Local Private Methods #
@@ -269,22 +272,27 @@ class Scheduler():
# automatically when Scheduler.run() is called initially,
#
def _sched(self):
- for job in self.waiting_jobs:
- self._resources.reserve_exclusive_resources(job)
+ def allocate_resources_and_spawn_jobs(job_list):
+ for job in job_list:
+ self._resources.reserve_exclusive_resources(job)
+
+ for job in job_list:
+ if not self._resources.reserve_job_resources(job):
+ continue
- for job in self.waiting_jobs:
- if not self._resources.reserve_job_resources(job):
- continue
+ job.spawn()
+ job_list.remove(job)
+ self.active_jobs.append(job)
- job.spawn()
- self.waiting_jobs.remove(job)
- self.active_jobs.append(job)
+ if self._job_start_callback:
+ self._job_start_callback(job)
- if self._job_start_callback:
- self._job_start_callback(job)
+ # Process jobs from the high priority list first
+ allocate_resources_and_spawn_jobs(self.waiting_priority_jobs)
+ allocate_resources_and_spawn_jobs(self.waiting_jobs)
# If nothings ticking, time to bail out
- if not self.active_jobs and not self.waiting_jobs:
+ if not self.active_jobs and not self.waiting_jobs and not self.waiting_priority_jobs:
self.loop.stop()
# _schedule_queue_jobs()
@@ -298,6 +306,7 @@ class Scheduler():
#
def _schedule_queue_jobs(self):
ready = []
+ ready_priority = []
process_queues = True
while self._queue_jobs and process_queues:
@@ -322,16 +331,19 @@ class Scheduler():
# to fetch tasks for elements which failed to pull, and
# thus need all the pulls to complete before ever starting
# a build
- ready.extend(chain.from_iterable(
- queue.pop_ready_jobs() for queue in reversed(self.queues)
- ))
+
+ for queue in reversed(self.queues):
+ if queue.high_priority:
+ ready_priority.extend(queue.pop_ready_jobs())
+ else:
+ ready.extend(queue.pop_ready_jobs())
# pop_ready_jobs() may have skipped jobs, adding them to
# the done_queue. Pull these skipped elements forward to
# 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, ready_priority)
self._sched()
# _run_cleanup()
[buildstream] 02/04: WIP: Prepend fetch jobs to waiting jobs
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch phil/712
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 7666f161465937e4b1d7217218df73ead394823b
Author: Phil Dawson <ph...@codethink.co.uk>
AuthorDate: Tue Nov 27 15:20:28 2018 +0000
WIP: Prepend fetch jobs to waiting jobs
---
buildstream/_scheduler/queues/queue.py | 1 +
buildstream/_scheduler/scheduler.py | 38 ++++++++++++++++++++--------------
2 files changed, 23 insertions(+), 16 deletions(-)
diff --git a/buildstream/_scheduler/queues/queue.py b/buildstream/_scheduler/queues/queue.py
index 909cebb..7df3bb1 100644
--- a/buildstream/_scheduler/queues/queue.py
+++ b/buildstream/_scheduler/queues/queue.py
@@ -58,6 +58,7 @@ class Queue():
action_name = None
complete_name = None
resources = [] # Resources this queues' jobs want
+ high_priority = False # If the jobs from this queue should be prioritised by the scheduler
def __init__(self, scheduler):
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index b76c730..cfaa0a2 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -25,6 +25,7 @@ from itertools import chain
import signal
import datetime
from contextlib import contextmanager
+from collections import deque
# Local imports
from .resources import Resources, ResourceType
@@ -71,16 +72,16 @@ class Scheduler():
#
# Public members
#
- self.active_jobs = [] # Jobs currently being run in the scheduler
- self.waiting_jobs = [] # Jobs waiting for resources
- self.queues = None # Exposed for the frontend to print summaries
- self.context = context # The Context object shared with Queues
- self.terminated = False # Whether the scheduler was asked to terminate or has terminated
- self.suspended = False # Whether the scheduler is currently suspended
+ self.active_jobs = [] # Jobs currently being run in the scheduler
+ self.waiting_jobs = deque() # Jobs waiting for resources
+ self.queues = None # Exposed for the frontend to print summaries
+ self.context = context # The Context object shared with Queues
+ self.terminated = False # Whether the scheduler was asked to terminate or has terminated
+ self.suspended = False # Whether the scheduler is currently suspended
# These are shared with the Job, but should probably be removed or made private in some way.
- self.loop = None # Shared for Job access to observe the message queue
- self.internal_stops = 0 # Amount of SIGSTP signals we've introduced, this is shared with job.py
+ self.loop = None # Shared for Job access to observe the message queue
+ self.internal_stops = 0 # Amount of SIGSTP signals we've introduced, this is shared with job.py
#
# Private members
@@ -215,14 +216,15 @@ class Scheduler():
#
# Args:
# jobs ([Job]): A list of jobs to schedule
+ # priority_jobs([Job]): A list of jobs which should be prioritised over those in jobs
#
# 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
+ # resources are available for them to run.
#
- def schedule_jobs(self, jobs):
- for job in jobs:
- self.waiting_jobs.append(job)
+ def schedule_jobs(self, jobs, priority_jobs):
+ self.waiting_jobs.extend(jobs)
+ self.waiting_jobs.extendleft(priority_jobs)
# job_completed():
#
@@ -298,6 +300,7 @@ class Scheduler():
#
def _schedule_queue_jobs(self):
ready = []
+ ready_priority = []
process_queues = True
while self._queue_jobs and process_queues:
@@ -322,16 +325,19 @@ class Scheduler():
# to fetch tasks for elements which failed to pull, and
# thus need all the pulls to complete before ever starting
# a build
- ready.extend(chain.from_iterable(
- queue.pop_ready_jobs() for queue in reversed(self.queues)
- ))
+ for queue in reversed(self.queues):
+ ready_jobs = queue.pop_ready_jobs()
+ if queue.high_priority:
+ ready_priority.extend(ready_jobs)
+ else:
+ ready.extend(ready_jobs)
# pop_ready_jobs() may have skipped jobs, adding them to
# the done_queue. Pull these skipped elements forward to
# 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, ready_priority)
self._sched()
# _run_cleanup()
[buildstream] 01/04: Revert "WIP: use priority queue in schedular"
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch phil/712
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 9e60c9ba226a7e7bc52c64f8155bc5105db7126e
Author: Phil Dawson <ph...@codethink.co.uk>
AuthorDate: Tue Nov 27 15:20:28 2018 +0000
Revert "WIP: use priority queue in schedular"
This reverts commit 34395d069f1fe78c0d50af6255d2a62c68fbde95.
---
buildstream/_scheduler/jobs/cachesizejob.py | 3 ---
buildstream/_scheduler/jobs/cleanupjob.py | 3 ---
buildstream/_scheduler/jobs/elementjob.py | 10 ----------
buildstream/_scheduler/jobs/job.py | 4 ----
buildstream/_scheduler/scheduler.py | 5 ++---
setup.py | 1 -
6 files changed, 2 insertions(+), 24 deletions(-)
diff --git a/buildstream/_scheduler/jobs/cachesizejob.py b/buildstream/_scheduler/jobs/cachesizejob.py
index 17ccb0a..d46fd4c 100644
--- a/buildstream/_scheduler/jobs/cachesizejob.py
+++ b/buildstream/_scheduler/jobs/cachesizejob.py
@@ -39,6 +39,3 @@ class CacheSizeJob(Job):
def child_process_data(self):
return {}
-
- def key(self):
- return (100, 'cache-size')
diff --git a/buildstream/_scheduler/jobs/cleanupjob.py b/buildstream/_scheduler/jobs/cleanupjob.py
index 3e860e2..8bdbba0 100644
--- a/buildstream/_scheduler/jobs/cleanupjob.py
+++ b/buildstream/_scheduler/jobs/cleanupjob.py
@@ -32,6 +32,3 @@ class CleanupJob(Job):
def parent_complete(self, success, result):
if success:
self._artifacts.set_cache_size(result)
-
- def key(self):
- return (0, 'cleanup')
diff --git a/buildstream/_scheduler/jobs/elementjob.py b/buildstream/_scheduler/jobs/elementjob.py
index 1d6305c..8ce5c06 100644
--- a/buildstream/_scheduler/jobs/elementjob.py
+++ b/buildstream/_scheduler/jobs/elementjob.py
@@ -22,13 +22,6 @@ from ..._message import Message, MessageType
from .job import Job
-_ACTIONS = {
- "Build": 10,
- "Fetch": 20,
- "Pull": 30,
- "Push": 40,
- "Track": 50,
-}
# ElementJob()
#
@@ -120,6 +113,3 @@ class ElementJob(Job):
data['workspace'] = workspace.to_dict()
return data
-
- def key(self):
- return (_ACTIONS.get(self.action_name, 100), self._element.name)
diff --git a/buildstream/_scheduler/jobs/job.py b/buildstream/_scheduler/jobs/job.py
index ff0ea83..60ae0d0 100644
--- a/buildstream/_scheduler/jobs/job.py
+++ b/buildstream/_scheduler/jobs/job.py
@@ -353,10 +353,6 @@ class Job():
def child_process_data(self):
return {}
- def key(self):
- raise ImplError("Job '{kind}' does not implement key()"
- .format(kind=type(self).__name__))
-
#######################################################
# Local Private Methods #
#######################################################
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index 6b849f3..b76c730 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -25,7 +25,6 @@ from itertools import chain
import signal
import datetime
from contextlib import contextmanager
-from sortedcontainers import SortedList
# Local imports
from .resources import Resources, ResourceType
@@ -73,7 +72,7 @@ class Scheduler():
# Public members
#
self.active_jobs = [] # Jobs currently being run in the scheduler
- self.waiting_jobs = SortedList([], key=lambda job: job.key()) # Jobs waiting for resources
+ self.waiting_jobs = [] # Jobs waiting for resources
self.queues = None # Exposed for the frontend to print summaries
self.context = context # The Context object shared with Queues
self.terminated = False # Whether the scheduler was asked to terminate or has terminated
@@ -223,7 +222,7 @@ class Scheduler():
#
def schedule_jobs(self, jobs):
for job in jobs:
- self.waiting_jobs.add(job)
+ self.waiting_jobs.append(job)
# job_completed():
#
diff --git a/setup.py b/setup.py
index 95d9204..76610f0 100755
--- a/setup.py
+++ b/setup.py
@@ -343,7 +343,6 @@ setup(name='BuildStream',
'jinja2 >= 2.10',
'protobuf >= 3.5',
'grpcio >= 1.10',
- 'sortedcontainers >= 1.5.7',
],
entry_points=bst_install_entry_points,
tests_require=dev_requires,
[buildstream] 03/04: WIP: _schedular: introduce second 'high
priority' waiting jobs queue
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch phil/712
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 391bda38013bfee52bb39c57588e66c536056b7d
Author: Phil Dawson <ph...@codethink.co.uk>
AuthorDate: Tue Nov 27 15:20:28 2018 +0000
WIP: _schedular: introduce second 'high priority' waiting jobs queue
This reverts commit b23e5d16aa0eb5e0ba41fddc41bc9d29b1cad8dc.
---
buildstream/_scheduler/queues/queue.py | 1 -
buildstream/_scheduler/scheduler.py | 38 ++++++++++++++--------------------
2 files changed, 16 insertions(+), 23 deletions(-)
diff --git a/buildstream/_scheduler/queues/queue.py b/buildstream/_scheduler/queues/queue.py
index 7df3bb1..909cebb 100644
--- a/buildstream/_scheduler/queues/queue.py
+++ b/buildstream/_scheduler/queues/queue.py
@@ -58,7 +58,6 @@ class Queue():
action_name = None
complete_name = None
resources = [] # Resources this queues' jobs want
- high_priority = False # If the jobs from this queue should be prioritised by the scheduler
def __init__(self, scheduler):
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index cfaa0a2..b76c730 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -25,7 +25,6 @@ from itertools import chain
import signal
import datetime
from contextlib import contextmanager
-from collections import deque
# Local imports
from .resources import Resources, ResourceType
@@ -72,16 +71,16 @@ class Scheduler():
#
# Public members
#
- self.active_jobs = [] # Jobs currently being run in the scheduler
- self.waiting_jobs = deque() # Jobs waiting for resources
- self.queues = None # Exposed for the frontend to print summaries
- self.context = context # The Context object shared with Queues
- self.terminated = False # Whether the scheduler was asked to terminate or has terminated
- self.suspended = False # Whether the scheduler is currently suspended
+ self.active_jobs = [] # Jobs currently being run in the scheduler
+ self.waiting_jobs = [] # Jobs waiting for resources
+ self.queues = None # Exposed for the frontend to print summaries
+ self.context = context # The Context object shared with Queues
+ self.terminated = False # Whether the scheduler was asked to terminate or has terminated
+ self.suspended = False # Whether the scheduler is currently suspended
# These are shared with the Job, but should probably be removed or made private in some way.
- self.loop = None # Shared for Job access to observe the message queue
- self.internal_stops = 0 # Amount of SIGSTP signals we've introduced, this is shared with job.py
+ self.loop = None # Shared for Job access to observe the message queue
+ self.internal_stops = 0 # Amount of SIGSTP signals we've introduced, this is shared with job.py
#
# Private members
@@ -216,15 +215,14 @@ class Scheduler():
#
# Args:
# jobs ([Job]): A list of jobs to schedule
- # priority_jobs([Job]): A list of jobs which should be prioritised over those in jobs
#
# 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.
+ # resources are available for them to run
#
- def schedule_jobs(self, jobs, priority_jobs):
- self.waiting_jobs.extend(jobs)
- self.waiting_jobs.extendleft(priority_jobs)
+ def schedule_jobs(self, jobs):
+ for job in jobs:
+ self.waiting_jobs.append(job)
# job_completed():
#
@@ -300,7 +298,6 @@ class Scheduler():
#
def _schedule_queue_jobs(self):
ready = []
- ready_priority = []
process_queues = True
while self._queue_jobs and process_queues:
@@ -325,19 +322,16 @@ class Scheduler():
# to fetch tasks for elements which failed to pull, and
# thus need all the pulls to complete before ever starting
# a build
- for queue in reversed(self.queues):
- ready_jobs = queue.pop_ready_jobs()
- if queue.high_priority:
- ready_priority.extend(ready_jobs)
- else:
- ready.extend(ready_jobs)
+ ready.extend(chain.from_iterable(
+ queue.pop_ready_jobs() for queue in reversed(self.queues)
+ ))
# pop_ready_jobs() may have skipped jobs, adding them to
# the done_queue. Pull these skipped elements forward to
# the next queue and process them.
process_queues = any(q.dequeue_ready() for q in self.queues)
- self.schedule_jobs(ready, ready_priority)
+ self.schedule_jobs(ready)
self._sched()
# _run_cleanup()