github-featured/ray-project--ray
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
master
ray-project--ray/python/ray/tests/test_basic.py
T

1238 lines
36 KiB
Python
Raw Permalink Blame History

# coding: utf-8
import logging
import os
import pickle
import platform
import random
import re
import sys
import time
import pytest
import ray
import ray.cluster_utils
from ray._common.test_utils import (
SignalActor,
run_string_as_driver,
)
from ray._private.test_utils import client_test_enabled
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
import psutil
logger = logging.getLogger(__name__)
# https://github.com/ray-project/ray/issues/6662
@pytest.mark.skipif(
os.environ.get("RAY_MINIMAL") == "1",
reason="This test is not supposed to work for minimal installation.",
)
@pytest.mark.skipif(client_test_enabled(), reason="interferes with grpc")
def test_http_proxy(start_http_proxy, shutdown_only):
# C++ config `grpc_enable_http_proxy` only initializes once, so we have to
# run driver as a separate process to make sure the correct config value
# is initialized.
script = """
import ray
ray.init(num_cpus=1)
@ray.remote
def f():
return 1
assert ray.get(f.remote()) == 1
"""
env = start_http_proxy
run_string_as_driver(script, dict(os.environ, **env))
def test_release_cpu_resources(shutdown_only):
ray.init(num_cpus=1)
@ray.remote(num_cpus=1)
def child():
return 3
@ray.remote(num_cpus=1)
def parent():
# Parent should release the CPU resource
# to run child.
return ray.get(child.remote())
assert ray.get(parent.remote()) == 3
# Make sure CPU resource inside PG can also be released properly.
pg = ray.util.placement_group(bundles=[{"CPU": 1}])
assert (
ray.get(
parent.options(
scheduling_strategy=PlacementGroupSchedulingStrategy(
placement_group=pg, placement_group_capture_child_tasks=True
)
).remote()
)
== 3
)
assert (
ray.get(
parent.options(
scheduling_strategy=PlacementGroupSchedulingStrategy(
placement_group=pg,
placement_group_bundle_index=0,
placement_group_capture_child_tasks=True,
)
).remote()
)
== 3
)
# https://github.com/ray-project/ray/issues/16025
def test_release_resources_race(shutdown_only):
ray.init(num_cpus=2)
refs = []
for _ in range(10):
refs.append(ray.put(bytearray(1024 * 1024)))
@ray.remote
def consume(refs):
# Should work without releasing resources!
ray.get(refs)
return os.getpid()
pids = set(ray.get([consume.remote(refs) for _ in range(10)]))
# Should not have started multiple workers.
assert len(pids) <= 2, pids
def test_not_release_resource(shutdown_only):
# Test to make sure we don't release CPU
# resource if the object is already fetched.
ray.init(num_cpus=1)
@ray.remote
def task1():
return [1] * (1024 * 1024)
o1 = task1.remote()
@ray.remote
def task2(*args, **kwargs):
# ray.get here should not release
# CPU resource since the object is already
# available in args[0]
assert args[0] == ray.get(kwargs["o"][0])
return os.getpid()
@ray.remote
def task3(*args):
return os.getpid()
o2 = task2.remote(o1, o=[o1])
# This should run after task2 finishes
o3 = task3.remote(o1)
assert len(set(ray.get([o2, o3]))) == 1
# https://github.com/ray-project/ray/issues/22504
def test_worker_isolation_by_resources(shutdown_only):
ray.init(num_cpus=1, num_gpus=1)
@ray.remote(num_gpus=1)
def gpu():
return os.getpid()
@ray.remote
def cpu():
return os.getpid()
pid1 = ray.get(cpu.remote())
pid2 = ray.get(gpu.remote())
assert pid1 != pid2, (pid1, pid2)
# https://github.com/ray-project/ray/issues/10960
def test_max_calls_releases_resources(shutdown_only):
ray.init(num_cpus=2, num_gpus=1)
@ray.remote(num_cpus=0)
def g():
return 0
@ray.remote(num_cpus=1, num_gpus=1, max_calls=1, max_retries=0)
def f():
return [g.remote()]
for i in range(10):
print(i)
ray.get(f.remote()) # This will hang if GPU resources aren't released.
# https://github.com/ray-project/ray/issues/7263
def test_grpc_message_size(shutdown_only):
ray.init(num_cpus=1)
@ray.remote
def bar(*a):
return
# 50KiB, not enough to spill to plasma, but will be inlined.
def f():
return bytearray(50000)
# Executes a 10MiB task spec
ray.get(bar.remote(*[f() for _ in range(200)]))
def test_default_worker_import_dependency(shutdown_only):
"""
Test ray's python worker import doesn't import the not-allowed dependencies.
"""
# We don't allow numpy to be imported in the worker script to avoid slow
# worker startup time, as well as interfering with OMP_NUM_THREADS which
# is used by numpy when imported.
# See https://github.com/ray-project/ray/issues/33891
blocked_deps = ["numpy"]
# Ray should not be importing pydantic (used in serialization) eagerly.
# This introduces regression in worker start up time.
# https://github.com/ray-project/ray/issues/41338
blocked_deps += ["pydantic"]
# Remove the ray module and the blocked deps from sys.modules.
sys.modules.pop("ray", None)
assert "ray" not in sys.modules
for dep in blocked_deps:
sys.modules.pop(dep, None)
assert dep not in sys.modules
# This imports the python worker.
import ray._private.workers.default_worker # noqa: F401
# Check that the ray module is imported.
assert "ray" in sys.modules
# Check that the blocked deps are not imported.
for dep in blocked_deps:
assert dep not in sys.modules
# Test starting a ray workers should not see unwanted deps loaded eagerly.
ray.init()
@ray.remote
def f():
import ray # noqa: F401
assert "ray" in sys.modules
for x in blocked_deps:
assert x not in sys.modules
ray.get(f.remote())
@pytest.mark.skipif(
sys.platform != "linux", reason="Windows/OSX thread count not policed yet."
)
@pytest.mark.parametrize(
"recorder_env, expected_thread_counts",
[
pytest.param(
{"RAY_enable_ray_task_event_recorder": "0"},
{21, 22, 23, 24},
id="task_event_recorder_off",
),
pytest.param(
{"RAY_enable_ray_event": "1", "RAY_enable_ray_task_event_recorder": "1"},
{22, 23, 24, 25},
id="task_event_recorder_on",
),
],
)
def test_worker_thread_count(
monkeypatch, shutdown_only, recorder_env, expected_thread_counts
):
"""This test will fail if the number of threads spawned by a worker process
increases. If you find that a patch is now causing this test to fail,
consider if this thread count change is expected and adjust the test
(or your patch) accordingly!
RayTaskEventRecorder spawns one dedicated io thread when active (which needs both
enable_ray_event and enable_ray_task_event_recorder), so the expected count is
parametrized on that flag instead of depending on the compiled default.
"""
@ray.remote
class Actor:
def get_thread_count(self):
try:
process = psutil.Process(os.getpid())
return process.num_threads()
except ImportError:
return None
# Set the environment variables used by the raylet and worker
monkeypatch.setenv("RAY_worker_num_grpc_internal_threads", "1")
monkeypatch.setenv("RAY_num_server_call_thread", "1")
monkeypatch.setenv("RAY_core_worker_num_server_call_thread", "1")
for k, v in recorder_env.items():
monkeypatch.setenv(k, v)
# TODO(#55215): The for loop and the 'assert ... in {..,..}' complicates this
# test unnecessarily. We should only need to call the assert after
# a single call to the worker. However, because the thread count
# per worker today isn't entirely static, we need to allow for this
# flexibility. https://github.com/ray-project/ray/issues/55215
actor = Actor.remote()
for _ in range(5):
ray.get(actor.get_thread_count.remote())
# Lowering these numbers in this assert should be celebrated,
# increasing these numbers should be scrutinized
assert ray.get(actor.get_thread_count.remote()) in expected_thread_counts
# https://github.com/ray-project/ray/issues/7287
def test_omp_threads_set(ray_start_cluster, monkeypatch):
cluster = ray_start_cluster
cluster.add_node(num_cpus=2)
ray.init(address=cluster.address)
@ray.remote
def f():
return os.environ.get("OMP_NUM_THREADS")
@ray.remote
class Actor:
def f(self):
return os.environ.get("OMP_NUM_THREADS")
###########################
# Test basic tasks
###########################
# Test override to num_cpus if OMP_NUM_THREADS not set
assert ray.get(f.options(num_cpus=2).remote()) == "2"
# Test override to default cpu number if OMP_NUM_THREADS not set
assert ray.get(f.remote()) == "1"
# Test set to 1 for fractional CPU
assert ray.get(f.options(num_cpus=0.25).remote()) == "1"
###########################
# Test not overriding env_variables
###########################
from ray.runtime_env import RuntimeEnv
assert (
ray.get(
f.options(
runtime_env=RuntimeEnv(env_vars={"OMP_NUM_THREADS": "2"})
).remote()
)
== "2"
)
assert (
ray.get(
f.options(
num_cpus=1, runtime_env=RuntimeEnv(env_vars={"OMP_NUM_THREADS": "2"})
).remote()
)
== "2"
)
###########################
# Test actor tasks
###########################
# Test actor tasks set OMP_NUM_THREADS correctly in a similar way.
assert ray.get(Actor.remote().f.remote()) == "1"
assert ray.get(Actor.options(num_cpus=2).remote().f.remote()) == "2"
assert ray.get(Actor.options(num_cpus=0.25).remote().f.remote()) == "1"
###########################
# Test setting and restoring of the environ after tasks run
###########################
@ray.remote
def g():
return os.getpid(), os.environ.get("OMP_NUM_THREADS")
# Set to 1
pid1, omp_num_threads = ray.get(g.remote())
assert omp_num_threads == "1"
# Set to 2
pid2, omp_num_threads = ray.get(g.options(num_cpus=2).remote())
assert pid1 == pid2
assert omp_num_threads == "2"
###########################
# Test not setting the value with environ already set to 1 in env
###########################
with monkeypatch.context() as m:
m.setenv("OMP_NUM_THREADS", "1")
cluster.add_node(num_cpus=4)
assert ray.get(f.options(num_cpus=4).remote()) == "1"
def test_submit_api(shutdown_only):
ray.init(num_cpus=2, num_gpus=1, resources={"Custom": 1})
@ray.remote
def f(n):
return list(range(n))
@ray.remote
def g():
return ray.get_gpu_ids()
assert f._remote([0], num_returns=0) is None
id1 = f._remote(args=[1], num_returns=1)
assert ray.get(id1) == [0]
id1, id2 = f._remote(args=[2], num_returns=2)
assert ray.get([id1, id2]) == [0, 1]
id1, id2, id3 = f._remote(args=[3], num_returns=3)
assert ray.get([id1, id2, id3]) == [0, 1, 2]
assert ray.get(
g._remote(args=[], num_cpus=1, num_gpus=1, resources={"Custom": 1})
) == [0]
infeasible_id = g._remote(args=[], resources={"NonexistentCustom": 1})
assert ray.get(g._remote()) == []
ready_ids, remaining_ids = ray.wait([infeasible_id], timeout=0.05)
assert len(ready_ids) == 0
assert len(remaining_ids) == 1
# Check mismatch with num_returns.
with pytest.raises(ValueError):
ray.get(f.options(num_returns=2).remote(3))
with pytest.raises(ValueError):
ray.get(f.options(num_returns=3).remote(2))
@ray.remote
class Actor:
def __init__(self, x, y=0):
self.x = x
self.y = y
def method(self, a, b=0):
return self.x, self.y, a, b
def gpu_ids(self):
return ray.get_gpu_ids()
@ray.remote
class Actor2:
def __init__(self):
pass
def method(self):
pass
a = Actor._remote(args=[0], kwargs={"y": 1}, num_gpus=1, resources={"Custom": 1})
a2 = Actor2._remote()
ray.get(a2.method._remote())
id1, id2, id3, id4 = a.method._remote(args=["test"], kwargs={"b": 2}, num_returns=4)
assert ray.get([id1, id2, id3, id4]) == [0, 1, "test", 2]
def test_invalid_arguments():
def f():
return 1
class A:
x = 1
template1 = (
"The type of keyword '{}' "
+ f"must be {(int, type(None))}, but received type {float}"
)
# Type check
for keyword in ("max_retries", "max_calls"):
with pytest.raises(TypeError, match=re.escape(template1.format(keyword))):
ray.remote(**{keyword: random.random()})(f)
num_returns_template = (
"The type of keyword 'num_returns' "
+ f"must be {(int, str, type(None))}, but received type {float}"
)
with pytest.raises(TypeError, match=re.escape(num_returns_template)):
ray.remote(**{"num_returns": random.random()})(f)
for keyword in ("max_restarts", "max_task_retries"):
with pytest.raises(TypeError, match=re.escape(template1.format(keyword))):
ray.remote(**{keyword: random.random()})(A)
# Value check for non-negative finite values
for v in (random.randint(-100, -2), -1):
keyword = "max_calls"
with pytest.raises(
ValueError,
match=f"The keyword '{keyword}' only accepts None, "
f"0 or a positive integer",
):
ray.remote(**{keyword: v})(f)
keyword = "num_returns"
with pytest.raises(
ValueError,
match=f"The keyword '{keyword}' only accepts None, "
"a non-negative integer, "
r"'streaming' \(for generators\), or 'dynamic'",
):
ray.remote(**{keyword: v})(f)
# Value check for non-negative and infinite values
template2 = (
"The keyword '{}' only accepts None, 0, -1 or a positive integer, "
"where -1 represents infinity."
)
with pytest.raises(ValueError, match=template2.format("max_retries")):
ray.remote(max_retries=random.randint(-100, -2))(f)
for keyword in ("max_restarts", "max_task_retries"):
with pytest.raises(ValueError, match=template2.format(keyword)):
ray.remote(**{keyword: random.randint(-100, -2)})(A)
# Check invalid resource quantity
with pytest.raises(
ValueError,
match=(
"The precision of the fractional quantity of resource num_gpus"
" cannot go beyond 0.0001"
),
):
ray.remote(num_gpus=0.0000001)(f)
with pytest.raises(
ValueError,
match=(
"The precision of the fractional quantity of resource custom_resource"
" cannot go beyond 0.0001"
),
):
ray.remote(resources={"custom_resource": 0.0000001})(f)
def test_options():
"""General test of option keywords in Ray."""
from ray._common import ray_option_utils
def f():
return 1
class A:
x = 1
task_defaults = {
k: v.default_value for k, v in ray_option_utils.task_options.items()
}
task_defaults_for_options = task_defaults.copy()
task_defaults_for_options.pop("max_calls")
ray.remote(f).options(**task_defaults_for_options)
ray.remote(**task_defaults)(f).options(**task_defaults_for_options)
with pytest.raises(
ValueError,
match=re.escape("Setting 'max_calls' is not supported in '.options()'."),
):
ray.remote(f).options(max_calls=1)
actor_defaults = {
k: v.default_value for k, v in ray_option_utils.actor_options.items()
}
actor_defaults_for_options = actor_defaults.copy()
actor_defaults_for_options.pop("concurrency_groups")
ray.remote(A).options(**actor_defaults_for_options)
ray.remote(**actor_defaults)(A).options(**actor_defaults_for_options)
with pytest.raises(
ValueError,
match=re.escape(
"Setting 'concurrency_groups' is not supported in '.options()'."
),
):
ray.remote(A).options(concurrency_groups=[])
unique_object = type("###", (), {})()
for k, v in ray_option_utils.task_options.items():
v.validate(k, v.default_value)
with pytest.raises(TypeError):
v.validate(k, unique_object)
for k, v in ray_option_utils.actor_options.items():
v.validate(k, v.default_value)
with pytest.raises(TypeError):
v.validate(k, unique_object)
@ray.remote(num_gpus=2)
def foo():
pass
assert foo._default_options == {
"max_calls": 1,
"num_gpus": 2,
}
f2 = foo.options(num_cpus=1, num_gpus=1)
# TODO(suquark): The current implementation of `.options()` is so bad that we
# cannot even access its options from outside. Here we hack the closures to
# achieve our goal. Need futher efforts to clean up the tech debt.
assert f2.remote.__closure__[2].cell_contents == {
"num_cpus": 1,
"num_gpus": 1,
}
# https://github.com/ray-project/ray/issues/17842
def test_disable_cuda_devices():
script = """
import ray
ray.init()
@ray.remote
def check():
import os
assert "CUDA_VISIBLE_DEVICES" not in os.environ
print("remote", ray.get(check.remote()))
"""
run_string_as_driver(
script, dict(os.environ, **{"RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES": "1"})
)
# https://github.com/ray-project/ray/issues/54868
@pytest.mark.skipif(
sys.platform == "darwin" and platform.machine() == "arm64",
reason=(
"On Apple Silicon the GPU resource is managed by AppleGPUAcceleratorManager, "
"which has no visible-devices env var (e.g. CUDA_VISIBLE_DEVICES) to set or "
"override, so this NVIDIA-specific override behavior does not apply. Manager "
"selection is hardware-based, so NVIDIA semantics can't be exercised on a Mac."
),
)
def test_not_override_accelerator_ids_when_num_accelerators_is_zero():
not_override_check_script = """
import ray
ray.init()
@ray.remote(num_gpus=0)
def check():
import os
assert "CUDA_VISIBLE_DEVICES" not in os.environ
@ray.remote(num_gpus=0)
class Actor:
def check(self):
import os
assert "CUDA_VISIBLE_DEVICES" not in os.environ
print("task check", ray.get(check.remote()))
print("actor check", ray.get(Actor.options(num_gpus=0).remote().check.remote()))
"""
run_string_as_driver(not_override_check_script)
override_check_script = """
import ray
ray.init()
@ray.remote(num_gpus=0)
def check():
import os
assert os.environ.get("CUDA_VISIBLE_DEVICES") == ""
@ray.remote(num_gpus=0)
class Actor:
def check(self):
import os
assert os.environ.get("CUDA_VISIBLE_DEVICES") == ""
print("task check", ray.get(check.remote()))
print("actor check", ray.get(Actor.options(num_gpus=0).remote().check.remote()))
"""
run_string_as_driver(
override_check_script,
dict(
os.environ,
**{"RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO": "1"},
),
)
def test_put_get(shutdown_only):
ray.init(num_cpus=0)
for i in range(100):
value_before = i * 10**6
object_ref = ray.put(value_before)
value_after = ray.get(object_ref)
assert value_before == value_after
for i in range(100):
value_before = i * 10**6 * 1.0
object_ref = ray.put(value_before)
value_after = ray.get(object_ref)
assert value_before == value_after
for i in range(100):
value_before = "h" * i
object_ref = ray.put(value_before)
value_after = ray.get(object_ref)
assert value_before == value_after
for i in range(100):
value_before = [1] * i
object_ref = ray.put(value_before)
value_after = ray.get(object_ref)
assert value_before == value_after
@pytest.mark.skipif(client_test_enabled(), reason="internal _raylet")
def test_function_descriptor():
python_descriptor = ray._raylet.PythonFunctionDescriptor(
"module_name", "function_name", "class_name", "function_hash"
)
python_descriptor2 = pickle.loads(pickle.dumps(python_descriptor))
assert python_descriptor == python_descriptor2
assert hash(python_descriptor) == hash(python_descriptor2)
assert python_descriptor.function_id == python_descriptor2.function_id
java_descriptor = ray._raylet.JavaFunctionDescriptor(
"class_name", "function_name", "signature"
)
java_descriptor2 = pickle.loads(pickle.dumps(java_descriptor))
assert java_descriptor == java_descriptor2
assert python_descriptor != java_descriptor
assert python_descriptor != object()
d = {python_descriptor: 123}
assert d.get(python_descriptor2) == 123
def test_ray_options(shutdown_only):
ray.init(num_cpus=10, num_gpus=10, resources={"custom1": 2})
@ray.remote(num_cpus=2, num_gpus=3, memory=150 * 2**20, resources={"custom1": 1})
def foo(expected_resources):
# Possibly wait until the available resources have been updated
# (there might be a delay due to heartbeats)
retries = 10
keys = ["CPU", "GPU", "custom1"]
while retries >= 0:
resources = ray.available_resources()
do_return = True
for key in keys:
if resources[key] != expected_resources[key]:
print(key, resources[key], expected_resources[key])
do_return = False
break
if do_return:
return resources["memory"]
time.sleep(0.1)
retries -= 1
raise RuntimeError("Number of retries exceeded")
expected_resources_without_options = {"CPU": 8.0, "GPU": 7.0, "custom1": 1.0}
memory_available_without_options = ray.get(
foo.remote(expected_resources_without_options)
)
expected_resources_with_options = {"CPU": 7.0, "GPU": 6.0, "custom1": 1.5}
memory_available_with_options = ray.get(
foo.options(
num_cpus=3, num_gpus=4, memory=50 * 2**20, resources={"custom1": 0.5}
).remote(expected_resources_with_options)
)
assert memory_available_without_options < memory_available_with_options
@pytest.mark.skipif(client_test_enabled(), reason="internal api")
@pytest.mark.parametrize(
"ray_start_cluster_head",
[
{
"num_cpus": 0,
"object_store_memory": 75 * 1024 * 1024,
"_system_config": {"automatic_object_spilling_enabled": False},
}
],
indirect=True,
)
def test_fetch_local(ray_start_cluster_head):
cluster = ray_start_cluster_head
cluster.add_node(num_cpus=2, object_store_memory=75 * 1024 * 1024)
signal_actor = SignalActor.remote()
@ray.remote
def put():
ray.wait([signal_actor.wait.remote()])
return bytearray(40 * 1024 * 1024) # 40 MB data
local_ref = ray.put(bytearray(40 * 1024 * 1024))
remote_ref = put.remote()
# Data is not ready in any node
(ready_ref, remaining_ref) = ray.wait([remote_ref], timeout=2, fetch_local=False)
assert (0, 1) == (len(ready_ref), len(remaining_ref))
ray.wait([signal_actor.send.remote()])
# Data is ready in some node, but not local node.
(ready_ref, remaining_ref) = ray.wait([remote_ref], fetch_local=False)
assert (1, 0) == (len(ready_ref), len(remaining_ref))
(ready_ref, remaining_ref) = ray.wait([remote_ref], timeout=2, fetch_local=True)
assert (0, 1) == (len(ready_ref), len(remaining_ref))
del local_ref
(ready_ref, remaining_ref) = ray.wait([remote_ref], fetch_local=True)
assert (1, 0) == (len(ready_ref), len(remaining_ref))
def test_nested_functions(ray_start_regular_shared):
# Make sure that remote functions can use other values that are defined
# after the remote function but before the first function invocation.
@ray.remote
def f():
return g(), ray.get(h.remote())
def g():
return 1
@ray.remote
def h():
return 2
assert ray.get(f.remote()) == (1, 2)
def test_recursive_remote_call(ray_start_regular_shared):
# Test a remote function that recursively calls itself.
@ray.remote
def factorial(n):
if n == 0:
return 1
return n * ray.get(factorial.remote(n - 1))
assert ray.get(factorial.remote(0)) == 1
assert ray.get(factorial.remote(1)) == 1
assert ray.get(factorial.remote(2)) == 2
assert ray.get(factorial.remote(3)) == 6
assert ray.get(factorial.remote(4)) == 24
assert ray.get(factorial.remote(5)) == 120
def test_mutually_recursive_functions(ray_start_regular_shared):
# Test remote functions that recursively call each other.
@ray.remote
def factorial_even(n):
assert n % 2 == 0
if n == 0:
return 1
return n * ray.get(factorial_odd.remote(n - 1))
@ray.remote
def factorial_odd(n):
assert n % 2 == 1
return n * ray.get(factorial_even.remote(n - 1))
assert ray.get(factorial_even.remote(4)) == 24
assert ray.get(factorial_odd.remote(5)) == 120
def test_ray_recursive_objects(ray_start_regular_shared):
class ClassA:
pass
# Make a list that contains itself.
lst = []
lst.append(lst)
# Make an object that contains itself as a field.
a1 = ClassA()
a1.field = a1
# Make two objects that contain each other as fields.
a2 = ClassA()
a3 = ClassA()
a2.field = a3
a3.field = a2
# Make a dictionary that contains itself.
d1 = {}
d1["key"] = d1
# Create a list of recursive objects.
recursive_objects = [lst, a1, a2, a3, d1]
# Serialize the recursive objects.
for obj in recursive_objects:
ray.put(obj)
def test_passing_arguments_by_value_out_of_the_box(ray_start_regular_shared):
@ray.remote
def f(x):
return x
# Test passing lambdas.
def temp():
return 1
assert ray.get(f.remote(temp))() == 1
assert ray.get(f.remote(lambda x: x + 1))(3) == 4
# Test sets.
assert ray.get(f.remote(set())) == set()
s = {1, (1, 2, "hi")}
assert ray.get(f.remote(s)) == s
# Test types.
assert ray.get(f.remote(int)) is int
assert ray.get(f.remote(float)) is float
assert ray.get(f.remote(str)) is str
class Foo:
def __init__(self):
pass
# Make sure that we can put and get a custom type. Note that the result
# won't be "equal" to Foo.
ray.get(ray.put(Foo))
def test_putting_object_that_closes_over_object_ref(ray_start_regular_shared):
# This test is here to prevent a regression of
# https://github.com/ray-project/ray/issues/1317.
class Foo:
def __init__(self):
self.val = ray.put(0)
def method(self):
_ = f
f = Foo()
ray.put(f)
def test_keyword_args(ray_start_regular_shared):
@ray.remote
def keyword_fct1(a, b="hello"):
return "{} {}".format(a, b)
@ray.remote
def keyword_fct2(a="hello", b="world"):
return "{} {}".format(a, b)
@ray.remote
def keyword_fct3(a, b, c="hello", d="world"):
return "{} {} {} {}".format(a, b, c, d)
x = keyword_fct1.remote(1)
assert ray.get(x) == "1 hello"
x = keyword_fct1.remote(1, "hi")
assert ray.get(x) == "1 hi"
x = keyword_fct1.remote(1, b="world")
assert ray.get(x) == "1 world"
x = keyword_fct1.remote(a=1, b="world")
assert ray.get(x) == "1 world"
x = keyword_fct2.remote(a="w", b="hi")
assert ray.get(x) == "w hi"
x = keyword_fct2.remote(b="hi", a="w")
assert ray.get(x) == "w hi"
x = keyword_fct2.remote(a="w")
assert ray.get(x) == "w world"
x = keyword_fct2.remote(b="hi")
assert ray.get(x) == "hello hi"
x = keyword_fct2.remote("w")
assert ray.get(x) == "w world"
x = keyword_fct2.remote("w", "hi")
assert ray.get(x) == "w hi"
x = keyword_fct3.remote(0, 1, c="w", d="hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(0, b=1, c="w", d="hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(a=0, b=1, c="w", d="hi")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(0, 1, d="hi", c="w")
assert ray.get(x) == "0 1 w hi"
x = keyword_fct3.remote(0, 1, c="w")
assert ray.get(x) == "0 1 w world"
x = keyword_fct3.remote(0, 1, d="hi")
assert ray.get(x) == "0 1 hello hi"
x = keyword_fct3.remote(0, 1)
assert ray.get(x) == "0 1 hello world"
x = keyword_fct3.remote(a=0, b=1)
assert ray.get(x) == "0 1 hello world"
# Check that we cannot pass invalid keyword arguments to functions.
@ray.remote
def f1():
return
@ray.remote
def f2(x, y=0, z=0):
return
# Make sure we get an exception if too many arguments are passed in.
with pytest.raises(TypeError):
f1.remote(3)
with pytest.raises(TypeError):
f1.remote(x=3)
with pytest.raises(TypeError):
f2.remote(0, w=0)
with pytest.raises(TypeError):
f2.remote(3, x=3)
# Make sure we get an exception if too many arguments are passed in.
with pytest.raises(TypeError):
f2.remote(1, 2, 3, 4)
@ray.remote
def f3(x):
return x
assert ray.get(f3.remote(4)) == 4
def test_args_starkwargs(ray_start_regular_shared):
def starkwargs(a, b, **kwargs):
return a, b, kwargs
class TestActor:
def starkwargs(self, a, b, **kwargs):
return a, b, kwargs
def test_function(fn, remote_fn):
assert fn(1, 2, x=3) == ray.get(remote_fn.remote(1, 2, x=3))
with pytest.raises(TypeError):
remote_fn.remote(3)
remote_test_function = ray.remote(test_function)
remote_starkwargs = ray.remote(starkwargs)
test_function(starkwargs, remote_starkwargs)
ray.get(remote_test_function.remote(starkwargs, remote_starkwargs))
remote_actor_class = ray.remote(TestActor)
remote_actor = remote_actor_class.remote()
actor_method = remote_actor.starkwargs
local_actor = TestActor()
local_method = local_actor.starkwargs
test_function(local_method, actor_method)
ray.get(remote_test_function.remote(local_method, actor_method))
def test_args_named_and_star(ray_start_regular_shared):
def hello(a, x="hello", **kwargs):
return a, x, kwargs
class TestActor:
def hello(self, a, x="hello", **kwargs):
return a, x, kwargs
def test_function(fn, remote_fn):
assert fn(1, x=2, y=3) == ray.get(remote_fn.remote(1, x=2, y=3))
assert fn(1, 2, y=3) == ray.get(remote_fn.remote(1, 2, y=3))
assert fn(1, y=3) == ray.get(remote_fn.remote(1, y=3))
assert fn(1,) == ray.get(
remote_fn.remote(
1,
)
)
assert fn(1) == ray.get(remote_fn.remote(1))
with pytest.raises(TypeError):
remote_fn.remote(1, 2, x=3)
remote_test_function = ray.remote(test_function)
remote_hello = ray.remote(hello)
test_function(hello, remote_hello)
ray.get(remote_test_function.remote(hello, remote_hello))
remote_actor_class = ray.remote(TestActor)
remote_actor = remote_actor_class.remote()
actor_method = remote_actor.hello
local_actor = TestActor()
local_method = local_actor.hello
test_function(local_method, actor_method)
ray.get(remote_test_function.remote(local_method, actor_method))
def test_oversized_function(ray_start_regular_shared):
bar = bytearray(800 * 1024 * 125)
@ray.remote
class Actor:
def foo(self):
return len(bar)
@ray.remote
def f():
return len(bar)
with pytest.raises(ValueError, match="The remote function .*f is too large"):
f.remote()
with pytest.raises(ValueError, match="The actor Actor is too large"):
Actor.remote()
def test_args_stars_after(ray_start_regular_shared):
def star_args_after(a="hello", b="heo", *args, **kwargs):
return a, b, args, kwargs
class TestActor:
def star_args_after(self, a="hello", b="heo", *args, **kwargs):
return a, b, args, kwargs
def test_function(fn, remote_fn):
assert fn("hi", "hello", 2) == ray.get(remote_fn.remote("hi", "hello", 2))
assert fn("hi", "hello", 2, hi="hi") == ray.get(
remote_fn.remote("hi", "hello", 2, hi="hi")
)
assert fn(hi="hi") == ray.get(remote_fn.remote(hi="hi"))
remote_test_function = ray.remote(test_function)
remote_star_args_after = ray.remote(star_args_after)
test_function(star_args_after, remote_star_args_after)
ray.get(remote_test_function.remote(star_args_after, remote_star_args_after))
remote_actor_class = ray.remote(TestActor)
remote_actor = remote_actor_class.remote()
actor_method = remote_actor.star_args_after
local_actor = TestActor()
local_method = local_actor.star_args_after
test_function(local_method, actor_method)
ray.get(remote_test_function.remote(local_method, actor_method))
@pytest.mark.skipif(client_test_enabled(), reason="internal api")
def test_object_id_backward_compatibility(ray_start_regular_shared):
# We've renamed Python's `ObjectID` to `ObjectRef`, and added a type
# alias for backward compatibility.
# This test is to make sure legacy code can still use `ObjectID`.
# TODO(hchen): once we completely remove Python's `ObjectID`,
# this test can be removed as well.
# Check that these 2 types are the same.
assert ray.ObjectID == ray.ObjectRef
object_ref = ray.put(1)
# Check that users can use either type in `isinstance`
assert isinstance(object_ref, ray.ObjectID)
assert isinstance(object_ref, ray.ObjectRef)
def test_nonascii_in_function_body(ray_start_regular_shared):
@ray.remote
def return_a_greek_char():
return "φ"
assert ray.get(return_a_greek_char.remote()) == "φ"
def test_failed_task(ray_start_regular_shared, error_pubsub):
@ray.remote
def throw_exception_fct1():
raise Exception("Test function 1 intentionally failed.")
@ray.remote
def throw_exception_fct2():
raise Exception("Test function 2 intentionally failed.")
@ray.remote(num_returns=3)
def throw_exception_fct3(x):
raise Exception("Test function 3 intentionally failed.")
throw_exception_fct1.remote()
throw_exception_fct1.remote()
x = throw_exception_fct2.remote()
try:
ray.get(x)
except Exception as e:
assert "Test function 2 intentionally failed." in str(e)
else:
# ray.get should throw an exception.
assert False
x, y, z = throw_exception_fct3.remote(1.0)
for ref in [x, y, z]:
try:
ray.get(ref)
except Exception as e:
assert "Test function 3 intentionally failed." in str(e)
else:
# ray.get should throw an exception.
assert False
class CustomException(ValueError):
def __init__(self, msg):
super().__init__(msg)
self.field = 1
def f(self):
return 2
@ray.remote
def f():
raise CustomException("This function failed.")
try:
ray.get(f.remote())
except Exception as e:
assert "This function failed." in str(e)
assert isinstance(e, ValueError)
assert isinstance(e, CustomException)
assert isinstance(e, ray.exceptions.RayTaskError)
assert "RayTaskError(CustomException)" in repr(e)
assert e.field == 1
assert e.f() == 2
else:
# ray.get should throw an exception.
assert False
def test_base_exception_raised(ray_start_regular_shared):
@ray.remote
def f():
raise BaseException("rip")
return 1
with pytest.raises(BaseException):
ray.get(f.remote())
def test_import_ray_does_not_import_grpc():
# First unload grpc and ray
if "grpc" in sys.modules:
del sys.modules["grpc"]
if "ray" in sys.modules:
del sys.modules["ray"]
# Then import ray from scratch
import ray # noqa: F401
# Make sure grpc did not get imported by "import ray"
assert "grpc" not in sys.modules
# Load grpc back so other tests will not be affected
try:
import grpc # noqa: F401
except ImportError:
# It's ok if we don't have grpc installed.
pass
if __name__ == "__main__":
sys.exit(pytest.main(["-sv", __file__]))
Reference in New Issue View Git Blame Copy Permalink