Switching to Cancel

python/ray/_raylet.pyx

@@ -975,10 +975,13 @@ cdef class CoreWorker:
     def kill_task(self, ObjectID object_id, c_bool force_kill):
         cdef:
             CObjectID c_object_id = object_id.native()
+            CRayStatus status = CRayStatus.OK()
 
-        with nogil:
-            check_status(CCoreWorkerProcess.GetCoreWorker().KillTask(
-                                                c_object_id, force_kill))
+        status = CCoreWorkerProcess.GetCoreWorker().KillTask(
+                                            c_object_id, force_kill)
+
+        if not status.ok():
+            raise ValueError(status.message().decode())
 
     def resource_ids(self):
         cdef:

python/ray/tests/BUILD

@@ -414,3 +414,11 @@ py_test(
     tags = ["exclusive"],
     deps = ["//:ray_lib"],
 )
+
+py_test(
+    name = "test_cancel",
+    size = "small",
+    srcs = ["test_cancel.py],
+    tags = ["exclusive"],
+    deps = ["//:ray_lib"],
+)

python/ray/tests/test_kill.py → python/ray/tests/test_cancel.py

@@ -8,7 +8,7 @@
 
 
 @pytest.mark.parametrize("use_force", [True, False])
-def test_kill_chain(ray_start_regular, use_force):
+def test_cancel_chain(ray_start_regular, use_force):
     """A helper method for chain of events tests"""
     signaler = SignalActor.remote()
 
@@ -50,7 +50,7 @@ def wait_for(t):
 
 
 @pytest.mark.parametrize("use_force", [True, False])
-def test_kill_multiple_dependents(ray_start_regular, use_force):
+def test_cancel_multiple_dependents(ray_start_regular, use_force):
     """A helper method for multiple waiters on events tests"""
     signaler = SignalActor.remote()
 
@@ -87,7 +87,7 @@ def wait_for(t):
 
 
 @pytest.mark.parametrize("use_force", [True, False])
-def test_single_cpu_kill(shutdown_only, use_force):
+def test_single_cpu_cancel(shutdown_only, use_force):
     ray.init(num_cpus=1)
     signaler = SignalActor.remote()
 
@@ -162,25 +162,25 @@ def infinite_sleep(y):
 
     sleep_or_no = [random.randint(0, 1) for _ in range(100)]
     tasks = [infinite_sleep.remote(i) for i in sleep_or_no]
-    killed = set()
+    cancelled = set()
     for t in tasks:
         if random.random() > 0.5:
             ray.cancel(t, use_force)
-            killed.add(t)
+            cancelled.add(t)
 
     ray.cancel(first, use_force)
-    killed.add(first)
+    cancelled.add(first)
 
-    for done in killed:
+    for done in cancelled:
         with pytest.raises((RayTaskError, RayCancellationError)):
             ray.get(done, 10)
 
     for indx in range(len(tasks)):
         t = tasks[indx]
         if sleep_or_no[indx]:
             ray.cancel(t, use_force)
-            killed.add(t)
-        if t in killed:
+            cancelled.add(t)
+        if t in cancelled:
             with pytest.raises((RayTaskError, RayCancellationError)):
                 ray.get(t, 10)
         else:

python/ray/worker.py

@@ -1663,8 +1663,9 @@ def cancel(object_id, force=False):
 
     if isinstance(object_id, ray.ObjectID) and object_id.task_id().actor_id(
     ).hex() not in ray.actors().keys():
-        worker.core_worker.kill_task(object_id, force)
-        return
+        if len(ray.wait([object_id], timeout=0)[1]) != 1:
+            return
+        return worker.core_worker.kill_task(object_id, force)
     raise ValueError("ray.cancel() only supported for non-actor object IDs. "
                      "Got: {}.".format(type(object_id)))
 

src/ray/core_worker/core_worker.cc

@@ -1200,11 +1200,14 @@ Status CoreWorker::SubmitActorTask(const ActorID &actor_id, const RayFunction &f
 }
 
 Status CoreWorker::KillTask(const ObjectID &object_id, bool force_kill) {
+  if (!object_id.CreatedByTask()) {
+    return Status::Invalid("Not created by a task.");
+  }
   auto task_spec = task_manager_->GetTaskSpec(object_id.TaskId());
   if (task_spec.has_value() && !task_spec.value().IsActorCreationTask()) {
-    RAY_RETURN_NOT_OK(direct_task_submitter_->KillTask(task_spec.value(), force_kill));
+    return direct_task_submitter_->KillTask(task_spec.value(), force_kill);
   }
-  return Status::OK();
+  return Status::Invalid("Task is not locally submitted.");
 }
 
 Status CoreWorker::KillActor(const ActorID &actor_id, bool force_kill,