Skip to content

taps.executor.ray

taps/executor/ray.py

RayExecutor 

RayExecutor(
    address: str | None = "local",
    num_cpus: int | None = None,
)

Bases: Executor

Ray execution engine.

Note

Ray will raise a serialization error if a Proxy[bytes] is passed to or returned by a function. This is because Ray skips serializing bytes instances. Ray works with all other types of proxies, so if you need to send bytes data, wrap the data in another type.

Parameters:

  • address (str | None, default: 'local' ) –

    Address to pass to ray.init().

  • num_cpus (int | None, default: None ) –

    Number of CPUs to use.

Source code in taps/executor/ray.py 
75
76
77
78
79
80
81
82
83
84
85
def __init__(
    self,
    address: str | None = 'local',
    num_cpus: int | None = None,
) -> None:
    if RAY_IMPORT_ERROR is not None:  # pragma: no cover
        raise RAY_IMPORT_ERROR

    ray.init(address=address, configure_logging=False, num_cpus=num_cpus)
    # Mapping of Python callables to Ray RemoteFunction types
    self._remote: dict[Any, Any] = {}

submit() 

submit(
    function: Callable[P, T],
    /,
    *args: args,
    **kwargs: kwargs,
) -> Future[T]

Schedule the callable to be executed.

Parameters:

  • function (Callable[P, T]) –

    Callable to execute.

  • args (args, default: () ) –

    Positional arguments.

  • kwargs (kwargs, default: {} ) –

    Keyword arguments.

Returns:

  • Future[T] –

    Future-like object representing the result of the execution of the callable.

Source code in taps/executor/ray.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def submit(
    self,
    function: Callable[P, T],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> Future[T]:
    """Schedule the callable to be executed.

    Args:
        function: Callable to execute.
        args: Positional arguments.
        kwargs: Keyword arguments.

    Returns:
        [`Future`][concurrent.futures.Future]-like object representing \
        the result of the execution of the callable.
    """
    args = cast(P.args, _parse_args(args))
    kwargs = cast(P.kwargs, _parse_kwargs(kwargs))

    if function in self._remote:
        remote = self._remote[function]
    else:
        wrapped = _wrap_function(function)
        remote = ray.remote(wrapped)
        self._remote[function] = remote

    object_ref = remote.remote(*args, **kwargs)

    return object_ref.future()

shutdown() 

shutdown(
    wait: bool = True, *, cancel_futures: bool = False
) -> None

Shutdown the client.

Source code in taps/executor/ray.py 
119
120
121
122
123
124
125
126
def shutdown(
    self,
    wait: bool = True,
    *,
    cancel_futures: bool = False,
) -> None:
    """Shutdown the client."""
    ray.shutdown()

RayConfig

Bases: ExecutorConfig

RayExecutor plugin configuration.

Attributes:

  • address (Optional[str]) –

    Address of the Ray cluster to run on.

  • num_cpus (Optional[int]) –

    Number of actor processes to start in the pool. Defaults to the number of cores in the Ray cluster, or the number of cores on this machine.

get_executor() 

get_executor() -> RayExecutor

Create an executor instance from the config.

Source code in taps/executor/ray.py 
150
151
152
def get_executor(self) -> RayExecutor:
    """Create an executor instance from the config."""
    return RayExecutor(address=self.address, num_cpus=self.num_cpus)