fix again on init all, hopefully
This commit is contained in:
1 parent
969475112d
commit
8b73ee2d8d
5 files changed
+542
-50
No files matched your search
@@ -4,6 +4,7 @@ import weakref
|
||||
from eco.acquisition.utilities import Acquisition
|
||||
from eco.elements.protocols import Detector, MonitorableValueUpdate, resolve_lazy
|
||||
from eco.utilities.datafiles import ensure_dir, ensure_group_writable
|
||||
from eco.utilities.utilities import is_notebook
|
||||
from collections import namedtuple
|
||||
from escape import ArrayTimestamps
|
||||
from matplotlib.animation import FuncAnimation
|
||||
@@ -18,6 +19,42 @@ DEFAULT_STORAGE_DIR = Path("./")
|
||||
StepTime = namedtuple("StepTime", "start stop")
|
||||
|
||||
|
||||
def _build_fit_icon(size=24):
|
||||
"""A small "scatter + fit line" QIcon for the toolbar button, drawn with
|
||||
QPainter instead of shipping a bitmap asset."""
|
||||
from qtpy import QtCore, QtGui
|
||||
|
||||
pixmap = QtGui.QPixmap(size, size)
|
||||
pixmap.fill(QtCore.Qt.transparent)
|
||||
painter = QtGui.QPainter(pixmap)
|
||||
painter.setRenderHint(QtGui.QPainter.Antialiasing)
|
||||
|
||||
pen = QtGui.QPen(QtGui.QColor("black"))
|
||||
pen.setWidthF(max(1.0, size / 10))
|
||||
pen.setCapStyle(QtCore.Qt.RoundCap)
|
||||
painter.setPen(pen)
|
||||
painter.drawLine(
|
||||
QtCore.QPointF(0.08 * size, 0.85 * size),
|
||||
QtCore.QPointF(0.92 * size, 0.15 * size),
|
||||
)
|
||||
|
||||
painter.setPen(QtCore.Qt.NoPen)
|
||||
painter.setBrush(QtGui.QColor("black"))
|
||||
dot_r = size * 0.06
|
||||
dots = [
|
||||
(0.10, 0.55), (0.14, 0.72), (0.18, 0.40), (0.22, 0.62), (0.26, 0.30),
|
||||
(0.30, 0.50), (0.34, 0.68), (0.38, 0.35), (0.42, 0.55), (0.46, 0.25),
|
||||
(0.54, 0.62), (0.58, 0.30), (0.62, 0.50), (0.66, 0.20),
|
||||
(0.70, 0.40), (0.74, 0.58), (0.78, 0.28), (0.82, 0.45), (0.86, 0.15),
|
||||
(0.90, 0.35), (0.20, 0.20), (0.60, 0.65), (0.35, 0.15),
|
||||
]
|
||||
for fx, fy in dots:
|
||||
painter.drawEllipse(QtCore.QPointF(fx * size, fy * size), dot_r, dot_r)
|
||||
|
||||
painter.end()
|
||||
return QtGui.QIcon(pixmap)
|
||||
|
||||
|
||||
|
||||
class CounterValue:
|
||||
def __init__(self, *detectors, name="value_counter"):
|
||||
@@ -199,6 +236,7 @@ class CounterValue:
|
||||
scan.fig = f
|
||||
if isinstance(axs, plt.Axes):
|
||||
axs = [axs]
|
||||
scan.axs = axs
|
||||
|
||||
def plotdat(n, *args):
|
||||
for ma, ax in zip(scan.monitor_scan_arrays.values(), axs):
|
||||
@@ -209,11 +247,47 @@ class CounterValue:
|
||||
fig=f, func=plotdat, cache_frame_data=False, interval=500
|
||||
)
|
||||
plt.show(block=False)
|
||||
self._add_fit_button(scan, axs)
|
||||
else:
|
||||
scan.fig.tight_layout()
|
||||
scan.fig.canvas.draw()
|
||||
scan.fig.canvas.flush_events()
|
||||
|
||||
def _add_fit_button(self, scan, axs):
|
||||
"""Add a UI trigger that opens `escape.fit_gui.AxesFitter` on demand,
|
||||
one per plotted axis -- a Qt toolbar icon if the figure has a Qt
|
||||
navigation toolbar, else an ipywidgets button if running in a
|
||||
Jupyter kernel. Nothing is added otherwise (e.g. inline backend).
|
||||
|
||||
The fit GUI draws its span selector/preview on top of `axs`, which
|
||||
the live animation clears every frame (`ax.cla()`), so starting a
|
||||
fit also stops that animation -- same effect `stopani` already has
|
||||
once the scan itself ends.
|
||||
"""
|
||||
|
||||
def start_fitters(*_):
|
||||
from escape.fit_gui import AxesFitter
|
||||
|
||||
if hasattr(scan, "animation"):
|
||||
scan.animation.event_source.stop()
|
||||
for ax in axs:
|
||||
if ax.lines:
|
||||
AxesFitter(ax)
|
||||
|
||||
toolbar = getattr(scan.fig.canvas.manager, "toolbar", None)
|
||||
if toolbar is not None and toolbar.__class__.__name__.startswith(
|
||||
"NavigationToolbar2"
|
||||
):
|
||||
action = toolbar.addAction(_build_fit_icon(), "Fit", start_fitters)
|
||||
action.setToolTip("Open interactive fit (escape.fit_gui.AxesFitter)")
|
||||
elif is_notebook():
|
||||
import ipywidgets
|
||||
from IPython.display import display
|
||||
|
||||
button = ipywidgets.Button(description="Fit", icon="line-chart")
|
||||
button.on_click(start_fitters)
|
||||
display(button)
|
||||
|
||||
def store_arrays(
|
||||
self, scan, filename="auto", directory="auto", elog=None, **kwargs
|
||||
):
|
||||
|
||||
@@ -59,6 +59,53 @@ def wait_for_enum_strs(pv, retries=10, delay=0.05):
|
||||
return pv.enum_strs
|
||||
|
||||
|
||||
def read_pv_value(adjustable, timeout=3.0, attempts=3, description=None):
|
||||
"""Read `adjustable`'s current value, treating a timed-out get as a
|
||||
failure instead of as the value `None`.
|
||||
|
||||
Why this is needed at all: pyepics' ``PV.get()`` returns ``None`` when
|
||||
``wait_for_connection()`` fails - it never raises (see
|
||||
``epics/pv.py``'s ``get_with_metadata``). eco's ``AdjustablePv`` builds
|
||||
every PV with ``connection_timeout=0.05`` and its readback with
|
||||
``auto_monitor=False``, so a read issued shortly after construction has
|
||||
a 50 ms budget on a brand-new channel and no monitor cache to fall back
|
||||
on. During a concurrent ``Namespace.init_all()`` pass - thousands of
|
||||
channels connecting at once - that budget is regularly missed, and the
|
||||
``None`` then flows on as if it were data: an event code of ``None``, a
|
||||
pulser number of ``None``. The resulting failure surfaces far from its
|
||||
cause, or does not surface at all (see ``EvrPulser``/``EvrOutput``).
|
||||
|
||||
This is the same failure shape as `wait_for_enum_strs` above (a value
|
||||
that is simply not there yet under load), and the same shape as the
|
||||
``Daq.get_pulse_id()`` fix - both boil down to "never let a timed-out
|
||||
CA get become a value".
|
||||
|
||||
Uses the underlying PV directly so a real timeout can be given, rather
|
||||
than inheriting the 50 ms connection timeout. Raises ``TimeoutError``
|
||||
naming the PV once the attempts are used up, so the caller fails with
|
||||
something readable instead of an ``AttributeError`` three frames later.
|
||||
"""
|
||||
pv = getattr(adjustable, "_pvreadback", None)
|
||||
if pv is None:
|
||||
pv = getattr(adjustable, "_pv", None)
|
||||
what = description or getattr(adjustable, "name", None) or repr(adjustable)
|
||||
pvname = getattr(pv, "pvname", None) or getattr(adjustable, "pvname", what)
|
||||
for _ in range(max(int(attempts), 1)):
|
||||
if pv is not None:
|
||||
pv.wait_for_connection(timeout=timeout)
|
||||
value = pv.get(timeout=timeout)
|
||||
else:
|
||||
value = adjustable.get_current_value()
|
||||
if value is not None:
|
||||
return value
|
||||
raise TimeoutError(
|
||||
f"could not read '{what}' ({pvname}) after {attempts} attempts of "
|
||||
f"{timeout}s each - the channel did not connect or did not return a "
|
||||
f"value. pyepics reports this as None rather than raising, so it is "
|
||||
f"raised here to keep it from being used as if it were data."
|
||||
)
|
||||
|
||||
|
||||
# Work in progress! TODO
|
||||
@spec_convenience
|
||||
@get_from_archive
|
||||
|
||||
@@ -1,7 +1,12 @@
|
||||
from epics import caget_many
|
||||
from ..elements.adjustable import AdjustableMemory, AdjustableVirtual
|
||||
from ..elements.detector import DetectorVirtual
|
||||
from ..epics_utils.adjustable import AdjustablePv, AdjustablePvEnum, AdjustablePvString
|
||||
from ..epics_utils.adjustable import (
|
||||
AdjustablePv,
|
||||
AdjustablePvEnum,
|
||||
AdjustablePvString,
|
||||
read_pv_value,
|
||||
)
|
||||
from ..epics_utils.detector import DetectorPvData, DetectorPvDataStream
|
||||
from ..detector.detectors_psi import DetectorBsStream
|
||||
from eco.epics_utils.utilities_epics import EpicsString
|
||||
@@ -11,6 +16,8 @@ from ..utilities.tables import format_table
|
||||
|
||||
logging.getLogger("cta_lib").setLevel(logging.WARNING)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TimingSystem(Assembly):
|
||||
"""This is a wrapper object for the global timing system at SwissFEL"""
|
||||
@@ -210,9 +217,46 @@ class MasterEventSystem(Assembly):
|
||||
)
|
||||
self.event_codes[code] = self.__dict__[f"code{code:03d}"]
|
||||
|
||||
def _get_slot_codes(self, slots=range(1, 257)):
|
||||
def _get_slot_codes(self, slots=range(1, 257), attempts=3, timeout=3.0):
|
||||
"""Read the master's slot->event-code table.
|
||||
|
||||
`caget_many` reports a slot it could not read as `None`, and those
|
||||
used to be dropped silently. Whichever init worker builds this
|
||||
component does so while the rest of the pass saturates channel
|
||||
access, so under `init_all()` a handful of slots could time out and
|
||||
the resulting `event_codes` dict stayed permanently incomplete for
|
||||
the life of the session - after which every EVR pulser wired to a
|
||||
dropped code came up without its delay/frequency chain, blaming "code
|
||||
missing in Timing Master" for what was really a timed-out read.
|
||||
Retry the missing ones (they are a handful, so this is cheap) and say
|
||||
so if any are still missing, rather than quietly shipping a partial
|
||||
table. Dependency-ordered init makes this much less likely to trigger
|
||||
in the first place, but the silent-drop path is worth closing anyway.
|
||||
"""
|
||||
slots = list(slots)
|
||||
pvs = [f"{self.pvname}:Evt-{slot}-Code-SP" for slot in slots]
|
||||
codes = caget_many(pvs)
|
||||
codes = list(caget_many(pvs, timeout=timeout))
|
||||
|
||||
for _ in range(max(int(attempts) - 1, 0)):
|
||||
missing = [i for i, c in enumerate(codes) if c is None]
|
||||
if not missing:
|
||||
break
|
||||
retried = caget_many([pvs[i] for i in missing], timeout=timeout)
|
||||
for i, c in zip(missing, retried):
|
||||
codes[i] = c
|
||||
still_missing = [slots[i] for i, c in enumerate(codes) if c is None]
|
||||
if still_missing:
|
||||
logger.warning(
|
||||
"timing master %s: %d of %d event-code slots could not be "
|
||||
"read after %d attempts (slots %s%s); event codes served by "
|
||||
"them will look missing to every EVR pulser using them.",
|
||||
self.pvname,
|
||||
len(still_missing),
|
||||
len(slots),
|
||||
attempts,
|
||||
", ".join(str(x) for x in still_missing[:10]),
|
||||
", ..." if len(still_missing) > 10 else "",
|
||||
)
|
||||
|
||||
slots_out = []
|
||||
codes_out = []
|
||||
@@ -326,16 +370,26 @@ class EvrPulser(Assembly):
|
||||
)
|
||||
self.description = EpicsString(pv_base + "-Name-I")
|
||||
|
||||
if self._eventcode is not None:
|
||||
# Resolve the event code ONCE and keep it. It used to be a property
|
||||
# re-issuing a CA get on every access, and __init__ touched it three
|
||||
# times (the `is not None` guard, then .frequency, then .delay): under
|
||||
# a concurrent init_all() the guard could pass and the next access
|
||||
# come back None (timed-out get -> None -> KeyError -> None), giving
|
||||
# `AttributeError: 'NoneType' object has no attribute 'frequency'` on
|
||||
# a pulser that initializes fine on its own. One read, one value, no
|
||||
# window between the check and the use.
|
||||
self._eventcode_resolved = self._resolve_eventcode()
|
||||
|
||||
if self._eventcode_resolved is not None:
|
||||
self._append(
|
||||
DetectorVirtual,
|
||||
[self._eventcode.frequency],
|
||||
[self._eventcode_resolved.frequency],
|
||||
lambda x: x,
|
||||
name="frequency",
|
||||
)
|
||||
self._append(
|
||||
DetectorVirtual,
|
||||
[self._eventcode.delay],
|
||||
[self._eventcode_resolved.delay],
|
||||
lambda x: x,
|
||||
name="delay_eventcode",
|
||||
)
|
||||
@@ -347,17 +401,55 @@ class EvrPulser(Assembly):
|
||||
name="delay",
|
||||
)
|
||||
else:
|
||||
print(f"Error initializing pulser {self.name} of EVR {self.pv_base}: Event code {self.eventcode.get_current_value()} is missing in Timing Master")
|
||||
logger.warning(
|
||||
"pulser %s of EVR %s: event code %s is missing in the timing "
|
||||
"master; delay/frequency are not available on it.",
|
||||
self.name,
|
||||
self.pv_base,
|
||||
self._eventcode_number,
|
||||
)
|
||||
|
||||
def _resolve_eventcode(self, timeout=3.0, attempts=3):
|
||||
"""The timing-master event code object this pulser is wired to.
|
||||
|
||||
The read is done through `read_pv_value` rather than
|
||||
`get_current_value()` so a channel that has not connected yet raises
|
||||
instead of yielding `None` - a `None` here used to be looked up in
|
||||
`event_codes`, miss, and silently leave the pulser without its
|
||||
delay/frequency chain, which then broke every EvrOutput pointing at
|
||||
it. Returns None only when the code is genuinely absent from the
|
||||
master.
|
||||
"""
|
||||
try:
|
||||
self._eventcode_number = read_pv_value(
|
||||
self.eventcode,
|
||||
timeout=timeout,
|
||||
attempts=attempts,
|
||||
description=f"event code of pulser {self.name} ({self.pv_base})",
|
||||
)
|
||||
except TimeoutError as e:
|
||||
# Degrade the way this always did (no delay/frequency chain)
|
||||
# rather than failing the whole pulser - a PV that genuinely
|
||||
# isn't there must not turn into a wall of new failures. The
|
||||
# difference is that it now says so instead of silently looking
|
||||
# up event code `None` and reporting it as missing from the
|
||||
# master.
|
||||
logger.warning("%s", e)
|
||||
self._eventcode_number = None
|
||||
return None
|
||||
if self._eventcode_number == 27:
|
||||
return self._parent_evr.sequencer
|
||||
return self._event_master.event_codes.get(self._eventcode_number)
|
||||
|
||||
def update_eventcode(self):
|
||||
"""Re-read the event code from the IOC (it is otherwise resolved once,
|
||||
at construction). Use after re-wiring the pulser externally."""
|
||||
self._eventcode_resolved = self._resolve_eventcode()
|
||||
return self._eventcode_resolved
|
||||
|
||||
@property
|
||||
def _eventcode(self):
|
||||
if self.eventcode.get_current_value() == 27:
|
||||
return self._parent_evr.sequencer
|
||||
else:
|
||||
try:
|
||||
return self._event_master.event_codes[self.eventcode.get_current_value()]
|
||||
except KeyError:
|
||||
return None
|
||||
return self._eventcode_resolved
|
||||
|
||||
|
||||
class DummyPulser(Assembly):
|
||||
@@ -398,6 +490,8 @@ class EvrOutput(Assembly):
|
||||
name="pulserA_number",
|
||||
is_setting=True,
|
||||
)
|
||||
# resolve once, before the eight virtuals below reach for it
|
||||
self._pulserA_resolved = self._resolve_pulser(self.pulserA_number, "pulserA")
|
||||
self._append(
|
||||
AdjustableVirtual,
|
||||
[self.pulserA.delay],
|
||||
@@ -466,6 +560,9 @@ class EvrOutput(Assembly):
|
||||
name="pulserB_number",
|
||||
is_setting=True,
|
||||
)
|
||||
# NB: this used to read pulserA_number - every pulserB_* virtual on
|
||||
# every output was silently tracking pulser A.
|
||||
self._pulserB_resolved = self._resolve_pulser(self.pulserB_number, "pulserB")
|
||||
self._append(
|
||||
AdjustableVirtual,
|
||||
[self.pulserB.delay],
|
||||
@@ -521,19 +618,62 @@ class EvrOutput(Assembly):
|
||||
name="pulserB_width",
|
||||
)
|
||||
|
||||
def _resolve_pulser(self, number_adj, which, timeout=3.0, attempts=3):
|
||||
"""The pulser object this output is wired to, read once.
|
||||
|
||||
Two bugs used to live here. The read went through
|
||||
`get_current_value()`, which returns None on a timed-out CA get, and
|
||||
`self._pulsers[None]` raised TypeError, which was caught and turned
|
||||
into a *fresh* `DummyPulser()`. And because this was a property,
|
||||
`__init__` re-read it once per virtual it builds - eight times for
|
||||
pulserA, eight for pulserB - so under a concurrent init_all() a
|
||||
single output could end up with its eight `pulserA_*` virtuals bound
|
||||
to several different objects, some of them throwaway dummies whose
|
||||
values are permanently None. That mis-wiring never failed and never
|
||||
printed anything.
|
||||
|
||||
`read_pv_value` raises rather than yielding None, and the result is
|
||||
resolved once and cached; an out-of-range number still degrades to a
|
||||
DummyPulser, since that is a real (if odd) IOC state rather than a
|
||||
transport failure.
|
||||
"""
|
||||
try:
|
||||
number = read_pv_value(
|
||||
number_adj,
|
||||
timeout=timeout,
|
||||
attempts=attempts,
|
||||
description=f"{which} number of output {self.name} ({self.pv_base})",
|
||||
)
|
||||
except TimeoutError as e:
|
||||
logger.warning("%s", e)
|
||||
number = None
|
||||
try:
|
||||
return self._pulsers[number]
|
||||
except (IndexError, TypeError):
|
||||
logger.warning(
|
||||
"output %s (%s): %s number %r does not address any of the %d "
|
||||
"pulsers of this EVR; using a dummy pulser.",
|
||||
self.name,
|
||||
self.pv_base,
|
||||
which,
|
||||
number,
|
||||
len(self._pulsers or ()),
|
||||
)
|
||||
return DummyPulser()
|
||||
|
||||
def update_pulsers(self):
|
||||
"""Re-read which pulsers this output is wired to (they are otherwise
|
||||
resolved once, at construction). Use after re-wiring the output."""
|
||||
self._pulserA_resolved = self._resolve_pulser(self.pulserA_number, "pulserA")
|
||||
self._pulserB_resolved = self._resolve_pulser(self.pulserB_number, "pulserB")
|
||||
|
||||
@property
|
||||
def pulserA(self):
|
||||
try:
|
||||
return self._pulsers[self.pulserA_number.get_current_value()]
|
||||
except (IndexError, TypeError):
|
||||
return DummyPulser()
|
||||
return self._pulserA_resolved
|
||||
|
||||
@property
|
||||
def pulserB(self):
|
||||
try:
|
||||
return self._pulsers[self.pulserA_number.get_current_value()]
|
||||
except (IndexError, TypeError):
|
||||
return DummyPulser()
|
||||
return self._pulserB_resolved
|
||||
|
||||
|
||||
class EventReceiver(Assembly):
|
||||
|
||||
+128
-28
@@ -837,6 +837,14 @@ class Namespace(Assembly):
|
||||
self._initialisation_start_time = {}
|
||||
self._init_priority = {}
|
||||
|
||||
# name -> [NamespaceComponent, ...] found among the args/kwargs it
|
||||
# was registered with, i.e. the other namespace items it declares a
|
||||
# dependency on. Recorded at append_obj time (the only moment the
|
||||
# arguments are visible as objects rather than already-substituted
|
||||
# proxies) so init_all can order the pass by dependency instead of
|
||||
# letting every consumer race to build the same shared component.
|
||||
self._declared_dependencies = {}
|
||||
|
||||
# name -> {"module_name": str|None, "obj_factory": callable|str}, as
|
||||
# passed to append_obj. Lets reinitialize(reload_modules=True) find
|
||||
# which module to importlib.reload() for a given name; see
|
||||
@@ -1381,6 +1389,71 @@ class Namespace(Assembly):
|
||||
print(f"Captured init log no longer exists at {path}.")
|
||||
return ""
|
||||
|
||||
def get_dependencies(self, name):
|
||||
"""Names this item declared a dependency on, via a
|
||||
`NamespaceComponent` among its `append_obj` arguments.
|
||||
|
||||
Only *declared* dependencies are visible here. A component that
|
||||
reaches into the namespace from inside its own `__init__` (rather
|
||||
than being handed a NamespaceComponent) is invisible to this, which
|
||||
is why dependency ordering is a scheduling hint and not a guarantee -
|
||||
see `_dependency_layers`.
|
||||
"""
|
||||
out = set()
|
||||
for dep in self._declared_dependencies.get(name, ()):
|
||||
try:
|
||||
dep._resolve()
|
||||
except KeyError:
|
||||
# points at something never registered - not our problem
|
||||
# here, it will fail loudly when the item is actually built.
|
||||
continue
|
||||
if dep.obj_name is not None and dep.obj_name != name:
|
||||
out.add(dep.obj_name)
|
||||
return out
|
||||
|
||||
def _dependency_layers(self, names, log=None):
|
||||
"""Split `names` into successive layers such that everything a name
|
||||
declares a dependency on is in an earlier layer.
|
||||
|
||||
The point is the shared-dependency stampede: at Bernina all nine
|
||||
EventReceivers take the same `event_master` NamespaceComponent, so a
|
||||
flat concurrent pass has up to `max_workers` of them reaching for it
|
||||
at once. One wins and builds it (a 256-PV `caget_many`); the rest sit
|
||||
in the lazy-init wait until `init_timeout` expires and come back as
|
||||
IsInitialisingError, to be retried a cycle later - which is both slow
|
||||
and, because all of that happens while channel access is saturated,
|
||||
the condition under which the EVRs' own reads start timing out.
|
||||
Building `event_master` in an earlier layer removes the collision
|
||||
entirely.
|
||||
|
||||
Dependencies pointing outside `names` (already initialized, excluded,
|
||||
or never registered) impose no constraint. A dependency cycle cannot
|
||||
be layered, so the members of the cycle are emitted together as one
|
||||
layer and left to the existing retry loop - the same behaviour as
|
||||
before this method existed, applied to just the cyclic group.
|
||||
"""
|
||||
remaining = set(names)
|
||||
edges = {n: self.get_dependencies(n) & remaining for n in remaining}
|
||||
layers = []
|
||||
placed = set()
|
||||
while remaining:
|
||||
ready = {n for n in remaining if not (edges[n] - placed)}
|
||||
if not ready:
|
||||
# cycle (or a dependency on something that can never be
|
||||
# placed): stop layering and hand the rest over as one group.
|
||||
if log is not None:
|
||||
log(
|
||||
f"dependency cycle among {len(remaining)} name(s), "
|
||||
f"initializing them together: "
|
||||
+ ", ".join(sorted(remaining))
|
||||
)
|
||||
layers.append(set(remaining))
|
||||
break
|
||||
layers.append(ready)
|
||||
placed |= ready
|
||||
remaining -= ready
|
||||
return layers
|
||||
|
||||
def _select_names_to_init(self, required_only, exclude_names, log):
|
||||
"""Which names this init_all() call needs to build, honoring
|
||||
required_only/exclude_names; warns (via `log`) about pre-existing
|
||||
@@ -1468,37 +1541,59 @@ class Namespace(Assembly):
|
||||
init_fn()
|
||||
|
||||
first_exception = None
|
||||
pending = set(names_to_init)
|
||||
cycles_left = max(int(N_cycles), 1)
|
||||
# Dependency-ordered: a name is only submitted once everything it
|
||||
# declared a NamespaceComponent dependency on has been built, so
|
||||
# consumers of a shared component no longer race each other to build
|
||||
# it. Names with no declared dependencies all land in the first
|
||||
# layer, so this costs essentially no parallelism - at Bernina it
|
||||
# moves the nine EventReceivers behind `event_master` and leaves the
|
||||
# other ~70 components exactly as concurrent as before.
|
||||
layers = self._dependency_layers(names_to_init, log=log)
|
||||
if len(layers) > 1:
|
||||
log(
|
||||
f"Initializing in {len(layers)} dependency layers: "
|
||||
+ ", ".join(str(len(la)) for la in layers)
|
||||
+ " name(s)"
|
||||
)
|
||||
givenup = set()
|
||||
with cap, ThreadPoolExecutor(
|
||||
max_workers=max_workers, initializer=_thread_initializer
|
||||
) as exc:
|
||||
while pending and cycles_left:
|
||||
cycles_left -= 1
|
||||
futs = {
|
||||
exc.submit(
|
||||
self.init_name,
|
||||
name,
|
||||
verbose=verbose,
|
||||
raise_errors=True,
|
||||
quiet=bool(silent),
|
||||
): name
|
||||
for name in pending
|
||||
}
|
||||
retry = set()
|
||||
for fut in as_completed(futs):
|
||||
name = futs[fut]
|
||||
try:
|
||||
fut.result()
|
||||
except IsInitialisingError:
|
||||
# Lost a race for a name someone else (this pass
|
||||
# itself, or the session directly) is already
|
||||
# initializing as a dependency; retry next round.
|
||||
retry.add(name)
|
||||
except Exception as exc_:
|
||||
if first_exception is None:
|
||||
first_exception = exc_
|
||||
pending = retry & (self.all_names - self.initialized_names)
|
||||
for layer in layers:
|
||||
pending = set(layer)
|
||||
cycles_left = max(int(N_cycles), 1)
|
||||
while pending and cycles_left:
|
||||
cycles_left -= 1
|
||||
futs = {
|
||||
exc.submit(
|
||||
self.init_name,
|
||||
name,
|
||||
verbose=verbose,
|
||||
raise_errors=True,
|
||||
quiet=bool(silent),
|
||||
): name
|
||||
for name in pending
|
||||
}
|
||||
retry = set()
|
||||
for fut in as_completed(futs):
|
||||
name = futs[fut]
|
||||
try:
|
||||
fut.result()
|
||||
except IsInitialisingError:
|
||||
# Lost a race for a name someone else (this pass
|
||||
# itself, or the session directly) is already
|
||||
# initializing as a dependency; retry next round.
|
||||
retry.add(name)
|
||||
except Exception as exc_:
|
||||
if first_exception is None:
|
||||
first_exception = exc_
|
||||
pending = retry & (self.all_names - self.initialized_names)
|
||||
# Whatever this layer could not finish is not allowed to hold
|
||||
# up the layers behind it: the dependents are attempted
|
||||
# anyway (they may not really need it), exactly as they were
|
||||
# before layering existed.
|
||||
givenup |= pending
|
||||
pending = givenup
|
||||
|
||||
if pending:
|
||||
# A warning rather than log(): log() is silenced by silent=True,
|
||||
@@ -1985,6 +2080,11 @@ class Namespace(Assembly):
|
||||
"module_name": module_name,
|
||||
"obj_factory": obj_factory,
|
||||
}
|
||||
self._declared_dependencies[name] = [
|
||||
a
|
||||
for a in list(args) + list(kwargs.values())
|
||||
if isinstance(a, NamespaceComponent)
|
||||
]
|
||||
if lazy:
|
||||
|
||||
def init_local():
|
||||
|
||||
@@ -317,3 +317,134 @@ def test_output_capture_survives_a_thread_that_outlives_the_pass():
|
||||
cap._write_sink("late output from a straggler thread")
|
||||
|
||||
assert "late output" in real.getvalue()
|
||||
|
||||
|
||||
class _DependentThing:
|
||||
def __init__(self, dep=None, name=None):
|
||||
self.name = name
|
||||
self.dep = dep
|
||||
|
||||
|
||||
def test_declared_dependencies_are_recorded_and_layered():
|
||||
"""init_all used to submit every name flat, so all thirteen consumers of
|
||||
bernina's `event_master` reached for it at once: one built it, the rest
|
||||
waited out init_timeout and came back as IsInitialisingError. Names are
|
||||
now ordered behind what they declare a NamespaceComponent dependency on.
|
||||
"""
|
||||
from eco.utilities.config import NamespaceComponent
|
||||
|
||||
ns = Namespace(name="test")
|
||||
ns.append_obj(_DependentThing, lazy=True, name="master")
|
||||
for n in range(3):
|
||||
ns.append_obj(
|
||||
_DependentThing,
|
||||
NamespaceComponent(ns, "master"),
|
||||
lazy=True,
|
||||
name=f"consumer{n}",
|
||||
)
|
||||
ns.append_obj(_DependentThing, lazy=True, name="unrelated")
|
||||
# a sub-attribute reference still resolves to the registered item
|
||||
ns.append_obj(
|
||||
_DependentThing,
|
||||
NamespaceComponent(ns, "master.some.child"),
|
||||
lazy=True,
|
||||
name="deep",
|
||||
)
|
||||
|
||||
assert ns.get_dependencies("consumer0") == {"master"}
|
||||
assert ns.get_dependencies("deep") == {"master"}
|
||||
assert ns.get_dependencies("master") == set()
|
||||
|
||||
layers = ns._dependency_layers(ns.all_names)
|
||||
assert layers[0] == {"master", "unrelated"}
|
||||
assert layers[1] == {"consumer0", "consumer1", "consumer2", "deep"}
|
||||
|
||||
|
||||
def test_init_all_builds_a_dependency_before_its_consumers(tmp_path):
|
||||
from eco.utilities.config import NamespaceComponent
|
||||
|
||||
ns = Namespace(name="test")
|
||||
ns.append_obj(_DependentThing, lazy=True, name="master")
|
||||
for n in range(4):
|
||||
ns.append_obj(
|
||||
_DependentThing,
|
||||
NamespaceComponent(ns, "master"),
|
||||
lazy=True,
|
||||
name=f"consumer{n}",
|
||||
)
|
||||
|
||||
order = []
|
||||
original = ns.init_name
|
||||
|
||||
def spy(name, **kwargs):
|
||||
order.append(name)
|
||||
return original(name, **kwargs)
|
||||
|
||||
ns.init_name = spy
|
||||
ns.init_all(
|
||||
required_only=False,
|
||||
background=False,
|
||||
silent=str(tmp_path / "i.log"),
|
||||
max_workers=4,
|
||||
)
|
||||
|
||||
assert order[0] == "master"
|
||||
assert set(order[1:]) == {f"consumer{n}" for n in range(4)}
|
||||
|
||||
|
||||
def test_dependency_cycle_does_not_hang_and_is_emitted_as_one_layer():
|
||||
from eco.utilities.config import NamespaceComponent
|
||||
|
||||
ns = Namespace(name="test")
|
||||
ns.append_obj(
|
||||
_DependentThing, NamespaceComponent(ns, "b"), lazy=True, name="a"
|
||||
)
|
||||
ns.append_obj(
|
||||
_DependentThing, NamespaceComponent(ns, "a"), lazy=True, name="b"
|
||||
)
|
||||
|
||||
assert ns._dependency_layers(ns.all_names) == [{"a", "b"}]
|
||||
|
||||
|
||||
def test_read_pv_value_retries_and_refuses_to_return_none():
|
||||
"""pyepics' PV.get() returns None on a timed-out get instead of raising,
|
||||
and eco's AdjustablePv gives every channel a 50 ms connection budget with
|
||||
auto_monitor=False on the readback - so under a concurrent init_all() a
|
||||
None routinely flowed on as if it were data (an event code of None, a
|
||||
pulser number of None)."""
|
||||
import pytest
|
||||
|
||||
from eco.epics_utils.adjustable import read_pv_value
|
||||
|
||||
class FakePV:
|
||||
def __init__(self, values):
|
||||
self.values = list(values)
|
||||
self.pvname = "FAKE:PV"
|
||||
self.gets = 0
|
||||
|
||||
def wait_for_connection(self, timeout=None):
|
||||
return True
|
||||
|
||||
def get(self, timeout=None):
|
||||
self.gets += 1
|
||||
return self.values.pop(0) if self.values else None
|
||||
|
||||
class FakeAdjustable:
|
||||
def __init__(self, pv):
|
||||
self._pvreadback = pv
|
||||
self.name = "fake"
|
||||
|
||||
# a transient timeout is retried, not taken at face value
|
||||
adj = FakeAdjustable(FakePV([None, None, 27]))
|
||||
assert read_pv_value(adj, timeout=0.01, attempts=3) == 27
|
||||
assert adj._pvreadback.gets == 3
|
||||
|
||||
# a persistent one raises with the PV named, instead of returning None
|
||||
adj = FakeAdjustable(FakePV([]))
|
||||
with pytest.raises(TimeoutError) as excinfo:
|
||||
read_pv_value(adj, timeout=0.01, attempts=2)
|
||||
assert "FAKE:PV" in str(excinfo.value)
|
||||
|
||||
# a legitimate falsy value is a value, not a failure
|
||||
adj = FakeAdjustable(FakePV([0]))
|
||||
assert read_pv_value(adj, timeout=0.01, attempts=2) == 0
|
||||
Reference in new issue
Block a user