Reuse Work Across Multiprocess Image Pipelines

Problem Statement

In high-throughput image processing systems, multiple workers frequently execute overlapping pipelines of deterministic transformations (e.g., resizing, normalizing, or filtering). Recomputing shared prefixes wastes CPU and GPU cycles.

Design and implement a thread-safe PipelineExecutor that executes a sequence of operations while maximizing intermediate result reuse. The executor must satisfy three core requirements:

1. Content-Addressed Prefix Caching: Intermediate states must be identified by a deterministic hash of the prior stage's identity, the operation name, and its arguments. If a pipeline shares a prefix with a previous execution, it must resume from the furthest cached intermediate state.
2. Single-Flight Execution: If multiple threads request the same intermediate stage concurrently and it is not yet cached, only one thread should execute the transformation. Other threads must wait for that result rather than duplicating work.
3. LRU Cache Eviction: The cache must hold at most capacity intermediate results. When the limit is reached, evict the least recently accessed completed result. In-flight operations must not be evicted.

Class Interface

from typing import Callable, Any

class PipelineExecutor:
    def __init__(self, capacity: int, apply_op: Callable[[str, bytes, dict[str, Any]], bytes]):
        """
        Args:
            capacity: Maximum number of intermediate results stored in the cache.
            apply_op: Deterministic function f(op_name, input_bytes, kwargs) -> output_bytes.
        """
        pass

    def run(self, image_bytes: bytes, stages: list[tuple[str, dict[str, Any]]]) -> bytes:
        """
        Executes the list of (op_name, kwargs) sequentially starting from image_bytes.
        Returns the final transformed bytes.
        """
        pass

Examples

Example 1: Prefix Reuse

executor = PipelineExecutor(capacity=4, apply_op=my_apply)

# Call 1
out1 = executor.run(img, [("resize", {"w": 256}), ("blur", {"r": 3})])
# Executes: resize(img) -> out_resize, blur(out_resize) -> out1

# Call 2 (different terminal step, same prefix)
out2 = executor.run(img, [("resize", {"w": 256}), ("grayscale", {})])
# Reuses: out_resize from cache
# Executes only: grayscale(out_resize) -> out2

Example 2: Concurrent Deduplication

# Threads T1 and T2 run simultaneously with the same initial operation
# T1: run(img, [("heavy_filter", {"k": 5})])
# T2: run(img, [("heavy_filter", {"k": 5})])
# Result: apply_op("heavy_filter", img, {"k": 5}) is invoked exactly once.

Constraints

• capacity >= 1
• 0 <= len(stages) <= 50
• image_bytes length up to 10 MB
• stages arguments are JSON-serializable dictionaries
• apply_op is deterministic and thread-safe, but may be slow (CPU-bound)
• Cache hits and single-flight waits must avoid busy waiting