| # Copyright lowRISC contributors. | |
| # Licensed under the Apache License, Version 2.0, see LICENSE for details. | |
| # SPDX-License-Identifier: Apache-2.0 | |
| import logging as log | |
| import threading | |
| from signal import SIGINT, SIGTERM, signal | |
| from Launcher import LauncherError | |
| from StatusPrinter import get_status_printer | |
| from Timer import Timer | |
| from utils import VERBOSE | |
| # Sum of lenghts of all lists in the given dict. | |
| def sum_dict_lists(d): | |
| '''Given a dict whose key values are lists, return sum of lengths of | |
| thoese lists.''' | |
| return sum([len(d[k]) for k in d]) | |
| def get_next_item(arr, index): | |
| '''Perpetually get an item from a list. | |
| Returns the next item on the list by advancing the index by 1. If the index | |
| is already the last item on the list, it loops back to the start, thus | |
| implementing a circular list. | |
| arr is a subscriptable list. | |
| index is the index of the last item returned. | |
| Returns (item, index) if successful. | |
| Raises IndexError if arr is empty. | |
| ''' | |
| index += 1 | |
| try: | |
| item = arr[index] | |
| except IndexError: | |
| index = 0 | |
| try: | |
| item = arr[index] | |
| except IndexError: | |
| raise IndexError("List is empty!") | |
| return item, index | |
| class Scheduler: | |
| '''An object that runs one or more Deploy items''' | |
| def __init__(self, items, launcher_cls, interactive): | |
| self.items = items | |
| # 'scheduled[target][cfg]' is a list of Deploy objects for the chosen | |
| # target and cfg. As items in _scheduled are ready to be run (once | |
| # their dependencies pass), they are moved to the _queued list, where | |
| # they wait until slots are available for them to be dispatched. | |
| # When all items (in all cfgs) of a target are done, it is removed from | |
| # this dictionary. | |
| self._scheduled = {} | |
| self.add_to_scheduled(items) | |
| # Print status periodically using an external status printer. | |
| self.status_printer = get_status_printer(interactive) | |
| self.status_printer.print_header( | |
| msg="Q: queued, D: dispatched, P: passed, F: failed, K: killed, " | |
| "T: total") | |
| # Sets of items, split up by their current state. The sets are | |
| # disjoint and their union equals the keys of self.item_to_status. | |
| # _queued is a list so that we dispatch things in order (relevant | |
| # for things like tests where we have ordered things cleverly to | |
| # try to see failures early). They are maintained for each target. | |
| # The list of available targets and the list of running items in each | |
| # target are polled in a circular fashion, looping back to the start. | |
| # This is done to allow us to poll a smaller subset of jobs rather than | |
| # the entire regression. We keep rotating through our list of running | |
| # items, picking up where we left off on the last poll. | |
| self._targets = list(self._scheduled.keys()) | |
| self._queued = {} | |
| self._running = {} | |
| self._passed = {} | |
| self._failed = {} | |
| self._killed = {} | |
| self._total = {} | |
| self.last_target_polled_idx = -1 | |
| self.last_item_polled_idx = {} | |
| for target in self._scheduled: | |
| self._queued[target] = [] | |
| self._running[target] = [] | |
| self._passed[target] = set() | |
| self._failed[target] = set() | |
| self._killed[target] = set() | |
| self._total[target] = sum_dict_lists(self._scheduled[target]) | |
| self.last_item_polled_idx[target] = -1 | |
| # Stuff for printing the status. | |
| width = len(str(self._total[target])) | |
| field_fmt = '{{:0{}d}}'.format(width) | |
| self.msg_fmt = 'Q: {0}, D: {0}, P: {0}, F: {0}, K: {0}, T: {0}'.format( | |
| field_fmt) | |
| msg = self.msg_fmt.format(0, 0, 0, 0, 0, self._total[target]) | |
| self.status_printer.init_target(target=target, msg=msg) | |
| # A map from the Deploy objects tracked by this class to their | |
| # current status. This status is 'Q', 'D', 'P', 'F' or 'K', | |
| # corresponding to membership in the dicts above. This is not | |
| # per-target. | |
| self.item_to_status = {} | |
| # Create the launcher instance for all items. | |
| for item in self.items: | |
| item.create_launcher() | |
| # The chosen launcher class. This allows us to access launcher | |
| # variant-specific settings such as max parallel jobs & poll rate. | |
| self.launcher_cls = launcher_cls | |
| def run(self): | |
| '''Run all scheduled jobs and return the results. | |
| Returns the results (status) of all items dispatched for all | |
| targets and cfgs. | |
| ''' | |
| timer = Timer() | |
| # Catch one SIGINT and tell the runner to quit. On a second, die. | |
| stop_now = threading.Event() | |
| old_handler = None | |
| def on_signal(signal_received, frame): | |
| log.info("Received signal %s. Exiting gracefully.", | |
| signal_received) | |
| if signal_received == SIGINT: | |
| log.info('Send another to force immediate quit (but you may ' | |
| 'need to manually kill child processes)') | |
| # Restore old handler to catch a second SIGINT | |
| assert old_handler is not None | |
| signal(signal_received, old_handler) | |
| stop_now.set() | |
| old_handler = signal(SIGINT, on_signal) | |
| # Install the SIGTERM handler before scheduling jobs. | |
| signal(SIGTERM, on_signal) | |
| # Enqueue all items of the first target. | |
| self._enqueue_successors(None) | |
| try: | |
| while True: | |
| if stop_now.is_set(): | |
| # We've had an interrupt. Kill any jobs that are running. | |
| self._kill() | |
| hms = timer.hms() | |
| changed = self._poll(hms) or timer.check_time() | |
| self._dispatch(hms) | |
| if changed: | |
| if self._check_if_done(hms): | |
| break | |
| # This is essentially sleep(1) to wait a second between each | |
| # polling loop. But we do it with a bounded wait on stop_now so | |
| # that we jump back to the polling loop immediately on a | |
| # signal. | |
| stop_now.wait(timeout=self.launcher_cls.poll_freq) | |
| finally: | |
| signal(SIGINT, old_handler) | |
| # Cleanup the status printer. | |
| self.status_printer.exit() | |
| # We got to the end without anything exploding. Return the results. | |
| return self.item_to_status | |
| def add_to_scheduled(self, items): | |
| '''Add items to the list of _scheduled. | |
| 'items' is a list of Deploy objects. | |
| ''' | |
| for item in items: | |
| target_dict = self._scheduled.setdefault(item.target, {}) | |
| cfg_list = target_dict.setdefault(item.sim_cfg, []) | |
| if item not in cfg_list: | |
| cfg_list.append(item) | |
| def _remove_from_scheduled(self, item): | |
| '''Removes the item from _scheduled[target][cfg] list. | |
| When all items in _scheduled[target][cfg] are finally removed, the cfg | |
| key is deleted. | |
| ''' | |
| target_dict = self._scheduled[item.target] | |
| cfg_list = target_dict.get(item.sim_cfg) | |
| if cfg_list is not None: | |
| try: | |
| cfg_list.remove(item) | |
| except ValueError: | |
| pass | |
| if not cfg_list: | |
| del target_dict[item.sim_cfg] | |
| def _get_next_target(self, curr_target): | |
| '''Returns the target that succeeds the current one. | |
| curr_target is the target of the job that just completed (example - | |
| build). If it is None, then the first target in _scheduled is returned. | |
| ''' | |
| if curr_target is None: | |
| return next(iter(self._scheduled)) | |
| assert curr_target in self._scheduled | |
| target_iterator = iter(self._scheduled) | |
| target = next(target_iterator) | |
| found = False | |
| while not found: | |
| if target == curr_target: | |
| found = True | |
| try: | |
| target = next(target_iterator) | |
| except StopIteration: | |
| return None | |
| return target | |
| def _enqueue_successors(self, item=None): | |
| '''Move an item's successors from _scheduled to _queued. | |
| 'item' is the recently run job that has completed. If None, then we | |
| move all available items in all available cfgs in _scheduled's first | |
| target. If 'item' is specified, then we find its successors and move | |
| them to _queued. | |
| ''' | |
| for next_item in self._get_successors(item): | |
| assert next_item not in self.item_to_status | |
| assert next_item not in self._queued[next_item.target] | |
| self.item_to_status[next_item] = 'Q' | |
| self._queued[next_item.target].append(next_item) | |
| self._remove_from_scheduled(next_item) | |
| def _cancel_successors(self, item): | |
| '''Cancel an item's successors recursively by moving them from | |
| _scheduled or _queued to _killed.''' | |
| items = self._get_successors(item) | |
| while items: | |
| next_item = items.pop() | |
| self._cancel_item(next_item, cancel_successors=False) | |
| items.extend(self._get_successors(next_item)) | |
| def _get_successors(self, item=None): | |
| '''Find immediate successors of an item. | |
| 'item' is a job that has completed. We choose the target that follows | |
| the 'item''s current target and find the list of successors whose | |
| dependency list contains 'item'. If 'item' is None, we pick successors | |
| from all cfgs, else we pick successors only from the cfg to which the | |
| item belongs. | |
| Returns the list of item's successors, or an empty list if there are | |
| none. | |
| ''' | |
| if item is None: | |
| target = self._get_next_target(None) | |
| cfgs = set(self._scheduled[target]) | |
| else: | |
| target = self._get_next_target(item.target) | |
| cfgs = {item.sim_cfg} | |
| if target is None: | |
| return [] | |
| # Find item's successors that can be enqueued. We assume here that | |
| # only the immediately succeeding target can be enqueued at this | |
| # time. | |
| successors = [] | |
| for cfg in cfgs: | |
| for next_item in self._scheduled[target][cfg]: | |
| if item is not None: | |
| # Something is terribly wrong if item exists but the | |
| # next_item's dependency list is empty. | |
| assert next_item.dependencies | |
| if item not in next_item.dependencies: | |
| continue | |
| if self._ok_to_enqueue(next_item): | |
| successors.append(next_item) | |
| return successors | |
| def _ok_to_enqueue(self, item): | |
| '''Returns true if ALL dependencies of item are complete.''' | |
| for dep in item.dependencies: | |
| # Ignore dependencies that were not scheduled to run. | |
| if dep not in self.items: | |
| continue | |
| # Has the dep even been enqueued? | |
| if dep not in self.item_to_status: | |
| return False | |
| # Has the dep completed? | |
| if self.item_to_status[dep] not in ["P", "F", "K"]: | |
| return False | |
| return True | |
| def _ok_to_run(self, item): | |
| '''Returns true if the required dependencies have passed. | |
| The item's needs_all_dependencies_passing setting is used to figure | |
| out whether we can run this item or not, based on its dependent jobs' | |
| statuses. | |
| ''' | |
| # 'item' can run only if its dependencies have passed (their results | |
| # should already show up in the item to status map). | |
| for dep in item.dependencies: | |
| # Ignore dependencies that were not scheduled to run. | |
| if dep not in self.items: | |
| continue | |
| dep_status = self.item_to_status[dep] | |
| assert dep_status in ['P', 'F', 'K'] | |
| if item.needs_all_dependencies_passing: | |
| if dep_status in ['F', 'K']: | |
| return False | |
| else: | |
| if dep_status in ['P']: | |
| return True | |
| return item.needs_all_dependencies_passing | |
| def _poll(self, hms): | |
| '''Check for running items that have finished | |
| Returns True if something changed. | |
| ''' | |
| max_poll = min(self.launcher_cls.max_poll, | |
| sum_dict_lists(self._running)) | |
| # If there are no jobs running, we are likely done (possibly because | |
| # of a SIGINT). Since poll() was called anyway, signal that something | |
| # has indeed changed. | |
| if not max_poll: | |
| return True | |
| changed = False | |
| while max_poll: | |
| target, self.last_target_polled_idx = get_next_item( | |
| self._targets, self.last_target_polled_idx) | |
| while self._running[target] and max_poll: | |
| max_poll -= 1 | |
| item, self.last_item_polled_idx[target] = get_next_item( | |
| self._running[target], self.last_item_polled_idx[target]) | |
| status = item.launcher.poll() | |
| level = VERBOSE | |
| assert status in ['D', 'P', 'F', 'K'] | |
| if status == 'D': | |
| continue | |
| elif status == 'P': | |
| self._passed[target].add(item) | |
| elif status == 'F': | |
| self._failed[target].add(item) | |
| level = log.ERROR | |
| else: | |
| self._killed[target].add(item) | |
| level = log.ERROR | |
| self._running[target].pop(self.last_item_polled_idx[target]) | |
| self.last_item_polled_idx[target] -= 1 | |
| self.item_to_status[item] = status | |
| log.log(level, "[%s]: [%s]: [status] [%s: %s]", hms, target, | |
| item.full_name, status) | |
| # Enqueue item's successors regardless of its status. | |
| # | |
| # It may be possible that a failed item's successor may not | |
| # need all of its dependents to pass (if it has other dependent | |
| # jobs). Hence we enqueue all successors rather than canceling | |
| # them right here. We leave it to _dispatch() to figure out | |
| # whether an enqueued item can be run or not. | |
| self._enqueue_successors(item) | |
| changed = True | |
| return changed | |
| def _dispatch(self, hms): | |
| '''Dispatch some queued items if possible.''' | |
| slots = self.launcher_cls.max_parallel - sum_dict_lists(self._running) | |
| if slots <= 0: | |
| return | |
| # Compute how many slots to allocate to each target based on their | |
| # weights. | |
| sum_weight = 0 | |
| slots_filled = 0 | |
| total_weight = sum(self._queued[t][0].weight for t in self._queued | |
| if self._queued[t]) | |
| for target in self._scheduled: | |
| if not self._queued[target]: | |
| continue | |
| # N slots are allocated to M targets each with W(m) weights with | |
| # the formula: | |
| # | |
| # N(m) = N * W(m) / T, where, | |
| # T is the sum total of all weights. | |
| # | |
| # This is however, problematic due to fractions. Even after | |
| # rounding off to the nearest digit, slots may not be fully | |
| # utilized (one extra left). An alternate approach that avoids this | |
| # problem is as follows: | |
| # | |
| # N(m) = (N * S(W(m)) / T) - F(m), where, | |
| # S(W(m)) is the running sum of weights upto current target m. | |
| # F(m) is the running total of slots filled. | |
| # | |
| # The computed slots per target is nearly identical to the first | |
| # solution, except that it prioritizes the slot allocation to | |
| # targets that are earlier in the list such that in the end, all | |
| # slots are fully consumed. | |
| sum_weight += self._queued[target][0].weight | |
| target_slots = round( | |
| (slots * sum_weight) / total_weight) - slots_filled | |
| if target_slots <= 0: | |
| continue | |
| slots_filled += target_slots | |
| to_dispatch = [] | |
| while self._queued[target] and target_slots > 0: | |
| next_item = self._queued[target].pop(0) | |
| if not self._ok_to_run(next_item): | |
| self._cancel_item(next_item, cancel_successors=False) | |
| self._enqueue_successors(next_item) | |
| continue | |
| to_dispatch.append(next_item) | |
| target_slots -= 1 | |
| if not to_dispatch: | |
| continue | |
| log.log(VERBOSE, "[%s]: [%s]: [dispatch]:\n%s", hms, target, | |
| ", ".join(item.full_name for item in to_dispatch)) | |
| for item in to_dispatch: | |
| self._running[target].append(item) | |
| self.item_to_status[item] = 'D' | |
| try: | |
| item.launcher.launch() | |
| except LauncherError as err: | |
| log.error('{}'.format(err)) | |
| self._kill_item(item) | |
| def _kill(self): | |
| '''Kill any running items and cancel any that are waiting''' | |
| # Cancel any waiting items. We take a copy of self._queued to avoid | |
| # iterating over the set as we modify it. | |
| for target in self._queued: | |
| for item in [item for item in self._queued[target]]: | |
| self._cancel_item(item) | |
| # Kill any running items. Again, take a copy of the set to avoid | |
| # modifying it while iterating over it. | |
| for target in self._running: | |
| for item in [item for item in self._running[target]]: | |
| self._kill_item(item) | |
| def _check_if_done(self, hms): | |
| '''Check if we are done executing all jobs. | |
| Also, prints the status of currently running jobs. | |
| ''' | |
| done = True | |
| for target in self._scheduled: | |
| done_cnt = sum([ | |
| len(self._passed[target]), | |
| len(self._failed[target]), | |
| len(self._killed[target]) | |
| ]) | |
| done = done and (done_cnt == self._total[target]) | |
| # Skip if a target has not even begun executing. | |
| if not (self._queued[target] or self._running[target] or | |
| done_cnt > 0): | |
| continue | |
| perc = done_cnt / self._total[target] * 100 | |
| running = ", ".join( | |
| [f"{item.full_name}" for item in self._running[target]]) | |
| msg = self.msg_fmt.format(len(self._queued[target]), | |
| len(self._running[target]), | |
| len(self._passed[target]), | |
| len(self._failed[target]), | |
| len(self._killed[target]), | |
| self._total[target]) | |
| self.status_printer.update_target(target=target, | |
| msg=msg, | |
| hms=hms, | |
| perc=perc, | |
| running=running) | |
| return done | |
| def _cancel_item(self, item, cancel_successors=True): | |
| '''Cancel an item and optionally all of its successors. | |
| Supplied item may be in _scheduled list or the _queued list. From | |
| either, we move it straight to _killed. | |
| ''' | |
| self.item_to_status[item] = 'K' | |
| self._killed[item.target].add(item) | |
| if item in self._queued[item.target]: | |
| self._queued[item.target].remove(item) | |
| else: | |
| self._remove_from_scheduled(item) | |
| if cancel_successors: | |
| self._cancel_successors(item) | |
| def _kill_item(self, item): | |
| '''Kill a running item and cancel all of its successors.''' | |
| item.launcher.kill() | |
| self.item_to_status[item] = 'K' | |
| self._killed[item.target].add(item) | |
| self._running[item.target].remove(item) | |
| self._cancel_successors(item) |