stouputils.parallel.subprocess module#

exception RemoteSubprocessError(
exc_type: str,
exc_repr: str,
traceback_str: str,
)[source]#

Bases: RuntimeError

Raised in the parent when the child raised an exception - contains the child’s formatted traceback.

run_in_subprocess(
func: Callable[[...], R],
*args: Any,
timeout: float | None = None,
no_join: bool = False,
capture_output: bool = False,
**kwargs: Any,
) → R[source]#

Execute a function in a subprocess with positional and keyword arguments.

This is useful when you need to run a function in isolation to avoid memory leaks, resource conflicts, or to ensure a clean execution environment. The subprocess will be created, run the function with the provided arguments, and return the result.

Parameters:

  • func (Callable) – The function to execute in a subprocess. (SHOULD BE A TOP-LEVEL FUNCTION TO BE PICKLABLE)

  • *args (Any) – Positional arguments to pass to the function.

  • timeout (float | None) – Maximum time in seconds to wait for the subprocess. If None, wait indefinitely. If the subprocess exceeds this time, it will be terminated.

  • no_join (bool) – If True, do not wait for the subprocess to finish (fire-and-forget) and return the Process object.

  • capture_output (bool) – If True, capture the subprocess’ stdout/stderr and relay it in real time to the parent’s stdout. This enables seeing print() output from the subprocess in the main process.

  • **kwargs (Any) – Keyword arguments to pass to the function.

Returns:

The return value of the function.

Return type:

R

Raises:

  • RemoteSubprocessError – If the child raised an exception - contains the child’s formatted traceback.

  • RuntimeError – If the subprocess exits with a non-zero exit code or did not return a result.

  • TimeoutError – If the subprocess exceeds the specified timeout.

Examples

> # Simple function execution
> run_in_subprocess(doctest_square, 5)
25

> # Function with multiple arguments
> def add(a: int, b: int, c: int) -> int:
.     return a + b + c
> run_in_subprocess(add, 10, 20, c=30)
60

> # Function with keyword arguments
> def greet(name: str, greeting: str = "Hello") -> str:
.     return f"{greeting}, {name}!"
> run_in_subprocess(greet, "World", greeting="Hi")
'Hi, World!'

> # With timeout to prevent hanging
> run_in_subprocess(some_gpu_func, data, timeout=300.0)
_subprocess_wrapper(
result_queue: Any,
func: Callable[[...], R],
args: tuple[Any, ...],
kwargs: dict[str, Any],
capturer: CaptureOutput | None = None,
) → None[source]#

Wrapper function to execute the target function and store the result in the queue.

Must be at module level to be pickable on Windows (spawn context).

Parameters:

  • result_queue (multiprocessing.Queue | None) – Queue to store the result or exception (None if detached).

  • func (Callable) – The target function to execute.

  • args (tuple) – Positional arguments for the function.

  • kwargs (dict) – Keyword arguments for the function.

  • capturer (CaptureOutput | None) – Optional CaptureOutput instance for stdout capture.