Instrument.h

Go to the documentation of this file.

// This file is part of DynExp.
 
#pragma once
 
#include "stdafx.h"
#include "Object.h"
 
namespace DynExp
{
    class InstrumentBase;
    class InstrumentInstance;
    class ExceptionContainer;
    class TaskBase;
    class InitTaskBase;
    class ExitTaskBase;
    class UpdateTaskBase;
 
    using InstrumentPtrType = std::shared_ptr<InstrumentBase>;
 
    template <typename InstrumentT>
    ConfiguratorBasePtrType MakeInstrumentConfig()
    {
        return std::make_shared<typename InstrumentT::ConfigType>();
    }
 
    template <typename InstrumentT>
    InstrumentPtrType MakeInstrument(const std::thread::id OwnerThreadID, ParamsBasePtrType&& Params)
    {
        dynamic_Params_cast<InstrumentT>(Params.get())->InstrumentData = std::make_unique<typename InstrumentT::InstrumentDataType>();
 
        return std::make_shared<InstrumentT>(OwnerThreadID, std::move(Params));
    }
 
    template <typename TaskT, typename... ArgTs>
    std::unique_ptr<TaskT> MakeTask(ArgTs&& ...Args)
    {
        return std::make_unique<TaskT>(std::forward<ArgTs>(Args)...);
    }
 
    int InstrumentThreadMain(std::unique_ptr<RunnableInstance>&& InstancePtr, RunnableObject* BaseObject);
 
    class ExceptionContainer
    {
    public:
        ExceptionContainer(const std::exception_ptr& Exception = nullptr) noexcept
            : Exception(Exception) {}
 
        void Throw() const { if (Exception) std::rethrow_exception(Exception); }
 
        bool IsError() const { return static_cast<bool>(Exception); }
 
        auto GetException() const { return Exception; }
 
        void ClearError() { Exception = nullptr; }
 
    private:
        std::exception_ptr Exception;   
    };
 
    class InstrumentDataBase : public Util::ISynchronizedPointerLockable
    {
    public:
        using TaskQueueType = std::list<std::unique_ptr<TaskBase>>;     
        using TaskQueueIteratorType = TaskQueueType::const_iterator;    
 
    protected:
        template <typename>
        struct dispatch_tag {};
 
    private:
        class InstrumentBaseOnlyType
        {
            friend class InstrumentDataBase;
            friend class InstrumentBase;
 
            constexpr InstrumentBaseOnlyType(InstrumentDataBase& Parent) noexcept : Parent(Parent) {}
 
            void Reset() { Parent.Reset(); }                                                                                                                                                                
            void IndicateException() noexcept { Parent.IndicateException(); }                                                                                                                               
            void SetException(std::exception_ptr Exception) noexcept { Parent.SetException(Exception); }                                                                                                    
            void EnqueueTask(std::unique_ptr<TaskBase>&& Task, bool CallFromInstrThread, bool NotifyReceiver) { Parent.EnqueueTask(std::move(Task), CallFromInstrThread, NotifyReceiver); }                 
            void EnqueuePriorityTask(std::unique_ptr<TaskBase>&& Task, bool CallFromInstrThread, bool NotifyReceiver) { Parent.EnqueuePriorityTask(std::move(Task), CallFromInstrThread, NotifyReceiver); } 
            void RemoveTaskFromQueue(TaskQueueIteratorType& Task) { Parent.RemoveTaskFromQueue(Task); }                                                                                                     
            void RemoveAllTasks() { Parent.RemoveAllTasks(); }                                                                                                                                              
            void RemoveAllTasksExceptFront() { Parent.RemoveAllTasksExceptFront(); }                                                                                                                        
            void CloseQueue() { Parent.CloseQueue(); }                                                                                                                                                      
 
            auto& GetNewTaskNotifier() noexcept { return Parent.GetNewTaskNotifier(); }                                                                                                                     
 
            InstrumentDataBase& Parent;     
        };
 
        class InstrumenThreadOnlyType
        {
            friend class InstrumentDataBase;
            friend int InstrumentThreadMain(std::unique_ptr<RunnableInstance>&&, RunnableObject*);
 
            constexpr InstrumenThreadOnlyType(InstrumentDataBase& Parent) noexcept : Parent(Parent) {}
 
            auto& GetNewTaskNotifier() noexcept { return Parent.GetNewTaskNotifier(); }                                                 
            void SetLastUpdateTime(std::chrono::system_clock::time_point LastUpdate) { Parent.LastUpdate = LastUpdate; }                
 
            InstrumentDataBase& Parent;     
        };
 
    public:
        InstrumentDataBase() : InstrumentBaseOnly(*this), InstrumentThreadOnly(*this), QueueClosed(false) {}
        virtual ~InstrumentDataBase() {}
 
    public:
 
        void EnqueueTask(std::unique_ptr<TaskBase>&& Task) { EnqueueTask(std::move(Task), false, true); }
 
        void EnqueuePriorityTask(std::unique_ptr<TaskBase>&& Task) { EnqueuePriorityTask(std::move(Task), false, true); }
 
        std::unique_ptr<TaskBase> PopTaskFront();
 
        std::unique_ptr<TaskBase> PopTaskBack();
 
        auto GetTaskFront() noexcept { return TaskQueue.begin(); }
 
        auto GetTaskBack() noexcept { return TaskQueue.end(); }
 
        size_t GetNumEnqueuedTasks() const noexcept { return TaskQueue.size(); }
 
        std::unique_ptr<TaskBase> PopFinishedTask();
 
        size_t GetNumFinishedTasks() const noexcept { return FinishedTasks.size(); }
 
        bool IsQueueClosed() const noexcept { return QueueClosed; }
 
        auto GetLastUpdateTime() const { return LastUpdate; }
 
        bool IsExceptionIndicated() const noexcept { return HasException; }
 
        std::exception_ptr GetException() const noexcept;
 
        InstrumentBaseOnlyType InstrumentBaseOnly;      
        InstrumenThreadOnlyType InstrumentThreadOnly;   
 
    private:
        void EnqueueTask(std::unique_ptr<TaskBase>&& Task, bool CallFromInstrThread, bool NotifyReceiver);
        
        void EnqueuePriorityTask(std::unique_ptr<TaskBase>&& Task, bool CallFromInstrThread, bool NotifyReceiver);
        
        void RemoveTaskFromQueue(TaskQueueIteratorType& Task);
 
        void RemoveAllTasks();
 
        void RemoveAllTasksExceptFront();
 
        void CloseQueue() { QueueClosed = true; }
 
        Util::OneToOneNotifier& GetNewTaskNotifier() noexcept { return NewTaskNotifier; }
 
 
        void Reset();
 
        virtual void ResetImpl(dispatch_tag<InstrumentDataBase>) {}
 
        void IndicateException() noexcept { HasException = true; }
 
        void SetException(std::exception_ptr Exception) noexcept;
 
        void CheckError() const;
 
        void CheckQueueState(bool CallFromInstrThread) const;
 
        TaskQueueType TaskQueue;                            
        TaskQueueType FinishedTasks;                        
        bool QueueClosed;                                   
 
        Util::OneToOneNotifier NewTaskNotifier;
 
        std::chrono::system_clock::time_point LastUpdate;   
 
        std::atomic<bool> HasException;
 
        std::exception_ptr InstrumentException;
    };
 
    class InstrumentParamsBase : public RunnableObjectParams
    {
        friend class InstrumentBase;
 
        template <typename>
        friend InstrumentPtrType MakeInstrument(const std::thread::id, ParamsBasePtrType&&);
 
    public:
        InstrumentParamsBase(ItemIDType ID, const DynExpCore& Core) : RunnableObjectParams(ID, Core) {}
 
        virtual ~InstrumentParamsBase() = 0;
 
        virtual const char* GetParamClassTag() const noexcept override { return "InstrumentParamsBase"; }
 
    private:
        void ConfigureParamsImpl(dispatch_tag<RunnableObjectParams>) override final { ConfigureParamsImpl(dispatch_tag<InstrumentParamsBase>()); }
        virtual void ConfigureParamsImpl(dispatch_tag<InstrumentParamsBase>) {}     
 
        std::unique_ptr<InstrumentDataBase> InstrumentData;
 
        DummyParam Dummy = { *this };                                               
    };
 
    class InstrumentConfiguratorBase : public RunnableObjectConfigurator
    {
    public:
        using ObjectType = InstrumentBase;
        using ParamsType = InstrumentParamsBase;
 
        InstrumentConfiguratorBase() = default;
        virtual ~InstrumentConfiguratorBase() = 0;
    };
 
    class InstrumentBase : public RunnableObject
    {
        class InstrumenThreadOnlyType
        {
            friend class InstrumentBase;
            friend int InstrumentThreadMain(std::unique_ptr<RunnableInstance>&&, RunnableObject*);
 
            constexpr InstrumenThreadOnlyType(InstrumentBase& Parent) noexcept : Parent(Parent) {}
 
            auto HandleTask(InstrumentInstance& Instance) { return Parent.HandleTask(Instance); }           
            void UpdateData() { Parent.UpdateDataInternal(); }                                              
            void SetException(std::exception_ptr Exception) noexcept { Parent.SetException(Exception); }    
            void OnError() { Parent.OnError(); }                                                            
            void SetInitialized() { Parent.Initialized = true; }                                            
 
            InstrumentBase& Parent;     
        };
 
    public:
        enum class TaskHandlingContinuationType { Continue, Terminate, Defer };
        using ParamsType = InstrumentParamsBase;                                                            
        using ConfigType = InstrumentConfiguratorBase;                                                      
        
        using InstrumentDataType = InstrumentDataBase;
 
        using InstrumentDataTypeSyncPtrType = Util::SynchronizedPointer<InstrumentDataType>;
 
        using InstrumentDataTypeSyncPtrConstType = Util::SynchronizedPointer<const InstrumentDataType>;
 
        
        using InitTaskType = InitTaskBase;          
        using ExitTaskType = ExitTaskBase;          
        using UpdateTaskType = UpdateTaskBase;      
 
        constexpr static auto Category() noexcept { return "General"; }
 
        InstrumentBase(const std::thread::id OwnerThreadID, ParamsBasePtrType&& Params);
 
        virtual ~InstrumentBase() = 0;
 
        virtual std::string GetCategory() const override { return Category(); }
 
        InstrumenThreadOnlyType InstrumentThreadOnly;       
 
 
        virtual std::chrono::milliseconds GetTaskQueueDelay() const { return std::chrono::milliseconds(std::chrono::milliseconds::max()); }
 
        static constexpr auto GetInstrumentDataTimeoutDefault = std::chrono::milliseconds(1000);
 
 
        InstrumentDataTypeSyncPtrType GetInstrumentData(const std::chrono::milliseconds Timeout = GetInstrumentDataTimeoutDefault);
 
        InstrumentDataTypeSyncPtrConstType GetInstrumentData(const std::chrono::milliseconds Timeout = GetInstrumentDataTimeoutDefault) const;
 
        using InstrumentDataGetterType = Util::CallableMemberWrapper<InstrumentBase,
            InstrumentDataTypeSyncPtrType (InstrumentBase::*)(const std::chrono::milliseconds)>;
 
        void UpdateData() const;
 
        void EnqueueArriveAtLatchTask(std::latch& Latch) const;
 
        bool IsInitialized() const { return Initialized; }
 
    private:
 
        InstrumentDataTypeSyncPtrType GetNonConstInstrumentData(const std::chrono::milliseconds Timeout = GetInstrumentDataTimeoutDefault) const;
 
    protected:
        static auto GetExceptionUnsafe(const InstrumentDataTypeSyncPtrConstType& InstrumentDataPtr) { return InstrumentDataPtr->GetException(); }
 
        template <typename TaskT, typename... ArgTs>
        void MakeAndEnqueueTask(ArgTs&& ...Args) const
        {
            auto Task = MakeTask<TaskT>(std::forward<ArgTs>(Args)...);
 
            // Locks InstrumentData
            GetNonConstInstrumentData()->InstrumentBaseOnly.EnqueueTask(std::move(Task), IsCallFromRunnableThread(), true);
        }
 
    public:
        template <typename DerivedInstrT, typename... TaskFuncArgTs, typename... ArgTs>
        ExceptionContainer AsSyncTask(void (DerivedInstrT::* TaskFunc)(TaskFuncArgTs...) const, ArgTs&& ...Args) const;
 
    private:
 
        TaskHandlingContinuationType HandleTask(InstrumentInstance& Instance);
 
        void UpdateDataInternal();
 
        void SetException(std::exception_ptr Exception) noexcept;
 
        void OnError();
 
        void ResetImpl(dispatch_tag<RunnableObject>) override final;
        virtual void ResetImpl(dispatch_tag<InstrumentBase>) = 0;       
 
        void RunChild() override final;
        void NotifyChild() override final;
        void TerminateChild(const std::chrono::milliseconds Timeout) override final;
 
        void OnPrepareExit();
 
        std::exception_ptr GetExceptionChild([[maybe_unused]] const std::chrono::milliseconds Timeout) const override final;
        bool IsReadyChild() const override final;
 
        virtual void OnErrorChild() const {}                            
        virtual void OnPrepareExitChild() const {}                      
 
        virtual bool HandleAdditionalTask() { return true; }
 
        virtual bool UpdateAdditionalData() { return true; }
 
        virtual std::unique_ptr<InitTaskBase> MakeInitTask() const { return nullptr; }
 
        virtual std::unique_ptr<ExitTaskBase> MakeExitTask() const { return nullptr; }
 
        virtual std::unique_ptr<UpdateTaskBase> MakeUpdateTask() const { return nullptr; }
 
        const std::unique_ptr<InstrumentDataType> InstrumentData;   
        std::atomic<bool> Initialized = false;                      
    };
 
    class InstrumentInstance : public RunnableInstance
    {
    public:
        InstrumentInstance(InstrumentBase& Owner, std::promise<void>&& ThreadExitedPromise,
            const InstrumentBase::InstrumentDataGetterType InstrumentDataGetter);
 
        InstrumentInstance(InstrumentInstance&& Other);
 
        ~InstrumentInstance() = default;
 
        const InstrumentBase::InstrumentDataGetterType InstrumentDataGetter;
    };
 
    template <typename To, typename From, std::enable_if_t<
        std::is_same_v<InstrumentDataBase, std::remove_cv_t<From>>, int> = 0
    >
    auto dynamic_InstrumentData_cast(Util::SynchronizedPointer<From>&& InstrumentDataPtr)
    {
        if (!InstrumentDataPtr)
            throw Util::InvalidArgException("InstrumentDataPtr must not be nullptr.");
 
        return Util::SynchronizedPointer<
            std::conditional_t<std::is_const_v<From>, std::add_const_t<typename To::InstrumentDataType>, typename To::InstrumentDataType>
        >(std::move(InstrumentDataPtr));
    }
 
    class TaskResultType
    {
    public:
        enum class ContinuationType : bool {
            Continue,   
            Terminate   
        };
 
        enum class AbortedType : bool {
            NotAborted, 
            Aborted     
        };
 
        constexpr TaskResultType(const ContinuationType Continue = ContinuationType::Continue, const AbortedType Aborted = AbortedType::NotAborted,
            const int ErrorCode = 0) noexcept
            : Continue(Continue), Aborted(Aborted), ErrorCode(ErrorCode) {}
 
        constexpr bool ShouldContinue() const noexcept { return Continue == ContinuationType::Continue; }
 
        constexpr InstrumentBase::TaskHandlingContinuationType ToTaskHandlingContinuationType() const noexcept;
 
        constexpr bool HasAborted() const noexcept { return Aborted == AbortedType::Aborted; }
 
        constexpr int GetErrorCode() const noexcept { return ErrorCode; }
 
    private:
        const ContinuationType Continue;    
        const AbortedType Aborted;          
        const int ErrorCode;                
    };
 
    class TaskBase
    {
        class InstrumentBaseOnlyType
        {
            friend class TaskBase;
            friend class InstrumentBase;
 
            constexpr InstrumentBaseOnlyType(TaskBase& Parent) noexcept : Parent(Parent) {}
 
            void Lock() { Parent.Lock(); }                                                  
            auto Run(InstrumentInstance& Instance) { return Parent.Run(Instance); }         
            void SetAborted() { Parent.State = TaskState::Aborted; }                        
 
            TaskBase& Parent;       
        };
 
        class InstrumentDataBaseOnlyType
        {
            friend class TaskBase;
            friend class InstrumentDataBase;
 
            constexpr InstrumentDataBaseOnlyType(TaskBase& Parent) noexcept : Parent(Parent) {}
 
            bool KeepFinishedTask() const noexcept { return Parent.KeepFinishedTask(); }    
 
            TaskBase& Parent;       
        };
 
    public:
        class CallbackType
        {
        public:
            using FuncType = std::function<void(const TaskBase*, ExceptionContainer&)>;
 
            CallbackType() : CallbackFunc() {}
            
            CallbackType(std::nullptr_t) : CallbackFunc() {}
            
            CallbackType(FuncType&& CallbackFunc) : CallbackFunc(CallbackFunc) {}
            
            CallbackType(CallbackType&& Other);
            
            ~CallbackType();
 
            template <typename... ArgTs>
            void operator()(ArgTs&& ...Args)
            {
                if (CallbackFunc && !HasBeenCalled)
                {
                    HasBeenCalled = true;
 
                    CallbackFunc(std::forward<ArgTs>(Args)...);
                }
            }
 
            operator bool() const noexcept { return static_cast<bool>(CallbackFunc); }
 
        private:
            const FuncType CallbackFunc;    
 
            bool HasBeenCalled = false;     
        };
 
        enum class TaskState { Waiting, Locked, Running, Finished, Failed, Aborted };
        TaskBase(CallbackType CallbackFunc = nullptr, std::chrono::system_clock::time_point DeferUntil = {}) noexcept
            : InstrumentBaseOnly(*this), InstrumentDataBaseOnly(*this),
            CallbackFunc(std::move(CallbackFunc)), DeferUntil(DeferUntil),
            State(TaskState::Waiting), ErrorCode(0), ShouldAbort(false) {}
 
        TaskBase(TaskBase& Other, std::chrono::system_clock::time_point DeferUntil = {}) noexcept
            : TaskBase(std::move(Other.CallbackFunc), DeferUntil) {}
 
        virtual ~TaskBase() = 0;
 
 
        auto GetDeferUntil() const noexcept { return DeferUntil; }
 
        TaskState GetState() const noexcept { return State; }
 
        bool IsLocked() const noexcept;
 
        bool IsAborting() const noexcept { return ShouldAbort; }
 
        int GetErrorCode() const noexcept { return ErrorCode; }
 
        void Abort() { ShouldAbort = true; }
 
        InstrumentBaseOnlyType InstrumentBaseOnly;                          
        InstrumentDataBaseOnlyType InstrumentDataBaseOnly;                  
        
    private:
        void Lock();
 
        InstrumentBase::TaskHandlingContinuationType Run(InstrumentInstance& Instance);
 
        virtual TaskResultType RunChild(InstrumentInstance& Instance) = 0;  
 
        virtual bool KeepFinishedTask() const noexcept { return false; }
 
        CallbackType CallbackFunc;
 
        const std::chrono::system_clock::time_point DeferUntil;
 
        std::atomic<TaskState> State;       
        std::atomic<int> ErrorCode;         
 
        std::atomic<bool> ShouldAbort;      
    };
 
    template <typename DerivedInstrT, typename... TaskFuncArgTs, typename... ArgTs>
    ExceptionContainer InstrumentBase::AsSyncTask(void (DerivedInstrT::* TaskFunc)(TaskFuncArgTs...) const, ArgTs&& ...Args) const
    {
        std::atomic<bool> FinishedFlag = false;
        ExceptionContainer Exception;
        auto CallbackFunc = TaskBase::CallbackType::FuncType([&FinishedFlag, &Exception](const TaskBase*, auto E) {
            Exception = E;
 
            // Must come last!
            FinishedFlag = true;
        });
 
        (dynamic_cast<const DerivedInstrT&>(*this).*TaskFunc)(std::forward<ArgTs>(Args)..., std::move(CallbackFunc));
 
        while (!FinishedFlag)
            std::this_thread::yield();
 
        return Exception;
    }
 
    class DefaultTask final : public TaskBase
    {
    public:
        DefaultTask(CallbackType CallbackFunc, std::chrono::system_clock::time_point DeferUntil = {}) noexcept
            : TaskBase(std::move(CallbackFunc), DeferUntil) {}
 
    private:
        virtual TaskResultType RunChild(InstrumentInstance& Instance) override { return {}; }
    };
 
    class InitTaskBase : public TaskBase
    {
    protected:
        template <typename>
        struct dispatch_tag {};
 
    private:
        TaskResultType RunChild(InstrumentInstance& Instance) override final;
 
        virtual void InitFuncImpl(dispatch_tag<InitTaskBase>, InstrumentInstance& Instance) = 0;
    };
 
    class ExitTaskBase : public TaskBase
    {
    protected:
        template <typename>
        struct dispatch_tag {};
 
    private:
        TaskResultType RunChild(InstrumentInstance& Instance) override final;
 
        virtual void ExitFuncImpl(dispatch_tag<ExitTaskBase>, InstrumentInstance& Instance) = 0;
    };
 
    class UpdateTaskBase : public TaskBase
    {
    protected:
        template <typename>
        struct dispatch_tag {};
 
    private:
        TaskResultType RunChild(InstrumentInstance& Instance) override final;
 
        virtual void UpdateFuncImpl(dispatch_tag<UpdateTaskBase>, InstrumentInstance& Instance) = 0;
    };
 
    class ArriveAtLatchTask final : public TaskBase
    {
    public:
        ArriveAtLatchTask(std::latch& Latch, CallbackType CallbackFunc = nullptr)
            : TaskBase(std::move(CallbackFunc)), Latch(Latch) {}
 
        ~ArriveAtLatchTask();
 
    private:
        virtual TaskResultType RunChild(InstrumentInstance& Instance) override;
 
        std::latch& Latch;              
        bool HasArrived = false;        
    };
 
    template <typename... InstrTs>
    void WaitForInstruments(InstrTs&... Instruments)
    {
        static_assert(std::conjunction_v<std::is_base_of<InstrumentBase, InstrTs>...>);
        static_assert(sizeof...(InstrTs) > 0);
        static_assert(sizeof...(InstrTs) <= std::latch::max());
 
        std::latch Latch(sizeof...(InstrTs));
        std::array<std::reference_wrapper<const InstrumentBase>, sizeof...(InstrTs)> Instrs{ Instruments... };
 
        for (auto& Instr : Instrs)
            Instr.get().EnqueueArriveAtLatchTask(Latch);
 
        Latch.wait();
    }
}
 
namespace DynExpInstr {};