ParallelSystemArray.hpp

#pragma once
#include <samchon/API.hpp>

#include <samchon/templates/external/ExternalSystemArray.hpp>
#   include <samchon/templates/parallel/ParallelSystem.hpp>
#include <samchon/templates/parallel/base/ParallelSystemArrayBase.hpp>

namespace samchon
{
namespace templates
{

namespace parallel
{
    template <class System = ParallelSystem>
    class ParallelSystemArray
        : public virtual external::ExternalSystemArray<System>,
        public base::ParallelSystemArrayBase
    {
    private:
        typedef external::ExternalSystemArray<System> super;

    public:
        /* ---------------------------------------------------------
            CONSTRUCTORS
        --------------------------------------------------------- */
        ParallelSystemArray()
            : super(),
            base::ParallelSystemArrayBase()
        {
        };
        virtual ~ParallelSystemArray() = default;

        /* =========================================================
            INVOKE MESSAGE CHAIN
                - SEND DATA
                - PERFORMANCE ESTIMATION
        ============================================================
            SEND & REPLY DATA
        --------------------------------------------------------- */
        auto sendSegmentData(std::shared_ptr<protocol::Invoke> invoke, size_t size) -> size_t
        {
            return sendPieceData(invoke, 0, size);
        };

        virtual auto sendPieceData(std::shared_ptr<protocol::Invoke> invoke, size_t first, size_t last) -> size_t
        {
            library::UniqueWriteLock uk(getMutex());

            if (invoke->has("_History_uid") == false)
                invoke->emplace_back(new protocol::InvokeParameter("_History_uid", _Fetch_history_sequence()));
            else
            {
                // INVOKE MESSAGE ALREADY HAS ITS OWN UNIQUE ID
                //  - THIS IS A TYPE OF ParallelSystemArrayMediator. THE MESSAGE HAS COME FROM ITS MASTER
                //  - A ParallelSystem HAS DISCONNECTED. THE SYSTEM SHIFTED ITS CHAIN TO OTHER SLAVES.
                size_t uid = invoke->get("_History_uid")->getValue<size_t>();

                // FOR CASE 1. UPDATE HISTORY_SEQUENCE TO MAXIMUM
                if (uid > _Get_history_sequence())
                    _Set_history_sequence(uid);
            }

            // TOTAL NUMBER OF PIECES TO DIVIDE
            size_t segment_size = last - first;

            // SYSTEMS TO BE GET DIVIDED PROCESSES AND
            std::vector<std::shared_ptr<ParallelSystem>> system_array;
            std::vector<std::thread> threads;
            
            system_array.reserve(size());
            threads.reserve(size());
            
            // POP EXCLUDEDS
            for (size_t i = 0; i < size(); i++)
            {
                std::shared_ptr<ParallelSystem> system = at(i);
                
                if (system->_Is_excluded() == false)
                    system_array.push_back(system);
            }

            // ORDERS
            for (size_t i = 0; i < system_array.size(); i++)
            {
                std::shared_ptr<ParallelSystem> system = system_array[i];
                
                // COMPUTE FIRST AND LAST INDEX TO ALLOCATE
                size_t piece_size = (i == system_array.size() - 1)
                    ? segment_size - first
                    : (size_t)(segment_size / system_array.size() * system->getPerformance());
                if (piece_size == 0)
                    continue;

                std::shared_ptr<protocol::Invoke> my_invoke(new protocol::Invoke(invoke->getListener()));
                {
                    // DUPLICATE INVOKE AND ATTACH PIECE INFO
                    my_invoke->assign(invoke->begin(), invoke->end());
                    my_invoke->emplace_back(new protocol::InvokeParameter("_Piece_first", first));
                    my_invoke->emplace_back(new protocol::InvokeParameter("_Piece_last", last));
                };

                // ENROLL TO PROGRESS LIST
                std::shared_ptr<slave::InvokeHistory> history(new PRInvokeHistory(my_invoke));
                system->_Get_progress_list().emplace(history->getUID(), std::make_pair(invoke, history));

                // ENROLL THE SEND DATA INTO THREADS
                threads.emplace_back(&ParallelSystem::sendData, system.get(), my_invoke);
                first += piece_size; // FOR THE NEXT STEP
            }
            uk.unlock();

            // JOIN THREADS
            for (auto it = threads.begin(); it != threads.end(); it++)
                it->join();

            return threads.size();
        };

        /* ---------------------------------------------------------
            PERFORMANCE ESTIMATION - INTERNAL METHODS
        --------------------------------------------------------- */
        virtual auto _Complete_history(std::shared_ptr<slave::InvokeHistory> history) -> bool
        {
            // WRONG TYPE
            if (std::dynamic_pointer_cast<PRInvokeHistory>(history) == nullptr)
                return false;

            //========
            // READ LOCK
            //========
            library::UniqueWriteLock uk(getMutex());
            size_t uid = history->getUID();

            // ALL THE SUB-TASKS ARE DONE?
            for (size_t i = 0; i < size(); i++)
                if (at(i)->_Get_progress_list().has(uid) == true)
                    return false; // IT'S ON A PROCESS IN SOME SYSTEM.

            //--------
            // RE-CALCULATE PERFORMANCE INDEX
            //--------
            // CONSTRUCT BASIC DATA
            std::vector<std::pair<std::shared_ptr<ParallelSystem>, double>> system_pairs;
            double performance_index_average = 0.0;

            system_pairs.reserve(size());

            for (size_t i = 0; i < size(); i++)
            {
                std::shared_ptr<ParallelSystem> system = at(i);
                if (system->_Get_history_list().has(uid) == false)
                    continue; // NO HISTORY (HAVE NOT PARTICIPATED IN THE PARALLEL PROCESS)

                // COMPUTE PERFORMANCE INDEX BASIS ON EXECUTION TIME OF THIS PARALLEL PROCESS
                std::shared_ptr<PRInvokeHistory> my_history = std::dynamic_pointer_cast<PRInvokeHistory>(system->_Get_history_list().get(uid));
                double performance_index = my_history->computeSize() / (double)my_history->computeElapsedTime();

                // PUSH TO SYSTEM PAIRS AND ADD TO AVERAGE
                system_pairs.emplace_back(system, performance_index);
                performance_index_average += performance_index;
            }
            performance_index_average /= system_pairs.size();

            // RE-CALCULATE PERFORMANCE INDEX
            for (size_t i = 0; i < system_pairs.size(); i++)
            {
                // SYSTEM AND NEW PERFORMANCE INDEX BASIS ON THE EXECUTION TIME
                auto system = system_pairs[i].first;
                if (system->_Is_enforced() == true)
                    continue; // PERFORMANCE INDEX IS ENFORCED. DOES NOT PERMIT REVALUATION

                double new_performance = system_pairs[i].second / performance_index_average;

                // DEDUCT RATIO TO REFLECT THE NEW PERFORMANCE INDEX
                double ordinary_ratio;
                if (system->_Get_history_list().size() < 2)
                    ordinary_ratio = .3;
                else
                    ordinary_ratio = std::min(0.7, 1.0 / (system->_Get_history_list().size() - 1.0));

                system->setPerformance((system->getPerformance() * ordinary_ratio) + (new_performance * (1 - ordinary_ratio)));
            }

            // AT LAST, NORMALIZE PERFORMANCE INDEXES OF ALL SLAVE SYSTEMS
            _Normalize_performance();
            return true;
        };

    protected:
        virtual void _Normalize_performance()
        {
            // COMPUTE AVERAGE
            double average = 0.0;
            size_t denominator = 0;

            for (size_t i = 0; i < size(); i++)
            {
                std::shared_ptr<ParallelSystem> system = at(i);
                if (system->_Is_enforced() == true)
                    continue; // PERFORMANCE INDEX IS ENFORCED. DOES NOT PERMIT REVALUATION

                average += system->getPerformance();
                denominator++;
            }
            average /= (double)denominator;

            // DIVIDE FROM THE AVERAGE
            for (size_t i = 0; i < size(); i++)
            {
                std::shared_ptr<ParallelSystem> system = at(i);
                if (system->_Is_enforced() == true)
                    continue; // PERFORMANCE INDEX IS ENFORCED. DOES NOT PERMIT REVALUATION

                system->setPerformance(system->getPerformance() / average);
            }
        };
    };
};
};
};