ARROW-17509: [C++] Simplify async scheduler by removing the need to call End #14524
Conversation
|
|
|
CC @save-buffer would appreciate a look |
| }); | ||
| } | ||
| #endif | ||
| // TODO(weston) The entire concept of ExecNode::finished() will hopefully go |
There was a problem hiding this comment.
yep my simplification PR gets rid of those finished futures.
| // part of StartProducing) then the plan may be finished before we return from this | ||
| // call. | ||
| Future<> scheduler_finished = | ||
| util::AsyncTaskScheduler::Make([this](util::AsyncTaskScheduler* async_scheduler) { |
There was a problem hiding this comment.
Could we move this giant lambda into its own function? The extra layer of indentation is pretty confusing
cpp/src/arrow/util/async_util.cc
Outdated
| } | ||
| std::shared_ptr<AsyncTaskScheduler> MakeSubScheduler( | ||
| std::unique_ptr<AsyncTaskScheduler::Holder> MakeSubScheduler( | ||
| FnOnce<Status(AsyncTaskScheduler*)> initial_task, |
There was a problem hiding this comment.
Could we guarantee that this initial_task gets run synchronously? I get why it's needed (the scheduler will end itself immediately after creation without it), but I would like the control flow for scheduling the initial tasks to be simple. We could make this function return a Result<unique_ptr> in case a node has a problem scheduling its initial tasks.
There was a problem hiding this comment.
Yes. initial_task is always run synchronously as part of the call to MakeSubScheduler (so it should be safe, e.g. to use [&]). Furthermore, when you add a task to a scheduler that does not have a throttle, the task submitter portion will always run synchronously as part of the call to AddTask. I'll update the docs to make this clearer.
You are now allowed to add tasks after a scheduler is "ended". The top-level scheduler basically starts "ended" (after the initial task runs). If you don't use or need sub-schedulers then we have what you describe. However, there are some cases where we don't want a sub-scheduler to end even though it has run out of tasks. For example, when writing to a file. We want to close the file when all write tasks have finished. However, we don't know, at the start, how many write tasks there will be, and we don't have all of them (sometimes InputReceived will add another write task and sometimes it won't, depends on partitioning). That being said, we still want to end the file writer no matter what when the plan itself is out of tasks. At that point there is no possible way we could add more write tasks so the fact that it hasn't been ended is probably a bug or an error condition we don't explicitly handle and so we should just abort what we were doing and close the file. |
|
…ire a call to End This call to end was a foot-gun because failing to call End on a scheduler would lead to deadlock. This led to numerous situations where one would accidentally fail to call End because of some kind of error or exceptional behavior. Now the call to end is no longer required. Schedulers are semantically divided (there aren't actually three different classes, just three different supported use cases) into three different types: * Destroy-when-done schedulers destroy themselves whenever they have finished running all of their tasks. These schedulers require an initial task to bootstrap the scheduler and then that task (and it's subtasks) can add more sub-tasks. This is the simplest and most basic kind of scheduler and the top-level scheduler must always be this kind of scheduler. * Peer schedulers will be ended when their parent scheduler runs out of tasks. These are useful for sinks which get tasks on a push-driven basis and don't have all their tasks up front. However, when the plan is done, we know for certain no more tasks will arrive. * Eager peer schedulers are the same as peer schedulers but they might be eagerly ended before their parent scheduler is finished. The file queue in the dataset writer is a good example. It receives tasks in an ad-hoc fashion so it is a peer scheduler. However, once we have written max_rows rows to a file we can end it early (no need to wait for the plan to finish). However, if we forget to end it early (or fail to due so in some error scenario) it will still end when the plan ends.
…tly. It now uses the plan's scheduler as a parent.
…pts at detecting deadlock but had little luck and have run out of energy to try further.
…heduler into three classes (scheduler, throttle, task group). Moved backpressure management into the dataset writer itself.
westonpace
force-pushed
the
feature/ARROW-17509--async-task-scheduler-without-end
branch
from
ca179e2
to
884d9e0
Compare
…ks. Created the MakeThrottledTaskGroup utility function which properly creates a task group around a throttle around a scheduler.
|
I reworked everything again so the above comments aren't really valid. I've reworked the PR description to reflect the new behavior. |
Co-authored-by: David Li <li.davidm96@gmail.com>
|
Benchmark runs are scheduled for baseline = a590b00 and contender = 3da803d. 3da803d is a master commit associated with this PR. Results will be available as each benchmark for each run completes. |
This call to end was a foot-gun because failing to call End on a scheduler would lead to deadlock. This led to numerous situations where one would accidentally fail to call End because of some kind of error or exceptional behavior.
Now the call to end is no longer required. The scheduler has also been simplified by breaking into three different concepts. The root scheduler runs a bunch of tasks until there are no more tasks and then automatically ends.
A throttle can wrap a scheduler and throttle task execution, deferring tasks to a queue if there is no space for new tasks.
A task group tracks a group of tasks and runs a finish callback when all the tasks in that group have finished.