DistributedProcess.hpp

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

#include <samchon/templates/distributed/base/DistributedProcessBase.hpp>
#include <samchon/protocol/Entity.hpp>

#include <samchon/protocol/Invoke.hpp>
#include <samchon/templates/distributed/DSInvokeHistory.hpp>

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

namespace samchon
{
namespace templates
{
namespace distributed
{
    class DistributedProcess
        : public virtual protocol::Entity<std::string>,
        public base::DistributedProcessBase
    {
    private:
        typedef protocol::Entity<std::string> super;

        base::DistributedSystemArrayBase *system_array_;

    protected:
        std::string name;

    public:
        /* ---------------------------------------------------------
            CONSTRUCTORS
        --------------------------------------------------------- */
        DistributedProcess(base::DistributedSystemArrayBase *systemArray)
            : super()
        {
            this->system_array_ = systemArray;

            resource = 1.0;
            enforced_ = false;
        };
        virtual ~DistributedProcess() = default;

        virtual void construct(std::shared_ptr<library::XML> xml) override
        {
            name = xml->getProperty<std::string>("name");
            resource = xml->getProperty<double>("resource");
        };

        /* ---------------------------------------------------------
            ACCESSORS
        --------------------------------------------------------- */
        virtual auto key() const -> std::string override
        {
            return name;
        };

        template <class SystemArray>
        auto getSystemArray() const -> SystemArray*
        {
            return (SystemArray*)system_array_;
        };

        auto getName() const -> std::string
        {
            return name;
        };

        auto getResource() const -> double
        {
            return resource;
        };

        void setResource(double val)
        {
            resource = val;
            enforced_ = false;
        };

        void enforceResource(double val)
        {
            resource = val;
            enforced_ = true;
        };

    public:
        /* ---------------------------------------------------------
            INVOKE MESSAGE CHAIN
        --------------------------------------------------------- */
        auto sendData(std::shared_ptr<protocol::Invoke> invoke) -> std::shared_ptr<DistributedSystem>
        {
            return sendData(invoke, 1.0);
        };

        virtual auto sendData(std::shared_ptr<protocol::Invoke> invoke, double weight) -> std::shared_ptr<DistributedSystem> override
        {
            library::UniqueReadLock uk(((external::base::ExternalSystemArrayBase*)system_array_)->getMutex());
            if (((protocol::SharedEntityDeque<external::ExternalSystem>*)system_array_)->empty() == true)
                return nullptr;

            // ADD UID FOR ARCHIVING HISTORY
            size_t uid;
            if (invoke->has("_History_uid") == false)
            {
                // ISSUE UID AND ATTACH IT TO INVOKE'S LAST PARAMETER
                uid = ((parallel::base::ParallelSystemArrayBase*)system_array_)->_Fetch_history_sequence();
                invoke->emplace_back(new protocol::InvokeParameter("_History_uid", uid));
            }
            else
            {
                // INVOKE MESSAGE ALREADY HAS ITS OWN UNIQUE ID
                //  - system_array_ IS A TYPE OF DistributedSystemArrayMediator. THE MESSAGE HAS COME FROM ITS MASTER
                //  - A Distributed HAS DISCONNECTED. THE SYSTEM SHIFTED ITS CHAIN TO ANOTHER SLAVE.
                uid = invoke->get("_History_uid")->getValue<size_t>();

                // FOR CASE 1. UPDATE HISTORY_SEQUENCE TO MAXIMUM
                if (uid > ((parallel::base::ParallelSystemArrayBase*)system_array_)->_Get_history_sequence())
                    ((parallel::base::ParallelSystemArrayBase*)system_array_)->_Set_history_sequence(uid);

                // FOR CASE 2. ERASE ORDINARY PROGRESSIVE HISTORY FROM THE DISCONNECTED
                progress_list_.erase(uid);
            }

            // ADD ROLE NAME FOR MEDIATOR
            if (invoke->has("_Process_name") == false)
                invoke->emplace_back(new protocol::InvokeParameter("_Process_name", name));
            if (invoke->has("_Process_weight") == false)
                invoke->emplace_back(new protocol::InvokeParameter("_Process_weight", weight));
            else
                weight = invoke->get("_Process_weight")->getValue<double>();

            // FIND THE MOST IDLE SYSTEM
            std::vector<std::shared_ptr<external::ExternalSystem>> children;
            std::shared_ptr<DistributedSystem> idle_system;

            for (size_t i = 0; i < children.size(); i++)
            {
                std::shared_ptr<DistributedSystem> system = std::dynamic_pointer_cast<DistributedSystem>(children.at(i));
                if (system->_Is_excluded() == true)
                    continue; // BEING REMOVED SYSTEM

                if (idle_system == nullptr
                    || system->_Get_progress_list().size() < idle_system->_Get_progress_list().size()
                    || system->getPerformance() < idle_system->getPerformance())
                    idle_system = system;
            }

            // ARCHIVE HISTORY ON PROGRESS_LIST (IN SYSTEM AND ROLE AT THE SAME TIME)
            std::shared_ptr<DSInvokeHistory> history(new DSInvokeHistory(idle_system.get(), this, invoke, weight));

            progress_list_.emplace(uid, history);
            idle_system->_Get_progress_list().emplace(uid, make_pair(invoke, history));

            uk.unlock(); // SELECTING IDLE AND ARCHIVING HISTORY ENTITY ARE COMPLETED.

            // SEND DATA
            idle_system->sendData(invoke);

            // RETURNS THE IDLE
            return idle_system;
        };

        virtual void replyData(std::shared_ptr<protocol::Invoke>) = 0;

    public:
        /* ---------------------------------------------------------
            EXPORTERS
        --------------------------------------------------------- */
        virtual auto TAG() const -> std::string override
        {
            return "process";
        };

        virtual auto toXML() const -> std::shared_ptr<library::XML> override
        {
            std::shared_ptr<library::XML> &xml = super::toXML();
            xml->setProperty("name", name);
            xml->setProperty("resource", resource);

            return xml;
        };

        /* ---------------------------------------------------------
            INTERNAL METHODS
        --------------------------------------------------------- */
        auto _Compute_average_elapsed_time() const -> double
        {
            double sum = 0.0;

            for (auto it = history_list_.begin(); it != history_list_.end(); it++)
            {
                std::shared_ptr<DSInvokeHistory> history = it->second;
                double elapsed_time = history->computeElapsedTime() / history->getWeight();

                // THE SYSTEM'S PERFORMANCE IS 5. THE SYSTEM CAN HANDLE A PROCESS VERY QUICKLY
                // AND ELAPSED TIME OF THE PROCESS IS 3 SECONDS
                // THEN I CONSIDER THE ELAPSED TIME AS 15 SECONDS.
                sum += elapsed_time * history->getSystem()->getPerformance();
            }
            return sum / history_list_.size();
        };

        auto _Get_progress_list() -> HashMap<size_t, std::shared_ptr<DSInvokeHistory>>& { return progress_list_; };
        auto _Get_progress_list() const -> const HashMap<size_t, std::shared_ptr<DSInvokeHistory>>& { return progress_list_; };

        auto _Get_history_list() -> HashMap<size_t, std::shared_ptr<DSInvokeHistory>>& { return history_list_; };
        auto _Get_history_list() const -> const HashMap<size_t, std::shared_ptr<DSInvokeHistory>>& { return history_list_; };

        auto _Is_enforced() const -> bool
        {
            return enforced_;
        };
    };
};
};
};