第26课 std::async异步任务

// CLASS TEMPLATE _Associated_state
template <class _Ty>
class _Associated_state { // class for managing associated synchronous state
public:
    using _State_type = _Ty;
    using _Mydel      = _Deleter_base<_Ty>;

    _Associated_state(_Mydel* _Dp = nullptr)
        : _Refs(1), // non-atomic initialization
          _Exception(), _Retrieved(false), _Ready(false), _Ready_at_thread_exit(false), _Has_stored_result(false),
          _Running(false), _Deleter(_Dp) { // construct
        // TODO: _Associated_state ctor assumes _Ty is default constructible
    }

    virtual ~_Associated_state() noexcept { // destroy
        if (_Has_stored_result && !_Ready) { // registered for release at thread exit
            _Cond._Unregister(_Mtx);
        }
    }

    void _Retain() { // increment reference count
        _MT_INCR(_Refs);
    }

    void _Release() { // decrement reference count and destroy when zero
        if (_MT_DECR(_Refs) == 0) {
            _Delete_this();
        }
    }

private:
    _Atomic_counter_t _Refs;

public:
    virtual void _Wait() { // wait for signal
        unique_lock<mutex> _Lock(_Mtx);
        _Maybe_run_deferred_function(_Lock);
        while (!_Ready) {
            _Cond.wait(_Lock);
        }
    }

    struct _Test_ready { // wraps _Associated_state
        _Test_ready(const _Associated_state* _St) : _State(_St) { // construct from associated state
        }

        bool operator()() const { // test state
            return _State->_Ready != 0;
        }
        const _Associated_state* _State;
    };

    template <class _Rep, class _Per>
    future_status _Wait_for(const chrono::duration<_Rep, _Per>& _Rel_time) { // wait for duration
        unique_lock<mutex> _Lock(_Mtx);
        if (_Has_deferred_function()) {
            return future_status::deferred;
        }

        if (_Cond.wait_for(_Lock, _Rel_time, _Test_ready(this))) {
            return future_status::ready;
        }

        return future_status::timeout;
    }

    template <class _Clock, class _Dur>
    future_status _Wait_until(const chrono::time_point<_Clock, _Dur>& _Abs_time) { // wait until time point
        unique_lock<mutex> _Lock(_Mtx);
        if (_Has_deferred_function()) {
            return future_status::deferred;
        }

        if (_Cond.wait_until(_Lock, _Abs_time, _Test_ready(this))) {
            return future_status::ready;
        }

        return future_status::timeout;
    }

    virtual _Ty& _Get_value(bool _Get_only_once) { // return the stored result or throw stored exception
        unique_lock<mutex> _Lock(_Mtx);
        if (_Get_only_once && _Retrieved) {
            _Throw_future_error(make_error_code(future_errc::future_already_retrieved));
        }

        if (_Exception) {
            _Rethrow_future_exception(_Exception);
        }

        _Retrieved = true;
        _Maybe_run_deferred_function(_Lock);
        while (!_Ready) {
            _Cond.wait(_Lock);
        }

        if (_Exception) {
            _Rethrow_future_exception(_Exception);
        }

        return _Result;
    }

    void _Set_value(const _Ty& _Val, bool _At_thread_exit) { // store a result
        unique_lock<mutex> _Lock(_Mtx);
        _Set_value_raw(_Val, &_Lock, _At_thread_exit);
    }

    void _Set_value_raw(const _Ty& _Val, unique_lock<mutex>* _Lock,
        bool _At_thread_exit) { // store a result while inside a locked block
        if (_Has_stored_result) {
            _Throw_future_error(make_error_code(future_errc::promise_already_satisfied));
        }

        _Result = _Val;
        _Do_notify(_Lock, _At_thread_exit);
    }

    void _Set_value(_Ty&& _Val, bool _At_thread_exit) { // store a result
        unique_lock<mutex> _Lock(_Mtx);
        _Set_value_raw(_STD forward<_Ty>(_Val), &_Lock, _At_thread_exit);
    }

    void _Set_value_raw(_Ty&& _Val, unique_lock<mutex>* _Lock,
        bool _At_thread_exit) { // store a result while inside a locked block
        if (_Has_stored_result) {
            _Throw_future_error(make_error_code(future_errc::promise_already_satisfied));
        }

        _Result = _STD forward<_Ty>(_Val);
        _Do_notify(_Lock, _At_thread_exit);
    }

    void _Set_value(bool _At_thread_exit) { // store a (void) result
        unique_lock<mutex> _Lock(_Mtx);
        _Set_value_raw(&_Lock, _At_thread_exit);
    }

    void _Set_value_raw(
        unique_lock<mutex>* _Lock, bool _At_thread_exit) { // store a (void) result while inside a locked block
        if (_Has_stored_result) {
            _Throw_future_error(make_error_code(future_errc::promise_already_satisfied));
        }

        _Do_notify(_Lock, _At_thread_exit);
    }

    void _Set_exception(exception_ptr _Exc, bool _At_thread_exit) { // store a result
        unique_lock<mutex> _Lock(_Mtx);
        _Set_exception_raw(_Exc, &_Lock, _At_thread_exit);
    }

    void _Set_exception_raw(exception_ptr _Exc, unique_lock<mutex>* _Lock,
        bool _At_thread_exit) { // store a result while inside a locked block
        if (_Has_stored_result) {
            _Throw_future_error(make_error_code(future_errc::promise_already_satisfied));
        }

        _Exception = _Exc;
        _Do_notify(_Lock, _At_thread_exit);
    }

    bool _Is_ready() const { // return ready status
        return _Ready != 0;
    }

    bool _Is_ready_at_thread_exit() const { // return ready at thread exit status
        return _Ready_at_thread_exit;
    }

    bool _Already_has_stored_result() const { // return presence of stored result
        return _Has_stored_result;
    }

    bool _Already_retrieved() const { // return retrieved status
        return _Retrieved;
    }

    void _Abandon() { // abandon shared state
        unique_lock<mutex> _Lock(_Mtx);
        if (!_Has_stored_result) { // queue exception
            future_error _Fut(make_error_code(future_errc::broken_promise));
            _Set_exception_raw(_STD make_exception_ptr(_Fut), &_Lock, false);
        }
    }

protected:
    void _Make_ready_at_thread_exit() { // set ready status at thread exit
        if (_Ready_at_thread_exit) {
            _Ready = true;
        }
    }

    void _Maybe_run_deferred_function(unique_lock<mutex>& _Lock) { // run a deferred function if not already done
        if (!_Running) { // run the function
            _Running = true;
            _Run_deferred_function(_Lock);
        }
    }

public:
    _Ty _Result;
    exception_ptr _Exception;
    mutex _Mtx;
    condition_variable _Cond;
    bool _Retrieved;
    int _Ready;
    bool _Ready_at_thread_exit;
    bool _Has_stored_result;
    bool _Running;

private:
    virtual bool _Has_deferred_function() const noexcept { // overridden by _Deferred_async_state
        return false;
    }

    virtual void _Run_deferred_function(unique_lock<mutex>&) { // do nothing
    }

    virtual void _Do_notify(unique_lock<mutex>* _Lock, bool _At_thread_exit) { // notify waiting threads
        _Has_stored_result = true;
        if (_At_thread_exit) { // notify at thread exit
            _Cond._Register(*_Lock, &_Ready);
        } else { // notify immediately
            _Ready = true;
            _Cond.notify_all();
        }
    }

    void _Delete_this() { // delete this object
        if (_Deleter) {
            _Deleter->_Delete(this);
        } else {
            delete this;
        }
    }

    _Mydel* _Deleter;

public:
    _Associated_state(const _Associated_state&) = delete;
    _Associated_state& operator=(const _Associated_state&) = delete;
};

// CLASS TEMPLATE _Packaged_state
template <class>
class _Packaged_state;

template <class _Ret,
    class... _ArgTypes>
class _Packaged_state<_Ret(_ArgTypes...)>
    : public _Associated_state<_Ret> { // class for managing associated asynchronous state for packaged_task
public:
    using _Mybase = _Associated_state<_Ret>;
    using _Mydel  = typename _Mybase::_Mydel;

    template <class _Fty2>
    _Packaged_state(const _Fty2& _Fnarg) : _Fn(_Fnarg) { // construct from function object
    }

#if _HAS_FUNCTION_ALLOCATOR_SUPPORT
    template <class _Fty2, class _Alloc>
    _Packaged_state(const _Fty2& _Fnarg, const _Alloc& _Al, _Mydel* _Dp)
        : _Mybase(_Dp), _Fn(allocator_arg, _Al, _Fnarg) { // construct from function object and allocator
    }
#endif // _HAS_FUNCTION_ALLOCATOR_SUPPORT

    template <class _Fty2>
    _Packaged_state(_Fty2&& _Fnarg) : _Fn(_STD forward<_Fty2>(_Fnarg)) { // construct from rvalue function object
    }

#if _HAS_FUNCTION_ALLOCATOR_SUPPORT
    template <class _Fty2, class _Alloc>
    _Packaged_state(_Fty2&& _Fnarg, const _Alloc& _Al, _Mydel* _Dp)
        : _Mybase(_Dp), _Fn(allocator_arg, _Al,
                            _STD forward<_Fty2>(_Fnarg)) { // construct from rvalue function object and allocator
    }
#endif // _HAS_FUNCTION_ALLOCATOR_SUPPORT

    void _Call_deferred(_ArgTypes... _Args) { // set deferred call
        _TRY_BEGIN
        // call function object and catch exceptions
        this->_Set_value(_Fn(_STD forward<_ArgTypes>(_Args)...), true);
        _CATCH_ALL
        // function object threw exception; record result
        this->_Set_exception(_STD current_exception(), true);
        _CATCH_END
    }

    void _Call_immediate(_ArgTypes... _Args) { // call function object
        _TRY_BEGIN
        // call function object and catch exceptions
        this->_Set_value(_Fn(_STD forward<_ArgTypes>(_Args)...), false);
        _CATCH_ALL
        // function object threw exception; record result
        this->_Set_exception(_STD current_exception(), false);
        _CATCH_END
    }

    const function<_Ret(_ArgTypes...)>& _Get_fn() { // return stored function object
        return _Fn;
    }

private:
    function<_Ret(_ArgTypes...)> _Fn;
};

// CLASS TEMPLATE _Deferred_async_state
template <class _Rx>
class _Deferred_async_state : public _Packaged_state<_Rx()> { // class for managing associated synchronous state for
                                                              // deferred execution from async
public:
    template <class _Fty2>
    _Deferred_async_state(const _Fty2& _Fnarg) : _Packaged_state<_Rx()>(_Fnarg) { // construct from function object
    }

    template <class _Fty2>
    _Deferred_async_state(_Fty2&& _Fnarg)
        : _Packaged_state<_Rx()>(_STD forward<_Fty2>(_Fnarg)) { // construct from rvalue function object
    }

private:
    virtual bool _Has_deferred_function() const
        noexcept { // this function is considered to be deferred until it‘s invoked
        return !this->_Running;
    }

    virtual void _Run_deferred_function(unique_lock<mutex>& _Lock) { // run the deferred function
        _Lock.unlock();
        _Packaged_state<_Rx()>::_Call_immediate();
        _Lock.lock();
    }
};

// CLASS TEMPLATE _Task_async_state
template <class _Rx>
class _Task_async_state : public _Packaged_state<_Rx()> { // class for managing associated synchronous state for
                                                          // asynchronous execution from async
public:
    using _Mybase     = _Packaged_state<_Rx()>;
    using _State_type = typename _Mybase::_State_type;

    template <class _Fty2>
    _Task_async_state(_Fty2&& _Fnarg) : _Mybase(_STD forward<_Fty2>(_Fnarg)) { // construct from rvalue function object
        _Task = ::Concurrency::create_task([this]() { // do it now
            this->_Call_immediate();
        });

        this->_Running = true;
    }

    virtual ~_Task_async_state() noexcept { // destroy
        _Wait();
    }

    virtual void _Wait() { // wait for completion
        _Task.wait();
    }

    virtual _State_type& _Get_value(bool _Get_only_once) { // return the stored result or throw stored exception
        _Task.wait();
        return _Mybase::_Get_value(_Get_only_once);
    }

private:
    ::Concurrency::task<void> _Task;
};