Fix resource bookkeeping for blocked actor methods. (#1766)

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<std::string, double> 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. */

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])