Execution
machinable.execution.Execution (inherits Interface) · source
The mechanism that runs interfaces (locally, Slurm, MPI, ...).
Its Config describes resources and mechanism; the interfaces to run arrive via add.
abandoned_at()
python
abandoned_at(self) -> arrow.arrow.Arrow | NoneWhen another dispatcher took over this run's claim, if it did.
add()
python
add(self, interface: machinable.interface.Interface | list[machinable.interface.Interface]) -> SelfQueue interface(s) to be computed by this execution.
cancel()
python
cancel(self) -> NoneAsk the run to stop, best-effort.
canonicalize_resources()
python
canonicalize_resources(self, resources: dict) -> dictNormalize a resources dict (override in backends).
collect()
python
collect(cls, executions) -> 'ExecutionCollection'Wrap executions in an ExecutionCollection.
compute_fingerprint()
python
compute_fingerprint(self) -> dict | NoneAlways None: an execution never reuses an existing run-record.
compute_predicate_key()
python
compute_predicate_key(self) -> strThe dispatch timestamp when set, else the predicate hash.
computed_resources()
python
computed_resources(self, interface: machinable.interface.Interface | None = None) -> dict | NoneThe effective resources for the bound interface.
Computed once and persisted alongside the run.
continue_run()
python
continue_run(self, interface: machinable.interface.Interface) -> NoneRun interface into this existing run-record, in place and index-free.
deferred()
python
deferred(self, defer: bool = True)Defer dispatch on context exit (to collect a grid without running it).
dispatch()
python
dispatch(self) -> SelfRun every queued interface that is not already cached.
dispatch_code()
python
dispatch_code(self, interface: machinable.interface.Interface, inline: bool = True, project_directory: str | None = None, python: str | None = None, interface_directory: str | None = None, run_record_directory: str | None = None) -> str | NonePython code (or an inline shell command) re-running interface.
dispatch_interface()
python
dispatch_interface(self, interface: machinable.interface.Interface) -> NoneEstablish a run-record for interface via the Index, then run it.
dispatched_at()
python
dispatched_at(self) -> arrow.arrow.Arrow | NoneReturns the dispatch (submission) time.
finished_at()
python
finished_at(self)Returns the finishing time.
heartbeat_at()
python
heartbeat_at(self)Returns the last heartbeat time.
host_info
Host metadata recorded at dispatch, if any.
interface
The Interface this Execution was created for.
Its index parent, set at dispatch via parent_id=interface.uuid.
interfaces
The interfaces queued on this execution.
is_abandoned()
python
is_abandoned(self) -> boolTrue if another dispatcher took over this run's claim.
is_active()
python
is_active(self)True if not finished and the heartbeat is within the liveness window.
is_dispatched()
python
is_dispatched(self)True if dispatch time has been written.
is_finished()
python
is_finished(self)True if finishing time has been written.
is_incomplete()
python
is_incomplete(self)Started (or abandoned) and no longer live.
is_live()
python
is_live(self)True if active or finished.
is_pending()
python
is_pending(self)True if submitted to a scheduler but not yet running.
is_resumed()
python
is_resumed(self)True if resumed time has been written.
is_started()
python
is_started(self)True if starting time has been written.
look_elsewhere()
python
look_elsewhere(self, interface: machinable.interface.Interface) -> str | NoneLook once for interface running or finished elsewhere, without waiting.
materialize()
python
materialize(self, parent_id: str | None = None, force: bool = False, record_id: str | None = None) -> SelfMaterialize the queued interfaces (and the container when standalone).
nickname
Memorable per-run name (e.g. chocolate_mosquito).
of()
python
of(self, interface: machinable.interface.Interface | None) -> SelfBind the interface this execution reports for.
on_add()
python
on_add(self, interface: machinable.interface.Interface)Event when an interface is added.
on_before_dispatch()
python
on_before_dispatch(self) -> bool | NoneHook: return False to abort dispatch of this execution.
on_compute_default_resources()
python
on_compute_default_resources(self, interface: machinable.interface.Interface | None) -> dict | NoneEvent triggered to compute default resources.
on_compute_predicate()
python
on_compute_predicate(self) -> dictResources plus the dispatch timestamp, when dispatching.
on_provenance_attributes()
python
on_provenance_attributes(self) -> dictHistory facet: extra() (seed/timing/resources), no config layers.
on_seeding()
python
on_seeding(self)Hook: default seeds random with self.seed.
on_write_meta_data()
python
on_write_meta_data(self) -> bool | NoneReturn False to prevent writing of meta-data.
output()
python
output(self, incremental: bool = False) -> str | NoneReturns the output log.
output_filepath()
python
output_filepath(self) -> strPath of the run's output.log.
pending_executables
The queued interfaces that are not yet computed.
The subset a submit-style execution backend (Slurm, MPI, ...) still needs to hand off.
prepare_dispatch()
python
prepare_dispatch(self, interface: machinable.interface.Interface)Phase 1 of serverless dispatch.
Materialize (or find continuable) the run-record for interface, persist it, and mark it dispatched, synchronously, before any payload supervision. The payload later continues exactly this record as an in-process/serverless payload via dispatch_interface (which finds it through _resumable_run), a handed-off index-free payload via continue_run (which is handed the record's directory directly). Returns the run's IndexEntry.
refresh_status()
python
refresh_status(self, force: bool = False) -> boolBring this run's status from its remote home, if it runs elsewhere.
A run whose home is not its working copy (an object store) and that is not finished here may be running (or have finished, or died) on another machine, whose markers reach this working copy only through the home. Reads the home's status markers (small files, no artifacts), at most every STATUS_REFRESH_SECONDS unless force: a newer heartbeat and markers missing here are written into the working copy; once the home has finished_at, the run is re-ingested (its catalog and its interface's), so the working copy and the index are complete. Nothing for a local home (it is the working copy), a run finished here, or a run whose payload runs in this process. Every status read calls it. Returns whether the working copy changed.
resumed_at()
python
resumed_at(self) -> arrow.arrow.Arrow | NoneReturns the resumed time.
retrieve_status()
python
retrieve_status(self, status: Literal['dispatched', 'started', 'heartbeat', 'finished', 'resumed', 'abandoned'] = 'heartbeat') -> arrow.arrow.Arrow | NoneRead one status marker as a datetime, or None if unwritten.
seed
The run's random seed.
started_at()
python
started_at(self) -> arrow.arrow.Arrow | NoneReturns the starting time.
status_snapshot()
python
status_snapshot(self) -> machinable.execution.ExecutionStatusRead all four status markers in one pass.
Returns an ExecutionStatus deriving every flag. Prefer this over the individual accessors when several are needed (e.g. when serializing a run).
stream_output()
python
stream_output(self, refresh_every: int | float = 1, stream=<built-in function print>)Follow the run's output log until it finishes.
update_status()
python
update_status(self, status: Literal['dispatched', 'started', 'heartbeat', 'finished', 'resumed', 'abandoned'] = 'heartbeat', timestamp: float | int | arrow.arrow.Arrow | None = None) -> float | int | arrow.arrow.ArrowWrite a status marker.
One of dispatched/started/heartbeat/finished/resumed.
wait_elsewhere()
python
wait_elsewhere(self, wait: bool = True, timeout: float | None = None, interval: float | None = None) -> SelfChoose what dispatch does with an interface running elsewhere.
Source
src/machinable/execution.py
py
"""Executions: the run-records and mechanisms that compute interfaces."""
import base64
import contextlib
import contextvars
import copy
import ctypes
import json
import os
import sys
import threading
import time
import traceback
import warnings
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any, Literal, Self, cast
import arrow
from machinable import schema
from machinable.collection import ExecutionCollection, InterfaceCollection
from machinable.errors import (
DispatchException,
ExecutionFailed,
ExecutionInterrupted,
)
from machinable.format import is_later
from machinable.interface import (
Interface,
_connections,
get_context,
get_dump,
get_inherits,
)
from machinable.storage import Storage, claim_storages
from machinable.types import (
DatetimeType,
TimestampType,
VersionType,
)
from machinable.utils import (
generate_seed,
load_file,
random_str,
save_file,
tee_output,
update_dict,
)
if TYPE_CHECKING:
from machinable.api.models import IndexEntry
from machinable.claims import Claim, Look, RunRecord
from machinable.manifest import Manifest
_allowed_status = [
"dispatched",
"started",
"heartbeat",
"finished",
"resumed",
"abandoned",
]
def _assert_allowed(status: str):
if status not in _allowed_status:
raise ValueError(f"Invalid status '{status}'; must be one of {_allowed_status}")
StatusType = Literal[
"dispatched", "started", "heartbeat", "finished", "resumed", "abandoned"
]
# Cadence at which a running payload writes its heartbeat marker
HEARTBEAT_INTERVAL = 15.0
# The liveness window
DEFAULT_HEARTBEAT_WINDOW = 6 * HEARTBEAT_INTERVAL
def heartbeat_window() -> float:
"""The liveness window in seconds, relaxable per store.
``MACHINABLE_HEARTBEAT_WINDOW`` widens it for stores whose markers reach
the reader on a delay (a node-local scratch rsynced to shared storage on
an interval, or a store polled over SSH), where transport latency rather
than the payload decides how stale a live run's heartbeat looks. A
malformed or non-positive value falls back to the default.
"""
value = os.environ.get("MACHINABLE_HEARTBEAT_WINDOW")
if value is None:
return DEFAULT_HEARTBEAT_WINDOW
try:
window = float(value)
except ValueError:
return DEFAULT_HEARTBEAT_WINDOW
return window if window > 0 else DEFAULT_HEARTBEAT_WINDOW
def heartbeat_interval() -> float:
"""The cadence of the heartbeat marker in seconds."""
return min(HEARTBEAT_INTERVAL, heartbeat_window() / 6)
def heartbeat_is_fresh(heartbeat: DatetimeType | None, window: float) -> bool:
"""Whether ``heartbeat`` is recent enough for the run to count as live.
The single freshness comparison behind both :attr:`ExecutionStatus.is_active`
and :meth:`Execution.is_active`. It measures the total elapsed seconds:
a ``timedelta``'s ``.seconds`` is only its seconds-of-day component, so a
day-old heartbeat would read as seconds-old. A negative age (the payload's
clock running ahead of the reader's, routine once the two are different
machines) is younger than any window, which is the safe direction as the
format prefers a false "live" over a false "died".
"""
if heartbeat is None:
return False
return (arrow.now() - heartbeat).total_seconds() < window
@dataclass(frozen=True)
class ExecutionStatus:
"""A one-shot read of a run's status markers.
Resolving them individually costs a directory lookup + file read *per* accessor;
this reads the five ``*_at`` markers once (see :meth:`Execution.status_snapshot`)
and derives every flag from them. The derived flags mirror the equivalent
:class:`Execution` methods exactly.
"""
dispatched_at: DatetimeType | None = None
started_at: DatetimeType | None = None
resumed_at: DatetimeType | None = None
heartbeat_at: DatetimeType | None = None
finished_at: DatetimeType | None = None
abandoned_at: DatetimeType | None = None
window: float = DEFAULT_HEARTBEAT_WINDOW
@property
def is_dispatched(self) -> bool:
"""True when ``dispatched_at`` has been written."""
return self.dispatched_at is not None
@property
def is_started(self) -> bool:
"""True when ``started_at`` has been written."""
return self.started_at is not None
@property
def is_finished(self) -> bool:
"""True when ``finished_at`` has been written."""
return self.finished_at is not None
@property
def is_resumed(self) -> bool:
"""True when the run was resumed at least once."""
return self.resumed_at is not None
@property
def is_abandoned(self) -> bool:
"""True when another dispatcher took over this run's claim."""
return self.abandoned_at is not None
@property
def is_active(self) -> bool:
"""True while the heartbeat is recent and the run unfinished."""
if self.is_finished or self.is_abandoned:
return False
return heartbeat_is_fresh(self.heartbeat_at, self.window)
@property
def is_pending(self) -> bool:
"""True while submitted but not yet running.
The scheduler-handoff window: ``prepare_dispatch`` writes
``dispatched_at`` synchronously, and the payload writes ``started_at``
only once it actually runs, so a queued job (Slurm, and any other
submit-style backend) is pending for as long as it sits in the queue.
Without this the state is indistinguishable from a never-dispatched
record (see docs/design/format.md).
"""
return (
self.is_dispatched
and not self.is_started
and not self.is_finished
and not self.is_abandoned
)
@property
def is_live(self) -> bool:
"""True when active or finished."""
return self.is_finished or self.is_active
@property
def is_incomplete(self) -> bool:
"""True when started (or abandoned) but no longer live."""
return (self.is_started or self.is_abandoned) and not self.is_live
@dataclass(frozen=True)
class _Hold:
"""A claim this dispatch holds while it runs an interface."""
storage: Storage
location: str
claim: "Claim"
resume: "IndexEntry | None"
_HELD_CLAIMS: contextvars.ContextVar[frozenset[str]] = contextvars.ContextVar(
"machinable_held_claims", default=frozenset()
)
# The least time between two reads of a run's markers from its remote home
# (:meth:`Execution.refresh_status`), so a status read in a loop is cheap.
STATUS_REFRESH_SECONDS = 1.0
# record id -> monotonic time of its last refresh from the home
_STATUS_REFRESHED: dict[str, float] = {}
# working copies a payload in this process writes: they are the run's truth
_RUNNING_HERE: set[str] = set()
def _context_key(context: Any) -> str:
return json.dumps(list(context), sort_keys=True, default=str)
def _enter_recorded_context(interface: Interface, stack: contextlib.ExitStack) -> None:
"""Enter the ``with``-contexts ``interface`` was created in that are not entered.
A run's predicate is computed from the ambient scopes, so a run must see
the scopes it was created in while it runs: an interface launched in its
``with`` block sees them already, but one dispatched later (an execution
dispatched by ``POST /v1/executions``, a script's ``continue_run``, a
resumed run) does not. The recorded context (``interface.context``) is
entered again in order, except the contexts already entered and an
``Execution`` (which is the runner, not something to run in): the very
instances it was queued in when this process has them, else made again
from the record. Nothing when the run's predicate is the one recorded.
"""
recorded = interface.context or []
if not recorded:
return
if interface.compute_predicate() == (interface.__model__.predicate or {}):
return
present = {_context_key(c) for c in get_context(interface)}
remembered = {
_context_key([c.module, c.version()]): c
for c in interface._cache.get("creation_context") or []
}
for context in recorded:
key = _context_key(context)
if key in present:
continue
if key in remembered:
stack.enter_context(remembered[key])
continue
module, version = context[0], context[1]
try:
made = Interface.make(module, version=version)
except Exception as ex: # noqa: BLE001 - the run goes on without it
warnings.warn(
f"Cannot enter {module} {version}, a context {interface!r} was "
f"created in: {type(ex).__name__}: {ex}",
stacklevel=2,
)
continue
if (made.kind or "Interface") == "Execution":
continue
stack.enter_context(made)
class Execution(Interface):
"""The mechanism that runs interfaces (locally, Slurm, MPI, ...).
Its ``Config`` describes resources and mechanism; the interfaces to run arrive via
:meth:`add`.
"""
kind = "Execution"
default = None
# Whether dispatch-time *submits* the payload to an external supervisor
# (Slurm, queues) rather than running it. `machinable dispatch` does not
# detach a handoff execution: the scheduler owns the payload, and the
# generated job re-enters via `machinable dispatch --foreground`.
# (Distinct from `deferred()`, which defers dispatch until context exit.)
handoff: bool = False
def __init__(
self,
version: VersionType = None,
resources: dict | None = None,
seed: int | None = None,
uses: None | Interface | list[Interface] = None,
derived_from: Interface | None = None,
):
super().__init__(version=version, uses=uses, derived_from=derived_from)
if seed is None:
seed = generate_seed()
self.__model__ = schema.Execution(
kind=self.kind,
module=self.__model__.module,
config=self.__model__.config,
version=self.__model__.version,
resources=resources,
seed=seed,
inherits=get_inherits(self),
)
self.__model__._dump = get_dump(self)
self._interface_: Interface | None = None
self._interfaces: list[Interface] = []
self._resources = {}
self._defer_dispatch = False
self._wait_elsewhere = True
self._wait_timeout: float | None = None
self._wait_interval: float | None = None
self.elsewhere: dict[str, tuple[str, str]] = {}
def deferred(self, defer: bool = True):
"""Defer dispatch on context exit (to collect a grid without running it)."""
self._defer_dispatch = defer
return self
def wait_elsewhere(
self,
wait: bool = True,
timeout: float | None = None,
interval: float | None = None,
) -> Self:
"""Choose what dispatch does with an interface running elsewhere."""
self._wait_elsewhere = wait
self._wait_timeout = timeout
self._wait_interval = interval
return self
@classmethod
def collect(cls, executions) -> "ExecutionCollection": # ty: ignore[invalid-method-override]
"""Wrap ``executions`` in an :class:`ExecutionCollection`."""
return ExecutionCollection(executions)
def compute_fingerprint(self) -> dict | None:
"""Always ``None``: an execution never reuses an existing run-record."""
return None # do not retrieve existing execution
def on_compute_predicate(self) -> dict:
"""Resources plus the dispatch timestamp, when dispatching."""
predicate: dict = {"resources": self._model.resources}
started_at = getattr(self, "_dispatch_started_at", None)
if started_at is not None:
predicate["started_at"] = started_at
return predicate
def compute_predicate_key(self) -> str:
"""The dispatch timestamp when set, else the predicate hash."""
predicate = self.compute_predicate()
started_at = predicate.get("started_at")
if started_at is not None:
return str(started_at)
return super().compute_predicate_key()
@property
def _model(self) -> schema.Execution:
return cast(schema.Execution, self.__model__)
@property
def seed(self) -> int:
"""The run's random seed."""
return self._model.seed
@property
def nickname(self) -> str:
"""Memorable per-run name (e.g. ``chocolate_mosquito``)."""
return self._model.nickname
@property
def interfaces(self) -> InterfaceCollection:
"""The interfaces queued on this execution."""
return InterfaceCollection(self._interfaces)
@property
def pending_executables(self) -> InterfaceCollection:
"""The queued interfaces that are not yet computed.
The subset a submit-style execution backend (Slurm, MPI, ...) still
needs to hand off.
"""
return self.interfaces.filter(lambda x: not x.cached())
@property
def interface(self) -> Interface | None:
"""The Interface this Execution was created for.
Its index parent, set at dispatch via ``parent_id=interface.uuid``.
"""
if self.uuid is not None:
from machinable.index import Index
entry = Index.get().get_by_id(self.uuid)
if entry is not None and entry.parent_id:
return Interface.find_by_id(entry.parent_id, stub=True)
return getattr(self, "_interface_", None)
def of(self, interface: Interface | None) -> Self:
"""Bind the interface this execution reports for."""
self._interface_ = interface
return self
def on_provenance_attributes(self) -> dict:
# history kind: carry extra() (seed/timing/resources), not config layers.
"""History facet: ``extra()`` (seed/timing/resources), no config layers."""
attrs = dict(self.__model__.extra() or {})
attrs.pop("_dump", None)
return attrs
def _capture_manifest(self) -> None:
"""Capture code/deps/environment provenance and record that this run.
``uses`` the resulting content-addressed Manifest node(s), one per
tool the project resolves via ``on_resolve_manifests`` (git by
default). Each is deduped by code state. The ``uses`` edge is
persisted via ``relate()`` to both the index and the run's on-disk
mirror, so it survives a reindex.
"""
from machinable.interface import Interface
from machinable.project import Project
if not Project.is_connected():
return
# hooks live on the provider (interface/project.py), like
# on_resolve_remotes and get_host_info
for module in Project.get().provider().on_resolve_manifests() or []:
try:
manifest = cast("Manifest", Interface.make(module)).capture()
except Exception: # noqa: BLE001 - a tool failing to capture never breaks a run
continue
if not manifest.entries:
continue
manifest.materialize()
# durable edge: index row + disk mirror (reindex reads the mirror).
self.relate("uses", manifest)
def on_add(self, interface: Interface):
"""Event when an interface is added."""
def add(
self,
interface: Interface | list[Interface],
) -> Self:
"""Queue interface(s) to be computed by this execution."""
if self.is_materialized():
from machinable.errors import MachinableError
raise MachinableError(
f"{repr(self)} already exists and cannot be modified."
)
if isinstance(interface, list | tuple):
for item in interface:
self.add(cast(Interface, item))
return self
if interface in self._interfaces:
return self
from machinable.data import Data
if isinstance(interface, Data):
from machinable.errors import MachinableError
raise MachinableError(
f"{interface.__class__.__name__} is machinable.Data: an input that"
" runs use, never executed itself"
)
if not interface.is_materialized():
# it materializes later (at dispatch, or by whoever collected it):
# in the contexts it is queued in, so its scopes still identify it
interface._remember_creation_context()
self.on_add(interface)
self._interfaces.append(interface)
return self
def materialize(
self,
parent_id: str | None = None,
force: bool = False,
record_id: str | None = None,
) -> Self:
"""Materialize the queued interfaces (and the container when standalone)."""
for interface in self._interfaces:
assert interface.config is not None
for interface in self._interfaces:
with interface._creation_context():
interface.materialize(parent_id=parent_id, force=force)
# Queued runs materialise under their interface in dispatch_interface().
if parent_id is not None or not self._interfaces:
return super().materialize(
parent_id=parent_id, force=force, record_id=record_id
)
return self
def _apply_index_entry(self, entry) -> None:
self._index_entry = entry # source of the record's id.json header
self.__model__.uuid = entry.record_id
self.__model__.created_at_ns = entry.created_at_ns
def canonicalize_resources(self, resources: dict) -> dict:
"""Normalize a resources dict (override in backends)."""
return resources
def computed_resources(self, interface: Interface | None = None) -> dict | None:
"""The effective resources for the bound interface.
Computed once and persisted alongside the run.
"""
if interface is not None:
self.of(interface)
cache_key = self._interface_.id if self._interface_ is not None else self.id
if cache_key not in self._resources:
# Persist per interface, not to one shared file. A submit-style
# runner (Slurm, MPI) resolves resources for many interfaces on a
# single Execution; a constant filename would hand every interface
# the first one's resources. On a run-record (one interface) this is
# simply that interface's file.
cache_file = f"computed_resources.{cache_key or 'default'}.json"
resources = self.load_file(cache_file, default=None)
if resources is None:
resources = self._compute_resources(self._interface_)
self.save_file(cache_file, resources)
self._resources[cache_key] = resources
return self._resources[cache_key]
def _compute_resources(self, interface: Interface | None = None) -> dict:
default_resources = self.on_compute_default_resources(interface)
model_resources = self._model.resources
if not model_resources and default_resources is not None:
return self.canonicalize_resources(default_resources)
if model_resources and not default_resources:
resources = copy.deepcopy(model_resources)
resources.pop("_inherit_defaults", None)
return self.canonicalize_resources(resources)
if model_resources and default_resources:
resources = copy.deepcopy(model_resources)
if resources.pop("_inherit_defaults", True) is False:
return self.canonicalize_resources(resources)
defaults = self.canonicalize_resources(default_resources)
update = self.canonicalize_resources(resources)
defaults_ = copy.deepcopy(defaults)
update_ = copy.deepcopy(update)
removals = [
k for k in update.keys() if isinstance(k, str) and k.startswith("#")
]
for removal in removals:
defaults_.pop(removal[1:], None)
update_.pop(removal, None)
return update_dict(defaults_, update_)
return {}
def dispatch(self) -> Self:
"""Run every queued interface that is not already cached."""
if not self._interfaces:
return self
uncached = False
for interface in self._interfaces:
if not interface.cached():
uncached = True
elif interface.is_materialized():
self._index_claimed_run(interface)
if not uncached:
return self
if self.on_before_dispatch() is False:
return self
for interface in self._interfaces:
if not interface.is_materialized():
with interface._creation_context():
interface.materialize()
if type(self).__call__ is not Execution.__call__:
for interface in self._interfaces:
if not interface.cached():
self.look_elsewhere(interface)
try:
self.__call__()
self.on_after_dispatch(success=True)
except BaseException as _ex: # pylint: disable=broad-except
self.on_after_dispatch(success=False)
raise ExecutionFailed("Execution failed") from _ex
return self
def __call__(self) -> None:
"""Run every queued interface (the default execution mechanism)."""
for interface in self._interfaces:
if not interface.cached():
self.dispatch_interface(interface)
else:
# cached since dispatch began, maybe by another index's run
self._index_claimed_run(interface)
def dispatch_interface(self, interface: Interface) -> None:
"""Establish a run-record for ``interface`` via the Index, then run it."""
if not interface.is_materialized():
with interface._creation_context():
interface.materialize()
hold = None
if (
self.on_write_meta_data() is not False
and interface.on_write_meta_data() is not False
):
# ranks that write no metadata (MPI workers) run under the
# controller's claim
run, hold = self._claim(interface)
if not run:
return
resume = hold.resume if hold is not None else self._resumable_run(interface)
token = None
try:
if resume is not None:
# Deliberate resume: the interface's latest run-record (same
# runner identity) is dispatched-or-started but unfinished,
# continue it in place; a merely-dispatched record (serverless
# phase 1) starts, a started one resumes.
self.__model__.predicate = copy.deepcopy(resume.predicate)
self._apply_index_entry(resume)
self._pull_working_copy(resume)
else:
self._apply_index_entry(
self._mint_run_record(
interface,
record_id=hold.claim.record_id if hold is not None else None,
)
)
self.to_directory(self.local_directory(create=True), relations=False)
if hold is not None:
# beat at once, so the claim timeout only covers the moment
# between claiming and this first heartbeat
self.update_status(status="heartbeat")
token = _HELD_CLAIMS.set(_HELD_CLAIMS.get() | {hold.location})
if resume is None:
self._capture_manifest()
self._run_payload(interface)
finally:
if token is not None:
_HELD_CLAIMS.reset(token)
if hold is not None:
# after _run_payload wrote finished_at (or failed)
self._release(hold)
def _pull_working_copy(self, entry: "IndexEntry") -> None:
"""Bring a resumed run's files from its remote home into its working copy."""
from machinable.storage import remote_home
home = remote_home(entry)
if home is None:
return
home.pull(entry.record_id)
self._home_pushed = True
def _cached_in_home(self, interface: Interface) -> bool:
"""The interface's ``cached`` marker, as its home has it.
A remote home's marker is brought into the working copy.
"""
if interface.cached():
return True
home = interface._remote_home()
if home is None:
return False
location = home.interface_location(interface)
reason = home.read_bytes(location, "cached") if location else None
if reason is None:
return False
save_file(interface.local_directory("cached"), reason.decode("utf-8"))
return True
def _looks(self, interface: Interface) -> "list[tuple[Storage, Look]]":
"""Look at ``interface`` in every storage that holds it (claim, then runs)."""
from machinable.claims import claim_location, look
looks = []
seen = set()
for storage in claim_storages():
directory = storage.interface_location(interface)
if directory is None:
continue
location = claim_location(directory)
if location in seen:
continue
seen.add(location)
looks.append((storage, look(storage, interface, location)))
return looks
def look_elsewhere(self, interface: Interface) -> str | None:
"""Look once for interface running or finished elsewhere, without waiting."""
waits = getattr(self, "_wait_elsewhere", True)
self._wait_elsewhere = False
try:
self._claim(interface, claim=False)
finally:
self._wait_elsewhere = waits
found = getattr(self, "elsewhere", {}).get(cast(str, interface.uuid))
return found[0] if found is not None else None
def _claim(
self, interface: Interface, claim: bool = True
) -> "tuple[bool, _Hold | None]":
"""Run the claim protocol for ``interface`` (see :mod:`machinable.claims`)."""
from machinable.claims import HELD, after, decide
from machinable.claims import Claim as ClaimFile
from machinable.index import Index
if not hasattr(self, "elsewhere"):
self.elsewhere = {}
uuid = cast(str, interface.uuid)
wait = getattr(self, "_wait_elsewhere", True)
timeout = getattr(self, "_wait_timeout", None)
interval = getattr(self, "_wait_interval", None) or max(
0.05, min(1.0, heartbeat_interval())
)
deadline = None if timeout is None else time.monotonic() + timeout
index = Index.get()
pending = self._resumable_run(interface)
dismissed: set[str] | None = None
rechecked = False
while True:
looks = self._looks(interface)
cached = self._cached_in_home(interface)
if dismissed is None:
dismissed = {
run.record_id
for _, seen in looks
for run in seen.runs
if run.finished
and not (
seen.claim is not None
and seen.claim.state == HELD
and seen.claim.record_id == run.record_id
)
and index.get_by_id(run.record_id) is not None
}
finished = [
(storage, run)
for storage, seen in looks
for run in seen.runs
if run.finished and run.record_id not in dismissed
]
primary = next(
(
(storage, seen)
for storage, seen in looks
if storage.exclusive_claims()
and (not seen.present or seen.version is not None)
),
None,
)
if primary is None:
# nowhere to claim exclusively: adopt what finished, else run
if finished and not cached:
self._adopt(interface, *finished[0])
return False, None
return not cached, None
storage, seen = primary
if (
seen.claim is not None
and seen.claim.state == HELD
and seen.location in _HELD_CLAIMS.get()
):
# this context's own run dispatches its interface again
return True, None
decision = decide(
cached=cached,
finished=bool(finished),
claim_file=seen.claim_file,
claim_state=seen.state,
wait=wait and claim,
)
if (
decision == "cached"
and not finished
and not rechecked
and interface._remote_home() is not None
):
rechecked = True
continue
if decision == "cached":
if finished:
# cached by the run this dispatch waited on, maybe made
# by another index: index it here too
self._adopt(interface, *finished[0])
self._abandon_pending(pending, finished[0][1].record_id)
else:
self.elsewhere.pop(uuid, None)
return False, None
if decision == "adopt":
if self._adopt(interface, *finished[0]):
self._abandon_pending(pending, finished[0][1].record_id)
return False, None
# it does not ingest (a damaged record): not a result to adopt
dismissed.add(finished[0][1].record_id)
continue
if decision in ("wait", "report"):
assert seen.claim is not None
self.elsewhere[uuid] = ("running elsewhere", seen.claim.record_id)
if decision == "report" or (
deadline is not None and time.monotonic() >= deadline
):
if claim:
self._abandon_pending(pending, seen.claim.record_id)
return False, None
time.sleep(interval)
continue
if not claim:
self.elsewhere.pop(uuid, None)
return False, None
resume = self._resumable_run(interface)
if (
decision == "take over"
and resume is not None
and seen.claim is not None
and resume.record_id == seen.claim.record_id
):
# a run whose claim is taken over is abandoned, never resumed
resume = None
record_id = resume.record_id if resume is not None else random_str(12)
generation = seen.claim.generation + 1 if seen.claim is not None else 1
data = ClaimFile.held(record_id, generation)
if decision == "create":
won = storage.create_if_absent(seen.location, data.to_bytes())
else:
won = storage.replace_if_unchanged(
seen.location, data.to_bytes(), cast(str, seen.version)
)
step = after(decision, won)
if step == "look":
continue
hold = _Hold(storage, seen.location, data, resume)
dead = seen.claimed_run
if (
step == "abandon"
and dead is not None
and dead.record_id != record_id
and not dead.finished
):
storage.mark_run(dead, "abandoned_at", str(arrow.now()))
# a finish racing the takeover wins: give the claim back
again = next(
(
r
for r in storage.run_records(interface)
if r.record_id == dead.record_id
),
None,
)
if again is not None and again.finished:
self._release(hold)
self._adopt(interface, storage, again)
return False, None
self.elsewhere.pop(uuid, None)
return True, hold
def _index_claimed_run(self, interface: Interface) -> None:
"""Index the finished run a cached interface's claim names, if unknown here."""
from machinable.claims import Claim as ClaimFile
from machinable.claims import claim_location
from machinable.index import Index
index = Index.get()
for storage in claim_storages():
directory = storage.interface_location(interface)
if directory is None:
continue
read = storage.read_versioned(claim_location(directory))
claim = ClaimFile.from_bytes(read[0]) if read is not None else None
if claim is None or index.get_by_id(claim.record_id) is not None:
return
for run in storage.run_records(interface):
if run.record_id == claim.record_id and run.finished:
storage.adopt(run)
self.elsewhere[cast(str, interface.uuid)] = (
"finished elsewhere",
run.record_id,
)
return
def _adopt(self, interface: Interface, storage: Storage, run: "RunRecord") -> bool:
"""End ``interface`` cached on ``run``, a run another dispatcher finished."""
from machinable.index import Index
known = Index.get().get_by_id(run.record_id) is not None
if not storage.adopt(run):
return False
if not interface.cached():
interface.cached(True, reason="finished" if known else "finished elsewhere")
self.elsewhere[cast(str, interface.uuid)] = (
"finished elsewhere",
run.record_id,
)
return True
def _abandon_pending(self, pending: "IndexEntry | None", other: str) -> None:
"""Mark this dispatch's own pending run-record abandoned."""
from machinable.claims import RunRecord
from machinable.storage import storage_for
if pending is None or pending.record_id == other:
return
try:
home = storage_for(pending.storage_uri)
except ValueError:
return
location = home.handle(pending.storage_uri)
markers = home.markers(location)
if not markers or any(
marker in markers
for marker in ("started_at", "finished_at", "abandoned_at")
):
return
home.mark_run(
RunRecord(pending.record_id, location), "abandoned_at", str(arrow.now())
)
@staticmethod
def _release(hold: _Hold) -> bool:
"""Give the claim up (state ``released``); False when it was taken over."""
from machinable.claims import HELD
from machinable.claims import Claim as ClaimFile
try:
current = hold.storage.read_versioned(hold.location)
if current is None:
return False
claim = ClaimFile.from_bytes(current[0])
if (
claim is None
or claim.state != HELD
or claim.record_id != hold.claim.record_id
):
return False # a waiter took it over; the run's result stands
version = current[1] if current[1] is not None else str(claim.generation)
return hold.storage.replace_if_unchanged(
hold.location, claim.released().to_bytes(), version
)
except Exception as ex: # noqa: BLE001 - a lost release must not fail the run
import warnings
warnings.warn(
f"Releasing the claim at {hold.location} failed: {ex}",
stacklevel=2,
)
return False
def continue_run(self, interface: Interface) -> None:
"""Run ``interface`` into this existing run-record, in place and index-free."""
if not self.is_materialized():
raise RuntimeError(
"continue_run requires an existing run-record (load it with "
"Execution.from_directory); use dispatch_interface to create one"
)
self._run_payload(interface)
def _run_payload(self, interface: Interface) -> None:
"""The execution lifecycle shared by dispatch_interface and continue_run."""
from machinable.interface import _IN_RUN
from machinable.project import Project
catalog_uuid = interface.uuid
self.of(interface)
self.computed_resources()
in_run = _IN_RUN.set(
_IN_RUN.get() | {cast(str, self.uuid), cast(str, interface.uuid)}
)
running_here = os.path.abspath(self.local_directory())
_RUNNING_HERE.add(running_here)
# the run whose results the interface saves (results.json's "run")
results_run = getattr(interface, "_results_run", None)
interface._results_run = self.uuid
# both the execution and the dispatched interface can veto (an MPI
# component may know that only its controller rank should write)
writes_meta_data = (
self.on_write_meta_data() is not False
and interface.on_write_meta_data() is not False
and self.is_mounted()
)
_watch_stop = (
threading.Event()
) # stops the best-effort cancel watcher (see below)
_beat_stop = threading.Event()
def _stop_beat():
_beat_stop.set()
# the run sees the contexts it was created in (its scopes, so its
# predicate), wherever it is dispatched from
contexts = contextlib.ExitStack()
try:
_enter_recorded_context(interface, contexts)
interface.on_before_dispatch()
self._invoke_seeding(interface)
if writes_meta_data:
status = "resumed" if self.is_started() else "started"
self.update_status(status=status)
self.save_file(
"host.json",
data=Project.get().provider().get_host_info(),
)
_beat_ctx = contextvars.copy_context()
def _write_beat():
interface.on_heartbeat()
self.update_status(status="heartbeat")
def beat():
while not _beat_stop.wait(heartbeat_interval()):
_write_beat()
if writes_meta_data:
_write_beat()
threading.Thread(
target=lambda: _beat_ctx.run(beat), daemon=True
).start()
# Best-effort cancellation: a watcher polls for a `cancelled`
# marker in this run's directory (written by
# POST /v1/executions/{uuid}/cancel) and injects
# ExecutionInterrupted into the dispatch thread; no
# execution-engine coupling, the cancel is just a file.
# Cooperative: raised at the next Python bytecode boundary.
_dispatch_tid = threading.get_ident()
_cancel_marker = self.local_directory("cancelled")
def watch_cancel() -> None:
if _watch_stop.is_set():
return
if os.path.exists(_cancel_marker):
ctypes.pythonapi.PyThreadState_SetAsyncExc(
ctypes.c_long(_dispatch_tid),
ctypes.py_object(ExecutionInterrupted),
)
return
t = threading.Timer(1.0, watch_cancel)
t.daemon = True
t.start()
# the 1s marker poll respects on_write_meta_data like every other
# filesystem touch, so N MPI ranks do not stat the (shared)
# filesystem N times a second; interrupting the metadata rank
# tears down the job through the launcher
if writes_meta_data:
watch_cancel()
# Do not use Execution.__call__ here as it would re-dispatch _interfaces.
if (
isinstance(interface, Execution)
and interface.__class__.__call__ is Execution.__call__
):
pass
else:
# capture the run's stdout/stderr into its output.log (thread-
# scoped tee, so concurrent in-process runs never interleave);
# script-based backends redirect at the shell level instead
if writes_meta_data:
with tee_output(self.output_filepath()):
try:
interface.__call__()
except BaseException:
traceback.print_exc()
raise
else:
interface.__call__()
# Send close sentinel so any server-side drain task can finish.
if interface._emit_queue is not None and interface._emit_loop is not None:
interface._emit_loop.call_soon_threadsafe(
interface._emit_queue.put_nowait, None
)
if writes_meta_data:
self._finish_results(interface)
interface.on_success()
interface.on_finish(success=True)
_stop_beat()
if writes_meta_data:
self.update_status(status="finished")
interface.cached(True, reason="finished", catalog_id=catalog_uuid)
interface.on_after_dispatch(success=True)
except BaseException as _ex:
# Ensure drain task unblocks on failure.
if interface._emit_queue is not None and interface._emit_loop is not None:
interface._emit_loop.call_soon_threadsafe(
interface._emit_queue.put_nowait, None
)
interface.on_failure(exception=_ex)
interface.on_finish(success=False)
interface.on_after_dispatch(success=False)
raise DispatchException(
f"{interface.__class__.__name__} dispatch failed"
) from _ex
finally:
_watch_stop.set()
_stop_beat()
contexts.close()
_IN_RUN.reset(in_run)
interface._results_run = results_run
_RUNNING_HERE.discard(running_here)
if writes_meta_data:
interface._write_through()
mirror_interface = interface.on_commit() is not False
for storage in Storage.connected():
storage.update(self)
if mirror_interface:
storage.update(interface)
def _finish_results(self, interface: Interface) -> None:
"""Save the interface's ``@result`` methods and mark its results complete."""
from machinable.results import declared_results, read_manifest
for name, options, method in declared_results(interface):
interface.save_result(name, method(interface), **options)
manifest = read_manifest(interface.local_directory())
if manifest is not None and manifest.get("run") == self.uuid:
interface.results_complete()
def _mint_run_record(self, interface: Interface, record_id: str | None = None):
"""Materialize a fresh run-record child of ``interface``.
Each new dispatch creates one; a prior runner uuid is only ever
continued via :meth:`_resumable_run`. ``record_id`` is the id the run
was claimed under (minted when not given). Returns the IndexEntry.
"""
from machinable.index import Index, materialize_keys_from_model
self._dispatch_started_at = datetime.now(UTC).strftime("%Y-%m-%d_%H%M%S")
try:
self.__model__.uuid = None
keys = materialize_keys_from_model(
self.__model__,
parent_id=interface.uuid,
identity_key=self.catalog_identity_key(),
predicate=self.predicate,
predicate_key=self.predicate_key,
record_id=record_id,
)
response = Index.get().materialize(keys, force=True)
finally:
self._dispatch_started_at = None
self.__model__.predicate = copy.deepcopy(response.entry.predicate)
return response.entry
def _resumable_run(self, interface: Interface):
"""The interface's latest continuable run-record, or ``None``.
Continuable means dispatched or started but not finished, and created
by the same runner identity (a differently configured Execution starts
its own run-record
instead of continuing a foreign one). A merely-dispatched record is the
serverless phase-1 "pending" state; the payload continues it and marks
it "started"; a started one resumes. A run-record marked
``abandoned_at`` (its claim was taken over) is never continued.
Returns the IndexEntry to continue, or ``None`` to mint a new
run-record. Markers cannot distinguish a very recently failed run
from a live one, so no liveness guard is applied here: concurrent
dispatch is coordinated by the claim :meth:`dispatch_interface` takes
before it continues or mints a run-record (``docs/design/format.md``,
"Claims"), so a run held by another thread, process or machine is
waited on rather than continued twice.
"""
from machinable.api.models import FindRequest
from machinable.index import Index
from machinable.storage import storage_for
from machinable.utils import sanitize_filename
index = Index.get()
latest = index.find(
FindRequest(
parent_id=interface.uuid,
kind="Execution",
include_hidden=True,
limit=1,
)
)
if not latest.items:
return None
entry = index.get_by_id(latest.items[0].record_id)
if entry is None:
return None
if entry.identity_key != sanitize_filename(
self.catalog_identity_key(), max_len=128
):
return None
try:
home = storage_for(entry.storage_uri)
markers = home.markers(home.handle(entry.storage_uri))
except ValueError:
return None
if "started_at" not in markers and "dispatched_at" not in markers:
return None
if "finished_at" in markers or "abandoned_at" in markers:
return None
return entry
def prepare_dispatch(self, interface: Interface):
"""Phase 1 of serverless dispatch.
Materialize (or find continuable) the run-record for ``interface``,
persist it, and mark it ``dispatched``, synchronously, before any
payload supervision. The payload later continues exactly this record as
an in-process/serverless payload via :meth:`dispatch_interface` (which
finds it through :meth:`_resumable_run`), a handed-off index-free
payload via :meth:`continue_run` (which is handed the record's
directory directly). Returns the run's IndexEntry.
"""
if not interface.is_materialized():
with interface._creation_context():
interface.materialize()
entry = self._resumable_run(interface)
minted = entry is None
if minted:
entry = self._mint_run_record(interface)
else:
self.__model__.predicate = copy.deepcopy(entry.predicate)
self._apply_index_entry(entry)
self.to_directory(self.local_directory(create=True), relations=False)
if minted:
self._capture_manifest()
self.of(interface)
self.update_status(status="dispatched")
return entry
def _invoke_seeding(self, interface: Interface) -> None:
# duck-typed: plain interfaces carry neither hook nor seed
target = cast("Any", interface)
if hasattr(interface, "on_seeding"):
target.on_seeding()
elif hasattr(interface, "seed"):
import random
random.seed(target.seed)
def dispatch_code(
self,
interface: Interface,
inline: bool = True,
project_directory: str | None = None,
python: str | None = None,
interface_directory: str | None = None,
run_record_directory: str | None = None,
) -> str | None:
"""Python code (or an inline shell command) re-running ``interface``."""
from machinable.index import Index
from machinable.project import Project
if project_directory is None:
project_directory = Project.get().path()
if python is None:
python = sys.executable
lines = ["from machinable import Project, Execution, Interface"]
# embed paths via repr so backslashes (Windows) and quotes/spaces are escaped
# into a valid Python string literal
lines.append(f"Project({project_directory!r}).__enter__()")
for kind, connected in list(_connections().items()):
# The payload is index-free as it reconstructs the interface and its
# run-record straight from their directories, so it needs neither
# the Storage nor the Index nor a Transport since this code is what
# the transport delivered, already running on the far side.
if kind in ["Project", "Execution", "Storage", "Index", "Transport"]:
continue
for connected_interface in connected:
# repr() escapes the JSON's quotes AND backslashes (Windows paths
# in the serialized config)
jn = connected_interface.as_json()
lines.append(f"Interface.from_json({jn!r}).__enter__()")
if not interface.is_materialized():
interface_directory = os.path.abspath(
os.path.join(project_directory, ".machinable", "nonexistent")
)
lines.append(
"interface__ = "
f"{interface.kind}.from_directory({interface_directory!r})"
)
lines.append("Execution().dispatch_interface(interface__)")
else:
if run_record_directory is None:
entry = Execution().prepare_dispatch(interface)
run_record_directory = os.path.abspath(
Index.get().working_directory(entry)
)
if interface_directory is None:
interface_directory = os.path.abspath(interface.local_directory())
lines.append(
"interface__ = "
f"{interface.kind}.from_directory({interface_directory!r})"
)
lines.append(f"run__ = Execution.from_directory({run_record_directory!r})")
lines.append("run__.continue_run(interface__)")
if inline:
# base64-wrap the program so the shell command carries no quotes or
# backslashes of its own, robust across shells and OSes (a raw
# `python -c "…"` breaks on the embedded JSON's quotes / Windows paths).
# Inline output targets a POSIX shell (the double-quote wrapper); a Windows
# interpreter path is forward-slashed + single-quoted so a bash on Windows
# (git-bash) doesn't strip the backslashes as escapes.
payload = base64.b64encode("\n".join(lines).encode()).decode()
python_arg = python.replace("\\", "/")
return (
f"'{python_arg}' -c "
f"\"import base64;exec(base64.b64decode('{payload}').decode())\""
)
return "\n".join(lines)
@property
def host_info(self) -> dict | None:
"""Host metadata recorded at dispatch, if any."""
return self.load_file("host.json", None)
def cancel(self) -> None:
"""Ask the run to stop, best-effort."""
from machinable.transport import Transport
self.save_file("cancelled", data="cancelled")
Transport.get().push(self.local_directory(), include=["cancelled"])
def output_filepath(self) -> str:
"""Path of the run's ``output.log``."""
return self.local_directory("output.log")
def output(
self,
incremental: bool = False,
) -> str | None:
"""Returns the output log."""
if not self.is_mounted():
return None
p = "output.log"
if incremental:
read_length = self._cache.get(f"{p}.read_length", 0)
output = self.load_file(p, None)
if output is None:
return None
if read_length >= len(output):
return ""
self._cache[f"{p}.read_length"] = len(output)
return output[read_length:]
if p in self._cache:
return self._cache[p]
output = self.load_file(p, None)
if self.is_finished() and output is not None:
self._cache[p] = output
return output
def stream_output(
self,
refresh_every: int | float = 1,
stream=print,
):
"""Follow the run's output log until it finishes."""
try:
while not self.is_started() or self.is_active():
output = self.output(incremental=True)
if output:
stream(output)
time.sleep(refresh_every)
except KeyboardInterrupt:
pass
def update_status(
self,
status: StatusType = "heartbeat",
timestamp: TimestampType | None = None,
) -> TimestampType:
"""Write a status marker.
One of ``dispatched``/``started``/``heartbeat``/``finished``/``resumed``.
"""
_assert_allowed(status)
if timestamp is None:
timestamp = arrow.now()
if isinstance(timestamp, arrow.Arrow):
timestamp = arrow.get(timestamp)
multiple = status == "resumed"
home = self._remote_home()
if home is not None and status == "finished":
home.push(self.local_directory(), cast(str, self.uuid))
interface = getattr(self, "_interface_", None) or self.interface
if interface is not None and interface.is_materialized():
interface._write_through()
save_file(
self.local_directory(f"{status}_at"),
str(timestamp) + ("\n" if multiple else ""),
mode="a" if multiple else "w",
)
# status markers are official record writes → bump updated_at
self.touch()
if home is not None:
self._write_status_through(home, status)
return timestamp
def _write_status_through(self, home: Storage, status: str) -> None:
"""Write a status marker through to the run's remote home."""
uuid = cast(str, self.uuid)
now = time.monotonic()
if not getattr(self, "_home_pushed", False):
home.push(self.local_directory(), uuid)
self._home_pushed = True
self._last_beat_through = now
return
if status == "heartbeat":
if now - getattr(self, "_last_beat_through", 0.0) < heartbeat_window() / 2:
return
self._last_beat_through = now
home.write_through(uuid, self.local_directory(), [f"{status}_at", "updated_at"])
def refresh_status(self, force: bool = False) -> bool:
"""Bring this run's status from its remote home, if it runs elsewhere.
A run whose home is not its working copy (an object store) and that
is not finished here may be running (or have finished, or died) on
another machine, whose markers reach this working copy only through
the home. Reads the home's status markers (small files, no
artifacts), at most every ``STATUS_REFRESH_SECONDS`` unless
``force``: a newer heartbeat and markers missing here are written
into the working copy; once the home has ``finished_at``, the run is
re-ingested (its catalog and its interface's), so the working copy
and the index are complete. Nothing for a local home (it is the
working copy), a run finished here, or a run whose payload runs in
this process. Every status read calls it. Returns whether the
working copy changed.
"""
uuid = self.uuid
if uuid is None or not self.is_materialized():
return False
directory = os.path.abspath(self.local_directory())
if (
directory in _RUNNING_HERE
or not os.path.isdir(directory)
or os.path.exists(os.path.join(directory, "finished_at"))
):
return False
now = time.monotonic()
last = _STATUS_REFRESHED.get(uuid)
if not force and last is not None and now - last < STATUS_REFRESH_SECONDS:
return False
_STATUS_REFRESHED[uuid] = now
try:
home = self._remote_home()
if home is None:
_STATUS_REFRESHED.pop(uuid, None)
return False
entry = cast("IndexEntry", self._resolve_index_entry())
location = home.handle(home.location(entry))
markers = home.markers(location)
except Exception: # noqa: BLE001 - an unreachable home leaves the copy as is
return False
if "finished_at" in markers:
from machinable.claims import RunRecord
try:
adopted = home.adopt(RunRecord(uuid, location, markers))
except Exception: # noqa: BLE001 - the markers below still tell the state
adopted = False
if adopted and os.path.exists(os.path.join(directory, "finished_at")):
self._index_entry = None
self._cache.pop("output.log", None)
_STATUS_REFRESHED.pop(uuid, None)
return True
changed = False
for name, value in markers.items():
path = os.path.join(directory, name)
local = load_file(path, default=None)
if local is not None and not (
name == "heartbeat_at" and is_later(value, str(local))
):
continue
save_file(path, value)
changed = True
return changed
def retrieve_status(
self,
status: StatusType = "heartbeat",
) -> DatetimeType | None:
"""Read one status marker as a datetime, or ``None`` if unwritten."""
_assert_allowed(status)
self.refresh_status()
status_value = self.load_file(f"{status}_at", default=None)
if status_value is None:
return None
multiple = status == "resumed"
if multiple:
status_value = status_value.strip("\n").split("\n")[-1]
try:
return arrow.get(status_value)
except arrow.ParserError:
return None
def status_snapshot(self) -> ExecutionStatus:
"""Read all four status markers in one pass.
Returns an :class:`ExecutionStatus` deriving every flag. Prefer this
over the individual accessors when several are needed (e.g. when
serializing a run).
"""
# one directory resolution + one read per marker; the individual
# accessors would re-resolve the directory on every call
if not self.is_mounted():
return ExecutionStatus()
self.refresh_status()
directory = self.local_directory()
def _read(status: str) -> DatetimeType | None:
value = load_file([directory, f"{status}_at"], default=None)
if value is None:
return None
if status == "resumed":
value = value.strip("\n").split("\n")[-1]
try:
return arrow.get(value)
except arrow.ParserError:
return None
return ExecutionStatus(
dispatched_at=_read("dispatched"),
started_at=_read("started"),
resumed_at=_read("resumed"),
heartbeat_at=_read("heartbeat"),
finished_at=_read("finished"),
abandoned_at=_read("abandoned"),
window=heartbeat_window(),
)
def dispatched_at(self) -> DatetimeType | None:
"""Returns the dispatch (submission) time."""
if not self.is_mounted():
return None
return self.retrieve_status("dispatched")
def started_at(self) -> DatetimeType | None:
"""Returns the starting time."""
if not self.is_mounted():
return None
return self.retrieve_status("started")
def resumed_at(self) -> DatetimeType | None:
"""Returns the resumed time."""
if not self.is_mounted():
return None
return self.retrieve_status("resumed")
def heartbeat_at(self):
"""Returns the last heartbeat time."""
if not self.is_mounted():
return None
return self.retrieve_status("heartbeat")
def finished_at(self):
"""Returns the finishing time."""
if not self.is_mounted():
return None
return self.retrieve_status("finished")
def abandoned_at(self) -> DatetimeType | None:
"""When another dispatcher took over this run's claim, if it did."""
if not self.is_mounted():
return None
return self.retrieve_status("abandoned")
def is_abandoned(self) -> bool:
"""True if another dispatcher took over this run's claim."""
return bool(self.abandoned_at())
def is_finished(self):
"""True if finishing time has been written."""
return bool(self.finished_at())
def is_dispatched(self):
"""True if dispatch time has been written."""
return bool(self.dispatched_at())
def is_started(self):
"""True if starting time has been written."""
return bool(self.started_at())
def is_pending(self):
"""True if submitted to a scheduler but not yet running."""
return (
self.is_dispatched()
and not self.is_started()
and not self.is_finished()
and not self.is_abandoned()
)
def is_resumed(self):
"""True if resumed time has been written."""
return bool(self.resumed_at())
def is_active(self):
"""True if not finished and the heartbeat is within the liveness window."""
if self.is_finished() or self.is_abandoned():
return False
return heartbeat_is_fresh(self.heartbeat_at(), heartbeat_window())
def is_live(self):
"""True if active or finished."""
return self.is_finished() or self.is_active()
def is_incomplete(self):
"""Started (or abandoned) and no longer live."""
return (self.is_started() or self.is_abandoned()) and not self.is_live()
def on_before_dispatch(self) -> bool | None:
"""Hook: return False to abort dispatch of this execution."""
def on_seeding(self):
"""Hook: default seeds random with self.seed."""
import random
random.seed(self.seed)
# on_success / on_failure / on_finish / on_after_dispatch / on_heartbeat
# are inherited from Interface (the dispatch lifecycle contract lives on
# the base class).
def on_write_meta_data(self) -> bool | None:
"""Return False to prevent writing of meta-data."""
def on_compute_default_resources(self, interface: Interface | None) -> dict | None:
"""Event triggered to compute default resources."""
def __iter__(self):
yield from self._interfaces
def __exit__(self, *args):
if len(args) == 3 and any(map(lambda x: x is not None, args)):
super().__exit__()
elif self._defer_dispatch:
super().__exit__()
else:
try:
self.dispatch()
finally:
super().__exit__()