[Feat] cancel sync actor by checking is_canceled()

[machichima]

Description

Enable sync actors to detect task cancellation during execution via ray.get_runtime_context().is_canceled() API for graceful termination

Main changes:

Record task cancellation in executor:

• core_worker.cc
  • Track canceled task with canceled_tasks_ and expose it to Python API through IsTaskCanceled()
• runtime_context.py
  • Add is_canceled() API that calls CoreWorker::IsTaskCanceled() following path worker.py -> _raylet.pyx -> core_worker.cc

Raise cancellation error when using ray.get() (on submitter side)

• actor_task_submitter.cc
  • For sync actor task that completed without error but marked as canceled (Get from TaskManager::IsTaskCanceled()), explicitly set TASK_CANCELLED error
• task_manager.cc
  • Add IsTaskCanceled() to expose is_canceled_ flag

Related issues

Related to #58213

Additional information

Tested with following example:

import ray
import time

ray.init()


@ray.remote(max_concurrency=10)
class ThreadedActor:
    def __init__(self):
        self.counter = 0

    def long_running_task(self, duration=10):
        for i in range(duration):
            print("counter: ", self.counter)
            if ray.get_runtime_context().is_canceled():
                return "canceled"

            self.counter += 1
            time.sleep(0.1)

        return "completed"


if __name__ == "__main__":
    actor = ThreadedActor.remote()
    task_ref = actor.long_running_task.remote(duration=10)

    time.sleep(0.3)
    ray.cancel(task_ref)

    try:
        result = ray.get(task_ref)
        print(f"Result: {result}")
    except ray.exceptions.TaskCancelledError as e:
        print(f"Cancelled: {e}")

    ray.shutdown()

Result screenshot:

[machichima]

I was trying to make ray.get() also raising TaskCanceled error for canceled sync task here.

I don't think this the best way handling it. Please let me know if there's a better way or where can I look into 🙏

[machichima]

cc @edoakes @Sparks0219 PTAL. Thank you!