source: mainline/uspace/lib/cpp/include/__bits/thread/shared_state.hpp@ a72f3b8

/*
 * Copyright (c) 2019 Jaroslav Jindrak
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * - Redistributions in binary form must reproduce the above copyright
 *   notice, this list of conditions and the following disclaimer in the
 *   documentation and/or other materials provided with the distribution.
 * - The name of the author may not be used to endorse or promote products
 *   derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#ifndef LIBCPP_BITS_THREAD_SHARED_STATE
#define LIBCPP_BITS_THREAD_SHARED_STATE

/**
 * 30.6.4, shared state:
 */

#include <__bits/exception.hpp>
#include <__bits/functional/function.hpp>
#include <__bits/functional/invoke.hpp>
#include <__bits/refcount_obj.hpp>
#include <__bits/thread/future_common.hpp>
#include <__bits/thread/threading.hpp>
#include <cerrno>
#include <thread>
#include <tuple>

namespace std::aux
{
    class shared_state_base: public aux::refcount_obj
    {
        public:
            shared_state_base()
                : mutex_{}, condvar_{}, value_set_{false},
                  exception_{}, has_exception_{false}
            {
                threading::mutex::init(mutex_);
                threading::condvar::init(condvar_);
            }

            void destroy() override
            {
                /**
                 * Note: No need to act in this case, async shared
                 *       state is the object that needs to sometimes
                 *       invoke its payload.
                 */
            }

            void mark_set(bool set = true) noexcept
            {
                value_set_ = set;
            }

            bool is_set() const noexcept
            {
                return value_set_;
            }

            void set_exception(exception_ptr ptr, bool set = true)
            {
                exception_ = ptr;
                has_exception_ = set;
            }

            bool has_exception() const noexcept
            {
                return has_exception_;
            }

            void throw_stored_exception() const
            {
                if (has_exception_)
                    rethrow_exception(exception_);
            }

            virtual void wait() const
            {
                aux::threading::mutex::lock(
                    const_cast<aux::mutex_t&>(mutex_)
                );

                while (!value_set_)
                {
                    aux::threading::condvar::wait(
                        const_cast<aux::condvar_t&>(condvar_),
                        const_cast<aux::mutex_t&>(mutex_)
                    );
                }

                aux::threading::mutex::unlock(
                    const_cast<aux::mutex_t&>(mutex_)
                );
            }

            template<class Rep, class Period>
            future_status
            wait_for(const chrono::duration<Rep, Period>& rel_time) const
            {
                if (value_set_)
                    return future_status::ready;

                aux::threading::mutex::lock(
                    const_cast<aux::mutex_t&>(mutex_)
                );

                auto res = timed_wait_(
                    aux::threading::time::convert(rel_time)
                );

                aux::threading::mutex::unlock(
                    const_cast<aux::mutex_t&>(mutex_)
                );

                return res;
            }

            template<class Clock, class Duration>
            future_status
            wait_until(const chrono::time_point<Clock, Duration>& abs_time)
            {
                if (value_set_)
                    return future_status::ready;

                aux::threading::mutex::lock(
                    const_cast<aux::mutex_t&>(mutex_)
                );

                auto res = timed_wait_(
                    aux::threading::time::convert(abs_time - Clock::now())
                );

                aux::threading::mutex::unlock(
                    const_cast<aux::mutex_t&>(mutex_)
                );

                return res;
            }

            ~shared_state_base() override = default;

        protected:
            aux::mutex_t mutex_;
            aux::condvar_t condvar_;

            bool value_set_;

            exception_ptr exception_;
            bool has_exception_;

            /**
             * Note: wait_for and wait_until are templates and as such
             *       cannot be virtual and overriden by the deferred_ and
             *       async_ children. However, we are using aux::time_unit_t
             *       in the end anyway, so we can work around that
             *       by using the 'template method' design pattern
             *       (i.e. by providing a virtual function called by these
             *       templates and then overriding that function in the
             *       children).
             */
            virtual future_status timed_wait_(aux::time_unit_t time) const
            {
                auto res = aux::threading::condvar::wait_for(
                    const_cast<aux::condvar_t&>(condvar_),
                    const_cast<aux::mutex_t&>(mutex_), time
                );

                return res == ETIMEOUT ? future_status::timeout
                                       : future_status::ready;
            }
    };

    template<class R>
    class shared_state: public shared_state_base
    {
        public:
            shared_state()
                : shared_state_base{}
            { /* DUMMY BODY */ }

            void set_value(const R& val, bool set)
            {
                /**
                 * Note: This is the 'mark ready' move described
                 *       in 30.6.4 (6).
                 */

                aux::threading::mutex::lock(mutex_);
                value_ = val;
                value_set_ = set;
                aux::threading::mutex::unlock(mutex_);

                if (set)
                    aux::threading::condvar::broadcast(condvar_);
            }

            void set_value(R&& val, bool set = true)
            {
                aux::threading::mutex::lock(mutex_);
                value_ = std::move(val);
                value_set_ = set;
                aux::threading::mutex::unlock(mutex_);

                if (set)
                    aux::threading::condvar::broadcast(condvar_);
            }

            R& get()
            {
                return value_;
            }

        protected:
            R value_;
    };

    template<>
    class shared_state<void>: public shared_state_base
    {
        public:
            shared_state()
                : shared_state_base{}
            { /* DUMMY BODY */ }

            void set_value()
            {
                value_set_ = true;
                aux::threading::condvar::broadcast(condvar_);
            }

            void get()
            { /* DUMMY BODY */ }
    };

    /**
     * We could make one state for both async and
     * deferred policies, but then we would be wasting
     * memory and the only benefit would be the ability
     * for additional implementation defined policies done
     * directly in that state (as opposed to making new
     * states for them).
     *
     * But since we have no plan (nor need) to make those,
     * this approach seems to be the best one.
     *
     * Also note that unlike the parent class shared_state,
     * we do not need to specialize these for void. This is because
     * the difference in class contents are handled by the parent
     * specialization and setting the value can be handled easily
     * by if constexpr and checking for the equivalence of the
     * R template parameter and void.
     */

    template<class R, class F, class... Args>
    class async_shared_state: public shared_state<R>
    {
        public:
            async_shared_state(F&& f, Args&&... args)
                : shared_state<R>{}, thread_{}
            {
                thread_ = thread{
                    [=](){
                        try
                        {
                            if constexpr (!is_same_v<R, void>)
                                this->set_value(invoke(f, args...));
                            else
                            {
                                invoke(f, args...);
                                this->mark_set(true);
                            }
                        }
                        catch(const exception& __exception)
                        {
                            this->set_exception(make_exception_ptr(__exception));
                        }
                    }
                };
            }

            void destroy() override
            {
                if (!this->is_set())
                    thread_.join();
            }

            void wait() const override
            {
                if (!this->is_set())
                    const_cast<thread&>(thread_).join();
            }

            ~async_shared_state() override
            {
                destroy();
            }

        protected:
            future_status timed_wait_(aux::time_unit_t time) const override
            {
                /**
                 * Note: Currently we have no timed join, but this
                 *       behaviour should be compliant.
                 */
                aux::threading::time::sleep(time);
                if (this->value_set_)
                    return future_status::ready;
                else
                    return future_status::timeout;
            }

        private:
            thread thread_;
    };

    template<class R, class F, class... Args>
    class deferred_shared_state: public shared_state<R>
    {
        public:
            template<class G>
            deferred_shared_state(G&& f, Args&&... args)
                : shared_state<R>{}, func_{forward<F>(f)},
                  args_{forward<Args>(args)...}
            { /* DUMMY BODY */ }

            void destroy() override
            {
                aux::threading::mutex::lock(this->mutex_);
                if (!this->is_set())
                    invoke_(make_index_sequence<sizeof...(Args)>{});
                aux::threading::mutex::unlock(this->mutex_);
            }

            void wait() const override
            {
                /**
                 * Note: Synchronization done in invoke_ -> set_value.
                 */
                if (!this->is_set())
                {
                    const_cast<
                        deferred_shared_state<R, F, Args...>*
                    >(this)->invoke_(make_index_sequence<sizeof...(Args)>{});
                }
            }

            ~deferred_shared_state() override
            {
                destroy();
            }

        protected:
            function<R(decay_t<Args>...)> func_;
            tuple<decay_t<Args>...> args_;

            template<size_t... Is>
            void invoke_(index_sequence<Is...>)
            {
                try
                {
                    if constexpr (!is_same_v<R, void>)
                        this->set_value(invoke(move(func_), get<Is>(move(args_))...));
                    else
                    {
                        invoke(move(func_), get<Is>(move(args_))...);
                        this->mark_set(true);
                    }
                }
                catch(const exception& __exception)
                {
                    this->set_exception(make_exception_ptr(__exception));
                }
            }

            future_status timed_wait_(aux::time_unit_t) const override
            {
                /**
                 * Note: Neither of the wait_ functions has any effect
                 *       for deferred functions spawned by async (which
                 *       are the only users of this state type).
                 */
                return future_status::deferred;
            }
    };

    /**
     * Note: The following two functions should:
     *   1) Increment refcount.
     *   2) Store ptr to a vector of shared_state_base ptrs
     *      (as those have ::mark_set member functions).
     *   3) If not done already, register a function
     *      executing all these in the thread_atexit function
     *      once that is implemented.
     */

    template<class R>
    void set_state_value_at_thread_exit(shared_state<R>* state)
    {
        // TODO: implement
        __unimplemented();
    }

    template<class R>
    void set_state_exception_at_thread_exit(shared_state<R>* state)
    {
        // TODO: implement
        __unimplemented();
    }
}

#endif