Non-polling implementation of std::when_any()

Question

The following — I claim — is a implementation of when_any from the C++ Concurrency TS, except without all the baggage around when_any_result. (This makes when_any and when_all have exactly the same function signature, which is elegant IMHO.)

My futures are just like std::future, except that they support one additional member function:

    auto then(F f) -> future<decltype(f(std::move(*this)))>;

(My futures also support next(F), recover(F), and fallback_to(V) as specified in N3865 "More improvements to std::future<T>", and also bool ready() const, but those aren't used in this code.)

The Concurrency TS's when_any philosophically-incorrectly returns a ready future when called with zero arguments. My version doesn't treat that case specially, and so the natural behavior falls out: the internal promise is destroyed before any 1 of the 0 provided futures has become ready, and so when_any(/*zero args*/) returns a ready future whose get() will throw broken_promise.

I'm a little concerned by the apparently necessary use of lock-free programming. Am I missing an "easier" (higher-level) way to do it?

#include "scratch/bits/concurrency/future-and-promise.h"
#include "scratch/bits/smart-ptrs/shared-ptr.h"
#include "scratch/bits/tuple/tuple.h"
#include "scratch/bits/type-traits/decay.h"

#include <atomic>
#include <utility>

namespace scratch::detail {

template<class... Futures>
struct when_any_shared_state {
    promise<tuple<Futures...>> m_promise;
    tuple<Futures...> m_tuple;
    std::atomic<bool> m_done;
    std::atomic<bool> m_count_to_two;

    when_any_shared_state(promise<tuple<Futures...>> p) :
        m_promise(std::move(p)), m_done(false), m_count_to_two(false) {}
};

} // namespace scratch::detail

namespace scratch {

template<class... Futures>
auto when_any(Futures... futures) -> future<tuple<Futures...>>
{
    using shared_state = detail::when_any_shared_state<Futures...>;
    using R = tuple<Futures...>;
    promise<R> p;
    future<R> result = p.get_future();

    auto sptr = make_shared<shared_state>(std::move(p));
    auto satisfy_combined_promise =
        [sptr](auto f) {
            if (sptr->m_done.exchange(true) == false) {
                if (sptr->m_count_to_two.exchange(true)) {
                    sptr->m_promise.set_value(std::move(sptr->m_tuple));
                }
            }
            return f.get();
        };
    sptr->m_tuple = tuple<Futures...>(futures.then(satisfy_combined_promise)...);
    if (sptr->m_count_to_two.exchange(true)) {
        sptr->m_promise.set_value(std::move(sptr->m_tuple));
    }
    return result;
}

} // namespace scratch

(GitHub link)

---

I've linked this post from over here. I wrote some external documentation there as well: :)

You attach a continuation to each incoming future (using then). This continuation holds a shared_ptr to a shared state. The shared state holds a count-to-one (m_done) and a count-to-two (m_count_to_two). Each continuation that executes will increment the count-to-one; if it's the winner, it will also increment the count-to-two. The main thread will also increment the count-to-two after it finishes setting up all this stuff. As soon as the count-to-two has reached 2 (indicating that the main thread finished setting up and at least one continuation has executed), we'll call set_value on the promise corresponding to when_any's return future. Ta-da!

---

To explicitly answer Harald's question in the comments

what is the purpose of the double check using m_done and m_count_to_two, what necessitates that ? Isn't checking on m_done enough?

• We need at least m_done, "intuitively".
• But if all we had was m_done, we wouldn't be able to bump it in the main thread.
• So when the first continuation finished and bumped m_done, it would have no way of knowing whether the main thread had finished writing to sptr->m_tuple.
• If the continuation tried to move-out-of sptr->m_tuple while the main thread was still writing into it, we'd have a race condition.
• So we need another counter to indicate that (exactly) one of the continuation threads is ready and that the main thread is ready.

...Now it does occur to me that we could use a mutex lock instead, like this, and then we could get rid of the count-to-two and make m_done non-atomic. But I currently think I prefer the lockfree version as easier to reason about.(!)

Answer 1

After Harald's questions, it occurred to me that I could have done it with a plain old mutex instead of all the lock-free stuff. But if I do it this way, then it starts to look like "blocking". I think it's fine — this use of a mutex isn't any more harmful than the use of a mutex in the innards of promise/future in the first place — but in a strange way I'm finding it harder to reason about this new version than about the version in the question.

#include "scratch/bits/concurrency/future-and-promise.h"
#include "scratch/bits/concurrency/lock-guard.h"
#include "scratch/bits/concurrency/mutex.h"
#include "scratch/bits/concurrency/unique-lock.h"
#include "scratch/bits/smart-ptrs/shared-ptr.h"
#include "scratch/bits/tuple/tuple.h"
#include "scratch/bits/type-traits/decay.h"

#include <utility>

namespace scratch::detail {

template<class... Futures>
struct when_any_shared_state {
    promise<tuple<Futures...>> m_promise;
    tuple<Futures...> m_tuple;
    mutex m_mtx;  // protects m_done and m_tuple
    bool m_done;

    when_any_shared_state(promise<tuple<Futures...>> p) :
        m_promise(std::move(p)), m_done(false) {}
};

} // namespace scratch::detail

namespace scratch {

template<class... Futures>
auto when_any(Futures... futures) -> future<tuple<Futures...>>
{
    using shared_state = detail::when_any_shared_state<Futures...>;
    using R = tuple<Futures...>;
    promise<R> p;
    future<R> result = p.get_future();

    auto sptr = make_shared<shared_state>(std::move(p));
    auto satisfy_combined_promise =
        [sptr](auto f) {
            unique_lock lk(sptr->m_mtx);
            if (!sptr->m_done) {
                sptr->m_done = true;
                lk.unlock();
                sptr->m_promise.set_value(std::move(sptr->m_tuple));
            }
            return f.get();
        };
    {
        lock_guard lk(sptr->m_mtx);
        sptr->m_tuple = tuple<Futures...>(futures.then(satisfy_combined_promise)...);
    }
    return result;
}

} // namespace scratch

Answer 2

It took me a while to figure this one out but thanks again for posting it was a good way to read up on these new concepts. I find the locking version easier to understand, as it expresses the key concept of having to protect the construction of the m_tuple much better. That was the part that I was struggling with on the atomic counter version. In that case a continuation function might have triggered while the constructor or the assignment operator are still running in the main thread.

I am torn about your change to the return value type, yes I agree that this makes the signatures the same but it now seems to require an iteration over the returned tuple to find the future that is ready. I can see where the standards committee is coming from.

I wonder if using the unique lock and unlocking early is worth the extra effort that might be in Instatiating a unique lock vs a lock guard.

As both versions seem to work the faster one should probably win ;)