FutureSharedState.h

Go to the documentation of this file.

 
#pragma once
#include "catapult/utils/NonCopyable.h"
#include "catapult/functions.h"
#include <future>
#include <memory>
#include <mutex>
 
namespace catapult { namespace thread { namespace detail {
 
    template<typename T>
    class shared_state : utils::NonCopyable {
    private:
        using ContinuationFunc = consumer<const std::shared_ptr<shared_state<T>>>;
 
        enum class future_status {
            pending,
            completed_success,
            completed_error
        };
 
    public:
        shared_state() : m_status(future_status::pending)
        {}
 
    public:
        bool is_ready() const {
            return future_status::pending != m_status;
        }
 
        T get() {
            if (m_status == future_status::pending) {
                std::unique_lock<std::mutex> lock(m_mutex);
                m_condition.wait(lock, [&status = m_status]() { return status != future_status::pending; });
            }
 
            if (future_status::completed_error == m_status)
                std::rethrow_exception(m_pException);
 
            return std::move(m_value);
        }
 
    public:
        void set_value(T&& value) {
            std::lock_guard<std::mutex> lock(m_mutex);
            assert_pending();
            m_value = std::move(value);
            signal(future_status::completed_success);
        }
 
        void set_exception(std::exception_ptr pException) {
            std::lock_guard<std::mutex> lock(m_mutex);
            assert_pending();
            m_pException = pException;
            signal(future_status::completed_error);
        }
 
        void set_continuation(const ContinuationFunc& continuation) {
            std::lock_guard<std::mutex> lock(m_mutex);
            if (m_continuation)
                throw std::logic_error("continuation already set");
 
            m_continuation = continuation;
            if (is_ready())
                invokeContinuation();
        }
 
    private:
        void assert_pending() const {
            if (is_ready())
                throw std::future_error(std::future_errc::promise_already_satisfied);
        }
 
        void signal(future_status status) {
            m_status = status;
            m_condition.notify_all();
            if (!m_continuation)
                return;
 
            invokeContinuation();
        }
 
        void invokeContinuation() {
            auto pStateCopy = std::make_shared<shared_state<T>>();
            pStateCopy->m_status = m_status.load();
            pStateCopy->m_value = std::move(m_value);
            pStateCopy->m_pException = m_pException;
            m_continuation(pStateCopy);
        }
 
    private:
        std::atomic<future_status> m_status;
        T m_value;
        std::exception_ptr m_pException;
        ContinuationFunc m_continuation;
 
        std::condition_variable m_condition;
        std::mutex m_mutex;
    };
}}}