ProcessScheduler.h

 
/*
 * SPDX-FileCopyrightText: 2005,2022 Daniel Barrientos
 * <danivillamanin@gmail.com>
 *
 * SPDX-License-Identifier: MIT
 */
 
#pragma once
 
#include <tasking/Process.h>
 
#include <forward_list>
using mel::tasking::Process;
using std::forward_list;
 
#include <utility>
 
/*
 * SPDX-FileCopyrightText: 2005,2022 Daniel Barrientos
 * <danivillamanin@gmail.com>
 *
 * SPDX-License-Identifier: MIT
 */
 
#include <core/Timer.h>
using mel::core::Timer;
#include <core/CallbackSubscriptor.h>
using mel::core::CallbackSubscriptor;
#include <mpl/Int2Type.h>
using ::mel::mpl::Int2Type;
#include <atomic>
#include <deque>
#include <memory>
#include <mutex>
#include <variant>
#include <vector>
namespace mel
{
    namespace tasking
    {
        typedef CallbackSubscriptor<::mel::core::CSMultithreadPolicy,
                                    std::shared_ptr<Process>>
            WakeSubscriptor;
        typedef CallbackSubscriptor<::mel::core::CSNoMultithreadPolicy,
                                    std::shared_ptr<Process>>
            SleepSubscriptor;
        typedef CallbackSubscriptor<::mel::core::CSNoMultithreadPolicy,
                                    std::shared_ptr<Process>>
            EvictSubscriptor;
        class MEL_API ProcessScheduler
        {
            typedef unsigned int ThreadID;
            friend class Process;
            // container for new tasks when using lock_Free
            class LockFreeTasksContainer
            {
              public:
                struct ElementType
                {
                    ElementType() : p( nullptr ), st( 0 ), valid( false ) {}
                    std::shared_ptr<Process> p;
                    unsigned int st;
                    std::atomic<bool> valid;
                };
 
              private:
                typedef std::vector<ElementType> PoolType;
                std::atomic<size_t> mCurrIdx = 0;
                std::atomic<size_t> mSize;
                std::deque<PoolType> mPool;
                size_t mChunkSize;
                size_t mMaxSize;
                volatile bool mInvalidate;
                std::mutex mSC;
 
              public:
                LockFreeTasksContainer( size_t chunkSize, size_t maxChunks );
                PoolType::value_type& operator[]( size_t idx );
                void add( std::shared_ptr<Process>& process,
                          unsigned int startTime );
                inline size_t
                getCurrIdx( std::memory_order mo = std::memory_order_relaxed )
                {
                    return mCurrIdx.load( mo );
                }
                void clear();
                size_t size( std::memory_order memOrder =
                                 std::memory_order_acquire ) const
                {
                    return mSize.load( memOrder );
                } // pruebas memoryorder
                // return previous value
                size_t exchangeIdx( size_t v, std::memory_order order =
                                                  std::memory_order_seq_cst );
                //@todo lock y unlock de pruebas
                void lock();
                void unlock();
                bool isInvalidate() const { return mInvalidate; }
                void setInvalidate( bool v ) { mInvalidate = v; }
                size_t getMaxSize() const { return mMaxSize; }
            };
 
          public:
            typedef std::pair<std::shared_ptr<Process>, unsigned int>
                ProcessElement;
            typedef forward_list<ProcessElement>
                TProcessList; // pairs of processes and starttime
            typedef std::deque<
                std::pair<std::shared_ptr<Process>, unsigned int>>
                TNewProcesses; // pairs of processes and starttime in blocking
                               // scheduler
 
            struct LockFreeOptions
            {
                size_t chunkSize = 512; 
                size_t maxChunks = 4; 
            };
            struct BlockingOptions
            {
                // empty now
            };
            typedef std::variant<BlockingOptions, LockFreeOptions>
                SchedulerOptions;
            ProcessScheduler( SchedulerOptions opts );
            ~ProcessScheduler( void );
 
            void insertProcess( std::shared_ptr<Process> process,
                                unsigned int startTime = 0 );
            void insertProcessNoLock( std::shared_ptr<Process> process,
                                      unsigned int startTime = 0 );
 
            inline unsigned int getProcessCount() const;
            /*
             * get number of processes running (not sleeped)
             */
            inline unsigned int getActiveProcessCount() const;
            void executeProcesses();
            void pauseProcesses();
            template <class T> void pauseProcesses( T& predicate );
 
            void activateProcesses();
            void killProcesses( bool deferred );
 
            // inline bool checkFor(unsigned int taskId);
 
            inline TProcessList& getProcesses();
            //      inline TNewProcesses& getPendingProcesses();
            void destroyAllProcesses();
 
            void setTimer( std::shared_ptr<Timer> timer );
            inline const std::shared_ptr<Timer> getTimer() const;
            inline std::shared_ptr<Timer> getTimer();
 
            static std::shared_ptr<Process> getCurrentProcess();
 
            template <class F> int susbcribeSleepEvent( F&& f )
            {
                return mSS.subscribeCallback( std::forward<F>( f ) );
            }
            template <class F> void unsusbcribeSleepEvent( F&& f )
            {
                mSS.unsubscribeCallback( std::forward<F>( f ) );
            }
            void unsusbcribeSleepEvent( int id )
            {
                mSS.unsubscribeCallback( id );
            }
            template <class F> int susbcribeWakeEvent( F&& f )
            {
                return mWS.subscribeCallback( std::forward<F>( f ) );
            }
            template <class F> void unsusbcribeWakeEvent( F&& f )
            {
                mWS.unsubscribeCallback( std::forward<F>( f ) );
            }
            template <class F> int subscribeProcessEvicted( F&& f )
            {
                return mES.subscribeCallback( std::forward<F>( f ) );
            }
            template <class F> void unsubscribeProcessEvicted( F&& f )
            {
                mES.unsubscribeCallback( std::forward<F>( f ) );
            }
            void unsubscribeProcessEvicted( int id )
            {
                mES.unsubscribeCallback( id );
            }
            struct ProcessInfo // for TLS
            {
                std::shared_ptr<Process> current;
            };
 
          private:
            SchedulerOptions mOpts;
            bool mIsLockFree; // same as checking index in mOpts but for
                              // performance
 
            ProcessInfo* mProcessInfo;
            TProcessList mProcessList;
            std::unique_ptr<LockFreeTasksContainer> mLockFreeTasks;
            // new processes to insert next time blocking variant
            TNewProcesses mBlockingTasks;
            size_t mLastIdx;
            std::mutex mCS;
            std::shared_ptr<Timer> mTimer;
            std::atomic<unsigned int> mProcessCount;
            std::atomic<int32_t> mInactiveProcessCount;
            bool mKillingProcess; // flag to mark when ther is a kill task
                                  // pending
#ifndef NDEBUG
            void* _stack = nullptr;
#endif
            WakeSubscriptor mWS;
            SleepSubscriptor mSS;
            EvictSubscriptor mES;
 
            void _executeProcesses( uint64_t time,
                                    TProcessList& processes ) OPTIMIZE_FLAGS;
 
            void _killTasks();
            void _triggerSleepEvents( std::shared_ptr<Process> p );
            void _triggerWakeEvents( std::shared_ptr<Process> p );
            static ProcessInfo* _getCurrentProcessInfo();
            void processAwakened( std::shared_ptr<Process> );
            void processAsleep( std::shared_ptr<Process> );
        };
 
        template <class T> void ProcessScheduler::pauseProcesses( T& predicate )
        {
            for ( auto i = mProcessList.begin(); i != mProcessList.end(); ++i )
            {
                if ( predicate( *i ) )
                {
                    ( *i ).first->pause();
                }
            }
        }
        ProcessScheduler::TProcessList& ProcessScheduler::getProcesses()
        {
            return mProcessList;
        }
        /*ProcessScheduler::TNewProcesses&
        ProcessScheduler::getPendingProcesses()
        {
                return mNewProcesses;
        }*/
 
        unsigned int ProcessScheduler::getProcessCount() const
        {
 
            return mProcessCount;
        }
        unsigned int ProcessScheduler::getActiveProcessCount() const
        {
            return getProcessCount() - ( (unsigned int)mInactiveProcessCount );
        }
 
        const std::shared_ptr<Timer> ProcessScheduler::getTimer() const
        {
            return mTimer;
        }
        std::shared_ptr<Timer> ProcessScheduler::getTimer() { return mTimer; }
        // static thread_local ProcessScheduler::ProcessInfo
        // tlCurrentProcess{nullptr};
    } // namespace tasking
} // namespace mel