337 lines
9.9 KiB
Python
337 lines
9.9 KiB
Python
"""Engine holds for a rename (section 2.2 of docs/private/id_refac_rename.md).
|
|
|
|
The pipeline refuses what it would run outside the queue, and the execution
|
|
service stops claiming and its timers once claimed work has finished. Both
|
|
wait without a limit and can be aborted.
|
|
"""
|
|
|
|
import asyncio
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow import executor
|
|
from fluksio.flow.controller import FlowController, LoadedNode, RebuildBusy
|
|
from fluksio.flow.executor import ExecutionService
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.namespace import Rename
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.pipeline import EngineHeld, Pipeline
|
|
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
|
|
from fluksio.flow.state import MemoryState
|
|
from fluksio.flow.store import FlowStore
|
|
|
|
|
|
def _engine(consume, queue=None) -> tuple[Pipeline, ExecutionService, Node]:
|
|
"""A source whose readings ``consume`` gets, and a service draining the
|
|
queue one cascade at a time, so order shows."""
|
|
|
|
def consumer_body(reading, params):
|
|
consume(reading)
|
|
|
|
source = Node(
|
|
f=lambda params: None,
|
|
provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
name="source",
|
|
)
|
|
consumer = Node(
|
|
f=consumer_body,
|
|
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
name="consumer",
|
|
)
|
|
source.assign_flow("f", "source")
|
|
consumer.assign_flow("f", "consumer")
|
|
queue = queue or MemoryWorkQueue()
|
|
service = ExecutionService(queue, max_cascades=1)
|
|
pipeline = Pipeline(
|
|
nodes=[source, consumer],
|
|
state=MemoryState(),
|
|
work_queue=queue,
|
|
node_pool=service.node_pool,
|
|
)
|
|
service.bind(pipeline)
|
|
return pipeline, service, source
|
|
|
|
|
|
def _in_thread(fn, *args) -> tuple[threading.Thread, list]:
|
|
result: list = []
|
|
thread = threading.Thread(target=lambda: result.append(fn(*args)), daemon=True)
|
|
thread.start()
|
|
return thread, result
|
|
|
|
|
|
def _until(check, timeout: float = 5.0) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while not check():
|
|
assert time.monotonic() < deadline, "timed out"
|
|
time.sleep(0.02)
|
|
|
|
|
|
def _blocking(seen: list) -> tuple[threading.Event, threading.Event, object]:
|
|
"""A consumer that records and then waits until told to finish."""
|
|
running, finish = threading.Event(), threading.Event()
|
|
|
|
def consume(reading):
|
|
seen.append(reading)
|
|
running.set()
|
|
finish.wait(5)
|
|
|
|
return running, finish, consume
|
|
|
|
|
|
def test_a_hold_waits_out_a_running_cascade_without_a_limit(monkeypatch):
|
|
# What a rebuild's drain gives up after, which a hold does not.
|
|
monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
|
|
seen: list = []
|
|
running, finish, consume = _blocking(seen)
|
|
_, service, source = _engine(consume)
|
|
service.start()
|
|
try:
|
|
source.inject({"reading": 1.0})
|
|
assert running.wait(5)
|
|
busy: list = []
|
|
thread, held = _in_thread(service.hold, threading.Event(), busy.append)
|
|
thread.join(0.3)
|
|
assert thread.is_alive()
|
|
assert busy == [["f.source"]]
|
|
|
|
source.inject({"reading": 2.0})
|
|
finish.set()
|
|
thread.join(5)
|
|
assert held == [True]
|
|
time.sleep(0.2)
|
|
assert seen == [1.0]
|
|
|
|
service.release()
|
|
_until(lambda: seen == [1.0, 2.0])
|
|
finally:
|
|
finish.set()
|
|
service.stop()
|
|
|
|
|
|
class SlowClaim(MemoryWorkQueue):
|
|
"""A queue whose claim is still under way when a hold begins."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.claiming = threading.Event()
|
|
self.go = threading.Event()
|
|
|
|
def claim(self, count, block_ms, due_only=False):
|
|
items = super().claim(count, block_ms, due_only)
|
|
if items:
|
|
self.claiming.set()
|
|
self.go.wait(5)
|
|
return items
|
|
|
|
|
|
def test_pause_intake_waits_out_a_claim_already_under_way(monkeypatch):
|
|
seen: list = []
|
|
queue = SlowClaim()
|
|
_, service, source = _engine(seen.append, queue)
|
|
wake = service._capacity_changed
|
|
|
|
def closing() -> None:
|
|
wake()
|
|
# The claim lands only once the intake has closed.
|
|
queue.go.set()
|
|
|
|
monkeypatch.setattr(service, "_capacity_changed", closing)
|
|
service.start()
|
|
try:
|
|
source.inject({"reading": 1.0})
|
|
assert queue.claiming.wait(5)
|
|
service.pause_intake()
|
|
assert seen == [1.0]
|
|
assert service.inflight == 0
|
|
finally:
|
|
queue.go.set()
|
|
service.stop()
|
|
|
|
|
|
def test_a_reload_that_cannot_drain_leaves_the_pipeline_as_it_was(
|
|
tmp_path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
|
|
seen: list = []
|
|
running, finish, consume = _blocking(seen)
|
|
pipeline, service, source = _engine(consume)
|
|
controller = FlowController(FlowStore(tmp_path / "flows"), execution=service)
|
|
controller.pipeline = pipeline
|
|
service.start()
|
|
try:
|
|
source.inject({"reading": 1.0})
|
|
assert running.wait(5)
|
|
with pytest.raises(RebuildBusy):
|
|
asyncio.run(controller.reload())
|
|
assert controller.pipeline is pipeline
|
|
assert service._pipeline is pipeline
|
|
assert service._intake.is_set()
|
|
finally:
|
|
finish.set()
|
|
service.stop()
|
|
|
|
|
|
def test_a_hold_waits_out_a_claim_already_under_way():
|
|
seen: list = []
|
|
queue = SlowClaim()
|
|
_, service, source = _engine(seen.append, queue)
|
|
service.start()
|
|
try:
|
|
source.inject({"reading": 1.0})
|
|
assert queue.claiming.wait(5)
|
|
thread, held = _in_thread(service.hold, threading.Event())
|
|
thread.join(0.2)
|
|
assert thread.is_alive()
|
|
|
|
queue.go.set()
|
|
thread.join(5)
|
|
assert held == [True]
|
|
assert seen == [1.0]
|
|
assert service.inflight == 0
|
|
service.release()
|
|
finally:
|
|
queue.go.set()
|
|
service.stop()
|
|
|
|
|
|
def test_an_aborted_hold_leaves_the_engine_as_it_was():
|
|
seen: list = []
|
|
running, finish, consume = _blocking(seen)
|
|
pipeline, service, source = _engine(consume)
|
|
service.start()
|
|
try:
|
|
source.inject({"reading": 1.0})
|
|
assert running.wait(5)
|
|
abort = threading.Event()
|
|
thread, held = _in_thread(service.hold, abort)
|
|
thread.join(0.2)
|
|
assert thread.is_alive()
|
|
|
|
abort.set()
|
|
thread.join(5)
|
|
assert held == [False]
|
|
finish.set()
|
|
|
|
# Claiming and promotion carry on.
|
|
assert pipeline.defer(source, {"f.reading": 2.0}, 0.05)
|
|
source.inject({"reading": 3.0})
|
|
_until(lambda: sorted(seen) == [1.0, 2.0, 3.0])
|
|
assert not service.queue.held
|
|
finally:
|
|
finish.set()
|
|
service.stop()
|
|
|
|
|
|
def test_an_aborted_pipeline_hold_lets_publishes_through_again():
|
|
seen: list = []
|
|
running, finish, consume = _blocking(seen)
|
|
pipeline, _, _ = _engine(consume)
|
|
publishing, _ = _in_thread(pipeline.publish, {"f.reading": 1.0})
|
|
try:
|
|
assert running.wait(5)
|
|
busy: list = []
|
|
abort = threading.Event()
|
|
thread, held = _in_thread(pipeline.hold, abort, busy.append)
|
|
thread.join(0.2)
|
|
assert thread.is_alive()
|
|
assert busy == [["f.reading"]]
|
|
|
|
abort.set()
|
|
thread.join(5)
|
|
assert held == [False]
|
|
finish.set()
|
|
publishing.join(5)
|
|
pipeline.publish({"f.reading": 2.0})
|
|
assert seen == [1.0, 2.0]
|
|
finally:
|
|
finish.set()
|
|
|
|
|
|
def test_publish_is_refused_while_held():
|
|
seen: list = []
|
|
pipeline, _, _ = _engine(seen.append)
|
|
assert pipeline.hold(threading.Event())
|
|
|
|
with pytest.raises(EngineHeld):
|
|
pipeline.publish({"f.reading": 1.0})
|
|
assert "f.reading" not in pipeline.state
|
|
|
|
pipeline.release()
|
|
pipeline.publish({"f.reading": 2.0})
|
|
assert seen == [2.0]
|
|
|
|
|
|
def test_a_trigger_by_hand_while_held_is_refused_not_failed(tmp_path, monkeypatch):
|
|
seen: list = []
|
|
pipeline, _, source = _engine(seen.append)
|
|
controller = FlowController(FlowStore(tmp_path / "flows"))
|
|
controller.pipeline = pipeline
|
|
controller.loaded[source.id] = LoadedNode(id=source.id, flow="f", node=source)
|
|
failures: list = []
|
|
monkeypatch.setattr(Pipeline, "publish_error", lambda *args: failures.append(args))
|
|
assert pipeline.hold(threading.Event())
|
|
|
|
with pytest.raises(EngineHeld):
|
|
controller.trigger_node(source.id, {"reading": 1.0})
|
|
assert failures == [] and seen == []
|
|
|
|
|
|
def test_work_arriving_during_a_hold_runs_after_it_in_order():
|
|
seen: list = []
|
|
pipeline, service, source = _engine(seen.append)
|
|
service.start()
|
|
try:
|
|
abort = threading.Event()
|
|
assert pipeline.hold(abort)
|
|
assert service.hold(abort)
|
|
source.inject({"reading": 1.0})
|
|
service.queue.hold()
|
|
source.inject({"reading": 2.0})
|
|
source.inject({"reading": 3.0})
|
|
time.sleep(0.2)
|
|
assert seen == []
|
|
|
|
service.queue.release()
|
|
service.release()
|
|
pipeline.release()
|
|
_until(lambda: seen == [1.0, 2.0, 3.0])
|
|
finally:
|
|
service.stop()
|
|
|
|
|
|
def test_a_queue_failure_while_held_drops_the_value_rather_than_run_it(
|
|
monkeypatch,
|
|
):
|
|
seen: list = []
|
|
pipeline, service, source = _engine(seen.append)
|
|
|
|
def down(item):
|
|
raise ConnectionError("queue down")
|
|
|
|
monkeypatch.setattr(service.queue, "add", down)
|
|
assert pipeline.hold(threading.Event())
|
|
source.inject({"reading": 1.0})
|
|
assert seen == []
|
|
|
|
pipeline.release()
|
|
source.inject({"reading": 2.0})
|
|
assert seen == [2.0]
|
|
|
|
|
|
def test_reclaimed_work_waiting_to_run_follows_a_rename():
|
|
service = ExecutionService(MemoryWorkQueue())
|
|
service._reclaimed.extend(
|
|
[
|
|
WorkItem(kind="cascade", node="heat.valve", flow="heat", entry_id="1-0"),
|
|
WorkItem(kind="cascade", node="heating.pump", flow="heating"),
|
|
]
|
|
)
|
|
service.rename_reclaimed(Rename("heat", "warm"))
|
|
assert [(i.node, i.flow) for i in service._reclaimed] == [
|
|
("warm.valve", "warm"),
|
|
("heating.pump", "heating"),
|
|
]
|
|
assert service._reclaimed[0].entry_id == "1-0"
|