ParallelSystem.hpp

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

#include <samchon/templates/external/ExternalSystem.hpp>
#include <samchon/protocol/IListener.hpp>

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

#include <thread>
#include <samchon/HashMap.hpp>
#include <samchon/templates/parallel/PRInvokeHistory.hpp>

namespace samchon
{
namespace templates
{
namespace parallel
{
    class ParallelSystem
        : public virtual external::ExternalSystem,
        public virtual protocol::IListener
    {
    private:
        typedef external::ExternalSystem super;

        HashMap<size_t, std::pair<std::shared_ptr<protocol::Invoke>, std::shared_ptr<slave::InvokeHistory>>> progress_list_;
        HashMap<size_t, std::shared_ptr<slave::InvokeHistory>> history_list_;

        double performance_{ 1.0 };
        bool enforced_{ false };
        bool excluded_{ false };

    public:
        /* ---------------------------------------------------------
            CONSTRUCTORS
        --------------------------------------------------------- */
        // TAKES SUPER'S CONSTRUCTORS
        using super::super;

        virtual ~ParallelSystem()
        {
            excluded_ = true;

            // SHIFT PARALLEL INVOKE MESSAGES HAD PROGRESSED TO OTHER SLAVES
            for (auto it = progress_list_.begin(); it != progress_list_.end(); it++)
            {
                // INVOKE MESSAGE AND ITS HISTORY ON PROGRESS
                std::shared_ptr<protocol::Invoke> invoke = it->second.first;
                std::shared_ptr<slave::InvokeHistory> history = it->second.second;

                // SEND THEM BACK
                _Send_back_history(invoke, history);
            }
        };

        virtual void construct(std::shared_ptr<library::XML> xml) override
        {
            super::construct(xml);

            performance_ = xml->getProperty<double>("performance");
            enforced_ = xml->getProperty<bool>("enforced");
        };

        /* ---------------------------------------------------------
            ACCESSORS
        --------------------------------------------------------- */
        auto getPerformance() const -> double
        {
            return performance_;
        };

        void setPerformance(double val)
        {
            performance_ = val;
            enforced_ = false;
        };

        void enforcePerformance(double val)
        {
            performance_ = val;
            enforced_ = true;
        };

        /* ---------------------------------------------------------
            INVOKE MESSAGE CHAIN - PERFORMANCE ESTIMATION
        --------------------------------------------------------- */
    protected:
        virtual void _Reply_data(std::shared_ptr<protocol::Invoke> invoke) override
        {
            if (invoke->getListener() == "_Report_history")
                _Report_history(invoke->front()->getValueAsXML());
            else if (invoke->getListener() == "_Send_back_history")
            {
                size_t uid = invoke->front()->getValue<size_t>();
                auto it = progress_list_.find(uid);

                if (it != progress_list_.end())
                    _Send_back_history(it->second.first, it->second.second);
            }
            else
                replyData(invoke);
        };

        virtual void _Report_history(std::shared_ptr<library::XML> xml)
        {
            library::UniqueWriteLock uk(system_array_->getMutex());

            //--------
            // CONSTRUCT HISTORY
            //--------
            std::shared_ptr<PRInvokeHistory> history(new PRInvokeHistory());
            history->construct(xml);

            // IF THE HISTORY IS NOT EXIST IN PROGRESS, THEN TERMINATE REPORTING
            auto progress_it = progress_list_.find(history->getUID());
            if (progress_it == progress_list_.end())
                return;

            // ARCHIVE FIRST AND LAST INDEX
            history->first_ = std::dynamic_pointer_cast<PRInvokeHistory>(progress_it->second.second)->getFirst();
            history->last_ = std::dynamic_pointer_cast<PRInvokeHistory>(progress_it->second.second)->getLast();

            // ERASE FROM ORDINARY PROGRESS AND MIGRATE TO THE HISTORY
            progress_list_.erase(progress_it);
            history_list_.insert({ history->getUID(), history });

            // NOTIFY TO THE MANAGER, SYSTEM_ARRAY
            ((base::ParallelSystemArrayBase*)system_array_)->_Complete_history(history);
        };

        virtual void _Send_back_history(std::shared_ptr<protocol::Invoke> invoke, std::shared_ptr<slave::InvokeHistory> $history)
        {
            std::shared_ptr<PRInvokeHistory> history = std::dynamic_pointer_cast<PRInvokeHistory>($history);
            if (history == nullptr)
                return;

            // REMOVE UID AND FIRST, LAST INDEXES
            for (size_t i = invoke->size(); i < invoke->size(); i--)
            {
                const std::string &name = invoke->at(i)->getName();

                if (name == "_History_uid" || name == "_Piece_first" || name == "_Piece_last")
                    invoke->erase(invoke->begin() + i);
            }

            // RE-SEND (DISTRIBUTE) THE PIECE TO OTHER SLAVES
            std::thread
            (
                &base::ParallelSystemArrayBase::sendPieceData, (base::ParallelSystemArrayBase*)system_array_,
                invoke, history->getFirst(), history->getLast()
            ).detach();

            // ERASE FROM THE PROGRESS LIST
            progress_list_.erase(history->getUID());
        };

    public:
        /* ---------------------------------------------------------
            EXPORTERS
        --------------------------------------------------------- */
        virtual auto toXML() const -> std::shared_ptr<library::XML> override
        {
            std::shared_ptr<library::XML> xml = super::toXML();
            xml->setProperty("performance", performance_);
            xml->setProperty("enforced", enforced_);

            return xml;
        };

        /* ---------------------------------------------------------
            INTERNAL ACCESSORS
        --------------------------------------------------------- */
        auto _Get_progress_list() -> HashMap<size_t, std::pair<std::shared_ptr<protocol::Invoke>, std::shared_ptr<slave::InvokeHistory>>>& { return progress_list_; };
        auto _Get_progress_list() const -> const HashMap<size_t, std::pair<std::shared_ptr<protocol::Invoke>, std::shared_ptr<slave::InvokeHistory>>>& { return progress_list_; };
        
        auto _Get_history_list() -> HashMap<size_t, std::shared_ptr<slave::InvokeHistory>>& { return history_list_; };
        auto _Get_history_list() const -> const HashMap<size_t, std::shared_ptr<slave::InvokeHistory>>& { return history_list_; };

        auto _Is_enforced() const -> bool
        {
            return enforced_;
        };
        auto _Is_excluded() const -> bool
        {
            return excluded_;
        };
        void _Set_excluded()
        {
            excluded_ = true;
        };
    };
};
};
};