backup_bundles: Optional[List[BundleDetails]]
failure_count: int
+ def __repr__(self):
+ uuid = self.uuid
+ if uuid[-9:-2] == '_backup':
+ uuid = uuid[:-9]
+ suffix = f'{uuid[-6:]}_b{self.uuid[-1:]}'
+ else:
+ suffix = uuid[-6:]
+
+ colorz = [
+ fg('violet red'),
+ fg('red'),
+ fg('orange'),
+ fg('peach orange'),
+ fg('yellow'),
+ fg('marigold yellow'),
+ fg('green yellow'),
+ fg('tea green'),
+ fg('cornflower blue'),
+ fg('turquoise blue'),
+ fg('tropical blue'),
+ fg('lavender purple'),
+ fg('medium purple'),
+ ]
+ c = colorz[int(uuid[-2:], 16) % len(colorz)]
+ fname = self.fname if self.fname is not None else 'nofname'
+ machine = self.machine if self.machine is not None else 'nomachine'
+ return f'{c}{suffix}/{fname}/{machine}{reset()}'
+
class RemoteExecutorStatus:
def __init__(self, total_worker_count: int) -> None:
bundle_uuid,
None
)
- pid = str(details.pid) if details is not None 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}): {bundle_uuid} for {sec:.1f}s so far '
+ ret += f' (pid={pid}): {details} for {sec:.1f}s so far '
else:
- ret += f' {bundle_uuid} setting up / copying data...'
+ ret += f' {details} setting up / copying data...'
sec = 0.0
if qworker is not None:
if sec > qall[1] * 1.5:
ret += f'{bg("red")}!!!{reset()}'
if details is not None:
- logger.debug(f'Flagging {details.uuid} for another backup')
+ logger.debug(f'Flagging {details} for another backup')
details.super_slow = True
elif sec > qall[1]:
ret += f'{bg("red")}>∀p95{reset()} '
if details is not None:
- logger.debug(f'Flagging {details.uuid} for a backup')
+ logger.debug(f'Flagging {details} for a backup')
details.too_slow = True
elif sec > qall[0]:
ret += f'{fg("red")}>∀p50{reset()}'
self.status = RemoteExecutorStatus(self.worker_count)
self.total_bundles_submitted = 0
- def bundle_prefix(self, bundle: BundleDetails) -> str:
- colorz = [
- fg('violet red'),
- fg('red'),
- fg('orange'),
- fg('peach orange'),
- fg('yellow'),
- fg('marigold yellow'),
- fg('green yellow'),
- fg('tea green'),
- fg('cornflower blue'),
- fg('turquoise blue'),
- fg('tropical blue'),
- fg('lavender purple'),
- fg('medium purple'),
- ]
- c = colorz[int(bundle.uuid[-2:], 16) % len(colorz)]
- fname = bundle.fname if bundle.fname is not None else 'nofname'
- machine = bundle.machine if bundle.machine is not None else 'nomachine'
- return f'{c}{bundle.uuid[-8:]}/{fname}/{machine}{reset()}'
-
def is_worker_available(self) -> bool:
return self.policy.is_worker_available()
bundle.too_slow
and len(bundle.backup_bundles) == 0 # one backup per
):
- msg = f"*** Rescheduling {bundle.pid}/{bundle.uuid} ***"
+ msg = f"*** Rescheduling {bundle} (first backup) ***"
logger.debug(msg)
self.schedule_backup_for_bundle(bundle)
return
and len(bundle.backup_bundles) < 2 # two backups in dire situations
and self.status.total_idle() > 4
):
- msg = f"*** Rescheduling {bundle.pid}/{bundle.uuid} ***"
+ msg = f"*** Rescheduling {bundle} (second backup) ***"
logger.debug(msg)
self.schedule_backup_for_bundle(bundle)
return
bundle.worker = worker
machine = bundle.machine = worker.machine
username = bundle.username = worker.username
- fname = bundle.fname
+
self.status.record_acquire_worker(worker, uuid)
- logger.debug(f'{self.bundle_prefix(bundle)}: Running bundle on {worker}...')
+ logger.debug(f'{bundle}: Running bundle on {worker}...')
# Before we do any work, make sure the bundle is still viable.
if self.check_if_cancelled(bundle):
except Exception as e:
logger.exception(e)
logger.error(
- f'{self.bundle_prefix(bundle)}: bundle says it\'s cancelled upfront but no results?!'
+ f'{bundle}: bundle says it\'s cancelled upfront but no results?!'
)
assert bundle.worker is not None
self.status.record_release_worker(
try:
cmd = f'{RSYNC} {bundle.code_file} {username}@{machine}:{bundle.code_file}'
start_ts = time.time()
- logger.info(f"{self.bundle_prefix(bundle)}: Copying work to {worker} via {cmd}.")
+ logger.info(f"{bundle}: Copying work to {worker} via {cmd}.")
run_silently(cmd)
xfer_latency = time.time() - start_ts
- logger.info(f"{self.bundle_prefix(bundle)}: Copying done to {worker} in {xfer_latency:.1f}s.")
+ logger.info(f"{bundle}: Copying done to {worker} in {xfer_latency:.1f}s.")
except Exception as e:
logger.exception(e)
logger.error(
- f'{self.bundle_prefix(bundle)}: failed to send instructions to worker machine?!?'
+ f'{bundle}: failed to send instructions to worker machine?!?'
)
assert bundle.worker is not None
self.status.record_release_worker(
f'"source py39-venv/bin/activate &&'
f' /home/scott/lib/python_modules/remote_worker.py'
f' --code_file {bundle.code_file} --result_file {bundle.result_file}"')
- logger.debug(f'{self.bundle_prefix(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 = pid = p.pid
- logger.debug(f'{self.bundle_prefix(bundle)}: Local ssh process pid={pid}; remote worker is {machine}.')
+ logger.debug(f'{bundle}: Local ssh process pid={pid}; remote worker is {machine}.')
return self.wait_for_process(p, bundle, 0)
def wait_for_process(self, p: subprocess.Popen, bundle: BundleDetails, depth: int) -> Any:
- uuid = bundle.uuid
machine = bundle.machine
- fname = bundle.fname
pid = p.pid
if depth > 3:
logger.error(
self.heartbeat()
if self.check_if_cancelled(bundle):
logger.info(
- f'{self.bundle_prefix(bundle)}: another worker finished bundle, checking it out...'
+ f'{bundle}: another worker finished bundle, checking it out...'
)
break
else:
logger.info(
- f"{self.bundle_prefix(bundle)}: pid {pid} ({machine}) our ssh finished, checking it out..."
+ f"{bundle}: pid {pid} ({machine}) our ssh finished, checking it out..."
)
p = None
break
# Otherwise, time for an emergency reschedule.
except Exception as e:
logger.exception(e)
- logger.error(f'{self.bundle_prefix(bundle)}: Something unexpected just happened...')
+ logger.error(f'{bundle}: Something unexpected just happened...')
if p is not None:
logger.warning(
- f"{self.bundle_prefix(bundle)}: Failed to wrap up \"done\" bundle, re-waiting on active ssh."
+ f"{bundle}: Failed to wrap up \"done\" bundle, re-waiting on active ssh."
)
return self.wait_for_process(p, bundle, depth + 1)
else:
machine = bundle.machine
result_file = bundle.result_file
code_file = bundle.code_file
- fname = bundle.fname
- uuid = bundle.uuid
# Whether original or backup, if we finished first we must
# fetch the results if the computation happened on a
if bundle.hostname not in bundle.machine:
cmd = f'{RSYNC} {username}@{machine}:{result_file} {result_file} 2>/dev/null'
logger.info(
- f"{self.bundle_prefix(bundle)}: Fetching results from {username}@{machine} via {cmd}"
+ f"{bundle}: Fetching results from {username}@{machine} via {cmd}"
)
# If either of these throw they are handled in
# if one of the backups finished first; it still must read the
# result from disk.
if is_original:
- logger.debug(f"{self.bundle_prefix(bundle)}: Unpickling {result_file}.")
+ logger.debug(f"{bundle}: Unpickling {result_file}.")
try:
with open(f'{result_file}', 'rb') as rb:
serialized = rb.read()
if bundle.backup_bundles is not None:
for backup in bundle.backup_bundles:
logger.debug(
- f'{self.bundle_prefix(bundle)}: Notifying backup {backup.uuid} that it\'s cancelled'
+ f'{bundle}: Notifying backup {backup.uuid} that it\'s cancelled'
)
backup.is_cancelled.set()
# Tell the original to stop if we finished first.
if not was_cancelled:
logger.debug(
- f'{self.bundle_prefix(bundle)}: Notifying original {bundle.src_bundle.uuid} we beat them to it.'
+ f'{bundle}: Notifying original {bundle.src_bundle.uuid} we beat them to it.'
)
bundle.src_bundle.is_cancelled.set()
failure_count = 0,
)
self.status.record_bundle_details(bundle)
- logger.debug(f'{self.bundle_prefix(bundle)}: Created an original bundle')
+ logger.debug(f'{bundle}: Created an original bundle')
return bundle
def create_backup_bundle(self, src_bundle: BundleDetails):
)
src_bundle.backup_bundles.append(backup_bundle)
self.status.record_bundle_details_already_locked(backup_bundle)
- logger.debug(f'{self.bundle_prefix(bundle)}: Created a backup bundle')
+ logger.debug(f'{backup_bundle}: Created a backup bundle')
return backup_bundle
def schedule_backup_for_bundle(self,
# the original pick them up and unpickle them.
def emergency_retry_nasty_bundle(self, bundle: BundleDetails) -> fut.Future:
- uuid = bundle.uuid
is_original = bundle.src_bundle is None
bundle.worker = None
avoid_last_machine = bundle.machine
if bundle.failure_count > retry_limit:
logger.error(
- f'{self.bundle_prefix(bundle)}: Tried this bundle too many times already ({retry_limit}x); giving up.'
+ f'{bundle}: Tried this bundle too many times already ({retry_limit}x); giving up.'
)
if is_original:
raise RemoteExecutorException(
- f'{self.bundle_prefix(bundle)}: This bundle can\'t be completed despite several backups and retries'
+ f'{bundle}: This bundle can\'t be completed despite several backups and retries'
)
else:
- logger.error(f'{self.bundle_prefix(bundle)}: At least it\'s only a backup; better luck with the others.')
+ logger.error(f'{bundle}: At least it\'s only a backup; better luck with the others.')
return None
else:
logger.warning(
- f'>>> Emergency rescheduling {self.bundle_prefix(bundle)} because of unexected errors (wtf?!) <<<'
+ f'>>> Emergency rescheduling {bundle} because of unexected errors (wtf?!) <<<'
)
return self.launch(bundle, avoid_last_machine)
RemoteWorkerRecord(
username = 'scott',
machine = 'cheetah.house',
- weight = 12,
+ weight = 14,
count = 4,
),
)
projects / python_utils.git / commitdiff