ARROW-12560: [C++] Add scheduling option for Future callbacks

[westonpace]

Previously a future's callbacks would always run synchronously, either as part of Future::MarkFinished or as part of Future::AddCallback. Executor::Transfer made it possible to schedule continuations on a new thread but it would only take effect if the transferred future's callbacks were added before the source future finished. There are times when the desired behavior is to spawn a new thread task even if the source future is finished already.

This PR adds three scheduling options:

• Never - The default (and existing) behavior, never spawn a new task
• IfUnfinished - Spawn a new task only if the future isn't already finished when the callback is added
• Always - Always spawn a new task, on both finished and unfinished futures, regardless of destination thread pool idleness.

The Never option doesn't make any sense for transferring so the transfer only has two choices (always or if unfinished).

[github-actions]

https://issues.apache.org/jira/browse/ARROW-12560

[westonpace]

CC @pitrou I think you referenced this in your latest execution engine PR.

[pitrou]

I'm not sure why you're suggesting to add so much sophistication. To me there are only two interesting options: "always" and "if unfinished". So we could have Transfer (transfer always) vs. TransferUnfinished.

[westonpace]

Since I'm working on work stealing at the thread pool level I agree that idle is no longer needed. I've cleaned this up and rebased. It's much simpler than it was before.

[westonpace]

Also, I ran into a bit of trouble with the future callback's weak reference to the future. Before we could just assume it was valid since all callbacks were completed before MarkFinished was completed. Now, it is possible for a future to schedule a callback and that callback to far outlive the call to MarkFinished. So now when a callback is scheduled (run on an executor) we make a copy of the FutureImpl's shared_ptr to keep it alive until that callback has a chance to run.

[pitrou]

Sorry for the delay. This looks good to me on the principle.

[pitrou]

nullptr

[westonpace]

Fixed.

[pitrou]

Why "a copy"? It's not clear to me where a copy is being made.

[westonpace]

The copy is a few lines down when we call shared_from_this. I'll move the comment and make it more explicit.

[pitrou]

If it's only the shared_ptr copy, then I'm not sure it's worth mentioning.

[westonpace]

I think the more important thing is that we are intentionally extending the lifetime of the future. I reworded the comment a bit and dropped the "copy". I can always remove it if we want.

[pitrou]

The coding conventions prohibit passing mutable lrefs. You could make this a CallbackRecord&&, for example.

[westonpace]

Done.

[pitrou]

We should avoid using ALL_CAPS names, because of potential clashes with macros (this is a common issue with Windows headers, unfortunately).

[westonpace]

Hmm, technically the style guide prefers kAlways but I see Always used more often in Arrow. Although some of the gandiva code uses kAlways. (https://google.github.io/styleguide/cppguide.html#Enumerator_Names). Any preference?

[pitrou]

Always sounds fine to me.

[westonpace]

Ok. I'll make a PR to add this to the style guide docs as well.

[pitrou]

It's a bit weird to have this in a private test file, and the mock executor in a .h.

[westonpace]

My rationale was only that DelayedExecutor is only used in future_test.cc while MockExecutor is used in future_test.cc and thread_pool_test.cc but I see your point. I'll move this into test_common.h.

[pitrou]

NEVER is never tested?

[westonpace]

Added a test.

[pitrou]

Nit, but it would probably be nicer to be able to spell this as TransferAlways(fut).

[westonpace]

Done.

[westonpace]

@pitrou Don't worry about the delay, I've been plenty busy elsewhere. I have a just a few follow-up questions and then I'll make the changes.

[westonpace]

Ok, I've addressed the comments and this is ready for review again.

[pitrou]

Thanks for the update @westonpace . I'll merge once CI passes.