Multiprocessing

Reference information for the utils Multiprocessing API.

eva.core.utils.multiprocessing.Process

Bases: Process

Multiprocessing wrapper with logic to propagate exceptions to the parent process.

Source: https://stackoverflow.com/a/33599967/4992248

Source code in src/eva/core/utils/multiprocessing.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the process."""
    multiprocessing.Process.__init__(self, *args, **kwargs)

    self._parent_conn, self._child_conn = multiprocessing.Pipe()
    self._exception = None

exception property

Property that contains exception information from the process.

run

Run the process.

Source code in src/eva/core/utils/multiprocessing.py

def run(self) -> None:
    """Run the process."""
    try:
        multiprocessing.Process.run(self)
        self._child_conn.send(None)
    except Exception as e:
        tb = traceback.format_exc()
        self._child_conn.send((e, tb))

check_exceptions

Check for exception propagate it to the parent process.

Source code in src/eva/core/utils/multiprocessing.py

def check_exceptions(self) -> None:
    """Check for exception propagate it to the parent process."""
    if not self.is_alive():
        if self.exception:
            error, traceback = self.exception
            sys.stderr.write(traceback + "\n")
            raise error