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 07:50:49 UTC
[buildstream] 03/04: buildstream/_scheduler/*: Add QueueRunner
This is an automated email from the ASF dual-hosted git repository.
tvb pushed a commit to branch 372-allow-queues-to-run-auxilliary-jobs-after-an-element-s-job-finishes
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 8e59ca0b205b5f84e1e9a88cd169bd4168a53ded
Author: Tristan Maat <tr...@codethink.co.uk>
AuthorDate: Fri Apr 20 15:35:25 2018 +0100
buildstream/_scheduler/*: Add QueueRunner
---
buildstream/_scheduler/queuerunner.py | 104 ++++++++++++++++++++++++++++++++++
buildstream/_scheduler/scheduler.py | 35 ++----------
2 files changed, 109 insertions(+), 30 deletions(-)
diff --git a/buildstream/_scheduler/queuerunner.py b/buildstream/_scheduler/queuerunner.py
new file mode 100644
index 0000000..aac2e40
--- /dev/null
+++ b/buildstream/_scheduler/queuerunner.py
@@ -0,0 +1,104 @@
+# Copyright (C) 2018 Codethink Limited
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+# Author:
+# Tristan Daniël Maat <tr...@codethink.co.uk>
+#
+from itertools import chain
+
+
+# QueueRunner()
+#
+# The queue runner manages queues and determines what jobs should be
+# run at any given point.
+#
+# `QueueRunner.schedule_jobs` will pull elements through queues as
+# appropriate, and return jobs for any "ready" elements to submit for
+# processing.
+#
+# Args:
+# scheduler (Scheduler) - The scheduler to provide jobs for.
+# queues ([Queue]) - The queues to manage.
+#
+class QueueRunner():
+ def __init__(self, scheduler, queues=None):
+ try:
+ queues = list(queues)
+ except TypeError:
+ queues = [queues] if queues is not None else []
+
+ self.queues = queues
+ self._scheduler = scheduler
+
+ # append()
+ #
+ # Append a queue to the list of queues.
+ #
+ # Args:
+ # queue (Queue) - The queue to append
+ #
+ def append(self, queue):
+ self.queues.append(queue)
+
+ # extend()
+ #
+ # Extend the list of queues.
+ #
+ # Args:
+ # queues (typing.Iterable[Queue]) - The queues to append
+ #
+ def extend(self, queues):
+ self.queues.extend(queues)
+
+ # schedule_jobs()
+ #
+ # Pull elements through the managed queues, and collect jobs to be
+ # executed by the scheduler
+ #
+ # Returns:
+ # (typing.List[Job]) jobs for the scheduler to execute.
+ #
+ def schedule_jobs(self):
+
+ ready = []
+ process_queues = True
+
+ while self._scheduler._queue_jobs and process_queues:
+
+ # Pull elements forward through queues
+ elements = []
+ for queue in self.queues:
+ # Enqueue elements complete from the last queue
+ queue.enqueue(elements)
+
+ # Dequeue processed elements for the next queue
+ elements = list(queue.dequeue())
+
+ # Kickoff whatever processes can be processed at this time
+ #
+ # We start by queuing from the last queue first, because we want to
+ # give priority to queues later in the scheduling process in the case
+ # that multiple queues share the same token type.
+ #
+ # This avoids starvation situations where we dont move on 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.process_ready() for queue in reversed(self.queues)))
+
+ # process_ready() 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)
+
+ return ready
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index a475722..61cfc11 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -28,6 +28,7 @@ from contextlib import contextmanager
# Local imports
from .queues import QueueType
+from .queuerunner import QueueRunner
# A decent return code for Scheduler.run()
@@ -82,6 +83,7 @@ class Scheduler():
#
# Private members
#
+ self._runners = []
self._interrupt_callback = interrupt_callback
self._ticker_callback = ticker_callback
self._job_start_callback = job_start_callback
@@ -115,6 +117,7 @@ class Scheduler():
def run(self, queues):
# Hold on to the queues to process
+ self._runners.append(QueueRunner(self, queues))
self.queues = queues
# Ensure that we have a fresh new event loop, in case we want
@@ -221,36 +224,8 @@ class Scheduler():
# and process anything that is ready.
#
def sched(self):
-
- process_queues = True
-
- while self._queue_jobs and process_queues:
-
- # Pull elements forward through queues
- elements = []
- for queue in self.queues:
- # Enqueue elements complete from the last queue
- queue.enqueue(elements)
-
- # Dequeue processed elements for the next queue
- elements = list(queue.dequeue())
- elements = list(elements)
-
- # Kickoff whatever processes can be processed at this time
- #
- # We start by queuing from the last queue first, because we want to
- # give priority to queues later in the scheduling process in the case
- # that multiple queues share the same token type.
- #
- # This avoids starvation situations where we dont move on 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):
- queue.process_ready()
-
- # process_ready() 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)
+ for runner in self._runners:
+ runner.schedule_jobs()
# If nothings ticking, time to bail out
ticking = 0