From 0c835a379f9aa230e3da8e2f6ec0632a8eaf6476 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Wed, 21 Mar 2018 20:48:04 -0700 Subject: [PATCH] Fix resource bookkeeping for blocked actor methods. (#1766) --- src/local_scheduler/local_scheduler.cc | 10 +++++++--- test/actor_test.py | 20 ++++++++++++++++++-- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/src/local_scheduler/local_scheduler.cc b/src/local_scheduler/local_scheduler.cc index 66ae094e6..ecf43e35f 100644 --- a/src/local_scheduler/local_scheduler.cc +++ b/src/local_scheduler/local_scheduler.cc @@ -1145,10 +1145,14 @@ void process_message(event_loop *loop, * already blocked on an object that's not locally available, update its * state to blocked. */ worker->is_blocked = true; - /* Return the CPU resources that the blocked worker was using, but not - * other resources. */ + // Return the CPU resources that the blocked worker was using, but not + // other resources. If the worker is an actor, this will not return the + // CPU resources that the worker has acquired for its lifetime. It will + // only return the ones associated with the current method. + TaskSpec *spec = + Task_task_execution_spec(worker->task_in_progress)->Spec(); std::unordered_map cpu_resources; - cpu_resources["CPU"] = worker->resources_in_use["CPU"]; + cpu_resources["CPU"] = TaskSpec_get_required_resource(spec, "CPU"); release_resources(state, worker, cpu_resources); /* Let the scheduling algorithm process the fact that the worker is * blocked. */ diff --git a/test/actor_test.py b/test/actor_test.py index c1544c8f1..83515531d 100644 --- a/test/actor_test.py +++ b/test/actor_test.py @@ -1162,6 +1162,22 @@ class ActorsWithGPUs(unittest.TestCase): actor = Foo.remote() ray.get(actor.blocking_method.remote()) + @ray.remote(num_cpus=1) + class CPUFoo(object): + def __init__(self): + pass + + def blocking_method(self): + ray.get(f.remote()) + + # Make sure that lifetime CPU resources are not released when actors + # block. + actor = CPUFoo.remote() + x_id = actor.blocking_method.remote() + ready_ids, remaining_ids = ray.wait([x_id], timeout=1000) + self.assertEqual(ready_ids, []) + self.assertEqual(remaining_ids, [x_id]) + @ray.remote(num_gpus=1) class GPUFoo(object): def __init__(self): @@ -1170,10 +1186,10 @@ class ActorsWithGPUs(unittest.TestCase): def blocking_method(self): ray.get(f.remote()) - # Make sure that we GPU resources are not released when actors block. + # Make sure that GPU resources are not released when actors block. actor = GPUFoo.remote() x_id = actor.blocking_method.remote() - ready_ids, remaining_ids = ray.wait([x_id], timeout=500) + ready_ids, remaining_ids = ray.wait([x_id], timeout=1000) self.assertEqual(ready_ids, []) self.assertEqual(remaining_ids, [x_id])