You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/10/26 23:03:05 UTC
[GitHub] [arrow] westonpace opened a new pull request, #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
westonpace opened a new pull request, #14524:
URL: https://github.com/apache/arrow/pull/14524
…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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] ursabot commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1310275960
Benchmark runs are scheduled for baseline = a590b00b7af552ec20a13c46c0d578b211bac0a0 and contender = 3da803db456536782e6ad1cd6cb4f5d08d6a5d6a. 3da803db456536782e6ad1cd6cb4f5d08d6a5d6a is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished :arrow_down:0.0% :arrow_up:0.0%] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/5c9a7ca895b54babb73788a5974501a4...3db54def2f7940fbbcb59376fc0ef1b9/)
[Failed :arrow_down:0.58% :arrow_up:0.17%] [test-mac-arm](https://conbench.ursa.dev/compare/runs/8da766f13f4b47f989bc8cd65a425085...5225c941108040ba924c7a6ce8a3b711/)
[Finished :arrow_down:0.27% :arrow_up:0.0%] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/f4724f99d73944548c7f7e5ef6c77f40...5dfb0a3e09be4d9e8e2a62e74956fda4/)
[Finished :arrow_down:1.06% :arrow_up:0.28%] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/09c48a269abd48d7b833dda558db1da7...d22ccad69a474b5e82c840151567a2bf/)
Buildkite builds:
[Finished] [`3da803db` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1844)
[Finished] [`3da803db` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1866)
[Finished] [`3da803db` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1832)
[Finished] [`3da803db` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1858)
[Finished] [`a590b00b` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1843)
[Failed] [`a590b00b` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1865)
[Finished] [`a590b00b` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1831)
[Finished] [`a590b00b` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1857)
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1018414412
##########
cpp/src/arrow/util/async_util.cc:
##########
@@ -100,345 +119,323 @@ class FifoQueue : public AsyncTaskScheduler::Queue {
class AlreadyFailedScheduler : public AsyncTaskScheduler {
Review Comment:
Nope, good catch. Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1018414137
##########
cpp/src/arrow/util/async_util.h:
##########
@@ -145,206 +188,216 @@ class ARROW_EXPORT AsyncTaskScheduler {
/// acquired and the caller can proceed. If a future is returned then the caller
/// should wait for the future to complete first. When the returned future completes
/// the permits have NOT been acquired and the caller must call Acquire again
+ ///
+ /// \param amt the number of permits to acquire
virtual std::optional<Future<>> TryAcquire(int amt) = 0;
/// Release amt permits
///
/// This will possibly complete waiting futures and should probably not be
/// called while holding locks.
+ ///
+ /// \param amt the number of permits to release
virtual void Release(int amt) = 0;
/// The size of the largest task that can run
///
/// Incoming tasks will have their cost latched to this value to ensure
- /// they can still run (although they will generally be the only thing allowed to
+ /// they can still run (although they will be the only thing allowed to
/// run at that time).
virtual int Capacity() = 0;
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
+ ///
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
};
- /// Create a throttle
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
///
- /// This throttle is used to limit how many tasks can run at once. The
- /// user should keep the throttle alive for the lifetime of the scheduler.
- /// The same throttle can be used in multiple schedulers.
- static std::unique_ptr<Throttle> MakeThrottle(int max_concurrent_cost);
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
- /// Add a task to the scheduler
+ /// Create a throttled view of a scheduler
///
- /// If the scheduler is in an aborted state this call will return false and the task
- /// will never be run. This is harmless and does not need to be guarded against.
+ /// Tasks added via this view will be subjected to the throttle and, if the tasks cannot
+ /// run immediately, will be placed into a queue.
///
- /// If the scheduler is in an ended state then this call will cause an program abort.
- /// This represents a logic error in the program and should be avoidable.
+ /// Using a throttled view after the underlying scheduler has finished is invalid.
Review Comment:
I removed this sentence. This is not a change in the API contracts. For all schedulers it is safe to access the scheduler from within a task submitted to that scheduler (e.g. follow-up tasks).
However, it is not safe to access a scheduler from a peer scheduler, though that concept doesn't have much meaning and adds confusion so probably doesn't need explanation. For example, it would not be safe to add a task to the task group for file A from a task in the task group for file B, because file A might have finished before file B.
The various returned pointers should make this fairly obvious (e.g. to do the above you would have to call `.get` on the `unique_ptr` returned when creating the task group for file A).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1292759746
CC @save-buffer would appreciate a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1292757084
:warning: Ticket **has not been started in JIRA**, please click 'Start Progress'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1292757069
https://issues.apache.org/jira/browse/ARROW-17509
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1295423847
> A different interface that could be simpler is to simply change the meaning of End to mean "End when empty but allow adding more tasks" instead of "End when empty but disallow adding more tasks".
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1298236528
- [ ] Reminder to myself to update the docs. They still mention that you don't need to dispose the holder but you do need to do so.
- [ ] Create a follow-up PR to add deadlock detection
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace merged pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace merged PR #14524:
URL: https://github.com/apache/arrow/pull/14524
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1018417571
##########
cpp/src/arrow/util/async_util.h:
##########
@@ -145,206 +188,216 @@ class ARROW_EXPORT AsyncTaskScheduler {
/// acquired and the caller can proceed. If a future is returned then the caller
/// should wait for the future to complete first. When the returned future completes
/// the permits have NOT been acquired and the caller must call Acquire again
+ ///
+ /// \param amt the number of permits to acquire
virtual std::optional<Future<>> TryAcquire(int amt) = 0;
/// Release amt permits
///
/// This will possibly complete waiting futures and should probably not be
/// called while holding locks.
+ ///
+ /// \param amt the number of permits to release
virtual void Release(int amt) = 0;
/// The size of the largest task that can run
///
/// Incoming tasks will have their cost latched to this value to ensure
- /// they can still run (although they will generally be the only thing allowed to
+ /// they can still run (although they will be the only thing allowed to
/// run at that time).
virtual int Capacity() = 0;
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
+ ///
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
};
- /// Create a throttle
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
///
- /// This throttle is used to limit how many tasks can run at once. The
- /// user should keep the throttle alive for the lifetime of the scheduler.
- /// The same throttle can be used in multiple schedulers.
- static std::unique_ptr<Throttle> MakeThrottle(int max_concurrent_cost);
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
- /// Add a task to the scheduler
+ /// Create a throttled view of a scheduler
///
- /// If the scheduler is in an aborted state this call will return false and the task
- /// will never be run. This is harmless and does not need to be guarded against.
+ /// Tasks added via this view will be subjected to the throttle and, if the tasks cannot
+ /// run immediately, will be placed into a queue.
///
- /// If the scheduler is in an ended state then this call will cause an program abort.
- /// This represents a logic error in the program and should be avoidable.
+ /// Using a throttled view after the underlying scheduler has finished is invalid.
Review Comment:
Sounds good to me, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #14524:
URL: https://github.com/apache/arrow/pull/14524#issuecomment-1308359622
I reworked everything again so the above comments aren't really valid. I've reworked the PR description to reflect the new behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1018083155
##########
cpp/src/arrow/util/async_util.h:
##########
@@ -48,50 +50,25 @@ namespace util {
/// By default the scheduler will submit the task (execute the synchronous part) as
/// soon as it is added, assuming the underlying thread pool hasn't terminated or the
/// scheduler hasn't aborted. In this mode, the scheduler is simply acting as
-/// a task group (keeping track of the ongoing work).
+/// a simple task group.
///
-/// This can be used to provide structured concurrency for asynchronous development.
-/// A task group created at a high level can be distributed amongst low level components
-/// which register work to be completed. The high level job can then wait for all work
-/// to be completed before cleaning up.
-///
-/// A task scheduler must eventually be ended when all tasks have been added. Once the
-/// scheduler has been ended it is an error to add further tasks. Note, it is not an
-/// error to add additional tasks after a scheduler has aborted (though these tasks
-/// will be ignored and never submitted). The scheduler has a future which will complete
-/// once the scheduler has been ended AND all remaining tasks have finished executing.
-/// Ending a scheduler will NOT cause the scheduler to flush existing tasks.
+/// A task scheduler starts with an initial task. That task, and all subsequent tasks
+/// are free to add subtasks. Once all submitted tasks finsih the scheduler will
Review Comment:
```suggestion
/// are free to add subtasks. Once all submitted tasks finish the scheduler will
```
##########
cpp/src/arrow/util/async_util.cc:
##########
@@ -100,345 +119,323 @@ class FifoQueue : public AsyncTaskScheduler::Queue {
class AlreadyFailedScheduler : public AsyncTaskScheduler {
Review Comment:
Is this actually used anymore?
##########
cpp/src/arrow/util/async_util.h:
##########
@@ -145,206 +188,216 @@ class ARROW_EXPORT AsyncTaskScheduler {
/// acquired and the caller can proceed. If a future is returned then the caller
/// should wait for the future to complete first. When the returned future completes
/// the permits have NOT been acquired and the caller must call Acquire again
+ ///
+ /// \param amt the number of permits to acquire
virtual std::optional<Future<>> TryAcquire(int amt) = 0;
/// Release amt permits
///
/// This will possibly complete waiting futures and should probably not be
/// called while holding locks.
+ ///
+ /// \param amt the number of permits to release
virtual void Release(int amt) = 0;
/// The size of the largest task that can run
///
/// Incoming tasks will have their cost latched to this value to ensure
- /// they can still run (although they will generally be the only thing allowed to
+ /// they can still run (although they will be the only thing allowed to
/// run at that time).
virtual int Capacity() = 0;
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
+ ///
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
};
- /// Create a throttle
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
///
- /// This throttle is used to limit how many tasks can run at once. The
- /// user should keep the throttle alive for the lifetime of the scheduler.
- /// The same throttle can be used in multiple schedulers.
- static std::unique_ptr<Throttle> MakeThrottle(int max_concurrent_cost);
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
- /// Add a task to the scheduler
+ /// Create a throttled view of a scheduler
///
- /// If the scheduler is in an aborted state this call will return false and the task
- /// will never be run. This is harmless and does not need to be guarded against.
+ /// Tasks added via this view will be subjected to the throttle and, if the tasks cannot
+ /// run immediately, will be placed into a queue.
///
- /// If the scheduler is in an ended state then this call will cause an program abort.
- /// This represents a logic error in the program and should be avoidable.
+ /// Using a throttled view after the underlying scheduler has finished is invalid.
///
- /// If there are no limits on the number of concurrent tasks then the submit function
- /// will be run immediately.
+ /// Although a shared_ptr is returned it should generally be assumed that the caller
+ /// is being given exclusive ownership. The shared_ptr is used to share the view with
+ /// queued and submitted tasks and the lifetime of those is unpredictable. It is
+ /// important the caller keep the returned pointer alive for as long as they plan to add
+ /// tasks to the view.
///
- /// Otherwise, if there is a throttle, and it is full, then this task will be inserted
- /// into the scheduler's queue and submitted when there is space.
+ /// \param scheduler a scheduler to submit tasks to after throttling
///
- /// The return value for this call can usually be ignored. There is little harm in
- /// attempting to add tasks to an aborted scheduler. It is only included for callers
- /// that want to avoid future task generation.
+ /// This can be the root scheduler, another throttled scheduler, or a task group. These
+ /// are all composable.
///
- /// \return true if the task was submitted or queued, false if the task was ignored
- virtual bool AddTask(std::unique_ptr<Task> task) = 0;
+ /// \param max_concurrent_cost the maximum amount of cost allowed to run at any one time
+ ///
+ /// If a task is added that has a cost greater than max_concurrent_cost then its cost
+ /// will be reduced to max_concurrent_cost so that it is still possible for the task to
+ /// run.
+ ///
+ /// \param queue the queue to use when tasks cannot be submitted
+ ///
+ /// By default a FIFO queue will be used. However, a custom queue can be provided if
+ /// some tasks have higher priority than other tasks.
+ static std::shared_ptr<ThrottledAsyncTaskScheduler> Make(
+ AsyncTaskScheduler* scheduler, int max_concurrent_cost,
+ std::unique_ptr<Queue> queue = NULLPTR);
- /// Adds an async generator to the scheduler
+ /// @brief Create a ThrottledAsyncTaskScheduler using a custom throttle
///
- /// The async generator will be visited, one item at a time. Submitting a task
- /// will consist of polling the generator for the next future. The generator's future
- /// will then represent the task itself.
+ /// \see Make
+ static std::shared_ptr<ThrottledAsyncTaskScheduler> MakeWithCustomThrottle(
+ AsyncTaskScheduler* scheduler, std::unique_ptr<Throttle> throttle,
+ std::unique_ptr<Queue> queue = NULLPTR);
+};
+
+/// A utility to keep track of a collection of tasks
+///
+/// Often it is useful to keep track of some state that only needs to stay alive
+/// for some small collection of tasks, or to perform some kind of final cleanup
+/// when a collection of tasks is finished.
+///
+/// For example, when scanning, we need to keep the file reader alive while all scan
+/// tasks run for a given file, and then we can gracefully close it when we finish the
+/// file.
+class ARROW_EXPORT AsyncTaskGroup : public AsyncTaskScheduler {
+ public:
+ /// Destructor for the task group
///
- /// This visits the task serially without readahead. If readahead or parallelism
- /// is desired then it should be added in the generator itself.
+ /// The finish callback will not run until the task group is destroyed and all
+ /// tasks are finished so you will generally want to eagerly call this at some point
Review Comment:
"eagerly call this" meaning "explicitly destroy the task group"?
##########
cpp/src/arrow/util/async_util.h:
##########
@@ -145,206 +188,216 @@ class ARROW_EXPORT AsyncTaskScheduler {
/// acquired and the caller can proceed. If a future is returned then the caller
/// should wait for the future to complete first. When the returned future completes
/// the permits have NOT been acquired and the caller must call Acquire again
+ ///
+ /// \param amt the number of permits to acquire
virtual std::optional<Future<>> TryAcquire(int amt) = 0;
/// Release amt permits
///
/// This will possibly complete waiting futures and should probably not be
/// called while holding locks.
+ ///
+ /// \param amt the number of permits to release
virtual void Release(int amt) = 0;
/// The size of the largest task that can run
///
/// Incoming tasks will have their cost latched to this value to ensure
- /// they can still run (although they will generally be the only thing allowed to
+ /// they can still run (although they will be the only thing allowed to
/// run at that time).
virtual int Capacity() = 0;
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
+ ///
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
};
- /// Create a throttle
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
///
- /// This throttle is used to limit how many tasks can run at once. The
- /// user should keep the throttle alive for the lifetime of the scheduler.
- /// The same throttle can be used in multiple schedulers.
- static std::unique_ptr<Throttle> MakeThrottle(int max_concurrent_cost);
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
- /// Add a task to the scheduler
+ /// Create a throttled view of a scheduler
///
- /// If the scheduler is in an aborted state this call will return false and the task
- /// will never be run. This is harmless and does not need to be guarded against.
+ /// Tasks added via this view will be subjected to the throttle and, if the tasks cannot
+ /// run immediately, will be placed into a queue.
///
- /// If the scheduler is in an ended state then this call will cause an program abort.
- /// This represents a logic error in the program and should be avoidable.
+ /// Using a throttled view after the underlying scheduler has finished is invalid.
Review Comment:
nit, but it may be a little confusing/error-prone to have different API contracts for a subclass of the main API
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] save-buffer commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
save-buffer commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1007212828
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -127,85 +127,89 @@ struct ExecPlanImpl : public ExecPlan {
}
Status StartProducing() {
- START_COMPUTE_SPAN(span_, "ExecPlan", {{"plan", ToString()}});
-#ifdef ARROW_WITH_OPENTELEMETRY
- if (HasMetadata()) {
- auto pairs = metadata().get()->sorted_pairs();
- opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span =
- ::arrow::internal::tracing::UnwrapSpan(span_.details.get());
- std::for_each(std::begin(pairs), std::end(pairs),
- [span](std::pair<std::string, std::string> const& pair) {
- span->SetAttribute(pair.first, pair.second);
- });
- }
-#endif
if (started_) {
return Status::Invalid("restarted ExecPlan");
}
-
- std::vector<Future<>> futures;
- for (auto& n : nodes_) {
- RETURN_NOT_OK(n->Init());
- futures.push_back(n->finished());
- }
-
- AllFinished(futures).AddCallback([this](const Status& st) {
- error_st_ = st;
- EndTaskGroup();
- });
-
- task_scheduler_->RegisterEnd();
- int num_threads = 1;
- bool sync_execution = true;
- if (auto executor = exec_context()->executor()) {
- num_threads = executor->GetCapacity();
- sync_execution = false;
- }
- RETURN_NOT_OK(task_scheduler_->StartScheduling(
- 0 /* thread_index */,
- [this](std::function<Status(size_t)> fn) -> Status {
- return this->ScheduleTask(std::move(fn));
- },
- /*concurrent_tasks=*/2 * num_threads, sync_execution));
-
started_ = true;
- // producers precede consumers
- sorted_nodes_ = TopoSort();
-
- Status st = Status::OK();
-
- using rev_it = std::reverse_iterator<NodeVector::iterator>;
- for (rev_it it(sorted_nodes_.end()), end(sorted_nodes_.begin()); it != end; ++it) {
- auto node = *it;
- EVENT(span_, "StartProducing:" + node->label(),
- {{"node.label", node->label()}, {"node.kind_name", node->kind_name()}});
- st = node->StartProducing();
- EVENT(span_, "StartProducing:" + node->label(), {{"status", st.ToString()}});
- if (!st.ok()) {
- // Stop nodes that successfully started, in reverse order
- stopped_ = true;
- StopProducingImpl(it.base(), sorted_nodes_.end());
- for (NodeVector::iterator fw_it = sorted_nodes_.begin(); fw_it != it.base();
- ++fw_it) {
- Future<> fut = (*fw_it)->finished();
- if (!fut.is_finished()) fut.MarkFinished();
- }
- return st;
- }
- }
- return st;
- }
-
- void EndTaskGroup() {
- bool expected = false;
- if (group_ended_.compare_exchange_strong(expected, true)) {
- async_scheduler_->End();
- async_scheduler_->OnFinished().AddCallback([this](const Status& st) {
- MARK_SPAN(span_, error_st_ & st);
- END_SPAN(span_);
- finished_.MarkFinished(error_st_ & st);
- });
+ // We call StartProducing on each of the nodes. The source nodes should generally
+ // start scheduling some tasks during this call.
+ //
+ // If no source node schedules any tasks (e.g. they do all their word synchronously as
+ // 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) {
+ this->async_scheduler_ = async_scheduler;
+ START_COMPUTE_SPAN(span_, "ExecPlan", {{"plan", ToString()}});
+#ifdef ARROW_WITH_OPENTELEMETRY
+ if (HasMetadata()) {
+ auto pairs = metadata().get()->sorted_pairs();
+ opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span =
+ ::arrow::internal::tracing::UnwrapSpan(span_.details.get());
+ std::for_each(std::begin(pairs), std::end(pairs),
+ [span](std::pair<std::string, std::string> const& pair) {
+ span->SetAttribute(pair.first, pair.second);
+ });
+ }
+#endif
+ // TODO(weston) The entire concept of ExecNode::finished() will hopefully go
Review Comment:
yep my simplification PR gets rid of those finished futures.
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -127,85 +127,89 @@ struct ExecPlanImpl : public ExecPlan {
}
Status StartProducing() {
- START_COMPUTE_SPAN(span_, "ExecPlan", {{"plan", ToString()}});
-#ifdef ARROW_WITH_OPENTELEMETRY
- if (HasMetadata()) {
- auto pairs = metadata().get()->sorted_pairs();
- opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span =
- ::arrow::internal::tracing::UnwrapSpan(span_.details.get());
- std::for_each(std::begin(pairs), std::end(pairs),
- [span](std::pair<std::string, std::string> const& pair) {
- span->SetAttribute(pair.first, pair.second);
- });
- }
-#endif
if (started_) {
return Status::Invalid("restarted ExecPlan");
}
-
- std::vector<Future<>> futures;
- for (auto& n : nodes_) {
- RETURN_NOT_OK(n->Init());
- futures.push_back(n->finished());
- }
-
- AllFinished(futures).AddCallback([this](const Status& st) {
- error_st_ = st;
- EndTaskGroup();
- });
-
- task_scheduler_->RegisterEnd();
- int num_threads = 1;
- bool sync_execution = true;
- if (auto executor = exec_context()->executor()) {
- num_threads = executor->GetCapacity();
- sync_execution = false;
- }
- RETURN_NOT_OK(task_scheduler_->StartScheduling(
- 0 /* thread_index */,
- [this](std::function<Status(size_t)> fn) -> Status {
- return this->ScheduleTask(std::move(fn));
- },
- /*concurrent_tasks=*/2 * num_threads, sync_execution));
-
started_ = true;
- // producers precede consumers
- sorted_nodes_ = TopoSort();
-
- Status st = Status::OK();
-
- using rev_it = std::reverse_iterator<NodeVector::iterator>;
- for (rev_it it(sorted_nodes_.end()), end(sorted_nodes_.begin()); it != end; ++it) {
- auto node = *it;
- EVENT(span_, "StartProducing:" + node->label(),
- {{"node.label", node->label()}, {"node.kind_name", node->kind_name()}});
- st = node->StartProducing();
- EVENT(span_, "StartProducing:" + node->label(), {{"status", st.ToString()}});
- if (!st.ok()) {
- // Stop nodes that successfully started, in reverse order
- stopped_ = true;
- StopProducingImpl(it.base(), sorted_nodes_.end());
- for (NodeVector::iterator fw_it = sorted_nodes_.begin(); fw_it != it.base();
- ++fw_it) {
- Future<> fut = (*fw_it)->finished();
- if (!fut.is_finished()) fut.MarkFinished();
- }
- return st;
- }
- }
- return st;
- }
-
- void EndTaskGroup() {
- bool expected = false;
- if (group_ended_.compare_exchange_strong(expected, true)) {
- async_scheduler_->End();
- async_scheduler_->OnFinished().AddCallback([this](const Status& st) {
- MARK_SPAN(span_, error_st_ & st);
- END_SPAN(span_);
- finished_.MarkFinished(error_st_ & st);
- });
+ // We call StartProducing on each of the nodes. The source nodes should generally
+ // start scheduling some tasks during this call.
+ //
+ // If no source node schedules any tasks (e.g. they do all their word synchronously as
+ // 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) {
Review 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:
##########
@@ -108,30 +108,37 @@ class AlreadyFailedScheduler : public AsyncTaskScheduler {
std::ignore = std::move(finish_callback_)(failure_reason_);
}
bool AddTask(std::unique_ptr<Task> task) override { return false; }
- void End() override {
- Status::UnknownError("Do not call End on a sub-scheduler.").Abort();
- }
- Future<> OnFinished() const override {
- Status::UnknownError(
- "You should not rely on sub-scheduler's OnFinished. Use a "
- "finished callback when creating the sub-scheduler instead")
- .Abort();
- }
- std::shared_ptr<AsyncTaskScheduler> MakeSubScheduler(
+ std::unique_ptr<AsyncTaskScheduler::Holder> MakeSubScheduler(
+ FnOnce<Status(AsyncTaskScheduler*)> initial_task,
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1008434063
##########
cpp/src/arrow/util/async_util.cc:
##########
@@ -108,30 +108,37 @@ class AlreadyFailedScheduler : public AsyncTaskScheduler {
std::ignore = std::move(finish_callback_)(failure_reason_);
}
bool AddTask(std::unique_ptr<Task> task) override { return false; }
- void End() override {
- Status::UnknownError("Do not call End on a sub-scheduler.").Abort();
- }
- Future<> OnFinished() const override {
- Status::UnknownError(
- "You should not rely on sub-scheduler's OnFinished. Use a "
- "finished callback when creating the sub-scheduler instead")
- .Abort();
- }
- std::shared_ptr<AsyncTaskScheduler> MakeSubScheduler(
+ std::unique_ptr<AsyncTaskScheduler::Holder> MakeSubScheduler(
+ FnOnce<Status(AsyncTaskScheduler*)> initial_task,
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14524: ARROW-17509: [C++] Simplify async scheduler by removing the need to call End
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14524:
URL: https://github.com/apache/arrow/pull/14524#discussion_r1018410231
##########
cpp/src/arrow/util/async_util.h:
##########
@@ -145,206 +188,216 @@ class ARROW_EXPORT AsyncTaskScheduler {
/// acquired and the caller can proceed. If a future is returned then the caller
/// should wait for the future to complete first. When the returned future completes
/// the permits have NOT been acquired and the caller must call Acquire again
+ ///
+ /// \param amt the number of permits to acquire
virtual std::optional<Future<>> TryAcquire(int amt) = 0;
/// Release amt permits
///
/// This will possibly complete waiting futures and should probably not be
/// called while holding locks.
+ ///
+ /// \param amt the number of permits to release
virtual void Release(int amt) = 0;
/// The size of the largest task that can run
///
/// Incoming tasks will have their cost latched to this value to ensure
- /// they can still run (although they will generally be the only thing allowed to
+ /// they can still run (although they will be the only thing allowed to
/// run at that time).
virtual int Capacity() = 0;
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
+ ///
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
};
- /// Create a throttle
+
+ /// Pause the throttle
+ ///
+ /// Any tasks that have been submitted already will continue. However, no new tasks
+ /// will be run until the throttle is resumed.
+ virtual void Pause() = 0;
+ /// Resume the throttle
///
- /// This throttle is used to limit how many tasks can run at once. The
- /// user should keep the throttle alive for the lifetime of the scheduler.
- /// The same throttle can be used in multiple schedulers.
- static std::unique_ptr<Throttle> MakeThrottle(int max_concurrent_cost);
+ /// Allows taks to be submitted again. If there is a max_concurrent_cost limit then
+ /// it will still apply.
+ virtual void Resume() = 0;
- /// Add a task to the scheduler
+ /// Create a throttled view of a scheduler
///
- /// If the scheduler is in an aborted state this call will return false and the task
- /// will never be run. This is harmless and does not need to be guarded against.
+ /// Tasks added via this view will be subjected to the throttle and, if the tasks cannot
+ /// run immediately, will be placed into a queue.
///
- /// If the scheduler is in an ended state then this call will cause an program abort.
- /// This represents a logic error in the program and should be avoidable.
+ /// Using a throttled view after the underlying scheduler has finished is invalid.
///
- /// If there are no limits on the number of concurrent tasks then the submit function
- /// will be run immediately.
+ /// Although a shared_ptr is returned it should generally be assumed that the caller
+ /// is being given exclusive ownership. The shared_ptr is used to share the view with
+ /// queued and submitted tasks and the lifetime of those is unpredictable. It is
+ /// important the caller keep the returned pointer alive for as long as they plan to add
+ /// tasks to the view.
///
- /// Otherwise, if there is a throttle, and it is full, then this task will be inserted
- /// into the scheduler's queue and submitted when there is space.
+ /// \param scheduler a scheduler to submit tasks to after throttling
///
- /// The return value for this call can usually be ignored. There is little harm in
- /// attempting to add tasks to an aborted scheduler. It is only included for callers
- /// that want to avoid future task generation.
+ /// This can be the root scheduler, another throttled scheduler, or a task group. These
+ /// are all composable.
///
- /// \return true if the task was submitted or queued, false if the task was ignored
- virtual bool AddTask(std::unique_ptr<Task> task) = 0;
+ /// \param max_concurrent_cost the maximum amount of cost allowed to run at any one time
+ ///
+ /// If a task is added that has a cost greater than max_concurrent_cost then its cost
+ /// will be reduced to max_concurrent_cost so that it is still possible for the task to
+ /// run.
+ ///
+ /// \param queue the queue to use when tasks cannot be submitted
+ ///
+ /// By default a FIFO queue will be used. However, a custom queue can be provided if
+ /// some tasks have higher priority than other tasks.
+ static std::shared_ptr<ThrottledAsyncTaskScheduler> Make(
+ AsyncTaskScheduler* scheduler, int max_concurrent_cost,
+ std::unique_ptr<Queue> queue = NULLPTR);
- /// Adds an async generator to the scheduler
+ /// @brief Create a ThrottledAsyncTaskScheduler using a custom throttle
///
- /// The async generator will be visited, one item at a time. Submitting a task
- /// will consist of polling the generator for the next future. The generator's future
- /// will then represent the task itself.
+ /// \see Make
+ static std::shared_ptr<ThrottledAsyncTaskScheduler> MakeWithCustomThrottle(
+ AsyncTaskScheduler* scheduler, std::unique_ptr<Throttle> throttle,
+ std::unique_ptr<Queue> queue = NULLPTR);
+};
+
+/// A utility to keep track of a collection of tasks
+///
+/// Often it is useful to keep track of some state that only needs to stay alive
+/// for some small collection of tasks, or to perform some kind of final cleanup
+/// when a collection of tasks is finished.
+///
+/// For example, when scanning, we need to keep the file reader alive while all scan
+/// tasks run for a given file, and then we can gracefully close it when we finish the
+/// file.
+class ARROW_EXPORT AsyncTaskGroup : public AsyncTaskScheduler {
+ public:
+ /// Destructor for the task group
///
- /// This visits the task serially without readahead. If readahead or parallelism
- /// is desired then it should be added in the generator itself.
+ /// The finish callback will not run until the task group is destroyed and all
+ /// tasks are finished so you will generally want to eagerly call this at some point
Review Comment:
Yes, good point. This comment made more sense on an older iteration of the code. I moved this paragraph to the `Make` call as that is the point where the unique_ptr is returned and made it clearer that the user should eagerly reset/destroy the unique_ptr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org