[Fix][Core] Execute user requested actor exit in C++ side

[MortalHappiness]

Why are these changes needed?

When an asyncio task creates another asyncio task, raising AsyncioActorExit cannot make the caller exit because they are not the same task. Therefore, this PR makes exit_actor to request actor exit in core worker context, which will be checked regularly by core worker.

This PR also removes AsyncioActorExit exception because it is no longer needed.

Related issue number

Closes: #49451

Follow-up: Unify actor exit codepaths

Core worker exit codepaths:

Either ForceExit or Exit:

• Cancel task (called by ray.cancel python function):

  ray/src/ray/core_worker/core_worker.cc

  Line 4348 in 9008eb6

  ForceExit(rpc::WorkerExitType::INTENDED_USER_EXIT,

• Requested to kill actor by GCS:

  • ray/src/ray/core_worker/core_worker.cc

    Line 4510 in 9008eb6

    Exit(rpc::WorkerExitType::INTENDED_SYSTEM_EXIT,

  • ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

    Line 1730 in 67f9490

    rpc::KillActorRequest request;

• Received ExitRequest RPC call:

  ray/src/ray/core_worker/core_worker.cc

  Line 4737 in 9008eb6

  Exit(rpc::WorkerExitType::INTENDED_SYSTEM_EXIT,

• Based on the status of task_execution_callback application-language callback:

  • ray/src/ray/core_worker/core_worker.cc

    Line 3351 in 9008eb6

    status = options_.task_execution_callback(

  • ray/src/ray/core_worker/core_worker.cc

    Lines 3424 to 3443 in 9008eb6

    	if (status.IsCreationTaskError()) {
    	Exit(rpc::WorkerExitType::USER_ERROR,
    	absl::StrCat(
    	"Worker exits because there was an exception in the initialization method "
    	"(e.g., __init__). Fix the exceptions from the initialization to resolve "
    	"the issue. ",
    	status.message()),
    	creation_task_exception_pb_bytes);
    	} else if (status.IsIntentionalSystemExit()) {
    	Exit(rpc::WorkerExitType::INTENDED_USER_EXIT,
    	absl::StrCat("Worker exits by an user request. ", status.message()),
    	creation_task_exception_pb_bytes);
    	} else if (status.IsUnexpectedSystemExit()) {
    	Exit(rpc::WorkerExitType::SYSTEM_ERROR,
    	absl::StrCat("Worker exits unexpectedly. ", status.message()),
    	creation_task_exception_pb_bytes);
    	} else if (!status.ok()) {
    	RAY_LOG(FATAL) << "Unexpected task status type : " << status;
    	}
    	return status;

• Based on the status of check_signals application-language callback:

  • ray/src/ray/core_worker/core_worker.cc

    Lines 3159 to 3170 in 9008eb6

    	auto status = options_.check_signals();
    	if (status.IsIntentionalSystemExit()) {
    	Exit(rpc::WorkerExitType::INTENDED_USER_EXIT,
    	absl::StrCat("Worker exits by a signal. ", status.message()),
    	nullptr);
    	}
    	if (status.IsUnexpectedSystemExit()) {
    	Exit(
    	rpc::WorkerExitType::SYSTEM_ERROR,
    	absl::StrCat("Worker exits unexpectedly by a signal. ", status.message()),
    	nullptr);
    	}

Actor exit codepaths (in GCS):

Either DestroyActor or KillActor

• Actor out of scope:

  ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

  Line 377 in 67f9490

  DestroyActor(actor_id,

• Poll owner for actor ref deletd:

  ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

  Line 1017 in 67f9490

  DestroyActor(actor_id,

• Called by ray.kill (CoreWorker::KillActor -> ActorInfoAccessor::AsyncKillActor -> KillActorViaGcs RPC call -> GcsActorManager::HandleKillActorViaGcs):

  ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

  Lines 694 to 698 in 67f9490

  	if (no_restart) {
  	DestroyActor(actor_id, GenKilledByApplicationCause(GetActor(actor_id)));
  	} else {
  	KillActor(actor_id, force_kill);
  	}

• Worker died:

  ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

  Line 1217 in 67f9490

  DestroyActor(child_id,

• Node died:

  ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

  Line 1294 in 67f9490

  DestroyActor(child_id,

• Actor scheduling failed:

  ray/src/ray/gcs/gcs_server/gcs_actor_manager.cc

  Line 1517 in 67f9490

  DestroyActor(actor->GetActorID(), death_cause);

Examples:

• Normal actor exit (process completed): GCS will poll owner for actor ref deleted, and then notify core worker to delete actor.
• exit_actor is called (this PR): core worker exits based on application-language callback.
• ray.kill is called: DestroyActor in GCS will be called, and then it'll notify core worker to exit.

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[edoakes]

@MortalHappiness let me know when this is ready for review

[edoakes]

At first glance, I think we need to use a threading.Event here, not an asyncio.Event. This is because we need to communicate across threads, not just across asyncio tasks within the same thread.

[MortalHappiness]

@edoakes What do you think would be a better approach? Implementing it purely on the Python side using threading.Event, or implementing it on the C++ side, as you mentioned in #49451 (comment)?

If implemented on the C++ side, I will export a function to Python and call this function when exit_actor is triggered. Inside this function, I will modify some state and periodically check whether this state indicates that an exit is required.

[edoakes]

@edoakes What do you think would be a better approach? Implementing it purely on the Python side using threading.Event, or implementing it on the C++ side, as you mentioned in #49451 (comment)?

If implemented on the C++ side, I will export a function to Python and call this function when exit_actor is triggered. Inside this function, I will modify some state and periodically check whether this state indicates that an exit is required.

If it's possible to do it in C++ without significantly more code change I'd prefer to do it there. That is more future proof and allows us to reuse it across different language frontends in the future.

[edoakes]

@MortalHappiness great start!

Let's please unify as many of the actor exit codepaths as possible to share the same logic. For example, can we also unify the single threaded and multi threaded actor exit logic to use the same flag?

[MortalHappiness]

Let's please unify as many of the actor exit codepaths as possible to share the same logic. For example, can we also unify the single threaded and multi threaded actor exit logic to use the same flag?

Should I do this in this PR or in follow-up PRs?

[edoakes]

Let's please unify as many of the actor exit codepaths as possible to share the same logic. For example, can we also unify the single threaded and multi threaded actor exit logic to use the same flag?

Should I do this in this PR or in follow-up PRs?

I'm ok to do it in follow-up PRs, but could you summarize for me all of the existing exit paths before this PR and what you see as the end goal state?

[MortalHappiness]

Hi @edoakes I've tried to summarize exit codepaths in PR description. Let me know if I missed something.

[edoakes]

Looks good now -- my only concern is that we don't have any behavior change for the sync worker case. Let's make sure that still raises immediately and fails the current task w/ an RayActorError

It seems like the "raising immediately" behavior might not be covered by a current test case (if my understanding is correct). If so, please add a test case for that. For example:

def method(self):
    ray.exit_actor()
    do_something() # This shouldn't be called and this method should raise `RayActorError` when `ray.get` is called.

[MortalHappiness]

Hi @edoakes I've addressed all comments. Ready for another review. Thanks.

[edoakes]

A little confused at why this is conditional because we should try to make the behavior as uniform as possible:

• For non-async actors, don't we still want to set the field in the core worker? Otherwise, if users accidentally or intentionally catch the SystemExit exception that is raised, the actor won't actually be exited.
• For async actors, don't we still want to raise the SystemExit exception immediately?

[edoakes]

I don't think this assertion would actually catch a regression because this line might execute and then the exit method would still return a RayActorError. For any condition like this, you should verify that if you intentionally break the behavior (in this case, make it so that sync actors don't raise immediately), the test fails. I don't think it will in this case.

Instead, what you will need to do is add an "out of bounds" check -- so for example catch the SystemExit exception and then send a signal on an actor or write a tempfile like the other test cases. Then after the actor exits, assert that the exception was raised and therefore the out of band behavior happened.

[edoakes]

A few things for this test:

1. We should add the same check that we have for the sync case that exit_actor() immediately raises an exception).
2. I might be misguided, but I think we can have two test cases here instead of just checking for both errors. The first one can call exit_actor on the task directly executing the actor method and that should always return "exit_actor() was called...". The second one can call exit_actor in a background asyncio task and then assert that `"User requested to exit the actor" is eventually raised.

[edoakes]

hm this test doesn't look correct. the method starts a background task so there is no guarantee that runs before the method finishes successfully. I think what you need to do is start the background task and have it block on a SignalActor before it actually calls exit_actor(), check that start_exit_task completes successfully, then unblock the background task so it calls exit_actor(), then use wait_for_condition on ray.get(a.noop_method.remote()) to check that eventually subsequent methods will return the actor error

[MortalHappiness]

@edoakes I've addressed all of your comments, but the actual implementation differs slightly. Let me describe the changes here:

• I added back the AsyncioActorExit exception. Since you mentioned that a normal async task should raise immediately, I can't just call set_current_actor_should_exit because this flag is checked periodically rather than triggering an immediate exception. I also can't raise SystemExit, as noted in the original comment

  ray/python/ray/actor.py

  Lines 1784 to 1786 in e151dc2

  	# In asyncio actor mode, we can't raise SystemExit because it will just
  	# quit the asycnio event loop thread, not the main thread. Instead, we
  	# raise a custom error to the main thread to tell it to exit.

• I split the test into 6 separate tests:

  1. test_exit_actor_invalid_usage_error:
     • This is the original test for invalid call of exit_actor in non-actor functions.
  2. test_exit_actor_normal_actor_raise_immediately:
     • This is the original test for a normal actor, but I added one line to ensure that any code after exit_actor is not executed.
  3. test_exit_actor_normal_actor_user_catch_err_should_still_exit:
     • This is an "out-of-bound" test for a normal actor.
     • In this case, the user catches SystemExit, so the program does not raise immediately.
     • However, the actor will still exit shortly after the flag is checked by the core worker, similar to the exit path for an async background task.
  4. test_exit_actor_async_actor_raise_immediately:
     • This is the original test for an async actor, but I added one line to verify that any code after exit_actor is not executed.
     • Why does this test still check both "exit_actor()" in str(exc_info.value)" and "User requested to exit the actor" in str(exc_info.value)"?
       • Most of the time, the first case occurs.
       • However, since we call set_current_actor_should_exit (which is checked periodically by the core worker), there is a small chance that the second case occurs instead.
  5. test_exit_actor_async_actor_user_catch_err_should_still_exit:
     • This is an "out-of-bound" test for an async actor.
     • In this case, the user catches AsyncioActorExit, so the program does not raise immediately.
     • However, the actor will still exit shortly after the flag is checked by the core worker, similar to the exit path for an async background task.
  6. test_exit_actor_async_actor_nested_task:
     • I updated this test to use SignalActor.
     • No error is raised in this case, but the actor still exits short after when the flag is checked by the core worker.
     • Therefore the atexit handler is triggered.

[edoakes]

@MortalHappiness why can't we set the should_exit flag and then raise the normal SystemExit? AFAICT this makes the behavior as consistent as possible:

• The currently-running task will raise and exit immediately (same as a sync actor method).
• The actor will exit soon after once the should_exit flag is checked.

Am I missing something?

[MortalHappiness]

@MortalHappiness why can't we set the should_exit flag and then raise the normal SystemExit? AFAICT this makes the behavior as consistent as possible:

• The currently-running task will raise and exit immediately (same as a sync actor method).
• The actor will exit soon after once the should_exit flag is checked.

Am I missing something?

@edoakes After some debug printing and some experiment, I found that asyncio.run_coroutine_thread_safe handles SystemExit and KeyboardInterrupt differently.

CPython source code: https://github.com/python/cpython/blob/e53d105872fafa77507ea33b7ecf0faddd4c3b60/Lib/asyncio/tasks.py#L991-L1011

The following is a toy example.

import time
import asyncio
import threading

async def f():
    print("Inside coroutine f()")
    # raise ValueError
    # raise BaseException
    raise SystemExit

def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

def main():
    loop = asyncio.new_event_loop()
    thread = threading.Thread(target=start_loop, args=(loop,))
    thread.start()

    future = asyncio.run_coroutine_threadsafe(f(), loop)

    try:
        time.sleep(2)
        print("loop alive?", loop.is_running())
        future.result()
    except BaseException as e:
        print("Caught exception:", repr(e))

    print("about to stop loop")
    loop.call_soon_threadsafe(loop.stop)
    thread.join()

if __name__ == "__main__":
    main()

Explanation:

• raise ValueError or BaseException in f: Loop alive, and exception was caught.
• raise SystemExit in f: Loop dead. Program hangs forever on future.result() because future.set_exception(exc) is not called inside asyncio.run_coroutine_thread_safe

And in Ray, we use run_coroutine_thread_safe to run async tasks:

ray/python/ray/_raylet.pyx

Line 4587 in b845581

future = asyncio.run_coroutine_threadsafe(async_func(), eventloop)

This is why we cannot raise SystemExit in exit_actor.

[edoakes]

@MortalHappiness I see, thanks for the explanation. Is it possible to add a top-level exception handler for SystemExit in the async_func() that will call future.set_exception()?

[MortalHappiness]

@MortalHappiness I see, thanks for the explanation. Is it possible to add a top-level exception handler for SystemExit in the async_func() that will call future.set_exception()?

@edoakes Not easy to do that because the future object is returned by asyncio.run_coroutine_threadsafe, so we cannot call future.set_exception inside async_func.

It's easier to raise a custom exception in this case. My idea is that exit_actor only raises SystemExit, then async_func catches SystemExit, raises a custom exception, and then future.result() catches the custom exception and raises SystemExit. Although this approach is similar to the previous custom exception class AsyncioActorExit, the advantage is that the external file will never see this custom exception class. It only needs to be declared inside this function. WDYT?

[edoakes]

Code changes LGTM, just a few nits

Your latest suggestion sounds better than what is in the code, but given that we already have the existing behavior with AsyncioActorExit and it would be a very minor API breakage to change, better to leave it as is.

[edoakes]

Please add header comments for these indicating what they're used for

[MortalHappiness]

added

[edoakes]

Can you add a test case for calling ray.exit_actor() within the constructor if there isn't one already?

[MortalHappiness]

Added 3 tests: test_exit_actor_normal_actor_in_constructor_should_exit, test_exit_actor_async_actor_in_constructor_should_exit, and test_exit_actor_async_actor_nested_task_in_constructor_should_exit

[edoakes]

🚢

[MortalHappiness]

@edoakes I found that you enabled auto-merged. But I found that I accidentally changed 1 line that should not be changed so I pushed it again and the auto-merge is disabled. Could you enable it again? Thanks.