mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(pipeline): set_parallelism now enables dependency-layer parallel execution (closes #1223) (#1226)

Browse Source 
PipelineBuilder.set_parallelism() validated and stored a level in
pipeline config, but ExecutionEngine._execute_steps() had no parallel
code path and no code ever read it back, so steps always ran strictly
sequentially regardless of the configured value. parallelism was also
lost across a serialize/deserialize round trip, since the nested
config key was never promoted to the top-level dict build_pipeline()
reads.

Steps are now grouped into dependency layers (declaration order
preserved within each layer). A layer runs concurrently, bounded by
ThreadPoolExecutor(max_workers=min(configured parallelism, engine
max_workers)), only when every one of the following holds: more than
one step in the layer, the shared input is a dict, every step is
opted in via the new PipelineStep.parallel_safe flag, and no step is
in delta_mode. Any layer that doesn't meet all four falls back to the
existing sequential path unchanged.

Each step's input is deep-copied before any handler in the layer
starts, so concurrent steps never share mutable state. Layer results
are merged back in declaration order, not completion order; keys
whose value is unchanged from the shared input are treated as an
echo rather than a write, so two handlers both returning {**data, ...}
don't spuriously conflict on keys neither of them actually touched.
Genuinely conflicting values for the same key raise ProcessingError
naming both the key and the two steps involved. Retry policy, step
status, result/error tracking, and progress reporting are shared
between the sequential and parallel paths so behavior stays identical
either way. On step failure, not yet started futures in the same
layer are cancelled and the error propagates, so no downstream layer
ever runs.

parallel_safe is opt-in per step because handlers that share mutable
state or depend on strict ordering are not safe to run concurrently.
ParallelismManager.execute_pipeline_steps_parallel() is deliberately
not reused here; the engine implements its own bounded layer
scheduler so retry/status semantics stay identical between the
sequential and parallel code paths instead of diverging.

fix(pipeline): address qodo review findings on PR #1226

- detect circular/unknown dependencies in parallel grouping
  (ValidationError instead of RecursionError/KeyError)
- skip unchanged echoed keys in parallel result merging to
  avoid false conflicts
- fail before COMPLETED status when a parallel step returns a
  non-dict; never retry such contract violations
- require strict bool parallel_safe in builder and engine gate
- make per-step progress tracking IDs unique across same-type
  parallel steps

---------

Co-authored-by: 江俊杰 <jiangjunjie.37@jd.com>
This commit is contained in:
cxzg007
2026-08-28 12:33:39 +05:00
committed by GitHub
co-authored by 江俊杰
parent e12eec40a1
commit cce5ea177c
4 changed files with 1201 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/guides/pipeline.md
+12 -1
View File
@@ -241,7 +241,18 @@ engine = ExecutionEngine(max_workers=2, retry_on_failure=True)
result = engine.execute_pipeline(pipeline)
```
`set_parallelism(n)` tells the engine how many steps it may run simultaneously. The topological sort guarantees that only steps whose dependencies are all completed are eligible for concurrent execution — you cannot accidentally run a step before its inputs are ready.
`set_parallelism(n)` tells the engine how many steps it may run simultaneously; `n` must be a positive integer. The topological sort guarantees that only steps whose dependencies are all completed are eligible for concurrent execution — you cannot accidentally run a step before its inputs are ready. The effective concurrency is capped at `min(n, max_workers)`, so the engine's `max_workers` setting remains a hard resource ceiling.
Concurrency is opt-in per step. A dependency layer only runs in parallel when every step in that layer is marked `parallel_safe`, the layer has more than one step, and the data flowing into the layer is a dict:
```python
builder.add_step("ner", "ner_extract", parallel_safe=True, confidence_threshold=0.75)
builder.add_step("triplets", "triplet_extract", parallel_safe=True, include_temporal=True)
```
If any step in a layer is not marked `parallel_safe`, or if a step runs in delta mode, the entire layer falls back to sequential execution — parallelism never silently bypasses a step that was not declared safe. `parallel_safe` is a control field: like `dependencies`, it is consumed by the builder and never reaches your handler's config.
Parallel-safe handlers must return a dict. Each step in a parallel layer receives an isolated deep copy of the layer's input, so steps cannot see each other's mutations. The per-step results are merged key by key in step declaration order: a key written by one step is added to the merged output, a key written by several steps with equal values is kept, and two steps writing different values for the same key fail the pipeline with a `ProcessingError` naming the conflicting key and both steps. Handlers that touch shared mutable resources — database connections, in-memory stores, global caches — should not be marked `parallel_safe`.
## Common Pitfalls
semantica/pipeline/execution_engine.py
+369 -59
View File
@@ -32,8 +32,10 @@ Author: Semantica Contributors
License: MIT
"""
import copy
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
@@ -59,6 +61,14 @@ class PipelineStatus(Enum):
STOPPED = "stopped"
class _ParallelResultContractError(ProcessingError):
"""Raised when a parallel step violates the dict-result contract.
Contract violations are deterministic: re-running the handler cannot
change its return type, so the retry loop must be skipped entirely.
"""
@dataclass
class ExecutionResult:
"""Pipeline execution result."""
@@ -242,11 +252,67 @@ class ExecutionEngine:
return ExecutionResult(success=False, output=None, errors=[str(e)])
def _execute_steps(self, pipeline: Pipeline, data: Any, **options) -> Any:
"""Execute pipeline steps."""
# Sort steps by dependencies (topological sort)
sorted_steps = self._topological_sort(pipeline.steps)
"""
Execute pipeline steps.
Steps are grouped into dependency layers. When a pipeline is
configured with parallelism > 1, a layer whose steps are all marked
``parallel_safe`` and whose input is a dict is executed concurrently
(bounded by the effective parallelism); every other layer runs
sequentially, preserving the default serial behaviour.
"""
effective_parallelism = self._get_effective_parallelism(pipeline)
if effective_parallelism <= 1:
return self._execute_steps_sequential(pipeline, data, **options)
layers = self._group_steps_by_dependency_level(pipeline.steps)
current_data = data
for layer in layers:
if self.pipeline_status.get(pipeline.name) == PipelineStatus.STOPPED:
break
# Wait if paused
while self.pipeline_status.get(pipeline.name) == PipelineStatus.PAUSED:
time.sleep(0.1)
if self._can_run_layer_in_parallel(layer, current_data):
merged = self._execute_parallel_group(
layer,
current_data,
effective_parallelism,
pipeline_name=pipeline.name,
**options,
)
if merged is None:
# Input isolation failed before any handler started;
# run this layer sequentially instead.
current_data = self._execute_steps_sequential(
pipeline, current_data, steps=layer, **options
)
else:
current_data = merged
else:
current_data = self._execute_steps_sequential(
pipeline, current_data, steps=layer, **options
)
return current_data
def _execute_steps_sequential(
self,
pipeline: Pipeline,
data: Any,
steps: Optional[List[PipelineStep]] = None,
**options,
) -> Any:
"""Execute steps sequentially following dependency order."""
if steps is None:
sorted_steps = self._topological_sort(pipeline.steps)
else:
sorted_steps = list(steps)
# Execute steps
current_data = data
total_steps = len(sorted_steps)
@@ -258,76 +324,320 @@ class ExecutionEngine:
while self.pipeline_status.get(pipeline.name) == PipelineStatus.PAUSED:
time.sleep(0.1)
# Track step execution
step_tracking_id = self.progress_tracker.start_tracking(
module="pipeline",
submodule=step.step_type or step.name,
message=f"Step {step_idx + 1}/{total_steps}: {step.name}",
current_data = self._execute_step_with_retries(
step,
current_data,
step_label=f"Step {step_idx + 1}/{total_steps}: {step.name}",
pipeline_name=pipeline.name,
**options,
)
try:
# Execute step
step.status = StepStatus.RUNNING
step_result = self._execute_step(step, current_data, **options)
step.status = StepStatus.COMPLETED
step.result = step_result
current_data = step_result
return current_data
def _execute_step_with_retries(
self,
step: PipelineStep,
data: Any,
step_label: Optional[str] = None,
pipeline_name: Optional[str] = None,
require_dict_result: bool = False,
**options,
) -> Any:
"""
Execute a single step with retry handling.
Shared by the sequential and parallel execution paths so that retry
policies, step status tracking and progress reporting behave
identically. Returns the step result, or raises the final error
after retries are exhausted.
When ``require_dict_result`` is set (parallel layers), a handler
returning a non-dict raises ProcessingError before the step is
marked completed, so status and progress reporting stay consistent.
Such contract violations are never retried: the handler's return
type cannot change between attempts.
"""
step_tracking_id = self.progress_tracker.start_tracking(
module="pipeline",
# Pipeline identity + step name keep tracking IDs unique so
# concurrent steps of the same step_type cannot overwrite each
# other's progress records.
submodule=(
f"{pipeline_name or 'pipeline'}:"
f"{step.step_type or 'step'}:{step.name}"
),
message=step_label or f"Executing step: {step.name}",
)
try:
step.status = StepStatus.RUNNING
step_result = self._execute_step(step, data, **options)
if require_dict_result and not isinstance(step_result, dict):
raise _ParallelResultContractError(
f"Step '{step.name}' is marked parallel_safe and must "
f"return a dict so parallel results can be merged, got "
f"{type(step_result).__name__}"
)
step.status = StepStatus.COMPLETED
step.result = step_result
self.progress_tracker.stop_tracking(
step_tracking_id,
status="completed",
message=f"Completed step: {step.name}",
)
return step_result
except Exception as e:
step.status = StepStatus.FAILED
step.error = e
# Contract violations are deterministic failures: re-running
# the handler cannot change its return type, so never consult
# the retry policy for them.
if isinstance(e, _ParallelResultContractError):
self.progress_tracker.stop_tracking(
step_tracking_id, status="failed", message=str(e)
)
raise
# Retry loop respecting max_retries from the policy
retry_policy = self.failure_handler.get_retry_policy(step.step_type)
max_retries = retry_policy.max_retries if retry_policy else 0
retry_count = 0
success = False
while retry_count < max_retries:
recovery_result = self.failure_handler.handle_step_failure(step, e)
if not recovery_result.get("retry", False):
break
retry_delay = recovery_result.get("retry_delay", 0.0)
if retry_delay > 0:
time.sleep(retry_delay)
self.progress_tracker.update_tracking(
step_tracking_id,
status="running",
message=f"Retrying step: {step.name} (attempt {retry_count + 1})",
)
step.status = StepStatus.RUNNING
try:
step_result = self._execute_step(step, data, **options)
if require_dict_result and not isinstance(
step_result, dict
):
raise _ParallelResultContractError(
f"Step '{step.name}' is marked parallel_safe "
f"and must return a dict so parallel results "
f"can be merged, got "
f"{type(step_result).__name__}"
)
step.status = StepStatus.COMPLETED
step.result = step_result
success = True
break
except Exception as retry_e:
step.status = StepStatus.FAILED
step.error = retry_e
e = retry_e
retry_count += 1
if success:
self.progress_tracker.stop_tracking(
step_tracking_id,
status="completed",
message=f"Completed step: {step.name}",
message=f"Retry successful: {step.name}",
)
return step_result
else:
self.progress_tracker.stop_tracking(
step_tracking_id, status="failed", message=str(e)
)
raise e
except Exception as e:
step.status = StepStatus.FAILED
step.error = e
def _get_effective_parallelism(self, pipeline: Pipeline) -> int:
"""Return the parallelism actually used for this pipeline."""
configured = pipeline.config.get("parallelism", 1)
if not isinstance(configured, int) or configured <= 0:
return 1
return min(configured, self.parallelism_manager.max_workers)
# Retry loop respecting max_retries from the policy
retry_policy = self.failure_handler.get_retry_policy(step.step_type)
max_retries = retry_policy.max_retries if retry_policy else 0
retry_count = 0
success = False
def _group_steps_by_dependency_level(
self, steps: List[PipelineStep]
) -> List[List[PipelineStep]]:
"""
Group steps into dependency layers, preserving declaration order.
while retry_count < max_retries:
recovery_result = self.failure_handler.handle_step_failure(step, e)
if not recovery_result.get("retry", False):
break
retry_delay = recovery_result.get("retry_delay", 0.0)
if retry_delay > 0:
time.sleep(retry_delay)
self.progress_tracker.update_tracking(
step_tracking_id,
status="running",
message=f"Retrying step: {step.name} (attempt {retry_count + 1})",
Circular or unknown dependencies raise ValidationError so the
parallel path fails deterministically, matching the validation
behaviour of the serial topological sort.
"""
step_map = {step.name: step for step in steps}
levels: Dict[str, int] = {}
visiting: set = set()
for step in steps:
for dep in step.dependencies:
if dep not in step_map:
raise ValidationError(
f"Step '{step.name}' depends on unknown step '{dep}'"
)
step.status = StepStatus.RUNNING
try:
step_result = self._execute_step(step, current_data, **options)
step.status = StepStatus.COMPLETED
step.result = step_result
current_data = step_result
success = True
break
except Exception as retry_e:
step.status = StepStatus.FAILED
step.error = retry_e
e = retry_e
retry_count += 1
if success:
self.progress_tracker.stop_tracking(
step_tracking_id,
status="completed",
message=f"Retry successful: {step.name}",
)
def get_level(step_name: str) -> int:
if step_name in levels:
return levels[step_name]
if step_name in visiting:
raise ValidationError(
"Circular dependency detected in pipeline "
f"(cycle passes through step '{step_name}')"
)
visiting.add(step_name)
step = step_map[step_name]
if not step.dependencies:
level = 0
else:
level = max(get_level(dep) for dep in step.dependencies) + 1
visiting.discard(step_name)
levels[step_name] = level
return level
for step in steps:
get_level(step.name)
grouped: Dict[int, List[PipelineStep]] = {}
for step in steps:
grouped.setdefault(levels[step.name], []).append(step)
return [grouped[level] for level in sorted(grouped)]
def _can_run_layer_in_parallel(self, layer: List[PipelineStep], data: Any) -> bool:
"""Check whether a dependency layer can safely run in parallel."""
if len(layer) <= 1:
return False
if not isinstance(data, dict):
return False
for step in layer:
# Strict boolean check: truthy non-bool values (e.g. the
# string "false") must never opt a step into concurrency.
if getattr(step, "parallel_safe", False) is not True:
return False
if getattr(step, "delta_mode", False):
return False
return True
def _execute_parallel_group(
self,
layer: List[PipelineStep],
data: Any,
effective_parallelism: int,
pipeline_name: Optional[str] = None,
**options,
) -> Optional[Any]:
"""
Execute a dependency layer concurrently.
Per-step inputs are deep-copied before any handler starts so that
parallel steps do not share mutable state. Returns the merged dict
result, or None when input isolation failed (before any handler
ran) and the layer should fall back to sequential execution.
"""
# Isolate per-step inputs before starting any handler
try:
step_inputs = {step.name: copy.deepcopy(data) for step in layer}
except Exception as e:
self.logger.warning(
f"Falling back to sequential execution: input for parallel "
f"layer could not be isolated ({e})"
)
return None
step_results: Dict[str, Any] = {}
failure: Optional[BaseException] = None
max_workers = min(effective_parallelism, len(layer))
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(
self._execute_step_with_retries,
step,
step_inputs[step.name],
# Validate the dict-result contract inside the shared
# lifecycle path, before completion is reported.
pipeline_name=pipeline_name,
require_dict_result=True,
**options,
): step
for step in layer
}
for future in as_completed(futures):
step = futures[future]
try:
step_result = future.result()
except Exception as e:
if failure is None:
failure = e
# Cancel steps that have not started yet
for pending in futures:
pending.cancel()
else:
self.progress_tracker.stop_tracking(
step_tracking_id, status="failed", message=str(e)
step_results[step.name] = step_result
if failure is not None:
raise failure
return self._merge_parallel_results(data, layer, step_results)
def _merge_parallel_results(
self,
base: Dict[str, Any],
layer: List[PipelineStep],
step_results: Dict[str, Any],
) -> Dict[str, Any]:
"""
Merge the results of a parallel layer into a single dict.
Steps are processed in declaration order (never by completion
order). Keys whose values are unchanged from the shared base input
are skipped (handlers commonly return complete dicts such as
``{**data, ...}``); only added or changed keys count as branch
writes. Keys written with equal values by multiple steps are
allowed; conflicting values for the same key raise a
ProcessingError naming the key and both steps. Ambiguous equality
comparisons count as changed, never as unchanged.
"""
merged = dict(base)
key_sources: Dict[str, str] = {}
for step in layer:
step_result = step_results.get(step.name)
if step_result is None:
continue
for key, value in step_result.items():
if key in base and self._values_equal(base[key], value):
# Echo of the shared input: not a branch write, so it
# cannot conflict with a sibling that changes the key.
continue
if key in key_sources and not self._values_equal(
merged.get(key), value
):
raise ProcessingError(
f"Conflicting values for key '{key}' in parallel step "
f"results: step '{step.name}' produced {value!r}, but "
f"step '{key_sources[key]}' previously produced "
f"{merged.get(key)!r}"
)
raise e
key_sources[key] = step.name
merged[key] = value
return merged
@staticmethod
def _values_equal(left: Any, right: Any) -> bool:
"""Safely compare two values; ambiguous comparisons count as conflicts."""
try:
return bool(left == right)
except (TypeError, ValueError):
return False
return current_data
def _execute_step(self, step: PipelineStep, data: Any, **options) -> Any:
"""
semantica/pipeline/pipeline_builder.py
+35
View File
@@ -67,6 +67,7 @@ class PipelineStep:
delta_mode: bool = False
base_version_id: Optional[str] = None
target_version_id: Optional[str] = None
parallel_safe: bool = False
@dataclass
@@ -133,6 +134,12 @@ class PipelineBuilder:
target_version_id = config.pop("target_version_id", None)
dependencies = config.pop("dependencies", [])
handler = config.pop("handler", None)
parallel_safe = config.pop("parallel_safe", False)
if not isinstance(parallel_safe, bool):
raise ValidationError(
f"parallel_safe must be a boolean, got "
f"{type(parallel_safe).__name__} for step '{step_name}'"
)
if handler is None:
handler = self.step_registry.get(step_type)
@@ -145,6 +152,7 @@ class PipelineBuilder:
delta_mode = delta_mode,
base_version_id=base_version_id,
target_version_id=target_version_id,
parallel_safe=parallel_safe,
)
self.steps.append(step)
@@ -186,6 +194,14 @@ class PipelineBuilder:
Returns:
Self for method chaining
"""
if (
isinstance(level, bool)
or not isinstance(level, int)
or level <= 0
):
raise ValidationError(
f"Parallelism level must be a positive integer, got {level!r}"
)
self.pipeline_config["parallelism"] = level
return self
@@ -289,6 +305,16 @@ class PipelineBuilder:
step.target_version_id = step_config.get(
"target_version_id", step.target_version_id
)
raw_parallel_safe = step_config.get(
"parallel_safe", step.parallel_safe
)
if not isinstance(raw_parallel_safe, bool):
raise ValidationError(
"parallel_safe must be a boolean for step "
f"'{step_name}', got "
f"{type(raw_parallel_safe).__name__}"
)
step.parallel_safe = raw_parallel_safe
# Set parallelism if specified
if "parallelism" in pipeline_config:
@@ -344,6 +370,7 @@ class PipelineBuilder:
"type": step.step_type,
"config": step.config,
"dependencies": step.dependencies,
"parallel_safe": step.parallel_safe,
}
for step in self.steps
],
@@ -425,6 +452,7 @@ class PipelineSerializer:
"delta_mode",
"base_version_id",
"target_version_id",
"parallel_safe",
}
pipeline_data = {
"name": pipeline.name,
@@ -441,6 +469,7 @@ class PipelineSerializer:
"delta_mode": getattr(step, "delta_mode", False),
"base_version_id": getattr(step, "base_version_id", None),
"target_version_id": getattr(step, "target_version_id", None),
"parallel_safe": getattr(step, "parallel_safe", False),
}
for step in pipeline.steps
],
@@ -488,6 +517,12 @@ class PipelineSerializer:
sanitized_steps.append(sanitized_step)
pipeline_data["steps"] = sanitized_steps
# Reapply pipeline-level config (e.g. parallelism) at the top level
# so build_pipeline picks it up
serialized_config = pipeline_data.pop("config", None) or {}
for key, value in serialized_config.items():
pipeline_data.setdefault(key, value)
# Reconstruct pipeline
builder = PipelineBuilder(**self.config)
pipeline = builder.build_pipeline(pipeline_data, **options)
tests/pipeline/test_pipeline_parallel.py
+785
View File
@@ -0,0 +1,785 @@
"""Focused tests for pipeline parallel execution (issue #1223).
Covers:
- ``PipelineBuilder.set_parallelism()`` validation
- dependency-layer parallel execution gated on ``parallel_safe``
- input isolation via deep copies
- incremental dict merging of parallel outputs
- failure handling and cancellation semantics
- ``parallel_safe`` not leaking into handler kwargs
- ``parallel_safe`` surviving serialization round-trips
"""
import threading
import time
import unittest
from unittest.mock import MagicMock, patch
from semantica.pipeline.execution_engine import ExecutionEngine
from semantica.pipeline.failure_handler import RetryPolicy, RetryStrategy
from semantica.pipeline.pipeline_builder import (
Pipeline,
PipelineBuilder,
PipelineSerializer,
PipelineStep,
StepStatus,
)
from semantica.utils.exceptions import ProcessingError, ValidationError
class ConcurrencyProbe:
"""Thread-safe tracker of handler invocations and concurrency level."""
def __init__(self):
self._lock = threading.Lock()
self.active = 0
self.max_active = 0
self.calls = 0
def __enter__(self):
with self._lock:
self.active += 1
self.calls += 1
self.max_active = max(self.max_active, self.active)
return self
def __exit__(self, *exc):
with self._lock:
self.active -= 1
return False
def branch_handler(probe, key, value=True, delay=0.0):
"""Build a handler that records concurrency and adds one output key."""
def handler(data, **kwargs):
with probe:
if delay:
time.sleep(delay)
return {**data, key: value}
return handler
class Undeepcopyable:
"""Object whose deepcopy always fails."""
def __deepcopy__(self, memo):
raise TypeError("cannot deepcopy this object")
class TestSetParallelismValidation(unittest.TestCase):
"""set_parallelism() must only accept positive integers."""
def test_rejects_invalid_levels(self):
builder = PipelineBuilder()
for invalid in (0, -1, 1.5, True, False, "2", None):
with self.assertRaises(ValidationError):
builder.set_parallelism(invalid)
def test_accepts_positive_integers(self):
builder = PipelineBuilder()
builder.set_parallelism(4)
self.assertEqual(builder.pipeline_config["parallelism"], 4)
class TestPipelineParallelExecution(unittest.TestCase):
def setUp(self):
self.mock_tracker_patcher = patch(
"semantica.utils.progress_tracker.get_progress_tracker"
)
self.mock_get_tracker = self.mock_tracker_patcher.start()
self.mock_get_tracker.return_value = MagicMock()
def tearDown(self):
self.mock_tracker_patcher.stop()
def _build(self, steps, parallelism=None, name="parallel_pipeline"):
"""steps: list of (step_name, step_type, handler, parallel_safe)."""
builder = PipelineBuilder()
for step_name, step_type, handler, parallel_safe in steps:
builder.add_step(
step_name, step_type, handler=handler, parallel_safe=parallel_safe
)
if parallelism is not None:
builder.set_parallelism(parallelism)
return builder.build(name)
def test_unconfigured_pipeline_stays_serial(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", branch_handler(probe, "b"), True),
]
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertEqual(probe.max_active, 1)
def test_parallelism_one_stays_serial(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", branch_handler(probe, "b"), True),
],
parallelism=1,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertEqual(probe.max_active, 1)
def test_parallel_safe_steps_overlap_with_barrier(self):
barrier = threading.Barrier(2)
done = {"a": False, "b": False}
def handler(key):
def inner(data, **kwargs):
barrier.wait(timeout=5)
done[key] = True
return {**data, key: True}
return inner
pipeline = self._build(
[
("a", "branch", handler("a"), True),
("b", "branch", handler("b"), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success, msg=str(result.errors))
self.assertTrue(done["a"])
self.assertTrue(done["b"])
def test_active_workers_do_not_exceed_parallelism(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a", delay=0.15), True),
("b", "branch", branch_handler(probe, "b", delay=0.15), True),
("c", "branch", branch_handler(probe, "c", delay=0.15), True),
("d", "branch", branch_handler(probe, "d", delay=0.15), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertLessEqual(probe.max_active, 2)
self.assertGreaterEqual(probe.max_active, 2)
def test_unmarked_steps_stay_serial(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), False),
("b", "branch", branch_handler(probe, "b"), False),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertEqual(probe.max_active, 1)
def test_layer_with_one_unsafe_step_is_serial(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", branch_handler(probe, "b"), False),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertEqual(probe.max_active, 1)
def test_pure_dependency_chain_stays_serial(self):
probe = ConcurrencyProbe()
def chain_handler(key):
def inner(data, **kwargs):
with probe:
return {**data, key: True}
return inner
builder = PipelineBuilder()
builder.add_step(
"a", "branch", handler=chain_handler("a"), parallel_safe=True
)
builder.add_step(
"b",
"branch",
handler=chain_handler("b"),
parallel_safe=True,
dependencies=["a"],
)
builder.add_step(
"c",
"branch",
handler=chain_handler("c"),
parallel_safe=True,
dependencies=["b"],
)
builder.set_parallelism(4)
pipeline = builder.build("chain_pipeline")
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertEqual(probe.max_active, 1)
self.assertEqual(result.output, {"text": "hi", "a": True, "b": True, "c": True})
def test_non_dict_input_falls_back_to_serial(self):
probe = ConcurrencyProbe()
def passthrough(data, **kwargs):
with probe:
return data
pipeline = self._build(
[
("a", "branch", passthrough, True),
("b", "branch", passthrough, True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(
pipeline, data="plain-text-input"
)
self.assertTrue(result.success)
self.assertEqual(probe.calls, 2)
self.assertEqual(probe.max_active, 1)
self.assertEqual(result.output, "plain-text-input")
def test_deepcopy_failure_falls_back_to_serial(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", branch_handler(probe, "b"), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(
pipeline, data={"text": "hi", "obj": Undeepcopyable()}
)
self.assertTrue(result.success, msg=str(result.errors))
self.assertEqual(probe.calls, 2)
self.assertEqual(probe.max_active, 1)
self.assertIn("a", result.output)
self.assertIn("b", result.output)
def test_parallel_results_merge_different_fields(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "entities", ["Alice"]), True),
("b", "branch", branch_handler(probe, "triplets", [(1, 2)]), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(
pipeline, data={"text": "Alice works at Acme"}
)
self.assertTrue(result.success)
self.assertEqual(
result.output,
{
"text": "Alice works at Acme",
"entities": ["Alice"],
"triplets": [(1, 2)],
},
)
def test_parallel_results_same_key_same_value_allowed(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "shared", [1, 2]), True),
("b", "branch", branch_handler(probe, "shared", [1, 2]), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success, msg=str(result.errors))
self.assertEqual(result.output["shared"], [1, 2])
def test_parallel_results_same_key_different_values_raises(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "shared", 1), True),
("b", "branch", branch_handler(probe, "shared", 2), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
error_text = result.errors[0] if result.errors else ""
self.assertIn("shared", error_text)
self.assertIn("a", error_text)
self.assertIn("b", error_text)
def test_parallel_handler_non_dict_return_fails(self):
probe = ConcurrencyProbe()
def non_dict_handler(data, **kwargs):
with probe:
return "not-a-dict"
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", non_dict_handler, True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
error_text = result.errors[0] if result.errors else ""
self.assertIn("b", error_text)
self.assertIn("str", error_text)
def test_non_dict_handler_executed_only_once(self):
probe = ConcurrencyProbe()
def non_dict_handler(data, **kwargs):
with probe:
return "not-a-dict"
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", non_dict_handler, True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
self.assertEqual(probe.calls, 2) # one invocation per handler, no re-runs
def test_parallel_step_retry_succeeds(self):
engine = ExecutionEngine()
engine.failure_handler.set_retry_policy(
"flaky",
RetryPolicy(
max_retries=2, initial_delay=0.0, strategy=RetryStrategy.FIXED
),
)
probe = ConcurrencyProbe()
attempts = {"flaky": 0}
def flaky_handler(data, **kwargs):
with probe:
attempts["flaky"] += 1
if attempts["flaky"] == 1:
raise RuntimeError("transient failure")
return {**data, "flaky": True}
pipeline = self._build(
[
("a", "flaky", flaky_handler, True),
("b", "branch", branch_handler(probe, "b"), True),
],
parallelism=2,
)
result = engine.execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success, msg=str(result.errors))
self.assertEqual(attempts["flaky"], 2)
def test_failed_branch_skips_downstream_layer(self):
engine = ExecutionEngine()
engine.failure_handler.set_retry_policy(
"always_fail", RetryPolicy(max_retries=0)
)
probe = ConcurrencyProbe()
def failing_handler(data, **kwargs):
with probe:
raise RuntimeError("permanent failure")
downstream_calls = {"c": 0}
def downstream_handler(data, **kwargs):
downstream_calls["c"] += 1
return {**data, "c": True}
builder = PipelineBuilder()
builder.add_step("a", "always_fail", handler=failing_handler, parallel_safe=True)
builder.add_step("b", "branch", handler=branch_handler(probe, "b"), parallel_safe=True)
builder.add_step(
"c", "branch", handler=downstream_handler, dependencies=["a", "b"]
)
builder.set_parallelism(2)
pipeline = builder.build("failure_pipeline")
result = engine.execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
self.assertEqual(downstream_calls["c"], 0)
failed_step = next(s for s in pipeline.steps if s.name == "a")
self.assertEqual(failed_step.status, StepStatus.FAILED)
self.assertIsNotNone(failed_step.error)
def test_parallel_safe_not_passed_to_handler_kwargs(self):
received_kwargs = {}
def capturing_handler(data, **kwargs):
received_kwargs.update(kwargs)
return data
builder = PipelineBuilder()
builder.add_step(
"a", "branch", handler=capturing_handler, parallel_safe=True, batch_size=2
)
pipeline = builder.build("kwargs_pipeline")
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertIn("batch_size", received_kwargs)
self.assertNotIn("parallel_safe", received_kwargs)
self.assertNotIn("parallel_safe", pipeline.steps[0].config)
def test_unchanged_echoed_key_does_not_conflict(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
# "a" echoes the base value of "shared" (unchanged),
# "b" legitimately changes it: no false conflict.
("a", "branch", branch_handler(probe, "shared", 1), True),
("b", "branch", branch_handler(probe, "shared", 2), True),
],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(
pipeline, data={"text": "hi", "shared": 1}
)
self.assertTrue(result.success, msg=str(result.errors))
self.assertEqual(result.output["shared"], 2)
def test_ambiguous_equality_counts_as_changed(self):
sentinel = Uncomparable()
def echo(data, **kwargs):
return {**data, "shared": data["shared"]}
def change(data, **kwargs):
return {**data, "shared": "changed"}
pipeline = self._build(
[("a", "branch", echo, True), ("b", "branch", change, True)],
parallelism=2,
)
result = ExecutionEngine().execute_pipeline(
pipeline, data={"text": "hi", "shared": sentinel}
)
# Ambiguous equality must not be treated as unchanged, so both
# branches count as writes and the merge reports the conflict.
self.assertFalse(result.success)
error_text = result.errors[0] if result.errors else ""
self.assertIn("shared", error_text)
def test_non_boolean_parallel_safe_stays_serial(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", branch_handler(probe, "b"), True),
],
parallelism=2,
)
# Simulate a truthy non-bool value reaching the engine (e.g. set
# directly on the step attribute): must not enable concurrency.
for step in pipeline.steps:
step.parallel_safe = "false"
result = ExecutionEngine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success)
self.assertEqual(probe.max_active, 1)
def test_non_dict_result_reports_failure_not_completion(self):
probe = ConcurrencyProbe()
tracking_ids = {}
counter = {"n": 0}
def fake_start(*args, **kwargs):
tid = f"tid_{counter['n']}"
counter["n"] += 1
tracking_ids[kwargs.get("submodule")] = tid
return tid
with patch(
"semantica.pipeline.execution_engine.get_progress_tracker"
) as mock_get:
tracker = MagicMock()
tracker.start_tracking.side_effect = fake_start
mock_get.return_value = tracker
engine = ExecutionEngine()
def non_dict_handler(data, **kwargs):
with probe:
return "not-a-dict"
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", non_dict_handler, True),
],
parallelism=2,
name="progress_pipeline",
)
result = engine.execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
failed_step = next(s for s in pipeline.steps if s.name == "b")
self.assertEqual(failed_step.status, StepStatus.FAILED)
self.assertIsInstance(failed_step.error, ProcessingError)
# Handler ran exactly once: never re-run serially.
self.assertEqual(probe.calls, 2)
b_tid = tracking_ids.get("progress_pipeline:branch:b")
self.assertIsNotNone(
b_tid, f"expected tracking for step b, got {tracking_ids}"
)
b_stops = [
c
for c in tracker.stop_tracking.call_args_list
if (c.args[0] if c.args else c.kwargs.get("tracking_id")) == b_tid
]
self.assertTrue(b_stops)
for call in b_stops:
status = call.kwargs.get("status")
self.assertEqual(status, "failed")
self.assertNotEqual(status, "completed")
def test_same_type_steps_get_distinct_tracking_records(self):
probe = ConcurrencyProbe()
pipeline = self._build(
[
("a", "branch", branch_handler(probe, "a"), True),
("b", "branch", branch_handler(probe, "b"), True),
],
parallelism=2,
name="tracking_pipeline",
)
with patch(
"semantica.pipeline.execution_engine.get_progress_tracker"
) as mock_get:
tracker = MagicMock()
mock_get.return_value = tracker
engine = ExecutionEngine()
result = engine.execute_pipeline(pipeline, data={"text": "hi"})
self.assertTrue(result.success, msg=str(result.errors))
submodules = [
call.kwargs.get("submodule")
for call in tracker.start_tracking.call_args_list
if call.kwargs.get("module") == "pipeline"
]
# Concurrent steps of the same step_type get distinct submodules
# (and therefore distinct tracking IDs), including pipeline identity.
self.assertIn("tracking_pipeline:branch:a", submodules)
self.assertIn("tracking_pipeline:branch:b", submodules)
class TestParallelSafeSerialization(unittest.TestCase):
"""parallel_safe must survive dict and JSON round-trips."""
def _build_pipeline(self):
builder = PipelineBuilder()
builder.add_step("extract", "source", parallel_safe=True, batch_size=10)
builder.add_step("index", "sink", dependencies=["extract"])
builder.set_parallelism(3)
return builder.build("parallel-serialization")
def test_serializer_roundtrip_preserves_parallel_safe(self):
pipeline = self._build_pipeline()
serializer = PipelineSerializer()
serialized = serializer.serialize_pipeline(pipeline, format="dict")
self.assertTrue(serialized["steps"][0]["parallel_safe"])
self.assertFalse(serialized["steps"][1]["parallel_safe"])
restored = serializer.deserialize_pipeline(serialized)
self.assertTrue(restored.steps[0].parallel_safe)
self.assertFalse(restored.steps[1].parallel_safe)
self.assertEqual(restored.config.get("parallelism"), 3)
def test_serializer_json_roundtrip_preserves_parallel_safe(self):
pipeline = self._build_pipeline()
serializer = PipelineSerializer()
serialized = serializer.serialize_pipeline(pipeline, format="json")
restored = serializer.deserialize_pipeline(serialized)
self.assertTrue(restored.steps[0].parallel_safe)
self.assertFalse(restored.steps[1].parallel_safe)
def test_builder_serialize_outputs_parallel_safe(self):
builder = PipelineBuilder()
builder.add_step("extract", "source", parallel_safe=True)
builder.add_step("index", "sink", dependencies=["extract"])
builder.set_parallelism(2)
serialized = builder.serialize(format="dict")
self.assertTrue(serialized["steps"][0]["parallel_safe"])
self.assertFalse(serialized["steps"][1]["parallel_safe"])
self.assertEqual(serialized["config"]["parallelism"], 2)
class Uncomparable:
"""Object whose equality comparison always raises TypeError."""
def __eq__(self, other):
raise TypeError("cannot compare")
class TestParallelDependencyValidation(unittest.TestCase):
"""Cyclic/unknown dependencies must fail with ValidationError in the
parallel path, not RecursionError/KeyError (review finding #1)."""
def _engine(self):
with patch(
"semantica.pipeline.execution_engine.get_progress_tracker"
) as mock_get:
mock_get.return_value = MagicMock()
engine = ExecutionEngine()
return engine
def _handler(self):
return lambda data, **kwargs: data
def test_cyclic_dependencies_raise_validation_error(self):
# Construct the pipeline directly: the builder already rejects
# cycles at build time, so bypassing it exercises the engine-level
# grouping guard (which otherwise hits RecursionError).
steps = [
PipelineStep(
name="a",
step_type="branch",
handler=self._handler(),
dependencies=["b"],
parallel_safe=True,
),
PipelineStep(
name="b",
step_type="branch",
handler=self._handler(),
dependencies=["a"],
parallel_safe=True,
),
]
pipeline = Pipeline(
name="cyclic_pipeline",
steps=steps,
config={"parallelism": 4},
)
result = self._engine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
error_text = result.errors[0] if result.errors else ""
self.assertIn("Circular dependency", error_text)
def test_unknown_dependency_raises_validation_error(self):
# Construct the pipeline directly so the engine-level unknown
# dependency guard is exercised (instead of a builder-time
# KeyError from the grouping DFS).
steps = [
PipelineStep(
name="a",
step_type="branch",
handler=self._handler(),
dependencies=["missing"],
parallel_safe=True,
),
]
pipeline = Pipeline(
name="unknown_dep_pipeline",
steps=steps,
config={"parallelism": 4},
)
result = self._engine().execute_pipeline(pipeline, data={"text": "hi"})
self.assertFalse(result.success)
error_text = result.errors[0] if result.errors else ""
self.assertIn("unknown step 'missing'", error_text)
self.assertIn("'a'", error_text)
class TestParallelSafeMustBeBoolean(unittest.TestCase):
"""parallel_safe must be an explicit boolean everywhere (review #4)."""
def test_add_step_rejects_non_boolean_values(self):
builder = PipelineBuilder()
for invalid in ("false", "true", 1, 0, None, [True]):
with self.assertRaises(ValidationError):
builder.add_step(
"a",
"branch",
handler=lambda data, **kwargs: data,
parallel_safe=invalid,
)
def test_build_pipeline_rejects_non_boolean_values(self):
builder = PipelineBuilder()
config = {
"name": "invalid_parallel_safe",
"steps": [
{"name": "a", "type": "branch", "parallel_safe": "false"}
],
}
with self.assertRaises(ValidationError):
builder.build_pipeline(config)
if __name__ == "__main__":
unittest.main()