westonpace commented on a change in pull request #10258:
URL: https://github.com/apache/arrow/pull/10258#discussion_r645780796
##########
File path: cpp/src/arrow/util/future.cc
##########
@@ -231,26 +232,68 @@ class ConcreteFutureImpl : public FutureImpl {
void DoMarkFailed() { DoMarkFinishedOrFailed(FutureState::FAILURE); }
- void AddCallback(Callback callback) {
+ void CheckOptions(const CallbackOptions& opts) {
+ if (opts.should_schedule != ShouldSchedule::NEVER) {
+ DCHECK_NE(opts.executor, NULL)
+ << "An executor must be specified when adding a callback that might
schedule";
+ }
+ }
+
+ void AddCallback(Callback callback, CallbackOptions opts) {
+ CheckOptions(opts);
std::unique_lock<std::mutex> lock(mutex_);
+ CallbackRecord callback_record{std::move(callback), opts};
if (IsFutureFinished(state_)) {
lock.unlock();
- std::move(callback)();
+ RunOrScheduleCallback(callback_record, /*from_unfinished=*/false);
} else {
- callbacks_.push_back(std::move(callback));
+ callbacks_.push_back(std::move(callback_record));
}
}
- bool TryAddCallback(const std::function<Callback()>& callback_factory) {
+ bool TryAddCallback(const std::function<Callback()>& callback_factory,
+ CallbackOptions opts) {
+ CheckOptions(opts);
std::unique_lock<std::mutex> lock(mutex_);
if (IsFutureFinished(state_)) {
return false;
} else {
- callbacks_.push_back(callback_factory());
+ callbacks_.push_back({callback_factory(), opts});
return true;
}
}
+ bool ShouldSchedule(const CallbackRecord& callback_record, bool
from_unfinished) {
+ switch (callback_record.options.should_schedule) {
+ case ShouldSchedule::NEVER:
+ return false;
+ case ShouldSchedule::ALWAYS:
+ return true;
+ case ShouldSchedule::IF_UNFINISHED:
+ return from_unfinished;
+ default:
+ DCHECK(false) << "Unrecognized ShouldSchedule option";
+ return false;
+ }
+ }
+
+ void RunOrScheduleCallback(CallbackRecord& callback_record, bool
from_unfinished) {
+ if (ShouldSchedule(callback_record, from_unfinished)) {
+ // Need to make a copy of this to keep it alive until the callback has a
chance
Review comment:
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.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
