# PipelineExecutionEngine Thread-Safety: Concurrency Controls in the Semantica Framework > Learn how PipelineExecutionEngine ensures thread-safety in Semantica Framework using RLock, immutable objects, and thread-local storage for concurrent pipeline execution. - Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica) - Tags: internals - Published: 2026-08-21 --- **The `ExecutionEngine` class ensures thread-safe pipeline execution by employing a `threading.RLock` to guard mutable shared state, utilizing immutable configuration objects, and isolating provenance data through thread-local storage.** The `ExecutionEngine` in [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py) serves as the core orchestrator within the `semantica-agi/semantica` repository, responsible for driving pipeline execution across diverse environments—from HTTP servers to background workers. When multiple threads invoke pipeline runs simultaneously, understanding the specific thread-safety mechanisms embedded in the engine's design is critical for preventing race conditions and ensuring deterministic behavior in concurrent workflows. ## Core Thread-Safety Mechanisms in ExecutionEngine ### Reentrant Lock Protection for Shared State According to the implementation in [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py), the `ExecutionEngine` initializes a `threading.RLock` instance (stored as `self._lock`) during construction. This reentrant lock protects all mutable attributes accessed by multiple threads, including the pipeline registry and internal step counters. Every public method that reads or writes shared state—such as `execute_pipeline()`—acquires this lock first, guaranteeing atomic operations and preventing data races during concurrent access. ### Immutable Configuration by Design The engine eliminates configuration-related race conditions through defensive copying. During initialization, the constructor executes `self.config = dict(config)`, creating a shallow copy of the configuration dictionary that is never modified thereafter. This immutability pattern ensures that concurrent threads cannot alter engine settings mid-execution, removing the need for locks on configuration access and preventing inconsistent state between pipeline runs. ### Synchronous Step Execution Model Individual pipeline steps execute synchronously within the engine's control flow via the `_execute_step` helper method. The engine maintains the lock throughout the step execution sequence, ensuring that modifications to internal tracking structures remain atomic. If a specific step requires internal parallelism—such as a vector-store search utilizing a thread pool—that step must manage its own synchronization, keeping the engine's contract simple and avoiding nested lock complications. ## Managing Concurrent Pipeline Registration and Execution ### Thread-Safe Registry Operations The `PipelineBuilder` class in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) interacts with the engine's shared registry, which is guarded by the same `RLock` instance. When pipelines are registered or unregistered while another thread is executing a pipeline, the registry operations block until the current execution completes. This coordination prevents scenarios where a pipeline definition could be modified or deleted mid-execution, ensuring registry consistency. ### Atomic Retry Logic and Error Handling The engine implements exponential back-off retry logic within the lock boundary. By executing retry attempts while holding `self._lock`, the engine ensures that only one thread can retry a failing step at any given time. This prevents duplicate work across concurrent threads and maintains a single source of truth for error states and retry counters, avoiding the "thundering herd" problem when transient failures occur. ## Provenance Tracking with Thread-Local Isolation The provenance subsystem in [`semantica/pipeline/pipeline_provenance.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_provenance.py) addresses thread-safety through thread-local storage rather than global locks. During pipeline execution, provenance metadata is written to thread-local buffers managed by `PipelineWithProvenance`. These buffers remain isolated per thread throughout the execution lifecycle and are flushed to persistent storage only after the pipeline completes. This design prevents metadata interleaving when multiple threads execute pipelines simultaneously, providing accurate lineage tracking without blocking concurrent operations. ## Practical Implementation Examples The following examples demonstrate safe concurrent usage patterns based on the engine's thread-safety guarantees: ```python # Example 1: Safe concurrent execution from multiple threads from semantica.pipeline import PipelineBuilder, ExecutionEngine import threading builder = PipelineBuilder() pipeline = builder.add_step("load_data", data_loader) \ .add_step("process", processor) \ .build(name="concurrent_pipeline") engine = ExecutionEngine() def run_pipeline(): # Each thread acquires the internal RLock automatically result = engine.execute_pipeline(pipeline, input_data={}) print(f"Success: {result.success}") # Spawn multiple threads accessing the same engine instance threads = [threading.Thread(target=run_pipeline) for _ in range(5)] for t in threads: t.start() for t in threads: t.join() ``` ```python # Example 2: Step-level concurrency with engine safety from semantica.pipeline import ExecutionEngine, PipelineBuilder from concurrent.futures import ThreadPoolExecutor def parallel_vector_search(context): # Step manages its own thread pool while engine lock is released with ThreadPoolExecutor(max_workers=4) as executor: futures = [ executor.submit(search_vector_store, item) for item in context["queries"] ] context["results"] = [f.result() for f in futures] return context builder = PipelineBuilder().add_step("search", parallel_vector_search) pipeline = builder.build(name="parallel_search") engine = ExecutionEngine() # Engine remains thread-safe even when steps spawn internal threads engine.execute_pipeline(pipeline, {"queries": ["q1", "q2", "q3"]}) ``` Real-world usage patterns can be observed in [`semantica/core/orchestrator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/core/orchestrator.py), which demonstrates how the `ExecutionEngine` integrates into larger asynchronous workflows while maintaining these safety guarantees. ## Summary - The `ExecutionEngine` utilizes `threading.RLock` (`self._lock`) in [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py) to serialize access to mutable shared state including registries and execution counters. - Configuration dictionaries are defensively copied during initialization and treated as immutable to prevent concurrent modification risks. - Pipeline steps execute synchronously under the engine's lock, while steps requiring internal concurrency manage their own synchronization. - The `PipelineBuilder` registry in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) shares the engine's lock to prevent modification during active executions. - Retry logic with exponential back-off executes within lock boundaries to prevent duplicate retry attempts across threads. - Provenance tracking uses thread-local storage in [`semantica/pipeline/pipeline_provenance.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_provenance.py) to isolate execution metadata between concurrent pipeline runs. ## Frequently Asked Questions ### Is the ExecutionEngine thread-safe for concurrent pipeline execution? Yes. The `ExecutionEngine` is explicitly designed for concurrent access from multiple threads. By utilizing a `threading.RLock` to protect all mutable internal state—including the pipeline registry and execution counters—the engine allows simultaneous calls to `execute_pipeline()` while guaranteeing that shared resources remain consistent. Each thread acquires the lock before accessing critical sections and releases it upon completion. ### Can I modify the engine configuration after initialization? No. The `ExecutionEngine` creates a defensive copy of the configuration dictionary during initialization (`self.config = dict(config)`) and treats it as immutable. This design intentionally prevents runtime configuration changes to eliminate race conditions. To use different configurations, instantiate separate `ExecutionEngine` objects rather than modifying existing instances. ### How does provenance tracking handle concurrent pipeline executions? The provenance system writes execution metadata to thread-local buffers during pipeline runs, as implemented in [`semantica/pipeline/pipeline_provenance.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_provenance.py). Because each thread maintains its own isolated buffer, provenance records from concurrent executions never intermix. The buffers are flushed to persistent storage only after pipeline completion, ensuring accurate lineage tracking without requiring global locks on the provenance store. ### What precautions are needed when writing custom pipeline steps? Custom steps that spawn their own threads—such as those using `ThreadPoolExecutor`—must manage their own internal synchronization. While the engine guarantees thread-safety for the broader pipeline state, it releases control during step execution if the step manages concurrency internally. Steps should treat input context objects as shared state requiring locks only if they modify the context across thread boundaries, and should never modify the `ExecutionEngine` instance directly.