Why are these changes needed?

We observe an issue that ray.wait hangs indefinitely in the following case:

• Driver sends a Ray task (task1) with a streaming generator enabled to actor A, and the task’s sequence number is 0.
• Actor A receives task1
• Actor A somehow becomes unavailable from the driver's perspective.
• Driver resubmits task1 again with the same sequence number (i.e. seq_no=0)
• Actor A finishes the first task1 and sends a RPC ReportGeneratorItemReturns to the driver.
• The driver replies with a Status::NotFound to Actor A because the driver has already made another attempt, and the previous attempt is now outdated.
  

  ray/src/ray/core_worker/task_manager.cc

  Lines 678 to 685 in c111d99

  	if (it->second.spec.AttemptNumber() > attempt_number) {
  	// Generator task reports can arrive at any time. If the first attempt
  	// fails, we may receive a report from the first executor after the
  	// second attempt has started. In this case, we should ignore the first
  	// attempt.
  	execution_signal_callback(
  	Status::NotFound("Stale object reports from the previous attempt."), -1);
  	return false;

• Actor A receives the second task1 again. However, the actor will not execute the request because the sequence number is the same.
• The driver will continue resubmitting task1 with seq_no=0, while the actor repeatedly cancels the tasks.

3 possible solutions

Solution 1: Always increment seq_no during resubmissions

• Pros:
  • Simplest implementation.
  • Solves a slightly smaller issue than Solution 2, but the implementation is much simpler.
• Cons:
  • Doesn't promise the execution order of the retryable actor tasks.
• Examples
  • Case 1
    • The driver submits task A (seq_no=0) to an actor, and B, C, D are in the queue.
    • Actor unavailable.
    • The actor doesn’t receive task A.
    • The network becomes healthy.
    • The driver resubmits task A (seq_no=4). Update client_processed_up_to_.
    • The actor executes B, C, D, A.
  • Case 2
    • The driver submits task A (seq_no=0) to an actor, and B, C, D are in the queue.
    • The actor receives task A.
    • Actor unavailable.
    • The network becomes healthy.
    • The driver resubmits task A (seq_no=4).
    • The actor executes A, B, C, D, A.

Solution 2: Resubmit with the same seq_no first. If this fails, resubmit with a new seq_no

• Pros:
  • If the actor doesn't receive any tasks, the task execution order remains the same for resubmission.
• Cons:
  • Doesn't promise the execution order of the actor tasks if the actor has already received some tasks before network issues.
  • The complexity is much higher than Solution 1. There has been a PR tried to fix it, but the PR failed to get merged (Introducing StaleTaskError #46705).
• Examples
  • Case 1
    • The driver submits task A (seq_no=0) to an actor, and B, C, D are in the queue.
    • Actor unavailable.
    • The actor doesn’t receive task A.
    • The network becomes healthy.
    • The driver resubmits task A (seq_no=0).
    • The actor executes A, B, C, D.
  • Case 2
    • The driver submits task A (seq_no=0) to an actor, and B, C, D are in the queue.
    • The actor receives task A.
    • Actor unavailable.
    • The network becomes healthy.
    • The actor resubmits task A (seq_no=0).
    • The actor cancels the second task A.
    • The actor resubmits task A (seq_no=4).
    • The actor executes A, B, C, D, A.

Solution 3: Actor caches the objects until the driver says that the objects are unnecessary.

• Pros
  • Promise the execution order
• Cons
  • Complex
  • It's possible to increase the possibility of an OOM error because actors need to cache the results. This could be solved by adding some kind of backpressure (if there are >N results pending due to network instability, pause generator execution). However, it increases the complexity further.
• Examples
  • Case 1
    • The driver submits task A (seq_no=0) to an actor, and B, C, D are in the queue.
    • Actor unavailable.
    • The actor doesn’t receive task A.
    • The network becomes healthy.
    • The driver resubmits task A (seq_no=0).
    • The actor executes A, B, C, D.
  • Case 2
    • The driver submits task A (seq_no=0) to an actor, and B, C, D are in the queue.
    • The actor receives task A.
    • Actor unavailable.
    • The network becomes healthy.
    • The driver resubmits task A (seq_no=0).
    • The actor executes A, B, C, D. When the actor receives the second task A, the actor returns the cache result to the driver.

Conclusion

3 solutions don't promise the execution order if take lineage reconstruction into consideration. Choose Solution 1 because it is the simplest.

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

To speed up the reproduction, you need to update the two configs:

// old: 300s, new: 30s
RAY_CONFIG(int64_t, grpc_client_keepalive_time_ms, 30000)
// old: 120s, new: 15s
RAY_CONFIG(int64_t, grpc_client_keepalive_timeout_ms, 15000)

The reproduction uses iptables to block the connections for HandlePushTask RPC between the driver and actor, so the driver can't receive the RPC reply from the actor after the network policies are applied.

The driver thinks the actor is "unavailable" because of gRPC keeplive watchdog timeout. The worst case it should happen after grpc_client_keepalive_time_ms + grpc_client_keepalive_timeout_ms (i.e. 45s in the above case).

• test.py
• network_policy.sh

RAY_BACKEND_LOG_LEVEL=debug python3 test.py
./network_policy.sh -A

# Wait until actor unavailable. Run the following command until the core-driver log file has some related logs
grep -rn "UNAVAIL"

# Remove network policy
./network_policy.sh -D

# Wait until the first actor task finishes and the io context starts to handle the resubmit task

Without this PR

ray.wait stucks forever.

With this PR

The resubmitted task will be executed after the old one finished, and the driver receives 0 ~ 7 from the first task and 8 ~ 19 from the resubmitted one.