DataServerProxy.hpp

Go to the documentation of this file.

#ifndef __HCE_DATASERVER_PROXY_INCLUDED__
#define __HCE_DATASERVER_PROXY_INCLUDED__

#include <sstream>
#include <map>

#include "Handler.hpp"
#include "DataNodeOptions.hpp"

//Node ini
#include <Poco/Util/IniFileConfiguration.h>
#include "Poco/String.h"
//Message cover object
#include "JsonMessageCover.hpp"
//Calculator balancing nodes for execution
#include "CalculatorNodeExecution.hpp"
//Checker of resource usage limits
#include "ResourceUsageLimitsChecker.hpp"

namespace HCE {
  namespace handlers {

    class DataServerProxy: public HCE::handlers::Handler, protected HCE::handlers::DataNodeOptions {
      public:
        DataServerProxy(std::string name, zmq::context_t& ctx, const std::string& clientIdentity, std::string connectionString,
            unsigned char nodeMode, int64_t startedAt, unsigned char logPriority, const std::string& iniFile="");
        ~DataServerProxy();

      protected:
        int initialize(void);
        void deinitialize(void);
        void* process(void);

        void handleInternalMessage(void);
        void handleExternalMessage(void);

        std::string processAdminCommands(std::string& command, std::string& parameters);

        void logPeriodicStatistics(void);
        void heartbit(void);
        void sendHeartbeat(const std::string& identity);
        void registerClient(const std::string& identity);

        void processControlMessage(zmsg& msg);
        void processDataMessage(zmsg& msg);

        std::string getPropertyInformation(void);

        enum class EmptyResponseType : std::int8_t
        {
          ertNoClients=0,
          ertBadNodeMode=1,
          ertNotEnoughResourceLimits=2
        };
        std::string makeEmptyResponseMessage(const std::string& messageBody,
            EmptyResponseType responseType,
            const std::string& messageRoute,
            const std::string& errorMsg="");

        void sendEmptyResponseMessage(zmsg& msg,
            const std::string& coverMessageBody,
            EmptyResponseType responseType,
            const std::string& messageId,
            const std::string& messageRoute,
            const std::string& errorMsg="");

        void sendResponseMessage(const std::string& identity,
            const std::string& messageId,
            const std::string& messageBody,
            const std::string& messageRoute);

        void nodeInit(const std::string& iniFile);

        void checkInit(void);

        zmq::socket_t* _inprocProxySock;
        zmq::socket_t* _inprocReducerInSock;
        zmq::socket_t* _dataServerSock;

        size_t _previousClientNumber;

        size_t getRandomNodeNumber(size_t maxNumber);
        std::string getIdentityRandom(std::vector<ClientWorkerItem>& clients,
            size_t& previousClientNumber);
        std::string getIdentityResourceUsage(std::vector<ClientWorkerItem>& clients,
            size_t& previousClientNumber,
            CalculatorAlgorithm::Types typeAlgorithm,
            const CalculatorAlgorithmData& algorithmData);

        CalculatorAlgorithmData calculatorAlgorithmData;
        ResourceUsageLimitsData resourceUsageLimitsData;
        CalculatorAlgorithm::Types typeCalculatorAlgorithm;

        ResourceUsageManager _resourceUsageManager;

        std::vector<ClientWorkerItem> _clientsQueue;
        ClientsQueueManager* _clientsManager;

        int64_t _heartbeatAt;

        void setNodeMode(unsigned char nodeMode);
        unsigned char getNodeMode(void);

        void setResourcesUsageAlg(unsigned int resourcesUsageAlg);
        unsigned int getResourcesUsageAlg(void);

        void setResourcesUsageAlg0Weights(const std::string& resourcesUsageAlg0Weights);
        std::string getResourcesUsageAlg0Weights(void);

        void setResourcesUsageOrder(unsigned int resourcesUsageOrder);
        unsigned int getResourcesUsageOrder(void);

        void setResourcesUsageLimits(const std::string& resourcesUsageLimits);
        std::string getResourcesUsageLimits(void);

        void setResourcesUsageCollectedSize(unsigned int resourcesUsageCollectedSize);
        unsigned int getResourcesUsageCollectedSize(void);

        void setResourcesUsageMaxAllowedRange(double resourcesUsageMaxAllowedRange);
        double getResourcesUsageMaxAllowedRange(void);

        void setResourcesExtractorMaxThreads(unsigned int resourcesExtractorMaxThreads);
        unsigned int getResourcesExtractorMaxThreads(void);

        void setResourcesExtractorMaxProcesses(unsigned int resourcesExtractorMaxProcesses);
        unsigned int getResourcesExtractorMaxProcesses(void);

        void setResourcesExtractorMaxDiskSize(unsigned int resourcesExtractorMaxDiskSize);
        unsigned int getResourcesExtractorMaxDiskSize(void);

        void setResourcesExtractorMaxVramSize(unsigned int resourcesExtractorMaxVramSize);
        unsigned int getResourcesExtractorMaxVramSize(void);

        void setResourcesExtractorMaxRramSize(unsigned int resourcesExtractorMaxRramSize);
        unsigned int getResourcesExtractorMaxRramSize(void);

        void setStatClientRequests(unsigned int statClientRequests);
        unsigned int getStatClientRequests(void);

        void setStatClientResponses(unsigned int statClientResponses);
        unsigned int getStatClientResponses(void);

        void setStatRequests(unsigned int statRequests);
        unsigned int getStatRequests(void);

        void setStatClientRequestsSize(unsigned int statRequestsSize);
        unsigned int getStatClientRequestsSize(void);

        void setStatClientResponsesSize(unsigned int statResponsesSize);
        unsigned int getStatClientResponsesSize(void);

        void setStatRequestsSize(unsigned int statRequestsSize);
        unsigned int getStatRequestsSize(void);

        void setPeriodicStatClientRequests(unsigned int periodicStatClientRequests);
        unsigned int getPeriodicStatClientRequests(void);

        void setPeriodicStatClientResponses(unsigned int periodicStatClientResponses);
        unsigned int getPeriodicStatClientResponses(void);

        void setStatRequestsProcessed(unsigned int statRequestsProcessed);
        unsigned int getStatRequestsProcessed(void);

        void setStatRequestsProcessedSize(unsigned int statRequestsProcessedSize);
        unsigned int getStatRequestsProcessedSize(void);

        void setPeriodicStatClientRequestsSize(unsigned int periodicStatClientRequestsSize);
        unsigned int getPeriodicStatClientRequestsSize(void);

        void setPeriodicStatClientResponsesSize(unsigned int periodicStatClientResponsesSize);
        unsigned int getPeriodicStatClientResponsesSize(void);

        void setStatQPSClient(float statQPSClient);
        float getStatQPSClient(void);

        void setStatQPSServer(float statQPSServer);
        float getStatQPSServer(void);

        void setStatRPSClient(float statRPSClient);
        float getStatRPSClient(void);

        void setStatRPSServer(float statRPSServer);
        float getStatRPSServer(void);

        void setStatRequestsBandwidthClient(float statRequestsBandwidthClient);
        float getStatRequestsBandwidthClient(void);

        void setStatRequestsBandwidthServer(float statRequestsBandwidthServer);
        float getStatRequestsBandwidthServer(void);

        void setStatResponsesBandwidthClient(float statResponsesBandwidthClient);
        float getStatResponsesBandwidthClient(void);

        void setStatResponsesBandwidthServer(float statResponsesBandwidthServer);
        float getStatResponsesBandwidthServer(void);

        int64_t _pstatsAt = s_clock() + HEARTBEAT_DELAY;
    };
  }
}

#endif /* __HCE_DATASERVER_PROXY_INCLUDED__ */