Fix thread panic when "unreachable" SpawnedTask code is reachable.

[wiedld]

Which issue does this PR close?

Closes #12089

We are occasionally seeing a thread panic "entered unreachable code: SpawnedTask was cancelled unexpectedly" in our server logs. This is because the "unreachable" is actually reachable.

Rationale for this change

The existing unreachable assumes that no polling will occur after the runtime begins a shutdown. However, we found that while running datafusion in it's own runtime (own threadpool) we can actually hit this unreachable code -- when we start shutting down the executor and an internal poll still occurs.

We think that how we are executing our datafusion queries is not uncommon, and therefore the shutdown behavior should not be causing a thread panic in datafusion.

What changes are included in this PR?

• Test to demonstrate the existing behavior (that we can reach the "unreachable").
• make SpawnedTask::join_unwind fallible
• make DataFusionError::ExecutionJoin(JoinError)
• handle use of new fallible API

Are these changes tested?

Yes.

Are there any user-facing changes?

Not substantially.
There will be a new enum option for DataFusionError, and any external caller of SpawnedTask::join_unwind will need to handle the error.

[crepererum]

The code change itself is good, but the test needs some work.

[crepererum]

Can we remove this hook? IIRC it is a global process state (similar to env variables) and is shared with all the other tests that run in the same process (so all unit tests in common-runtime). This will be a pain to debug for others.

See my comment below on an alternative.

[crepererum]

Suggested change

	let join = rt.spawn(async {
	let task = SpawnedTask::spawn(async {
	let fut: Pending<()> = pending();
	fut.await;
	unreachable!("should never return");
	});
	let _ = task.join_unwind().await;
	});
	// caller shutdown their DF runtime (e.g. timeout, error in caller, etc)
	rt.shutdown_background();
	// race condition
	// poll occurs during shutdown (buffered stream poll calls, etc)
	let _ = join.await;
	let task = rt.spawn(async {
	SpawnedTask::spawn(async {
	let fut: Pending<()> = pending();
	fut.await;
	unreachable!("should never return");
	})
	}).await;
	// caller shutdown their DF runtime (e.g. timeout, error in caller, etc)
	rt.shutdown_background();
	// race condition
	// poll occurs during shutdown (buffered stream poll calls, etc)
	let _ = task.join_unwind().await;

I think the current setup is a bit flaky:

• it depends on your panic hook, see my concerns above
• it depends on the fact that your sub-runtime makes progress in spawn. this is also why you need to loop a bunch of times

I think this change also allows you to remove the loop and the panic hook.

[wiedld]

Test change made. Confirmed that it does properly find the panic (without the hook, without looping) if we have not made the code changes. Thank you. ❤️

[DDtKey]

Join error is either is_panic or is_cancelled. However it's not exhaustive checks - without pattern matching (may become reachable at the end of the day if something changes).

Logic is in general: resume_unwind if panic or else return an error
I'd suggest to simplify this and keep the comment

            // `JoinError` can be caused either by panic or cancellation. We have to handle panics:
            if e.is_panic() {
                std::panic::resume_unwind(e.into_panic());
            } else {
                // Cancellation may be caused by two reasons:
                // 1. Abort is called, but since we consumed `self`, it's not our case (`JoinHandle` not accessible outside).
                // 2. The runtime is shutting down.
                log::warn!("SpawnedTask was polled during shutdown");
                e
            }

[crepererum]

This idea came a bit late, so this is optional:

Could you add a test that shows that the panic is propagated, e.g.:

#[tokio::test]
#[should_panic(expected = "foo")]
async fn panic_resume() {
    // this should panic w/o an `unwrap`
    SpawnedTask::spawn(async {panic!("foo")}).join_unwind().await.ok();
}

[alamb]

This is a good idea -- I'll make a follow on PR to do so (as it adds additional test coverage for an existing feature rather than something that was changed in this PR)

[alamb]

#12134

[alamb]

Thank you @wiedld and @crepererum and @DDtKey