pydvl.parallel.backends.ray

RayParallelBackend

RayParallelBackend(config: ParallelConfig | None = None)

Bases: ParallelBackend

Class used to wrap ray to make it transparent to algorithms.

Example

import ray
from pydvl.parallel import RayParallelBackend
ray.init()
parallel_backend = RayParallelBackend()

Source code in src/pydvl/parallel/backends/ray.py

@deprecated(
    target=True,
    args_mapping={"config": None},
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def __init__(self, config: ParallelConfig | None = None) -> None:
    if not ray.is_initialized():
        raise RuntimeError(
            "Starting from v0.9.0, ray is no longer automatically initialized. "
            "Please use `ray.init()` with the desired configuration "
            "before using this class."
        )
    # Register ray joblib backend
    register_ray()

executor classmethod

executor(
    max_workers: int | None = None,
    *,
    config: ParallelConfig | None = None,
    cancel_futures: CancellationPolicy | bool = PENDING,
) -> Executor

Returns a futures executor for the parallel backend.

Example

import ray
from pydvl.parallel import RayParallelBackend
ray.init()
parallel_backend = RayParallelBackend()
with parallel_backend.executor() as executor:
    executor.submit(...)

PARAMETER	DESCRIPTION
max_workers	Maximum number of parallel workers. TYPE: int | None DEFAULT: None
config	(DEPRECATED) Object configuring parallel computation, with cluster address, number of cpus, etc. TYPE: ParallelConfig | None DEFAULT: None
cancel_futures	Policy to use when cancelling futures after exiting an Executor. TYPE: CancellationPolicy | bool DEFAULT: PENDING

RETURNS	DESCRIPTION
Executor	Instance of RayExecutor.

Source code in src/pydvl/parallel/backends/ray.py

@classmethod
def executor(
    cls,
    max_workers: int | None = None,
    *,
    config: ParallelConfig | None = None,
    cancel_futures: CancellationPolicy | bool = CancellationPolicy.PENDING,
) -> Executor:
    """Returns a futures executor for the parallel backend.

    !!! Example
        ``` python
        import ray
        from pydvl.parallel import RayParallelBackend
        ray.init()
        parallel_backend = RayParallelBackend()
        with parallel_backend.executor() as executor:
            executor.submit(...)
        ```

    Args:
        max_workers: Maximum number of parallel workers.
        config: (**DEPRECATED**) Object configuring parallel computation,
            with cluster address, number of cpus, etc.
        cancel_futures: Policy to use when cancelling futures
            after exiting an Executor.

    Returns:
        Instance of [RayExecutor][pydvl.parallel.futures.ray.RayExecutor].
    """
    # Imported here to avoid circular import errors
    from pydvl.parallel.futures.ray import RayExecutor

    if config is not None:
        warnings.warn(
            "The `RayParallelBackend` uses deprecated arguments: "
            "`config`. They were deprecated since v0.9.0 "
            "and will be removed in v0.10.0.",
            FutureWarning,
        )

    return RayExecutor(max_workers, cancel_futures=cancel_futures)  # type: ignore

wrap

wrap(fun: Callable, **kwargs: dict[str, Any]) -> Callable

Wraps a function as a ray remote.

PARAMETER	DESCRIPTION
fun	the function to wrap TYPE: Callable
kwargs	keyword arguments to pass to @ray.remote TYPE: dict[str, Any] DEFAULT: {}

RETURNS	DESCRIPTION
Callable	The .remote method of the ray RemoteFunction.

Source code in src/pydvl/parallel/backends/ray.py

def wrap(self, fun: Callable, **kwargs: dict[str, Any]) -> Callable:
    """Wraps a function as a ray remote.

    Args:
        fun: the function to wrap
        kwargs: keyword arguments to pass to @ray.remote

    Returns:
        The `.remote` method of the ray `RemoteFunction`.
    """
    if len(kwargs) > 0:
        return ray.remote(**kwargs)(fun).remote  # type: ignore
    return ray.remote(fun).remote  # type: ignore