[Core] Fix check failure RAY_CHECK(it != current_tasks_.end());

[jjyao]

Why are these changes needed?

The check failure can happen under the following sequence of events:

1. An async or threaded actor is launched.
2. An actor task is submitted.
3. Actor task is received by the actor and is being executed. The task_id is added to CoreWorker.current_tasks_.
4. A transient network error happens and caller retries the actor task.
5. The second attempt of the actor task is received by the actor and is being executed.
6. The first attempt of the actor task finishes and the task_id is removed from CoreWorker.current_tasks_.
7. The second attempt of the actor task finishes and the check failure happens when we try to erase the same task_id from CoreWorker.current_tasks_ again:

auto it = current_tasks_.find(task_spec.TaskId());
RAY_CHECK(it != current_tasks_.end());
current_tasks_.erase(it);

This PR fixes the issue by making sure different attempts of the same task are executed sequentially. The reason why we don't support running them in parallel is that it's not safe to assume user's code can handle concurrent execution of the same actor method.

Related issue number

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 :(

[rkooo567]

the approach lgtm. lmk when I can review the PR!

[rynewang]

We may want to change pending_task_id_to_is_canceled value to a state enum.

enum TaskState {
// Waiting for deps, can be cancelled.
// Invariant: queued_actor_tasks_[task_id] does not exist.
WAITING_FOR_DEPS = 0;
// Waiting for deps, but cancelled by user.
CANCELLED_BY_USER;
// Waiting for deps, but cancelled by newer attempts.
CANCELLED_BY_NEW_ATTEMPTS;
// Running, can't be cancelled.
RUNNING;
};

On cancel:

• case WAITING_FOR_DEPS, CANCELLED_BY_NEW_ATTEMPTS: change to CANCELLED_BY_USER, cancel queued_actor_tasks_[task_id]
• else: cancel queued_actor_tasks_[task_id]

On new attempt Add:

• case WAITING_FOR_DEPS: change to CANCELLED_BY_NEW_ATTEMPTS, add new attempt to queued_actor_tasks_
• case CANCELLED_BY_USER: also cancel new attempt.
• else: add new attempt to queued_actor_tasks_

On dep resolved:

• case WAITING_FOR_DEPS: run
• case CANCELLED_*: Cancel(reason)
• case RUNNING: check fail

On run finished:

• case RUNNING: erase the state, wait-dep for queued_actor_tasks_[task_id] if any
• else: check fail

[jjyao]

pending_task_id_to_is_canceled cancel here only means user triggered cancellation.