From 36fea7f15ed17150691b5b3ead75450e575229ef Mon Sep 17 00:00:00 2001 From: Scott Date: Wed, 26 Jan 2022 21:34:26 -0800 Subject: Ran black code formatter on everything. --- executors.py | 67 ++++++++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 49 insertions(+), 18 deletions(-) (limited to 'executors.py') diff --git a/executors.py b/executors.py index 3786954..46812c2 100644 --- a/executors.py +++ b/executors.py @@ -160,7 +160,9 @@ class ProcessExecutor(BaseExecutor): self.adjust_task_count(+1) pickle = make_cloud_pickle(function, *args, **kwargs) result = self._process_executor.submit(self.run_cloud_pickle, pickle) - result.add_done_callback(lambda _: self.histogram.add_item(time.time() - start)) + result.add_done_callback( + lambda _: self.histogram.add_item(time.time() - start) + ) return result @overrides @@ -256,7 +258,9 @@ class RemoteExecutorStatus: self.finished_bundle_timings_per_worker: Dict[ RemoteWorkerRecord, List[float] ] = {} - self.in_flight_bundles_by_worker: Dict[RemoteWorkerRecord, Set[str]] = {} + self.in_flight_bundles_by_worker: Dict[ + RemoteWorkerRecord, Set[str] + ] = {} self.bundle_details_by_uuid: Dict[str, BundleDetails] = {} self.finished_bundle_timings: List[float] = [] self.last_periodic_dump: Optional[float] = None @@ -266,7 +270,9 @@ class RemoteExecutorStatus: # as a memory fence for modifications to bundle. self.lock: threading.Lock = threading.Lock() - def record_acquire_worker(self, worker: RemoteWorkerRecord, uuid: str) -> None: + def record_acquire_worker( + self, worker: RemoteWorkerRecord, uuid: str + ) -> None: with self.lock: self.record_acquire_worker_already_locked(worker, uuid) @@ -284,7 +290,9 @@ class RemoteExecutorStatus: with self.lock: self.record_bundle_details_already_locked(details) - def record_bundle_details_already_locked(self, details: BundleDetails) -> None: + def record_bundle_details_already_locked( + self, details: BundleDetails + ) -> None: assert self.lock.locked() self.bundle_details_by_uuid[details.uuid] = details @@ -295,7 +303,9 @@ class RemoteExecutorStatus: was_cancelled: bool, ) -> None: with self.lock: - self.record_release_worker_already_locked(worker, uuid, was_cancelled) + self.record_release_worker_already_locked( + worker, uuid, was_cancelled + ) def record_release_worker_already_locked( self, @@ -367,7 +377,11 @@ class RemoteExecutorStatus: ret += f' ...{in_flight} bundles currently in flight:\n' for bundle_uuid in self.in_flight_bundles_by_worker[worker]: details = self.bundle_details_by_uuid.get(bundle_uuid, None) - pid = str(details.pid) if (details and details.pid != 0) else "TBD" + pid = ( + str(details.pid) + if (details and details.pid != 0) + else "TBD" + ) if self.start_per_bundle[bundle_uuid] is not None: sec = ts - self.start_per_bundle[bundle_uuid] ret += f' (pid={pid}): {details} for {sec:.1f}s so far ' @@ -398,7 +412,10 @@ class RemoteExecutorStatus: assert self.lock.locked() self.total_bundles_submitted = total_bundles_submitted ts = time.time() - if self.last_periodic_dump is None or ts - self.last_periodic_dump > 5.0: + if ( + self.last_periodic_dump is None + or ts - self.last_periodic_dump > 5.0 + ): print(self) self.last_periodic_dump = ts @@ -412,7 +429,9 @@ class RemoteWorkerSelectionPolicy(ABC): pass @abstractmethod - def acquire_worker(self, machine_to_avoid=None) -> Optional[RemoteWorkerRecord]: + def acquire_worker( + self, machine_to_avoid=None + ) -> Optional[RemoteWorkerRecord]: pass @@ -425,7 +444,9 @@ class WeightedRandomRemoteWorkerSelectionPolicy(RemoteWorkerSelectionPolicy): return False @overrides - def acquire_worker(self, machine_to_avoid=None) -> Optional[RemoteWorkerRecord]: + def acquire_worker( + self, machine_to_avoid=None + ) -> Optional[RemoteWorkerRecord]: grabbag = [] for worker in self.workers: for x in range(0, worker.count): @@ -482,7 +503,9 @@ class RoundRobinRemoteWorkerSelectionPolicy(RemoteWorkerSelectionPolicy): class RemoteExecutor(BaseExecutor): def __init__( - self, workers: List[RemoteWorkerRecord], policy: RemoteWorkerSelectionPolicy + self, + workers: List[RemoteWorkerRecord], + policy: RemoteWorkerSelectionPolicy, ) -> None: super().__init__() self.workers = workers @@ -562,7 +585,9 @@ class RemoteExecutor(BaseExecutor): break for uuid in bundle_uuids: - bundle = self.status.bundle_details_by_uuid.get(uuid, None) + bundle = self.status.bundle_details_by_uuid.get( + uuid, None + ) if ( bundle is not None and bundle.src_bundle is None @@ -653,7 +678,9 @@ class RemoteExecutor(BaseExecutor): logger.critical(msg) raise Exception(msg) - def release_worker(self, bundle: BundleDetails, *, was_cancelled=True) -> None: + def release_worker( + self, bundle: BundleDetails, *, was_cancelled=True + ) -> None: worker = bundle.worker assert worker is not None logger.debug(f'Released worker {worker}') @@ -737,14 +764,14 @@ class RemoteExecutor(BaseExecutor): # Send input code / data to worker machine if it's not local. if hostname not in machine: try: - cmd = ( - f'{SCP} {bundle.code_file} {username}@{machine}:{bundle.code_file}' - ) + cmd = f'{SCP} {bundle.code_file} {username}@{machine}:{bundle.code_file}' start_ts = time.time() logger.info(f"{bundle}: Copying work to {worker} via {cmd}.") run_silently(cmd) xfer_latency = time.time() - start_ts - logger.debug(f"{bundle}: Copying to {worker} took {xfer_latency:.1f}s.") + logger.debug( + f"{bundle}: Copying to {worker} took {xfer_latency:.1f}s." + ) except Exception as e: self.release_worker(bundle) if is_original: @@ -777,7 +804,9 @@ class RemoteExecutor(BaseExecutor): f' /home/scott/lib/python_modules/remote_worker.py' f' --code_file {bundle.code_file} --result_file {bundle.result_file}"' ) - logger.debug(f'{bundle}: Executing {cmd} in the background to kick off work...') + logger.debug( + f'{bundle}: Executing {cmd} in the background to kick off work...' + ) p = cmd_in_background(cmd, silent=True) bundle.pid = p.pid logger.debug( @@ -906,7 +935,9 @@ class RemoteExecutor(BaseExecutor): # Re-raise the exception; the code in wait_for_process may # decide to emergency_retry_nasty_bundle here. raise Exception(e) - logger.debug(f'Removing local (master) {code_file} and {result_file}.') + logger.debug( + f'Removing local (master) {code_file} and {result_file}.' + ) os.remove(f'{result_file}') os.remove(f'{code_file}') -- cgit v1.3