Melvin Strobl
8be7f9770e
The submodule collapse was only half applied: .gitmodules was deleted but backend/ and frontend/ were still recorded as gitlinks, so none of their files were tracked. Replace the gitlinks with the real trees. Also untrack .env (it carried placeholder secrets) in favour of a tracked .env.example, drop the committed __pycache__, and narrow the blanket *.png ignore that would have swallowed design assets.
513 lines
16 KiB
Python
513 lines
16 KiB
Python
"""
|
|
Test synchronous node execution in pipelines.
|
|
|
|
This module tests the "synchronous" flag which ensures a node only executes
|
|
when ALL its inputs have changed since the last execution.
|
|
|
|
Synchronous nodes are useful for:
|
|
- Aggregation nodes that need all inputs to be "fresh"
|
|
- Nodes that should only process complete "batches" of data
|
|
- Avoiding redundant executions when multiple inputs update at different rates
|
|
"""
|
|
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# Add flow directory to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from state import MemoryState, RedisState
|
|
from pipeline import Pipeline
|
|
from nodes import Node
|
|
from util import Message
|
|
|
|
|
|
def test_synchronous_basic():
|
|
"""Test basic synchronous node behavior."""
|
|
print("=" * 60)
|
|
print("Testing Basic Synchronous Node Behavior")
|
|
print("=" * 60)
|
|
|
|
execution_log = []
|
|
|
|
def sensor_a(params, **kwargs):
|
|
value = params.get("value", 1.0)
|
|
execution_log.append(f"sensor_a -> {value}")
|
|
return {"temp_a": value}
|
|
|
|
def sensor_b(params, **kwargs):
|
|
value = params.get("value", 2.0)
|
|
execution_log.append(f"sensor_b -> {value}")
|
|
return {"temp_b": value}
|
|
|
|
def sync_processor(params, temp_a=0, temp_b=0, **kwargs):
|
|
result = temp_a + temp_b
|
|
execution_log.append(f"run_sync_processor({temp_a}, {temp_b}) -> {result}")
|
|
return {"combined": result}
|
|
|
|
def async_processor(params, temp_a=0, temp_b=0, **kwargs):
|
|
result = temp_a * temp_b
|
|
execution_log.append(f"run_async_processor({temp_a}, {temp_b}) -> {result}")
|
|
return {"product": result}
|
|
|
|
# Create nodes
|
|
node_a = Node(
|
|
f=sensor_a,
|
|
requires=[],
|
|
provides=[Message(name="temp_a", dtype=float)],
|
|
params={"value": 10.0},
|
|
name="sensor_a",
|
|
)
|
|
|
|
node_b = Node(
|
|
f=sensor_b,
|
|
requires=[],
|
|
provides=[Message(name="temp_b", dtype=float)],
|
|
params={"value": 20.0},
|
|
name="sensor_b",
|
|
)
|
|
|
|
# Synchronous node - only executes when BOTH inputs change
|
|
sync_node = Node(
|
|
f=sync_processor,
|
|
requires=[
|
|
Message(name="temp_a", dtype=float),
|
|
Message(name="temp_b", dtype=float),
|
|
],
|
|
provides=[Message(name="combined", dtype=float)],
|
|
params={"synchronous": True},
|
|
name="sync_processor",
|
|
)
|
|
|
|
# Non-synchronous node - executes whenever any input is available
|
|
async_node = Node(
|
|
f=async_processor,
|
|
requires=[
|
|
Message(name="temp_a", dtype=float),
|
|
Message(name="temp_b", dtype=float),
|
|
],
|
|
provides=[Message(name="product", dtype=float)],
|
|
params={"synchronous": False},
|
|
name="async_processor",
|
|
)
|
|
|
|
# Build pipeline
|
|
state = MemoryState()
|
|
pipeline = Pipeline(
|
|
nodes=[node_a, node_b, sync_node, async_node],
|
|
state=state,
|
|
max_workers=1, # Single worker for predictable ordering
|
|
)
|
|
|
|
print("\n--- Triggering sensor_a (first time) ---")
|
|
execution_log.clear()
|
|
node_a.inject({})
|
|
print(f"Execution log: {execution_log}")
|
|
print(f"State: temp_a={state.get('temp_a')}, temp_b={state.get('temp_b')}")
|
|
|
|
# Neither processor should run yet (temp_b missing)
|
|
assert "run_async_processor" not in str(
|
|
execution_log
|
|
), "async_processor ran too early"
|
|
assert "run_sync_processor" not in str(
|
|
execution_log
|
|
), "sync_processor ran too early"
|
|
print("✓ Neither processor ran (temp_b not yet available)")
|
|
|
|
print("\n--- Triggering sensor_b (first time) ---")
|
|
execution_log.clear()
|
|
node_b.inject({})
|
|
print(f"Execution log: {execution_log}")
|
|
|
|
# Both should run now (first time both inputs are available)
|
|
assert "run_async_processor" in str(
|
|
execution_log
|
|
), "async_processor should have run"
|
|
assert "run_sync_processor" in str(execution_log), "sync_processor should have run"
|
|
print("✓ Both processors ran (first time both inputs available)")
|
|
|
|
print("\n--- Triggering sensor_a (second time) ---")
|
|
node_a.params["value"] = 15.0
|
|
execution_log.clear()
|
|
node_a.inject({})
|
|
print(f"Execution log: {execution_log}")
|
|
|
|
# async_processor SHOULD run (any input change triggers it)
|
|
# sync_processor should NOT run (only temp_a changed, not temp_b)
|
|
assert "run_async_processor" in str(
|
|
execution_log
|
|
), "async_processor should have run"
|
|
assert "run_sync_processor" not in str(
|
|
execution_log
|
|
), "sync_processor should NOT have run (only temp_a changed)"
|
|
print("✓ Only async_processor ran (temp_b hasn't changed)")
|
|
|
|
print("\n--- Triggering sensor_b (second time) ---")
|
|
node_b.params["value"] = 25.0
|
|
execution_log.clear()
|
|
node_b.inject({})
|
|
print(f"Execution log: {execution_log}")
|
|
|
|
# Both should run now (both inputs have changed since last sync execution)
|
|
assert "run_async_processor" in str(
|
|
execution_log
|
|
), "async_processor should have run"
|
|
assert "run_sync_processor" in str(
|
|
execution_log
|
|
), "sync_processor should have run (both inputs changed)"
|
|
print("✓ Both processors ran (both inputs have changed)")
|
|
|
|
# Verify final state values
|
|
assert (
|
|
pipeline.state.get("combined") == 40.0
|
|
), f"Expected combined=40.0, got {pipeline.state.get('combined')}"
|
|
assert (
|
|
pipeline.state.get("product") == 375.0
|
|
), f"Expected product=375.0, got {pipeline.state.get('product')}"
|
|
print(
|
|
f"\n✓ Final state: combined={pipeline.state.get('combined')}, product={pipeline.state.get('product')}"
|
|
)
|
|
|
|
print("\n✓ Basic synchronous test passed!\n")
|
|
|
|
|
|
def test_synchronous_multiple_triggers():
|
|
"""Test that synchronous nodes handle rapid successive triggers correctly."""
|
|
print("=" * 60)
|
|
print("Testing Synchronous Node with Multiple Rapid Triggers")
|
|
print("=" * 60)
|
|
|
|
sync_execution_count = {"count": 0}
|
|
async_execution_count = {"count": 0}
|
|
|
|
def source_x(params, **kwargs):
|
|
return {"x": params.get("value", 1)}
|
|
|
|
def source_y(params, **kwargs):
|
|
return {"y": params.get("value", 2)}
|
|
|
|
def sync_consumer(params, x=0, y=0, **kwargs):
|
|
sync_execution_count["count"] += 1
|
|
return {"sync_out": x + y}
|
|
|
|
def async_consumer(params, x=0, y=0, **kwargs):
|
|
async_execution_count["count"] += 1
|
|
return {"async_out": x * y}
|
|
|
|
node_x = Node(
|
|
f=source_x,
|
|
requires=[],
|
|
provides=[Message(name="x", dtype=int)],
|
|
params={"value": 1},
|
|
name="source_x",
|
|
)
|
|
|
|
node_y = Node(
|
|
f=source_y,
|
|
requires=[],
|
|
provides=[Message(name="y", dtype=int)],
|
|
params={"value": 1},
|
|
name="source_y",
|
|
)
|
|
|
|
sync_node = Node(
|
|
f=sync_consumer,
|
|
requires=[Message(name="x", dtype=int), Message(name="y", dtype=int)],
|
|
provides=[Message(name="sync_out", dtype=int)],
|
|
params={"synchronous": True},
|
|
name="sync_consumer",
|
|
)
|
|
|
|
async_node = Node(
|
|
f=async_consumer,
|
|
requires=[Message(name="x", dtype=int), Message(name="y", dtype=int)],
|
|
provides=[Message(name="async_out", dtype=int)],
|
|
params={"synchronous": False},
|
|
name="async_consumer",
|
|
)
|
|
|
|
state = MemoryState()
|
|
pipeline = Pipeline(
|
|
nodes=[node_x, node_y, sync_node, async_node],
|
|
state=state,
|
|
max_workers=1,
|
|
)
|
|
|
|
# Initial trigger to get both inputs
|
|
node_x.inject({})
|
|
node_y.inject({})
|
|
|
|
initial_sync = sync_execution_count["count"]
|
|
initial_async = async_execution_count["count"]
|
|
|
|
print(f"After initial triggers: sync={initial_sync}, async={initial_async}")
|
|
assert initial_sync == 1, "sync_consumer should have run once initially"
|
|
assert initial_async == 1, "async_consumer should have run once initially"
|
|
|
|
# Now trigger X multiple times without triggering Y
|
|
print("\nTriggering source_x 5 times without changing source_y...")
|
|
for i in range(5):
|
|
node_x.params["value"] = 10 + i
|
|
node_x.inject({})
|
|
|
|
final_sync = sync_execution_count["count"]
|
|
final_async = async_execution_count["count"]
|
|
|
|
print(f"After 5 more X triggers: sync={final_sync}, async={final_async}")
|
|
|
|
# sync_consumer should NOT have run again (Y didn't change)
|
|
assert (
|
|
final_sync == initial_sync
|
|
), f"sync_consumer should still be at {initial_sync}, got {final_sync}"
|
|
# async_consumer should have run 5 more times
|
|
assert (
|
|
final_async == initial_async + 5
|
|
), f"async_consumer should be at {initial_async + 5}, got {final_async}"
|
|
|
|
print("✓ Synchronous node correctly waited for both inputs to change")
|
|
|
|
# Now trigger Y once - sync should run
|
|
print("\nTriggering source_y once...")
|
|
node_y.params["value"] = 100
|
|
node_y.inject({})
|
|
|
|
after_y_sync = sync_execution_count["count"]
|
|
after_y_async = async_execution_count["count"]
|
|
|
|
print(f"After Y trigger: sync={after_y_sync}, async={after_y_async}")
|
|
|
|
assert after_y_sync == initial_sync + 1, f"sync_consumer should have run once more"
|
|
assert after_y_async == final_async + 1, f"async_consumer should have run once more"
|
|
|
|
print("\n✓ Multiple triggers test passed!\n")
|
|
|
|
|
|
def test_synchronous_race_condition():
|
|
"""Test that synchronous nodes handle concurrent triggers correctly."""
|
|
print("=" * 60)
|
|
print("Testing Synchronous Node Race Condition Handling")
|
|
print("=" * 60)
|
|
|
|
execution_count = {"sync": 0, "async": 0}
|
|
execution_lock = threading.Lock()
|
|
|
|
def source_a(params, **kwargs):
|
|
return {"data_a": params.get("value", 1)}
|
|
|
|
def source_b(params, **kwargs):
|
|
return {"data_b": params.get("value", 2)}
|
|
|
|
def sync_consumer(params, data_a=0, data_b=0, **kwargs):
|
|
with execution_lock:
|
|
execution_count["sync"] += 1
|
|
time.sleep(0.01) # Small delay to increase chance of race conditions
|
|
return {"sync_result": data_a + data_b}
|
|
|
|
def async_consumer(params, data_a=0, data_b=0, **kwargs):
|
|
with execution_lock:
|
|
execution_count["async"] += 1
|
|
return {"async_result": data_a * data_b}
|
|
|
|
node_a = Node(
|
|
f=source_a,
|
|
requires=[],
|
|
provides=[Message(name="data_a", dtype=int)],
|
|
params={"value": 10},
|
|
name="source_a",
|
|
)
|
|
|
|
node_b = Node(
|
|
f=source_b,
|
|
requires=[],
|
|
provides=[Message(name="data_b", dtype=int)],
|
|
params={"value": 20},
|
|
name="source_b",
|
|
)
|
|
|
|
sync_node = Node(
|
|
f=sync_consumer,
|
|
requires=[Message(name="data_a", dtype=int), Message(name="data_b", dtype=int)],
|
|
provides=[Message(name="sync_result", dtype=int)],
|
|
params={"synchronous": True},
|
|
name="sync_consumer",
|
|
)
|
|
|
|
async_node = Node(
|
|
f=async_consumer,
|
|
requires=[Message(name="data_a", dtype=int), Message(name="data_b", dtype=int)],
|
|
provides=[Message(name="async_result", dtype=int)],
|
|
params={"synchronous": False},
|
|
name="async_consumer",
|
|
)
|
|
|
|
state = MemoryState()
|
|
pipeline = Pipeline(
|
|
nodes=[node_a, node_b, sync_node, async_node],
|
|
state=state,
|
|
max_workers=4, # Multiple workers for concurrency
|
|
)
|
|
|
|
# Trigger both sensors multiple times concurrently
|
|
num_rounds = 5
|
|
print(f"\nTriggering both sensors {num_rounds} times concurrently...")
|
|
|
|
threads = []
|
|
for i in range(num_rounds):
|
|
node_a.params["value"] = 10 + i
|
|
node_b.params["value"] = 20 + i
|
|
|
|
t1 = threading.Thread(target=lambda: node_a.inject({}))
|
|
t2 = threading.Thread(target=lambda: node_b.inject({}))
|
|
threads.extend([t1, t2])
|
|
t1.start()
|
|
t2.start()
|
|
|
|
# Wait for all threads
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Give pipeline time to process
|
|
time.sleep(0.5)
|
|
|
|
print(f"\nExecution counts:")
|
|
print(f" sync_consumer: {execution_count['sync']}")
|
|
print(f" async_consumer: {execution_count['async']}")
|
|
|
|
# sync_consumer should execute at most num_rounds times
|
|
# (could be less if some triggers happened before the other input updated)
|
|
assert (
|
|
execution_count["sync"] <= num_rounds + 1
|
|
), f"sync_consumer ran too many times: {execution_count['sync']} (expected <= {num_rounds + 1})"
|
|
|
|
# async_consumer will run more frequently
|
|
assert (
|
|
execution_count["async"] >= execution_count["sync"]
|
|
), "async_consumer should run at least as often as sync_consumer"
|
|
|
|
print(
|
|
f"\n✓ sync_consumer ran {execution_count['sync']} times (max expected: {num_rounds + 1})"
|
|
)
|
|
print(f"✓ async_consumer ran {execution_count['async']} times")
|
|
print("\n✓ Race condition test passed!\n")
|
|
|
|
|
|
def test_state_backend_atomic_operations():
|
|
"""Test the atomic operations used for synchronous node support."""
|
|
print("=" * 60)
|
|
print("Testing State Backend Atomic Operations")
|
|
print("=" * 60)
|
|
|
|
state = MemoryState()
|
|
|
|
# Test increment
|
|
print("\nTesting increment...")
|
|
assert state.increment("counter") == 1
|
|
assert state.increment("counter") == 2
|
|
assert state.increment("counter") == 3
|
|
print("✓ increment works correctly")
|
|
|
|
# Test get_multi
|
|
print("\nTesting get_multi...")
|
|
state.set("a", 1)
|
|
state.set("b", 2)
|
|
state.set("c", 3)
|
|
result = state.get_multi(["a", "b", "c", "missing"])
|
|
assert result == {"a": 1, "b": 2, "c": 3, "missing": None}
|
|
print("✓ get_multi works correctly")
|
|
|
|
# Test compare_and_swap_multi - success case
|
|
print("\nTesting compare_and_swap_multi (success)...")
|
|
success = state.compare_and_swap_multi(
|
|
expected={"a": 1, "b": 2}, updates={"a": 10, "b": 20, "new_key": 100}
|
|
)
|
|
assert success, "compare_and_swap_multi should succeed"
|
|
assert state.get("a") == 10
|
|
assert state.get("b") == 20
|
|
assert state.get("new_key") == 100
|
|
print("✓ compare_and_swap_multi succeeded and applied updates")
|
|
|
|
# Test compare_and_swap_multi - failure case
|
|
print("\nTesting compare_and_swap_multi (failure)...")
|
|
success = state.compare_and_swap_multi(
|
|
expected={"a": 1, "b": 20}, updates={"a": 999, "b": 999} # a is now 10, not 1
|
|
)
|
|
assert not success, "compare_and_swap_multi should fail"
|
|
assert state.get("a") == 10, "a should be unchanged"
|
|
assert state.get("b") == 20, "b should be unchanged"
|
|
print("✓ compare_and_swap_multi correctly rejected mismatched expectation")
|
|
|
|
print("\n✓ State backend atomic operations test passed!\n")
|
|
|
|
|
|
def test_concurrent_compare_and_swap():
|
|
"""Test that compare_and_swap_multi handles concurrent access correctly."""
|
|
print("=" * 60)
|
|
print("Testing Concurrent Compare-and-Swap")
|
|
print("=" * 60)
|
|
|
|
state = MemoryState()
|
|
state.set("version", 0)
|
|
|
|
success_count = {"count": 0}
|
|
failure_count = {"count": 0}
|
|
count_lock = threading.Lock()
|
|
|
|
def try_swap(thread_id: int):
|
|
"""Try to atomically increment the version."""
|
|
for _ in range(10):
|
|
current = state.get("version")
|
|
success = state.compare_and_swap_multi(
|
|
expected={"version": current}, updates={"version": current + 1}
|
|
)
|
|
with count_lock:
|
|
if success:
|
|
success_count["count"] += 1
|
|
else:
|
|
failure_count["count"] += 1
|
|
|
|
# Start multiple threads trying to increment
|
|
threads = [threading.Thread(target=try_swap, args=(i,)) for i in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
final_version = state.get("version")
|
|
total_attempts = success_count["count"] + failure_count["count"]
|
|
|
|
print(f"Total attempts: {total_attempts}")
|
|
print(f"Successful swaps: {success_count['count']}")
|
|
print(f"Failed swaps (race lost): {failure_count['count']}")
|
|
print(f"Final version: {final_version}")
|
|
|
|
# The final version should equal the number of successful swaps
|
|
assert (
|
|
final_version == success_count["count"]
|
|
), f"Version mismatch: {final_version} != {success_count['count']}"
|
|
|
|
print("\n✓ Concurrent compare-and-swap test passed!\n")
|
|
|
|
|
|
def main():
|
|
"""Run all synchronous node tests."""
|
|
print("\n" + "=" * 60)
|
|
print("Synchronous Node Test Suite")
|
|
print("=" * 60 + "\n")
|
|
|
|
# Run all tests
|
|
test_state_backend_atomic_operations()
|
|
test_concurrent_compare_and_swap()
|
|
test_synchronous_basic()
|
|
test_synchronous_multiple_triggers()
|
|
test_synchronous_race_condition()
|
|
|
|
print("=" * 60)
|
|
print("All synchronous node tests completed!")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|