diff options
Diffstat (limited to 'executors.py')
| -rw-r--r-- | executors.py | 21 |
1 files changed, 14 insertions, 7 deletions
diff --git a/executors.py b/executors.py index e95ed71..990df03 100644 --- a/executors.py +++ b/executors.py @@ -248,7 +248,7 @@ class BundleDetails: end_ts: float slower_than_local_p95: bool slower_than_global_p95: bool - src_bundle: BundleDetails + src_bundle: Optional[BundleDetails] is_cancelled: threading.Event was_cancelled: bool backup_bundles: Optional[List[BundleDetails]] @@ -288,7 +288,7 @@ class RemoteExecutorStatus: self.worker_count: int = total_worker_count self.known_workers: Set[RemoteWorkerRecord] = set() self.start_time: float = time.time() - self.start_per_bundle: Dict[str, float] = defaultdict(float) + self.start_per_bundle: Dict[str, Optional[float]] = defaultdict(float) self.end_per_bundle: Dict[str, float] = defaultdict(float) self.finished_bundle_timings_per_worker: Dict[ RemoteWorkerRecord, List[float] @@ -345,7 +345,9 @@ class RemoteExecutorStatus: self.end_per_bundle[uuid] = ts self.in_flight_bundles_by_worker[worker].remove(uuid) if not was_cancelled: - bundle_latency = ts - self.start_per_bundle[uuid] + start = self.start_per_bundle[uuid] + assert start + bundle_latency = ts - start x = self.finished_bundle_timings_per_worker.get(worker, list()) x.append(bundle_latency) self.finished_bundle_timings_per_worker[worker] = x @@ -836,9 +838,10 @@ class RemoteExecutor(BaseExecutor): return self.wait_for_process(p, bundle, 0) def wait_for_process( - self, p: subprocess.Popen, bundle: BundleDetails, depth: int + self, p: Optional[subprocess.Popen], bundle: BundleDetails, depth: int ) -> Any: machine = bundle.machine + assert p pid = p.pid if depth > 3: logger.error( @@ -981,10 +984,12 @@ class RemoteExecutor(BaseExecutor): # Tell the original to stop if we finished first. if not was_cancelled: + orig_bundle = bundle.src_bundle + assert orig_bundle logger.debug( - f'{bundle}: Notifying original {bundle.src_bundle.uuid} we beat them to it.' + f'{bundle}: Notifying original {orig_bundle.uuid} we beat them to it.' ) - bundle.src_bundle.is_cancelled.set() + orig_bundle.is_cancelled.set() self.release_worker(bundle, was_cancelled=was_cancelled) return result @@ -1068,7 +1073,9 @@ class RemoteExecutor(BaseExecutor): # they will move the result_file to this machine and let # the original pick them up and unpickle them. - def emergency_retry_nasty_bundle(self, bundle: BundleDetails) -> fut.Future: + def emergency_retry_nasty_bundle( + self, bundle: BundleDetails + ) -> Optional[fut.Future]: is_original = bundle.src_bundle is None bundle.worker = None avoid_last_machine = bundle.machine |