PipelineExecutionEngine Thread-Safety: Concurrency Controls in the Semantica Framework

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 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, 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 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 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:

# 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()

# 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, 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 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 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 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. 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}