Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 34 additions & 3 deletions src/dls_tiled/visr.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,27 @@ async def _shared(key: tuple, compute):
return await asyncio.shield(task)


async def _run_progress(root, uid) -> tuple:
"""How far a run has got, as its catalog records it: whether it has a stop
document, and the shape of its camera's ``RedTotal`` (one entry per point).

Part of the key of a shared result, so a result is never reused once the run
has moved on. Without it, the poll a UI makes when a scan ends could be
given the result of the poll just before, computed without the last points.
Anything that can't be looked up counts as unknown (``None``)."""
try:
run = await root.lookup_adapter([uid])
stopped = bool((run.metadata() or {}).get("stop"))
except Exception:
return (None, None)
try:
camera = await root.lookup_adapter([uid, "primary", "RedTotal"])
shape = tuple(camera.structure().shape)
except Exception:
shape = None
return (stopped, shape)


async def _compute_and_keep(key: tuple, compute):
try:
result = await compute()
Expand Down Expand Up @@ -653,12 +674,17 @@ async def binned( # type: ignore

# A UI polls this several times a second per client while points arrive only
# every couple of seconds, so identical requests share one computation (and
# one read of the files) and its result is reused for a moment.
# one read of the files) and its result is reused for a moment, but only
# while the run is where it was (see _run_progress).
# The key leaves out who is asking. That is safe only because the data is
# read through root_tree.lookup_adapter, which bypasses Tiled's access policy
# anyway; if per-run access control is added, the key needs the principal
# (or its access tags), or one user's result could be served to another.
key = (uid, tuple(sorted(request.query_params.multi_items())))
key = (
uid,
tuple(sorted(request.query_params.multi_items())),
await _run_progress(root, uid),
)

async def compute():
output = await _binned_output(
Expand Down Expand Up @@ -1167,7 +1193,12 @@ async def render( # type: ignore
root = request.app.state.root_tree
uid = [s for s in path.strip("/").split("/") if s][0]
# Shared like /binned's results, and with the same caveat about access control.
key = ("render", uid, tuple(sorted(request.query_params.multi_items())))
key = (
"render",
uid,
tuple(sorted(request.query_params.multi_items())),
await _run_progress(root, uid),
)

async def compute():
points = await load_points(
Expand Down
57 changes: 57 additions & 0 deletions tests/test_visr.py
Original file line number Diff line number Diff line change
Expand Up @@ -636,6 +636,63 @@ async def scenario():
assert compute.calls == 2


def grow_grid_scan(client, uid, n_from, n_to):
"""Append points n_from..n_to of write_grid_scan's raster to a run, as a
scan in progress does"""
primary = client[uid]["primary"]
x = numpy.array([0.0, 1.0, 2.0] * 3)[n_from:n_to]
y = numpy.repeat([0.0, 1.0, 2.0], 3)[n_from:n_to]
primary["X"].patch(x, offset=n_from, extend=True)
primary["Y"].patch(y, offset=n_from, extend=True)
for channel in ("RedTotal", "GreenTotal", "BlueTotal"):
primary[channel].patch(
numpy.arange(9.0)[n_from:n_to], offset=n_from, extend=True
)


@pytest.mark.parametrize("route", ["binned", "render"])
def test_a_shared_result_is_not_reused_once_the_run_has_grown(
client, monkeypatch, route
):
monkeypatch.setattr(visr, "BINNED_CACHE_SECONDS", 60.0)
write_grid_scan(client, "growing", 7)
http = client.context.http_client

before = http.get(f"/api/v1/{route}/growing").json()
grow_grid_scan(client, "growing", 7, 9)
after = http.get(f"/api/v1/{route}/growing").json()

assert (before["n_points"], after["n_points"]) == (7, 9)


def test_a_shared_result_is_not_reused_once_the_run_has_stopped(client, monkeypatch):
monkeypatch.setattr(visr, "BINNED_CACHE_SECONDS", 60.0)
write_grid_scan(client, "ending", 9)
http = client.context.http_client
compute = []
real = visr._binned_output

async def counting(*args):
compute.append(1)
return await real(*args)

monkeypatch.setattr(visr, "_binned_output", counting)

http.get("/api/v1/binned/ending")
http.get("/api/v1/binned/ending") # reused
run = client["ending"]
run.update_metadata(metadata={**run.metadata, "stop": {"exit_status": "success"}})
http.get("/api/v1/binned/ending")

assert len(compute) == 2


def test_run_progress_is_unknown_for_a_run_that_is_not_there(client):
root = client.context.http_client.app.state.root_tree

assert asyncio.run(visr._run_progress(root, "nowhere")) == (None, None)


def test_binned_has_an_etag_and_answers_304_when_nothing_changed(client, monkeypatch):
monkeypatch.setattr(visr, "BINNED_CACHE_SECONDS", 0.5)
write_grid_scan(client, "grid", 9)
Expand Down
Loading