7
\$\begingroup\$

I'm reinventing std::future from scratch (for an upcoming talk). I'd like to incorporate as many of the latest and most likely-to-succeed proposals as possible, and then also at least one additional feature that I haven't seen in any proposal yet: cancellation of tasks.

The idea is that if I spawn a chain of tasks via async(x).then(y).then(z), and then (maybe after a while) drop the resulting future on the floor, I am indicating that I no longer care about the result of the chain of computations, and therefore the executor should not execute any of them that aren't already in progress. (For example, if it's still in the middle of executing x, it should finish x but not start y or z).

I'd like feedback on the particular way I've chosen to implement this feature. Namely, I've added a new member to Future<T>: a shared_ptr named cancellable_task_state_. The caller can set this pointer via Future<T>::attach_cancellable_task_state(). When you create a PackagedTask from a function object, we make_shared a copy of your function object so that its lifetime is now controlled by the PackagedTask's stored Future; and then the thing that we enqueue with the scheduler is simply a wrapper around a weak_ptr.

This way, the scheduler itself doesn't have to worry about descheduling tasks from the middle of the queue; but we still get deterministic destruction of the function objects controlled by a future.

One downside (in my current way of thinking) is that destroying a Future can now cause a lot of code execution, because it might destroy a whole chain of user-defined function objects. These objects' destructors might even throw exceptions! This feels too much like spooky action at a distance for my taste. How does std::async currently deal with the fact that the user might supply a function object with a destructor that throws? Where would that exception show up?

(FYI, all this code is also available on GitHub, although it may eventually bit-rot since I rewrite my git history frequently.)

The tricky bits of cancellation are all in PackagedTask:

template<class F>
struct PackagedTask;

template<class R, class... A>
struct PackagedTask<R(A...)> {

    UniqueFunction<void(A...)> task_;
    Future<R> future_;
    bool promise_already_satisfied_ = false;

    PackagedTask() = default;

    template<class F>
    PackagedTask(F&& f) {
        Promise<R> p;
        future_ = p.get_future();

        auto f_holder = [f = std::forward<F>(f)]() mutable { return std::move(f); };

        auto sptr = std::make_shared<decltype(f_holder)>(std::move(f_holder));
        std::weak_ptr<decltype(f_holder)> wptr = sptr;

        future_.attach_cancellable_task_state(sptr);

        task_ = [p = std::move(p), wptr = std::move(wptr)](A... args) mutable {
            if (auto sptr = wptr.lock()) {
                auto f = (*sptr)();
                try {
                    p.set_value(f(std::forward<A>(args)...));
                } catch (...) {
                    p.set_exception(std::current_exception());
                }
            }
        };
    }

    bool valid() const { return task_; }

    Future<R> get_future() {
        if (!task_) throw "no_state";
        if (!future_.valid()) throw "future_already_retrieved";
        return std::move(future_);
    }

    void operator()(A... args) {
        if (!task_) throw "no_state";
        if (promise_already_satisfied_) throw "promise_already_satisfied";
        promise_already_satisfied_ = true;
        task_(std::forward<A>(args)...);
    }
};

Future<T> is derived from SharedFuture<T>, which looks like this. Note that SharedState<T> is defined below; and that both the Promise and the Future (or the Promise and the several SharedFutures) hold shared_ptrs to the SharedState. This is for simplicity; I'm not interested in micro-optimizing the non-shared variant of Future unless it also reduces the absolute amount of source code.

template<class R>
struct SharedFuture {

    std::shared_ptr<SharedState<R>> state_;
    std::shared_ptr<void> cancellable_task_state_;

    SharedFuture() {}
    SharedFuture(std::shared_ptr<SharedState<R>> s) : state_(s) {}

    R& get() const { ... }

    bool valid() const { return (state_ != nullptr); }

    bool ready() const { ... }

    void wait() const { ... }

    void attach_cancellable_task_state(std::shared_ptr<void> sptr) {
        cancellable_task_state_ = std::move(sptr);
    }

    template<class F>
    auto then(F func)
    {
        if (this->state_ == nullptr) throw "no_state";
        auto sp = this->state_;
        using R2 = decltype(func(*this));
        PackagedTask<R2()> task([func = std::move(func), fut = *this]() mutable {
            return func(std::move(fut));
        });
        Future<R2> result = task.get_future();
        std::lock_guard<std::mutex> lock(sp->mtx_);
        if (sp->ready_) {
            SystemScheduler().schedule(std::move(task));
        } else {
            sp->continuations_.emplace_back(std::move(task));
        }
        return result;
    }
};

template<class R>
struct Future : private SharedFuture<R> {
    // differs only in minor details, such as the signature of get()
};

And here's Promise<T>:

template<class R>
struct SharedState {
    R value_;
    std::exception_ptr exception_;
    bool ready_ = false;
    std::mutex mtx_;
    std::condition_variable cv_;
    std::list<UniqueFunction<void()>> continuations_;
};

template<class R>
struct Promise {

    std::shared_ptr<SharedState<R>> state_;
    bool future_already_retrieved_ = false;

    Promise() : state_(new SharedState<R>) {}

    Promise(const Promise&) = delete;
    Promise& operator=(const Promise&) = delete;
    Promise(Promise&&) = default;

    Promise& operator=(Promise&& rhs) {
        if (this != &rhs) abandon_state();
        state_ = std::move(rhs.state_);
        return *this;
    }

    ~Promise() { abandon_state(); }

    Future<R> get_future() {
        if (state_ == nullptr) throw "no_state";
        if (future_already_retrieved_) throw "future_already_retrieved";
        future_already_retrieved_ = true;
        return Future<R>(state_);
    }

    void set_value(R r) {
        if (state_ == nullptr) throw "no_state";
        if (state_->ready_) throw "promise_already_satisfied";
        state_->value_ = std::move(r);
        set_ready();
    }

    void set_exception(std::exception_ptr p) {
        if (state_ == nullptr) throw "no_state";
        if (state_->ready_) throw "promise_already_satisfied";
        state_->exception_ = std::move(p);
        set_ready();
    }

    bool has_extant_future() const {
        if (state_ == nullptr) return false;
        return future_already_retrieved_ && !state_.unique();
    }

  private:
    void set_ready() {
        std::lock_guard<std::mutex> lock(state_->mtx_);
        state_->ready_ = true;
        for (auto& task : state_->continuations_) {
            SystemScheduler().schedule(std::move(task));
        }
        state_->continuations_.clear();
        state_->cv_.notify_all();
    }

    void abandon_state() {
        if (state_ != nullptr && !state_->ready_) {
            set_exception(std::make_exception_ptr("broken_promise"));
        }
    }
};
\$\endgroup\$
  • \$\begingroup\$ Interestingly, I have thought about the same thing recently. But my approach would have been to put the logic into the scheduler. If it is about to execute a task, it would first check whether the result is still wanted and if not, immediately “break” the promise without executing anything. This would have had different destruction semantics than your solution. I wouldn't worry about exceptions thrown from destructors, though. If a destructor throws, immediate program termination will usually be just around the corner. \$\endgroup\$ – 5gon12eder Aug 17 '15 at 23:00
  • \$\begingroup\$ @5gon12eder I actually implemented that version first; see github.com/Quuxplusone/futures-from-scratch/commit/… ! But see the commit message for several strikes against that approach, one of which is that it hangs onto the function object's state for longer than I want it to. Re "I wouldn't worry about [throwing dtors]": In real life I agree with you, dtors should never throw; but the whole point of this question is to learn what I should do when I am worried about throwing dtors! \$\endgroup\$ – Quuxplusone Aug 17 '15 at 23:06
  • \$\begingroup\$ There's only one thing to do when coming over throwing dtors: despair! \$\endgroup\$ – Deduplicator Oct 30 '15 at 20:05
1
\$\begingroup\$

If you uncomment the destructor in the following code example, it won't compile with my GCC and libstdc++.

#include <future>

struct MyCallable
{
  void operator()() const {}
  //~MyCallable() noexcept(false) {}
};

int
main()
{
  std::async(MyCallable {}).get();
}

I'm not sure if this is required or even permitted by the standard (After all, the destructor of std::promise is shown without noexcept specification in § 30.6.5.) but at least it shows that a heavily used standard library implementation got away with assuming enforcing that function objects have non-throwing destructors. I'd recommend you do the same for your implementation. Throwing destructors are extremely poor coding style anyway.

\$\endgroup\$
  • 1
    \$\begingroup\$ The whole standard library assumes that destructors do not throw. Without that assumption, only some trivial functions can even provide the basic exception-safety guarantee, and without it where would we be? Up the creek without a paddle... Also, destructors are noexcept if all sub-objects' destructors are noexcept, unless there's an exception-specification. \$\endgroup\$ – Deduplicator Oct 20 '15 at 12:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.