|
| 1 | +from multiprocessing.managers import ListProxy, ValueProxy |
| 2 | +import sys |
| 3 | +from multiprocessing import Pool, cpu_count, Manager |
| 4 | +import time |
| 5 | +from typing import Callable, List, Any, Union |
| 6 | +from threading import Lock, Thread, Event |
| 7 | +import shutil |
| 8 | + |
| 9 | +from .runner_arguments import RunnerArguments, AdditionalSwiftSourcesArguments |
| 10 | + |
| 11 | + |
| 12 | +class MonitoredFunction: |
| 13 | + def __init__( |
| 14 | + self, |
| 15 | + fn: Callable, |
| 16 | + running_tasks: ListProxy, |
| 17 | + updated_repos: ValueProxy, |
| 18 | + lock: Lock |
| 19 | + ): |
| 20 | + self.fn = fn |
| 21 | + self.running_tasks = running_tasks |
| 22 | + self.updated_repos = updated_repos |
| 23 | + self._lock = lock |
| 24 | + |
| 25 | + def __call__(self, *args: Union[RunnerArguments, AdditionalSwiftSourcesArguments]): |
| 26 | + task_name = args[0].repo_name |
| 27 | + self.running_tasks.append(task_name) |
| 28 | + result = None |
| 29 | + try: |
| 30 | + result = self.fn(*args) |
| 31 | + except Exception as e: |
| 32 | + print(e) |
| 33 | + finally: |
| 34 | + self._lock.acquire() |
| 35 | + if task_name in self.running_tasks: |
| 36 | + self.running_tasks.remove(task_name) |
| 37 | + self.updated_repos.set(self.updated_repos.get() + 1) |
| 38 | + self._lock.release() |
| 39 | + return result |
| 40 | + |
| 41 | + |
| 42 | +class ParallelRunner: |
| 43 | + def __init__( |
| 44 | + self, |
| 45 | + fn: Callable, |
| 46 | + pool_args: List[Union[RunnerArguments, AdditionalSwiftSourcesArguments]], |
| 47 | + n_processes: int = 0, |
| 48 | + ): |
| 49 | + self._monitor_polling_period = 0.1 |
| 50 | + if n_processes == 0: |
| 51 | + n_processes = cpu_count() * 2 |
| 52 | + self._terminal_width = shutil.get_terminal_size().columns |
| 53 | + self._n_processes = n_processes |
| 54 | + self._pool_args = pool_args |
| 55 | + manager = Manager() |
| 56 | + self._lock = manager.Lock() |
| 57 | + self._running_tasks = manager.list() |
| 58 | + self._updated_repos = manager.Value("i", 0) |
| 59 | + self._fn = fn |
| 60 | + self._pool = Pool(processes=self._n_processes) |
| 61 | + self._verbose = pool_args[0].verbose |
| 62 | + self._output_prefix = pool_args[0].output_prefix |
| 63 | + self._nb_repos = len(pool_args) |
| 64 | + self._stop_event = Event() |
| 65 | + self._monitored_fn = MonitoredFunction( |
| 66 | + self._fn, self._running_tasks, self._updated_repos, self._lock |
| 67 | + ) |
| 68 | + |
| 69 | + def run(self) -> List[Any]: |
| 70 | + print( |
| 71 | + "Running ``%s`` with up to %d processes." |
| 72 | + % (self._fn.__name__, self._n_processes) |
| 73 | + ) |
| 74 | + if self._verbose: |
| 75 | + results = self._pool.map_async( |
| 76 | + func=self._fn, iterable=self._pool_args |
| 77 | + ).get(timeout=1800) |
| 78 | + self._pool.close() |
| 79 | + self._pool.join() |
| 80 | + else: |
| 81 | + monitor_thread = Thread(target=self._monitor, daemon=True) |
| 82 | + monitor_thread.start() |
| 83 | + results = self._pool.map_async( |
| 84 | + func=self._monitored_fn, iterable=self._pool_args |
| 85 | + ).get(timeout=1800) |
| 86 | + self._pool.close() |
| 87 | + self._pool.join() |
| 88 | + self._stop_event.set() |
| 89 | + monitor_thread.join() |
| 90 | + return results |
| 91 | + |
| 92 | + def _monitor(self): |
| 93 | + last_output = "" |
| 94 | + while not self._stop_event.is_set(): |
| 95 | + self._lock.acquire() |
| 96 | + current = list(self._running_tasks) |
| 97 | + current_line = ", ".join(current) |
| 98 | + updated_repos = self._updated_repos.get() |
| 99 | + self._lock.release() |
| 100 | + |
| 101 | + if current_line != last_output: |
| 102 | + truncated = (f"{self._output_prefix} [{updated_repos}/{self._nb_repos}] ({current_line})") |
| 103 | + if len(truncated) > self._terminal_width: |
| 104 | + ellipsis_marker = " ..." |
| 105 | + truncated = ( |
| 106 | + truncated[: self._terminal_width - len(ellipsis_marker)] |
| 107 | + + ellipsis_marker |
| 108 | + ) |
| 109 | + sys.stdout.write("\r" + truncated.ljust(self._terminal_width)) |
| 110 | + sys.stdout.flush() |
| 111 | + last_output = current_line |
| 112 | + |
| 113 | + time.sleep(self._monitor_polling_period) |
| 114 | + |
| 115 | + sys.stdout.write("\r" + " " * len(last_output) + "\r\n") |
| 116 | + sys.stdout.flush() |
| 117 | + |
| 118 | + @staticmethod |
| 119 | + def check_results(results, op) -> int: |
| 120 | + """Function used to check the results of ParallelRunner. |
| 121 | +
|
| 122 | + NOTE: This function was originally located in the shell module of |
| 123 | + swift_build_support and should eventually be replaced with a better |
| 124 | + parallel implementation. |
| 125 | + """ |
| 126 | + |
| 127 | + fail_count = 0 |
| 128 | + if results is None: |
| 129 | + return 0 |
| 130 | + for r in results: |
| 131 | + if r is not None: |
| 132 | + if fail_count == 0: |
| 133 | + print("======%s FAILURES======" % op) |
| 134 | + fail_count += 1 |
| 135 | + if isinstance(r, str): |
| 136 | + print(r) |
| 137 | + continue |
| 138 | + print("%s failed (ret=%d): %s" % (r.repo_path, r.ret, r)) |
| 139 | + if r.stderr: |
| 140 | + print(r.stderr.decode()) |
| 141 | + return fail_count |
| 142 | + |
0 commit comments