Adding Ray Backend

docs/src/code/executor/ray.rst

@@ -0,0 +1,5 @@
+Ray Executor
+=============
+
+.. automodule:: orion.executor.ray_backend
+ :members:

docs/src/user/parallel.rst

@@ -74,3 +74,10 @@ For more control over Dask, you should prefer using Dask executor backend direct
 The executor configuration is used to create the Dask Client. See Dask's documentation
 `here <https://distributed.dask.org/en/latest/api.html#distributed.Client>`__ for
 more information on possible arguments.
+
+Ray
+----
+
+We can also use the ray executor backend. For more control with ray, you can see ray's
+documentation `here <https://docs.ray.io/en/latest/ray-core/package-ref.html#python-api>`__ for
+more information on ray and it's python api.

setup.py

@@ -56,6 +56,7 @@
  "sphinx_gallery",
  ],
  "dask": ["dask[complete]"],
+ "ray": ["ray"],
  "track": ["track @ git+https:/Delaunay/track@master#egg=track"],
  "profet": ["emukit", "GPy", "torch", "pybnn"],
  "configspace": ["ConfigSpace"],
@@ -145,6 +146,7 @@
  "joblib = orion.executor.joblib_backend:Joblib",
  "poolexecutor = orion.executor.multiprocess_backend:PoolExecutor",
  "dask = orion.executor.dask_backend:Dask",
+ "ray = orion.executor.ray_backend:Ray",
  ],
  },
  install_requires=[

src/orion/executor/ray_backend.py

@@ -0,0 +1,113 @@
+import logging
+import traceback
+
+from orion.core.utils.module_import import ImportOptional
+from orion.executor.base import (
+ AsyncException,
+ AsyncResult,
+ BaseExecutor,
+ ExecutorClosed,
+ Future,
+)
+
+with ImportOptional("ray") as import_optional:
+ import ray
+
+HAS_RAY = not import_optional.failed
+
+logger = logging.getLogger(__name__)
+
+
+class _Future(Future):
+ def __init__(self, future):
+ self.future = future
+ self.exception = None
+
+ def get(self, timeout=None):
+ if self.exception:
+ raise self.exception
+ try:
+ return ray.get(self.future, timeout=timeout)
+ except ray.exceptions.GetTimeoutError as e:
+ raise TimeoutError() from e
+
+ def wait(self, timeout=None):
+ try:
+ ray.get(self.future, timeout=timeout)
+ except ray.exceptions.GetTimeoutError:
+ pass
+ except Exception as e:
+ self.exception = e
+
+ def ready(self):
+ obj_ready = ray.wait([self.future])
+ return len(obj_ready[0]) == 1
+
+ def successful(self):
+ # Python 3.6 raise assertion error
+ if not self.ready():
+ raise ValueError()
+
+ return self.future.successful()
+
+
+class Ray(BaseExecutor):
+ def __init__(
+ self,
+ n_workers=-1,
+ **config,
+ ):
+ super().__init__(n_workers=n_workers)
+ self.initialized = False
+ if not HAS_RAY:
+ raise ImportError("Ray must be installed to use Ray executor.")
+ self.config = config
+
+ if not ray.is_initialized():
+ ray.init(**self.config)
+ self.initialized = True
+ logger.debug("Ray was initiated with runtime_env : %s", **config)
+
+ def close(self):
+ if self.initialized:
+ self.initialized = False
+ ray.shutdown()
+
+ def __del__(self):
+ self.close()
+
+ def __enter__(self):
+ return self
+
+ def submit(self, function, *args, **kwargs):
+ if not ray.is_initialized():
+ raise ExecutorClosed()
+
+ remote_g = ray.remote(function)
+ return _Future(remote_g.remote(*args, **kwargs))
+
+ def wait(self, futures):
+ return [future.get() for future in futures]
+
+ def async_get(self, futures, timeout=None):
+ results = []
+ tobe_deleted = []
+ for i, future in enumerate(futures):
+ if timeout and i == 0:
+ future.wait(timeout)
+
+ if future.ready():
+ try:
+ results.append(AsyncResult(future, future.get()))
+ except Exception as err:
+ results.append(AsyncException(future, err, traceback.format_exc()))
+
+ tobe_deleted.append(future)
+ for future in tobe_deleted:
+ futures.remove(future)
+
+ return results
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self.close()
+ super().__exit__(exc_type, exc_value, traceback)

tests/unittests/executor/test_executor.py

@@ -1,10 +1,12 @@
+import os
 import time
 
 import pytest
 
 from orion.executor.base import AsyncException, ExecutorClosed, executor_factory
 from orion.executor.dask_backend import HAS_DASK, Dask
 from orion.executor.multiprocess_backend import PoolExecutor
+from orion.executor.ray_backend import HAS_RAY, Ray
 from orion.executor.single_backend import SingleExecutor
 
 
@@ -16,6 +18,11 @@ def thread(n):
  return PoolExecutor(n, "threading")
 
 
+def ray(n):
+ test_working_dir = os.path.dirname(os.path.abspath(__file__))
+ return Ray(n, runtime_env={"working_dir": test_working_dir})
+
+
 def skip_dask_if_not_installed(
  value, reason="Dask dependency is required for these tests."
 ):
@@ -39,18 +46,43 @@ def xfail_dask_if_not_installed(
  )
 
 
+def skip_ray_if_not_installed(
+ value, reason="Ray dependency is required for these tests."
+):
+ return pytest.param(
+ value,
+ marks=pytest.mark.skipif(
+ not HAS_RAY,
+ reason=reason,
+ ),
+ )
+
+
+def xfail_ray_if_not_installed(
+ value, reason="Ray dependency is required for these tests."
+):
+ return pytest.param(
+ value,
+ marks=pytest.mark.xfail(
+ condition=not HAS_RAY, reason=reason, raises=ImportError
+ ),
+ )
+
+
 executors = [
  "joblib",
  "poolexecutor",
  "singleexecutor",
  skip_dask_if_not_installed("dask"),
+ skip_ray_if_not_installed("ray"),
 ]
 
 backends = [
  thread,
  multiprocess,
  SingleExecutor,
  skip_dask_if_not_installed(Dask),
+ skip_ray_if_not_installed(ray),
 ]
 
 
@@ -191,9 +223,7 @@ def test_execute_async_bad(backend):
 
 def nested_jobs(executor):
  with executor:
- print("nested_jobs sub")
  futures = [executor.submit(function, 1, 2, i) for i in range(10)]
- print("nested_jobs wait")
  all_results = executor.wait(futures)
  return sum(all_results)