sgn.subprocess

Parallelize

Bases: SignalEOS

              flowchart TD
              sgn.subprocess.Parallelize[Parallelize]
              sgn.sources.SignalEOS[SignalEOS]

                              sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                


              click sgn.subprocess.Parallelize href "" "sgn.subprocess.Parallelize"
              click sgn.sources.SignalEOS href "" "sgn.sources.SignalEOS"
            

A context manager for running SGN pipelines with elements that implement separate processes or threads.

This class manages the lifecycle of workers (processes or threads) in an SGN pipeline, handling worker creation, execution, and cleanup. It also supports shared memory objects that will be automatically cleaned up on exit through the to_shm() method (only applicable for process mode).

Key features include: - Automatic management of worker lifecycle (creation, starting, joining, cleanup) - Shared memory management for efficient data sharing (process mode only) - Signal handling coordination between main process/thread and workers - Resilience against KeyboardInterrupt (Ctrl+C) - workers catch and ignore these signals, allowing the main process to coordinate a clean shutdown - Orderly shutdown to ensure all resources are properly released - Support for both multiprocessing and threading concurrency models - Automatic detection and invocation when pipeline.run() is called

IMPORTANT: When using process mode, code using Parallelize MUST be wrapped within an if name == "main": block. This is required because SGN uses Python's multiprocessing module with the 'spawn' start method, which requires that the main module be importable.

Example with automatic parallelization (RECOMMENDED): def main(): pipeline = Pipeline() # Add ParallelizeTransformElement, ParallelizeSinkElement, etc. pipeline.run() # Automatically detects and enables parallelization

if __name__ == "__main__":
    main()

Example with manual context manager (LEGACY): def main(): pipeline = Pipeline() with Parallelize(pipeline) as parallelize: parallelize.run()

if __name__ == "__main__":
    main()

Source code in sgn/subprocess.py

class Parallelize(SignalEOS):
    """
    A context manager for running SGN pipelines with elements that implement
    separate processes or threads.

    This class manages the lifecycle of workers (processes or threads) in an SGN
    pipeline, handling worker creation, execution, and cleanup. It also supports
    shared memory objects that will be automatically cleaned up on exit through the
    to_shm() method (only applicable for process mode).

    Key features include:
    - Automatic management of worker lifecycle (creation, starting, joining, cleanup)
    - Shared memory management for efficient data sharing (process mode only)
    - Signal handling coordination between main process/thread and workers
    - Resilience against KeyboardInterrupt (Ctrl+C) - workers catch and ignore these
      signals, allowing the main process to coordinate a clean shutdown
    - Orderly shutdown to ensure all resources are properly released
    - Support for both multiprocessing and threading concurrency models
    - Automatic detection and invocation when pipeline.run() is called

    IMPORTANT: When using process mode, code using Parallelize MUST be
    wrapped within an if __name__ == "__main__": block. This is required because SGN
    uses Python's multiprocessing module with the 'spawn' start method, which requires
    that the main module be importable.

    Example with automatic parallelization (RECOMMENDED):
        def main():
            pipeline = Pipeline()
            # Add ParallelizeTransformElement, ParallelizeSinkElement, etc.
            pipeline.run()  # Automatically detects and enables parallelization

        if __name__ == "__main__":
            main()

    Example with manual context manager (LEGACY):
        def main():
            pipeline = Pipeline()
            with Parallelize(pipeline) as parallelize:
                parallelize.run()

        if __name__ == "__main__":
            main()
    """

    shm_list: list = []
    instance_list: list = []
    enabled: bool = False
    # The hard timeout before a worker gets terminated.
    # Workers should cleanup after themselves within this time and exit cleanly.
    # This is a "global" property applied to all subprocesses / subthreads
    join_timeout: float = 10.0  # Increased for CI environments
    # Default flag for whether to use threading (False means use multiprocessing)
    use_threading_default: bool = False
    # Instance variable for thread mode
    use_threading: bool = False

    def __init__(self, pipeline=None, use_threading: bool | None = None):
        """
        Initialize the Parallelize context manager.

        Args:
            pipeline: The pipeline to run
            use_threading: Whether to use threading instead of multiprocessing.
                          If not specified, uses the use_threading_default
        """
        self.pipeline = pipeline
        # Use the specified mode, or fall back to the class default
        self.use_threading = (
            use_threading
            if use_threading is not None
            else Parallelize.use_threading_default
        )

    def __enter__(self):
        try:
            multiprocessing.set_start_method("spawn")
        except RuntimeError:
            pass
        super().__enter__()
        for e in Parallelize.instance_list:
            e.worker.start()
        Parallelize.enabled = True
        return self

    def __exit__(self, exc_type, exc_value, exc_traceback):
        super().__exit__(exc_type, exc_value, exc_traceback)
        # rejoin all the workers
        for e in Parallelize.instance_list:
            if e.in_queue is not None:
                e.in_queue.cancel_join_thread()
            if e.out_queue is not None:
                e.out_queue.cancel_join_thread()

            if (
                e.worker is not None
                and hasattr(e.worker, "is_alive")
                and e.worker.is_alive()
            ):
                e.worker.join(Parallelize.join_timeout)
                # Only processes can be killed, threads will naturally terminate
                if hasattr(e.worker, "kill") and e.worker.is_alive():
                    e.worker.kill()

        Parallelize.instance_list = []

        # Clean up shared memory (only applicable for process mode)
        if not self.use_threading:
            for d in Parallelize.shm_list:
                multiprocessing.shared_memory.SharedMemory(name=d["name"]).unlink()
            Parallelize.shm_list = []

        Parallelize.enabled = False

    @staticmethod
    def to_shm(name, bytez, **kwargs):
        """
        Create a shared memory object that can be accessed by subprocesses.

        Note: This is only applicable in process mode. In thread mode, shared memory
        is not necessary since threads share the same address space.

        This method creates a shared memory segment that will be automatically
        cleaned up when the Parallelize context manager exits. The shared memory can be
        used to efficiently share large data between processes without serialization
        overhead.

        Args:
            name (str): Unique identifier for the shared memory block
            bytez (bytes or bytearray): Data to store in shared memory
            **kwargs: Additional metadata to store with the shared memory reference

        Returns:
            dict: A dictionary containing the shared memory object and metadata
                  with keys:
                - "name": The name of the shared memory block
                - "shm": The SharedMemory object
                - Any additional key-value pairs from kwargs

        Raises:
            FileExistsError: If shared memory with the given name already exists

        Example:
            shared_data = bytearray("Hello world", "utf-8")
            shm_ref = SubProcess.to_shm("example_data", shared_data)
        """
        try:
            shm = multiprocessing.shared_memory.SharedMemory(
                name=name, create=True, size=len(bytez)
            )
        except FileExistsError as e:
            print(f"Shared memory: {name} already exists")
            print(
                "You can clear the memory by doing "
                f"multiprocessing.shared_memory.SharedMemory(name='{name}').unlink()\n"
            )
            for d in Parallelize.shm_list:
                multiprocessing.shared_memory.SharedMemory(name=d["name"]).unlink()
            Parallelize.shm_list = []
            raise e

        shm.buf[: len(bytez)] = bytez
        out = {"name": name, "shm": shm, **kwargs}
        Parallelize.shm_list.append(out)
        return out

    def run(self):
        """
        Run the pipeline managed by this Parallelize instance.

        This method executes the associated pipeline and ensures proper cleanup
        of worker resources, even in the case of exceptions. It signals all
        workers to stop when the pipeline execution completes or if an exception
        occurs.

        Raises:
            RuntimeError: If an exception occurs during pipeline execution
            AssertionError: If no pipeline was provided to the SubProcess
        """
        assert (
            self.pipeline is not None
        ), "Pipeline must be provided to Parallelize constructor before running"
        try:
            # Disable auto-parallelization since we're already in a Parallelize context
            self.pipeline.run(auto_parallelize=False)
        except Exception:
            # Signal all workers to stop when an exception occurs
            for p in Parallelize.instance_list:
                p.worker_stop.set()

            # Clean up all workers
            for p in Parallelize.instance_list:
                if p.in_queue is not None:
                    p.in_queue.cancel_join_thread()
                if p.out_queue is not None:
                    p.out_queue.cancel_join_thread()

                if (
                    p.worker is not None
                    and hasattr(p.worker, "is_alive")
                    and p.worker.is_alive()
                ):
                    p.worker.join(Parallelize.join_timeout)
                    if hasattr(p.worker, "kill") and p.worker.is_alive():
                        p.worker.kill()
            raise

        # Signal all workers to stop when pipeline completes normally
        for p in Parallelize.instance_list:
            p.worker_stop.set()

    @staticmethod
    def needs_parallelization(pipeline):
        """
        Check if a pipeline contains any elements that require parallelization.

        Args:
            pipeline: The Pipeline instance to check

        Returns:
            bool: True if the pipeline contains any Parallelize* elements
        """
        # Check if any element is a subclass of _ParallelizeBase
        return any(
            isinstance(element, _ParallelizeBase) for element in pipeline.elements
        )

__init__(pipeline=None, use_threading=None)

Initialize the Parallelize context manager.

Parameters:

Name	Type	Description	Default
pipeline		The pipeline to run	None
use_threading	bool | None	Whether to use threading instead of multiprocessing. If not specified, uses the use_threading_default	None

Source code in sgn/subprocess.py

def __init__(self, pipeline=None, use_threading: bool | None = None):
    """
    Initialize the Parallelize context manager.

    Args:
        pipeline: The pipeline to run
        use_threading: Whether to use threading instead of multiprocessing.
                      If not specified, uses the use_threading_default
    """
    self.pipeline = pipeline
    # Use the specified mode, or fall back to the class default
    self.use_threading = (
        use_threading
        if use_threading is not None
        else Parallelize.use_threading_default
    )

needs_parallelization(pipeline) staticmethod

Check if a pipeline contains any elements that require parallelization.

Parameters:

Name	Type	Description	Default
pipeline		The Pipeline instance to check	required

Returns:

Name	Type	Description
bool		True if the pipeline contains any Parallelize* elements

Source code in sgn/subprocess.py

@staticmethod
def needs_parallelization(pipeline):
    """
    Check if a pipeline contains any elements that require parallelization.

    Args:
        pipeline: The Pipeline instance to check

    Returns:
        bool: True if the pipeline contains any Parallelize* elements
    """
    # Check if any element is a subclass of _ParallelizeBase
    return any(
        isinstance(element, _ParallelizeBase) for element in pipeline.elements
    )

run()

Run the pipeline managed by this Parallelize instance.

This method executes the associated pipeline and ensures proper cleanup of worker resources, even in the case of exceptions. It signals all workers to stop when the pipeline execution completes or if an exception occurs.

Raises:

Type	Description
RuntimeError	If an exception occurs during pipeline execution
AssertionError	If no pipeline was provided to the SubProcess

Source code in sgn/subprocess.py

def run(self):
    """
    Run the pipeline managed by this Parallelize instance.

    This method executes the associated pipeline and ensures proper cleanup
    of worker resources, even in the case of exceptions. It signals all
    workers to stop when the pipeline execution completes or if an exception
    occurs.

    Raises:
        RuntimeError: If an exception occurs during pipeline execution
        AssertionError: If no pipeline was provided to the SubProcess
    """
    assert (
        self.pipeline is not None
    ), "Pipeline must be provided to Parallelize constructor before running"
    try:
        # Disable auto-parallelization since we're already in a Parallelize context
        self.pipeline.run(auto_parallelize=False)
    except Exception:
        # Signal all workers to stop when an exception occurs
        for p in Parallelize.instance_list:
            p.worker_stop.set()

        # Clean up all workers
        for p in Parallelize.instance_list:
            if p.in_queue is not None:
                p.in_queue.cancel_join_thread()
            if p.out_queue is not None:
                p.out_queue.cancel_join_thread()

            if (
                p.worker is not None
                and hasattr(p.worker, "is_alive")
                and p.worker.is_alive()
            ):
                p.worker.join(Parallelize.join_timeout)
                if hasattr(p.worker, "kill") and p.worker.is_alive():
                    p.worker.kill()
        raise

    # Signal all workers to stop when pipeline completes normally
    for p in Parallelize.instance_list:
        p.worker_stop.set()

to_shm(name, bytez, **kwargs) staticmethod

Create a shared memory object that can be accessed by subprocesses.

Note: This is only applicable in process mode. In thread mode, shared memory is not necessary since threads share the same address space.

This method creates a shared memory segment that will be automatically cleaned up when the Parallelize context manager exits. The shared memory can be used to efficiently share large data between processes without serialization overhead.

Parameters:

Name	Type	Description	Default
name	str	Unique identifier for the shared memory block	required
bytez	bytes or bytearray	Data to store in shared memory	required
**kwargs		Additional metadata to store with the shared memory reference	{}

Returns:

Name	Type	Description
dict		A dictionary containing the shared memory object and metadata with keys: - "name": The name of the shared memory block - "shm": The SharedMemory object - Any additional key-value pairs from kwargs

Raises:

Type	Description
FileExistsError	If shared memory with the given name already exists

Example

shared_data = bytearray("Hello world", "utf-8") shm_ref = SubProcess.to_shm("example_data", shared_data)

Source code in sgn/subprocess.py

@staticmethod
def to_shm(name, bytez, **kwargs):
    """
    Create a shared memory object that can be accessed by subprocesses.

    Note: This is only applicable in process mode. In thread mode, shared memory
    is not necessary since threads share the same address space.

    This method creates a shared memory segment that will be automatically
    cleaned up when the Parallelize context manager exits. The shared memory can be
    used to efficiently share large data between processes without serialization
    overhead.

    Args:
        name (str): Unique identifier for the shared memory block
        bytez (bytes or bytearray): Data to store in shared memory
        **kwargs: Additional metadata to store with the shared memory reference

    Returns:
        dict: A dictionary containing the shared memory object and metadata
              with keys:
            - "name": The name of the shared memory block
            - "shm": The SharedMemory object
            - Any additional key-value pairs from kwargs

    Raises:
        FileExistsError: If shared memory with the given name already exists

    Example:
        shared_data = bytearray("Hello world", "utf-8")
        shm_ref = SubProcess.to_shm("example_data", shared_data)
    """
    try:
        shm = multiprocessing.shared_memory.SharedMemory(
            name=name, create=True, size=len(bytez)
        )
    except FileExistsError as e:
        print(f"Shared memory: {name} already exists")
        print(
            "You can clear the memory by doing "
            f"multiprocessing.shared_memory.SharedMemory(name='{name}').unlink()\n"
        )
        for d in Parallelize.shm_list:
            multiprocessing.shared_memory.SharedMemory(name=d["name"]).unlink()
        Parallelize.shm_list = []
        raise e

    shm.buf[: len(bytez)] = bytez
    out = {"name": name, "shm": shm, **kwargs}
    Parallelize.shm_list.append(out)
    return out

ParallelizeSinkElement dataclass

Bases: SinkElement, _ParallelizeBase, Parallelize

              flowchart TD
              sgn.subprocess.ParallelizeSinkElement[ParallelizeSinkElement]
              sgn.base.SinkElement[SinkElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]
              sgn.subprocess._ParallelizeBase[_ParallelizeBase]
              sgn.subprocess.Parallelize[Parallelize]
              sgn.sources.SignalEOS[SignalEOS]

                              sgn.base.SinkElement --> sgn.subprocess.ParallelizeSinkElement
                                sgn.base.ElementLike --> sgn.base.SinkElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                


                sgn.subprocess._ParallelizeBase --> sgn.subprocess.ParallelizeSinkElement
                                sgn.subprocess.Parallelize --> sgn.subprocess._ParallelizeBase
                                sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                


                sgn.subprocess.Parallelize --> sgn.subprocess.ParallelizeSinkElement
                                sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                



              click sgn.subprocess.ParallelizeSinkElement href "" "sgn.subprocess.ParallelizeSinkElement"
              click sgn.base.SinkElement href "" "sgn.base.SinkElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"
              click sgn.subprocess._ParallelizeBase href "" "sgn.subprocess._ParallelizeBase"
              click sgn.subprocess.Parallelize href "" "sgn.subprocess.Parallelize"
              click sgn.sources.SignalEOS href "" "sgn.sources.SignalEOS"
            

A Sink element that runs data consumption logic in a separate process or thread.

This class extends the standard SinkElement to execute its processing in a separate worker (process or thread). It communicates with the main process/thread through input and output queues, and manages the worker lifecycle. Subclasses must implement the worker_process method to define the consumption logic that runs in the worker.

The design intentionally avoids passing class or instance references to the worker to prevent pickling issues when using process mode. Instead, it passes all necessary data and resources via function arguments.

The implementation includes special handling for KeyboardInterrupt signals. When Ctrl+C is pressed in the terminal, workers will catch and ignore the KeyboardInterrupt, allowing them to continue processing while the main process coordinates a graceful shutdown. This prevents data loss and ensures all resources are properly cleaned up.

Attributes:

Name	Type	Description
queue_maxsize	int	Maximum size of the communication queues
err_maxsize	int	Maximum size for error data
_use_threading_override	bool	Set to True to use threading or False to use multiprocessing. If not specified, uses the Parallelize.use_threading_default

Example with default process mode

@dataclass class MyLoggingSinkElement(ParallelizeSinkElement): def pull(self, pad, frame): if frame.EOS: self.mark_eos(pad) # Send the frame to the worker self.in_queue.put((pad.name, frame))

def worker_process(self, context: WorkerContext):
    try:
        # Get data from the main process/thread
        pad_name, frame = context.input_queue.get(timeout=0.1)

        # Process or log the data
        if not frame.EOS:
            print(f"Sink received on {pad_name}: {frame.data}")
        else:
            print(f"Sink received EOS on {pad_name}")

    except queue.Empty:
        pass

Example with thread mode

@dataclass class MyThreadedSinkElement(ParallelizeSinkElement): _use_threading_override = True # Implementation same as above

Source code in sgn/subprocess.py

@dataclass
class ParallelizeSinkElement(SinkElement, _ParallelizeBase, Parallelize):
    """
    A Sink element that runs data consumption logic in a separate process or thread.

    This class extends the standard SinkElement to execute its processing in a
    separate worker (process or thread). It communicates with the main process/thread
    through input and output queues, and manages the worker lifecycle. Subclasses must
    implement the worker_process method to define the consumption logic that runs
    in the worker.

    The design intentionally avoids passing class or instance references to the
    worker to prevent pickling issues when using process mode. Instead, it passes all
    necessary data and resources via function arguments.

    The implementation includes special handling for KeyboardInterrupt signals.
    When Ctrl+C is pressed in the terminal, workers will catch and ignore the
    KeyboardInterrupt, allowing them to continue processing while the main process
    coordinates a graceful shutdown. This prevents data loss and ensures all resources
    are properly cleaned up.

    Attributes:
        queue_maxsize (int, optional): Maximum size of the communication queues
        err_maxsize (int): Maximum size for error data
        _use_threading_override (bool, optional): Set to True to use threading or
            False to use multiprocessing. If not specified, uses the
            Parallelize.use_threading_default

    Example with default process mode:
        @dataclass
        class MyLoggingSinkElement(ParallelizeSinkElement):
            def pull(self, pad, frame):
                if frame.EOS:
                    self.mark_eos(pad)
                # Send the frame to the worker
                self.in_queue.put((pad.name, frame))

            def worker_process(self, context: WorkerContext):
                try:
                    # Get data from the main process/thread
                    pad_name, frame = context.input_queue.get(timeout=0.1)

                    # Process or log the data
                    if not frame.EOS:
                        print(f"Sink received on {pad_name}: {frame.data}")
                    else:
                        print(f"Sink received EOS on {pad_name}")

                except queue.Empty:
                    pass

    Example with thread mode:
        @dataclass
        class MyThreadedSinkElement(ParallelizeSinkElement):
            _use_threading_override = True
            # Implementation same as above
    """

    internal = _ParallelizeBase.internal

    def __post_init__(self):
        SinkElement.__post_init__(self)
        _ParallelizeBase.__post_init__(self)

ParallelizeSourceElement dataclass

Bases: SourceElement, _ParallelizeBase, Parallelize

              flowchart TD
              sgn.subprocess.ParallelizeSourceElement[ParallelizeSourceElement]
              sgn.base.SourceElement[SourceElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]
              sgn.subprocess._ParallelizeBase[_ParallelizeBase]
              sgn.subprocess.Parallelize[Parallelize]
              sgn.sources.SignalEOS[SignalEOS]

                              sgn.base.SourceElement --> sgn.subprocess.ParallelizeSourceElement
                                sgn.base.ElementLike --> sgn.base.SourceElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                


                sgn.subprocess._ParallelizeBase --> sgn.subprocess.ParallelizeSourceElement
                                sgn.subprocess.Parallelize --> sgn.subprocess._ParallelizeBase
                                sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                


                sgn.subprocess.Parallelize --> sgn.subprocess.ParallelizeSourceElement
                                sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                



              click sgn.subprocess.ParallelizeSourceElement href "" "sgn.subprocess.ParallelizeSourceElement"
              click sgn.base.SourceElement href "" "sgn.base.SourceElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"
              click sgn.subprocess._ParallelizeBase href "" "sgn.subprocess._ParallelizeBase"
              click sgn.subprocess.Parallelize href "" "sgn.subprocess.Parallelize"
              click sgn.sources.SignalEOS href "" "sgn.sources.SignalEOS"
            

A Source element that generates data in a separate process or thread.

This class extends the standard SourceElement to execute its data generation logic in a separate worker (process or thread). It communicates with the main process through output queues, and manages the worker lifecycle. Subclasses must implement the worker_process method to define the data generation logic that runs in the worker.

The design intentionally avoids passing class or instance references to the worker to prevent pickling issues when using process mode. Instead, it passes all necessary data and resources via function arguments.

The implementation includes special handling for KeyboardInterrupt signals. When Ctrl+C is pressed in the terminal, workers will catch and ignore the KeyboardInterrupt, allowing them to continue processing while the main process coordinates a graceful shutdown. This prevents data loss and ensures all resources are properly cleaned up.

Attributes:

Name	Type	Description
queue_maxsize	int	Maximum size of the communication queues
err_maxsize	int	Maximum size for error data
frame_factory	Callable	Function to create Frame objects
at_eos	bool	Flag indicating if End-Of-Stream has been reached
_use_threading_override	bool	Set to True to use threading or False to use multiprocessing. If not specified, uses the Parallelize.use_threading_default

Example with default process mode

@dataclass class MyDataSourceElement(ParallelizeSourceElement): def post_init(self): super().post_init() # Dictionary to track EOS status for each pad self.pad_eos = {pad.name: False for pad in self.source_pads}

def new(self, pad):
    # Check if this pad has already reached EOS
    if self.pad_eos[pad.name]:
        return Frame(data=None, EOS=True)

    try:
        # Get data generated by the worker
        # In a real implementation, you might use pad-specific queues
        # or have the worker send pad-specific data
        data = self.out_queue.get(timeout=1)

        # Check for EOS signal (None typically indicates EOS)
        if data is None:
            self.pad_eos[pad.name] = True
            # If all pads have reached EOS, set global EOS flag
            if all(self.pad_eos.values()):
                self.at_eos = True
            return Frame(data=None, EOS=True)

        # For data intended for other pads, you might implement
        # custom routing logic here

        return Frame(data=data)
    except queue.Empty:
        # Return an empty frame if no data is available
        return Frame(data=None)

def worker_process(self, context: WorkerContext):
    # Generate data and send it back to the main process/thread
    for i in range(10):
        if context.should_stop():
            break
        context.output_queue.put(f"Generated data {i}")
        time.sleep(0.5)

    # Signal end of stream with None
    context.output_queue.put(None)

    # Wait for worker_stop before terminating
    # This prevents "worker stopped before EOS" errors
    while not context.should_stop():
        time.sleep(0.1)

Example with thread mode

@dataclass class MyThreadedSourceElement(ParallelizeSourceElement): _use_threading_override = True

def __post_init__(self):
    super().__post_init__()
    # Dictionary to track EOS status for each pad
    self.pad_eos = {pad.name: False for pad in self.source_pads}

def new(self, pad):
    # Similar implementation as in the process mode example,
    # but might use threading-specific features if needed
    if self.pad_eos[pad.name]:
        return Frame(data=None, EOS=True)

    # Rest of implementation same as the process mode example

Source code in sgn/subprocess.py

@dataclass
class ParallelizeSourceElement(SourceElement, _ParallelizeBase, Parallelize):
    """
    A Source element that generates data in a separate process or thread.

    This class extends the standard SourceElement to execute its data generation logic
    in a separate worker (process or thread). It communicates with the main process
    through output queues, and manages the worker lifecycle. Subclasses must implement
    the worker_process method to define the data generation logic that runs in
    the worker.

    The design intentionally avoids passing class or instance references to the
    worker to prevent pickling issues when using process mode. Instead, it passes all
    necessary data and resources via function arguments.

    The implementation includes special handling for KeyboardInterrupt signals.
    When Ctrl+C is pressed in the terminal, workers will catch and ignore the
    KeyboardInterrupt, allowing them to continue processing while the main process
    coordinates a graceful shutdown. This prevents data loss and ensures all resources
    are properly cleaned up.

    Attributes:
        queue_maxsize (int, optional): Maximum size of the communication queues
        err_maxsize (int): Maximum size for error data
        frame_factory (Callable, optional): Function to create Frame objects
        at_eos (bool): Flag indicating if End-Of-Stream has been reached
        _use_threading_override (bool, optional): Set to True to use threading or
            False to use multiprocessing. If not specified, uses the
            Parallelize.use_threading_default

    Example with default process mode:
        @dataclass
        class MyDataSourceElement(ParallelizeSourceElement):
            def __post_init__(self):
                super().__post_init__()
                # Dictionary to track EOS status for each pad
                self.pad_eos = {pad.name: False for pad in self.source_pads}

            def new(self, pad):
                # Check if this pad has already reached EOS
                if self.pad_eos[pad.name]:
                    return Frame(data=None, EOS=True)

                try:
                    # Get data generated by the worker
                    # In a real implementation, you might use pad-specific queues
                    # or have the worker send pad-specific data
                    data = self.out_queue.get(timeout=1)

                    # Check for EOS signal (None typically indicates EOS)
                    if data is None:
                        self.pad_eos[pad.name] = True
                        # If all pads have reached EOS, set global EOS flag
                        if all(self.pad_eos.values()):
                            self.at_eos = True
                        return Frame(data=None, EOS=True)

                    # For data intended for other pads, you might implement
                    # custom routing logic here

                    return Frame(data=data)
                except queue.Empty:
                    # Return an empty frame if no data is available
                    return Frame(data=None)

            def worker_process(self, context: WorkerContext):
                # Generate data and send it back to the main process/thread
                for i in range(10):
                    if context.should_stop():
                        break
                    context.output_queue.put(f"Generated data {i}")
                    time.sleep(0.5)

                # Signal end of stream with None
                context.output_queue.put(None)

                # Wait for worker_stop before terminating
                # This prevents "worker stopped before EOS" errors
                while not context.should_stop():
                    time.sleep(0.1)

    Example with thread mode:
        @dataclass
        class MyThreadedSourceElement(ParallelizeSourceElement):
            _use_threading_override = True

            def __post_init__(self):
                super().__post_init__()
                # Dictionary to track EOS status for each pad
                self.pad_eos = {pad.name: False for pad in self.source_pads}

            def new(self, pad):
                # Similar implementation as in the process mode example,
                # but might use threading-specific features if needed
                if self.pad_eos[pad.name]:
                    return Frame(data=None, EOS=True)

                # Rest of implementation same as the process mode example
    """

    frame_factory: Callable = Frame
    at_eos: bool = False

    internal = _ParallelizeBase.internal

    def __post_init__(self):
        SourceElement.__post_init__(self)
        _ParallelizeBase.__post_init__(self)

ParallelizeTransformElement dataclass

Bases: TransformElement, _ParallelizeBase, Parallelize

              flowchart TD
              sgn.subprocess.ParallelizeTransformElement[ParallelizeTransformElement]
              sgn.base.TransformElement[TransformElement]
              sgn.base.ElementLike[ElementLike]
              sgn.base.UniqueID[UniqueID]
              sgn.subprocess._ParallelizeBase[_ParallelizeBase]
              sgn.subprocess.Parallelize[Parallelize]
              sgn.sources.SignalEOS[SignalEOS]

                              sgn.base.TransformElement --> sgn.subprocess.ParallelizeTransformElement
                                sgn.base.ElementLike --> sgn.base.TransformElement
                                sgn.base.UniqueID --> sgn.base.ElementLike
                


                sgn.subprocess._ParallelizeBase --> sgn.subprocess.ParallelizeTransformElement
                                sgn.subprocess.Parallelize --> sgn.subprocess._ParallelizeBase
                                sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                


                sgn.subprocess.Parallelize --> sgn.subprocess.ParallelizeTransformElement
                                sgn.sources.SignalEOS --> sgn.subprocess.Parallelize
                



              click sgn.subprocess.ParallelizeTransformElement href "" "sgn.subprocess.ParallelizeTransformElement"
              click sgn.base.TransformElement href "" "sgn.base.TransformElement"
              click sgn.base.ElementLike href "" "sgn.base.ElementLike"
              click sgn.base.UniqueID href "" "sgn.base.UniqueID"
              click sgn.subprocess._ParallelizeBase href "" "sgn.subprocess._ParallelizeBase"
              click sgn.subprocess.Parallelize href "" "sgn.subprocess.Parallelize"
              click sgn.sources.SignalEOS href "" "sgn.sources.SignalEOS"
            

A Transform element that runs processing logic in a separate process or thread.

This class extends the standard TransformElement to execute its processing in a separate worker (process or thread). It communicates with the main process/thread through input and output queues, and manages the worker lifecycle. Subclasses must implement the worker_process method to define the processing logic that runs in the worker.

The design intentionally avoids passing class or instance references to the worker to prevent pickling issues when using process mode. Instead, it passes all necessary data and resources via function arguments.

The implementation includes special handling for KeyboardInterrupt signals. When Ctrl+C is pressed in the terminal, workers will catch and ignore the KeyboardInterrupt, allowing them to continue processing while the main process coordinates a graceful shutdown. This prevents data loss and ensures all resources are properly cleaned up.

Attributes:

Name	Type	Description
queue_maxsize	int	Maximum size of the communication queues
err_maxsize	int	Maximum size for error data
at_eos	bool	Flag indicating if End-Of-Stream has been reached
_use_threading_override	bool	Set to True to use threading or False to use multiprocessing. If not specified, uses the Parallelize.use_threading_default

Example with default process mode

@dataclass class MyProcessingElement(ParallelizeTransformElement): multiplier: int = 2 # Instance attributes become worker parameters

def pull(self, pad, frame):
    # Send the frame to the worker
    self.in_queue.put(frame)
    if frame.EOS:
        self.at_eos = True

def worker_process(self, context: WorkerContext, multiplier: int):
    # Process data in the worker using the clean context
    try:
        frame = context.input_queue.get(timeout=0.1)
        if frame and not frame.EOS:
            frame.data *= multiplier
            context.output_queue.put(frame)
    except queue.Empty:
        pass

def new(self, pad):
    # Get processed data from the worker
    return self.out_queue.get()

Example with thread mode

@dataclass class MyThreadedElement(ParallelizeTransformElement): _use_threading_override = True # Implementation same as above

Example

@dataclass class MyProcessingElement(ParallelizeTransformElement): multiplier: int = 2 threshold: float = 0.5

def pull(self, pad, frame):
    self.in_queue.put(frame)
    if frame.EOS:
        self.at_eos = True

def worker_process(
    self, context: WorkerContext, multiplier: int, threshold: float
):
    try:
        frame = context.input_queue.get(timeout=0.1)
        if frame and not frame.EOS and frame.data > threshold:
            frame.data *= multiplier
            context.output_queue.put(frame)
    except queue.Empty:
        pass

def new(self, pad):
    return self.out_queue.get()

Source code in sgn/subprocess.py

@dataclass
class ParallelizeTransformElement(TransformElement, _ParallelizeBase, Parallelize):
    """
    A Transform element that runs processing logic in a separate process or thread.

    This class extends the standard TransformElement to execute its processing in a
    separate worker (process or thread). It communicates with the main process/thread
    through input and output queues, and manages the worker lifecycle. Subclasses must
    implement the worker_process method to define the processing logic that runs
    in the worker.

    The design intentionally avoids passing class or instance references to the
    worker to prevent pickling issues when using process mode. Instead, it passes all
    necessary data and resources via function arguments.

    The implementation includes special handling for KeyboardInterrupt signals.
    When Ctrl+C is pressed in the terminal, workers will catch and ignore the
    KeyboardInterrupt, allowing them to continue processing while the main process
    coordinates a graceful shutdown. This prevents data loss and ensures all resources
    are properly cleaned up.

    Attributes:
        queue_maxsize (int, optional): Maximum size of the communication queues
        err_maxsize (int): Maximum size for error data
        at_eos (bool): Flag indicating if End-Of-Stream has been reached
        _use_threading_override (bool, optional): Set to True to use threading or
            False to use multiprocessing. If not specified, uses the
            Parallelize.use_threading_default

    Example with default process mode:
        @dataclass
        class MyProcessingElement(ParallelizeTransformElement):
            multiplier: int = 2  # Instance attributes become worker parameters

            def pull(self, pad, frame):
                # Send the frame to the worker
                self.in_queue.put(frame)
                if frame.EOS:
                    self.at_eos = True

            def worker_process(self, context: WorkerContext, multiplier: int):
                # Process data in the worker using the clean context
                try:
                    frame = context.input_queue.get(timeout=0.1)
                    if frame and not frame.EOS:
                        frame.data *= multiplier
                        context.output_queue.put(frame)
                except queue.Empty:
                    pass

            def new(self, pad):
                # Get processed data from the worker
                return self.out_queue.get()

    Example with thread mode:
        @dataclass
        class MyThreadedElement(ParallelizeTransformElement):
            _use_threading_override = True
            # Implementation same as above

    Example:
        @dataclass
        class MyProcessingElement(ParallelizeTransformElement):
            multiplier: int = 2
            threshold: float = 0.5

            def pull(self, pad, frame):
                self.in_queue.put(frame)
                if frame.EOS:
                    self.at_eos = True

            def worker_process(
                self, context: WorkerContext, multiplier: int, threshold: float
            ):
                try:
                    frame = context.input_queue.get(timeout=0.1)
                    if frame and not frame.EOS and frame.data > threshold:
                        frame.data *= multiplier
                        context.output_queue.put(frame)
                except queue.Empty:
                    pass

            def new(self, pad):
                return self.out_queue.get()
    """

    at_eos: bool = False

    internal = _ParallelizeBase.internal

    def __post_init__(self):
        TransformElement.__post_init__(self)
        _ParallelizeBase.__post_init__(self)

QueueProtocol

Bases: Protocol

              flowchart TD
              sgn.subprocess.QueueProtocol[QueueProtocol]

              

              click sgn.subprocess.QueueProtocol href "" "sgn.subprocess.QueueProtocol"
            

Protocol defining a common Queue interface.

Source code in sgn/subprocess.py

class QueueProtocol(Protocol):
    """Protocol defining a common Queue interface."""

    def get(self, block: bool = True, timeout: float | None = None) -> Any:
        """Get an item from the queue."""
        ...

    def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:
        """Put an item into the queue."""
        ...

    def empty(self) -> bool:
        """Return True if the queue is empty."""
        ...

    def get_nowait(self) -> Any:
        """Get an item from the queue without blocking."""
        ...

    def put_nowait(self, item: Any) -> None:
        """Put an item into the queue without blocking."""
        ...

empty()

Return True if the queue is empty.

Source code in sgn/subprocess.py

def empty(self) -> bool:
    """Return True if the queue is empty."""
    ...

get(block=True, timeout=None)

Get an item from the queue.

Source code in sgn/subprocess.py

def get(self, block: bool = True, timeout: float | None = None) -> Any:
    """Get an item from the queue."""
    ...

get_nowait()

Get an item from the queue without blocking.

Source code in sgn/subprocess.py

def get_nowait(self) -> Any:
    """Get an item from the queue without blocking."""
    ...

put(item, block=True, timeout=None)

Put an item into the queue.

Source code in sgn/subprocess.py

def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:
    """Put an item into the queue."""
    ...

put_nowait(item)

Put an item into the queue without blocking.

Source code in sgn/subprocess.py

def put_nowait(self, item: Any) -> None:
    """Put an item into the queue without blocking."""
    ...

QueueWrapper

A wrapper that provides a unified interface for both Queue implementations.

This abstraction handles the differences between multiprocessing.Queue and queue.Queue APIs, specifically providing no-op implementations for multiprocessing-specific methods when wrapping a queue.Queue.

Source code in sgn/subprocess.py

class QueueWrapper:
    """
    A wrapper that provides a unified interface for both Queue implementations.

    This abstraction handles the differences between multiprocessing.Queue and
    queue.Queue APIs, specifically providing no-op implementations for
    multiprocessing-specific methods when wrapping a queue.Queue.
    """

    def __init__(self, queue_instance: QueueProtocol):
        self._queue = queue_instance

    def get(self, block: bool = True, timeout: float | None = None) -> Any:
        """Get an item from the queue."""
        return self._queue.get(block=block, timeout=timeout)

    def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:
        """Put an item into the queue."""
        self._queue.put(item, block=block, timeout=timeout)

    def empty(self) -> bool:
        """Return True if the queue is empty."""
        return self._queue.empty()

    def get_nowait(self) -> Any:
        """Get an item from the queue without blocking."""
        return self._queue.get_nowait()

    def put_nowait(self, item: Any) -> None:  # pragma: no cover
        """Put an item into the queue without blocking."""
        self._queue.put_nowait(item)

    def cancel_join_thread(self) -> None:  # pragma: no cover
        """Cancel the background thread (no-op for queue.Queue)."""
        if hasattr(self._queue, "cancel_join_thread"):
            self._queue.cancel_join_thread()

    def close(self) -> None:  # pragma: no cover
        """Close the queue (no-op for queue.Queue)."""
        if hasattr(self._queue, "close"):
            self._queue.close()

cancel_join_thread()

Cancel the background thread (no-op for queue.Queue).

Source code in sgn/subprocess.py

def cancel_join_thread(self) -> None:  # pragma: no cover
    """Cancel the background thread (no-op for queue.Queue)."""
    if hasattr(self._queue, "cancel_join_thread"):
        self._queue.cancel_join_thread()

close()

Close the queue (no-op for queue.Queue).

Source code in sgn/subprocess.py

def close(self) -> None:  # pragma: no cover
    """Close the queue (no-op for queue.Queue)."""
    if hasattr(self._queue, "close"):
        self._queue.close()

empty()

Return True if the queue is empty.

Source code in sgn/subprocess.py

def empty(self) -> bool:
    """Return True if the queue is empty."""
    return self._queue.empty()

get(block=True, timeout=None)

Get an item from the queue.

Source code in sgn/subprocess.py

def get(self, block: bool = True, timeout: float | None = None) -> Any:
    """Get an item from the queue."""
    return self._queue.get(block=block, timeout=timeout)

get_nowait()

Get an item from the queue without blocking.

Source code in sgn/subprocess.py

def get_nowait(self) -> Any:
    """Get an item from the queue without blocking."""
    return self._queue.get_nowait()

put(item, block=True, timeout=None)

Put an item into the queue.

Source code in sgn/subprocess.py

def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:
    """Put an item into the queue."""
    self._queue.put(item, block=block, timeout=timeout)

put_nowait(item)

Put an item into the queue without blocking.

Source code in sgn/subprocess.py

def put_nowait(self, item: Any) -> None:  # pragma: no cover
    """Put an item into the queue without blocking."""
    self._queue.put_nowait(item)

WorkerContext

Context object passed to worker methods with clean access to resources.

Source code in sgn/subprocess.py

class WorkerContext:
    """Context object passed to worker methods with clean access to resources."""

    def __init__(
        self,
        input_queue=None,
        output_queue=None,
        worker_stop=None,
        worker_shutdown=None,
        **kwargs,
    ):
        self.input_queue = input_queue
        self.output_queue = output_queue
        self.stop_event = worker_stop
        self.shutdown_event = worker_shutdown
        self.shared_memory = kwargs.get("shm_list", [])
        self.state = {}  # Persistent dict for worker state across calls
        self.worker_exception = None
        self._raw_kwargs = kwargs

    def should_stop(self) -> bool:
        """Check if the worker should stop processing."""
        return self.stop_event and self.stop_event.is_set()

    def should_shutdown(self) -> bool:
        """Check if the worker should shutdown gracefully."""
        return self.shutdown_event and self.shutdown_event.is_set()

should_shutdown()

Check if the worker should shutdown gracefully.

Source code in sgn/subprocess.py

def should_shutdown(self) -> bool:
    """Check if the worker should shutdown gracefully."""
    return self.shutdown_event and self.shutdown_event.is_set()

should_stop()

Check if the worker should stop processing.

Source code in sgn/subprocess.py

def should_stop(self) -> bool:
    """Check if the worker should stop processing."""
    return self.stop_event and self.stop_event.is_set()