Fixing segfaults in Async Streaming Generator

[alexeykudinkin]

Related issue number

Addresses #43771

This change makes sure that no side-effects outlive invocation of execute_task method:

Previously, after scheduling tasks onto Core Worker's ThreadPoolExecutor these could have continued executing, even after the request has been cancelled (cancelling of the future wouldn't cancel already running task), leading to SIGSEGV when the task running in TPE would try to access data-structures that were already cleaned up after returning from this method.

With this change:

• Upon encountering any failure, we'd set an interrupt_signal_event interrupting already scheduled, but not yet executed tasks (preventing them from modifying externally passed in data-structures)

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

[jjyao]

LG. Thanks for fixing it.

[jjyao]

should we call it thread_pool_executor_for_asnc_event_loop?

[alexeykudinkin]

Not a big fan of adding sentences as names (event_loop_executor captures the intent perfectly)

[jjyao]

You need to add it to bazel BUILD file otherwise it won't run by CI.

Do you want to add it to test_streaming_generator_4.py or create a _5.py?

[alexeykudinkin]

Good call!

[alexeykudinkin]

@jjyao have to revisit the approach:

• TPE is used cooperatively by multiple requests (hence can't shut it down)
• Running multiple TPEs proves costly in the benchmarks (streaming_core_throughput.py)
• Instead, revisited approach to use an external interruption (using threading.Event) that only causes very minor effect on performance (see below)

# Before (master)
Core Actors streaming throughput (ASYNC) (num_replicas=1, tokens_per_request=1000, batch_size=10): 13052.62 +- 372.43 tokens/s
(CallerActor pid=56100) Individual request quantiles:
(CallerActor pid=56100) 	P50=497.5112084999988
(CallerActor pid=56100) 	P75=648.4424480000001
(CallerActor pid=56100) 	P99=807.8154376599998

# After 
Core Actors streaming throughput (ASYNC) (num_replicas=1, tokens_per_request=1000, batch_size=10): 12782.83 +- 216.97 tokens/s
(CallerActor pid=51634) Individual request quantiles:
(CallerActor pid=51634) 	P50=504.2683960000014
(CallerActor pid=51634) 	P75=667.4028960000006
(CallerActor pid=51634) 	P99=811.2643177199991

[rkooo567]

IIUC the issue is if any of coroutine is failed (cancelled), it raises an exception before all tasks in the thread pool is finished -> ExecuteTask finishes before task inside thread pool finishes right?

Hmm actually I am curious why don't we just use gather(return_exceptions=True)?

In this case, gather will not return until everything in the coroutine is finished, and the race condition won't happen?

[rkooo567]

Hmm theoretically, isn't this possible the code is already executing the code below?

Is it okay because none of other functions are not accessing data structure that could be deleted when a task finishes?

[jjyao]

Yea, I have the same question: isn't there still a race condition where the signal event is set between line 1301 and 1304?

[alexeykudinkin]

Theoretically -- yes, RC is still possible, but is much, much less likely now as the task will have to be paused by the interpreter right at the spot where we're pushing into the vector, then switch over to the event-loop thread and let the caller complete freeing the vector.

This interrupting signal event though serves as a barrier guaranteeing that any tasks that still gonna reach this point won't proceed if the caller completed

[alexeykudinkin]

We'll fully eliminate RC only with removing pushing into vector completely (which i'm doing in #42443)

[rkooo567]

Q: isn't using await asyncio.gather(*futures, return_exceptions=True) solve the issue?

[alexeykudinkin]

Answered here: #43775 (comment)

[alexeykudinkin]

IIUC the issue is if any of coroutine is failed (cancelled), it raises an exception before all tasks in the thread pool is finished -> ExecuteTask finishes before task inside thread pool finishes right?

Hmm actually I am curious why don't we just use gather(return_exceptions=True)?

In this case, gather will not return until everything in the coroutine is finished, and the race condition won't happen?

That's not sufficient unfortunately, b/c of cancellation -- in this case there will be CancelledError thrown at await asyncio.gather(...) which we have to handle.

On top of that -- cancelling future of the task already running in the executor won't interrupt the task, hence we have to "enforce" the interruption via signal to guarantee that the vector isn't being pushed into once we return from this method