ClientsQueueManager.hpp

Go to the documentation of this file.

#ifndef __HCE_CLIENTS_QUEUE_MANAGER_INCLUDED__
#define __HCE_CLIENTS_QUEUE_MANAGER_INCLUDED__

#include "HandlerTypes.hpp"
#include "AdminClientsSerializator.hpp"
#include "ResourceUsageManager.hpp"

namespace HCE {
  namespace handlers {

    class ClientsQueueManager {
      public:
        //Purge mode: 0 - purge client from queue if expired, 1 - not purge but increment counter only
        ClientsQueueManager(std::vector<ClientWorkerItem>* queue, unsigned int purgeMode, HCE::handlers::ResourceUsageManager* resourceUsageManager=nullptr) :
            _refreshNotFoundCounter(0), _purgeMode(purgeMode),
            _purgeCounter(0), _heartbeatTimeout(HEARTBEAT_TIMEOUT),
            _maxResourcesUsageCollectedSize(HCE::handlers::RESOURCES_USAGE_COLLECTED_SIZE){
          _queue = queue;
          _resourceUsageManager=resourceUsageManager;
        }

        void append(const std::string &identity, bool checkPresence = true){
          bool found = false;

          if(checkPresence){
            //Check allworkers by identity
            for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
              if(it->identity.compare(identity) == 0){
                found = true;
                break;
              }
            }
          }

          //Not found
          if(!found){
            //Add new worker
            ClientWorkerItem worker(identity, (s_clock() + _heartbeatTimeout));
            _queue->push_back(worker);

            //Reset counters to balance requests proportional way after new worker insertion
            resetFrequencies();
          }
        }

        void append(const std::string& identity, const std::vector<ClientWorkerItem>& route, const std::string& resources, bool checkPresence = true){
          bool found = false;

          if(checkPresence){
            //Check allworkers by identity
            for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
              if(it->identity.compare(identity) == 0){
                it->route = route;
                if (!resources.empty() && _resourceUsageManager){
                                _resourceUsageManager->add(it->identity, it->resources, resources, _maxResourcesUsageCollectedSize);
                }
                found = true;
                break;
              }
            }
          }

          //Not found
          if(!found){
            //Add new worker
            ClientWorkerItem worker(identity, (s_clock() + getHeartbeatTimeout()));
            worker.route = route;
            if (!resources.empty() && _resourceUsageManager){
                        _resourceUsageManager->add(worker.identity, worker.resources, resources, _maxResourcesUsageCollectedSize);
            }
            _queue->push_back(worker);

            //Reset counters to balance requests proportional way after new worker insertion
            resetFrequencies();
          }
        }

        void append(const std::string& identity, const std::vector<ClientWorkerItem>& route, const ResourceUsageCollection& resources, bool checkPresence = true){
          bool found = false;

          if(checkPresence){
            //Check allworkers by identity
            for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
              if(it->identity.compare(identity) == 0){
                it->route = route;
                it->resources = resources;
                found = true;
                break;
              }
            }
          }

          //Not found
          if(!found){
            //Add new worker
            ClientWorkerItem worker(identity, (s_clock() + getHeartbeatTimeout()));
            worker.route = route;
            worker.resources = resources;
            _queue->push_back(worker);

            //Reset counters to balance requests proportional way after new worker insertion
            resetFrequencies();
          }
        }

        void append(const std::vector<ClientWorkerItem>& queue){
          for (size_t i=0;i<queue.size();++i)
          {
            append(queue[i].identity, queue[i].route, queue[i].resources);
          }
        }

        void append(const std::string& identity, const std::string& property, bool checkPresence = true){
          bool found = false;

          if(checkPresence){
            //Check allworkers by identity
            for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
              if(it->identity.compare(identity) == 0){
                it->property = property;
                found = true;
                break;
              }
            }
          }

          //Not found
          if(!found){
            //Add new worker
            ClientWorkerItem worker(identity, (s_clock() + getHeartbeatTimeout()));
            worker.property = property;
            _queue->push_back(worker);

            //Reset counters to balance requests proportional way after new worker insertion
            resetFrequencies();
          }
        }

        void remove(const std::string &identity){
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              it = _queue->erase(it);
              break;
            }
          }
        }

        bool refresh(const std::string &identity){
          bool found = false;

          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              it->expiry = s_clock() + _heartbeatTimeout;
              it->rtimestamp = s_clock();
              found = true;
              break;
            }
          }

          //TODO: experimental
          if(!found){
            //If not found - add as new and fresh
            append(identity);

            _refreshNotFoundCounter++;
          }

          return found;
        }

        bool refresh(const std::string &identity, const std::vector<ClientWorkerItem>& route, const std::string& resources=""){
          bool found = false;

          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              it->expiry = s_clock() + _heartbeatTimeout;
              it->rtimestamp = s_clock();
              it->route = route;
              if (!resources.empty() && _resourceUsageManager){
                        _resourceUsageManager->add(it->identity, it->resources, resources, _maxResourcesUsageCollectedSize);
              }
              found = true;
              break;
            }
          }

          //TODO: experimental
          if(!found){
            //If not found - add as new and fresh
            append(identity, route, resources);

            _refreshNotFoundCounter++;
          }

          return found;
        }

        bool refresh(const std::string& identity, const std::string& property){
          bool found = false;

          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              it->property = property;
              found = true;
              break;
            }
          }

          //TODO: experimental
          if(!found){
            //If not found - add as new and fresh
            append(identity, property);

            _refreshNotFoundCounter++;
          }
          return found;
        }

        std::string purge(void){
          std::stringstream ret("");

          int64_t clock = s_clock();
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it != _queue->end(); ++it){

            it->shiftCounter.responceTimeStamp(it->rtimestamp);
            it->shiftCounter.recount(_heartbeatTimeout);

//            if(clock > it->expiry){ //old
            if(clock > it->expiry+_heartbeatTimeout + it->shiftCounter.getShift()){
              ret << "{\"identity\":\"" << it->identity << "\",\"expiry\":" << it->expiry
                  << ",\"count\":" << it->count << ",\"frequency\":" << it->frequency
                  << ",\"shift\":" << it->shiftCounter.getShift()
                  << "},";
              it->shiftCounter.reset();

              if(_purgeMode == 0){
                it = _queue->erase(it) - 1;
              }
              _purgeCounter++;
            }
          }
          return ret.str();
        }

        bool exists(const std::string &identity){
          bool ret = false;

          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              ret = true;
              break;
            }
          }

          return ret;
        }

        std::string getDump(void) const {
          std::stringstream stream;
          stream << _purgeCounter << "," << _refreshNotFoundCounter << "," << _queue->size() << ",";
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            stream << it->identity << " " << it->count << " " << it->frequency << " " << it->expiry << " " << it->cpu << " " << it->iowait << ",";
          }
          return stream.str();
        }

        std::string getAdminClients(void){
          std::string json;
          AdminClientsSerializator adminClientsSerializator(*_queue, _purgeCounter, _refreshNotFoundCounter);
          adminClientsSerializator.serialize(json);
          return json;
        }

        void resetFrequencies(void){
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            it->count += it->frequency;
            it->frequency = 0;
          }
        }

        void incrementFrequency(const std::string &identity){
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              ++(it->frequency);
              break;
            }
          }
        }

        std::string getProperty(const std::string &identity){
          std::string result;
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              result = it->property;
              break;
            }
          }
          return result;
        }

        ResourceUsageCollection getResources(const std::string &identity){
          ResourceUsageCollection result;
          for(std::vector<ClientWorkerItem>::iterator it = _queue->begin(); it < _queue->end(); ++it){
            if(it->identity.compare(identity) == 0){
              result = it->resources;
              break;
            }
          }
          return result;
        }

        unsigned int getPurgeMode(void) const{
          return _purgeMode;
        }

        void setPurgeMode(const unsigned int purgeMode){
          _purgeMode = purgeMode;
        }

        unsigned int getHeartbeatTimeout(void) const{
          return _heartbeatTimeout;
        }

        void setHeartbeatTimeout(unsigned int heartbeatTimeout){
          _heartbeatTimeout = heartbeatTimeout;
        }

        unsigned int getMaxResourcesUsageCollectedSize(void) const{
          return _maxResourcesUsageCollectedSize;
        }

        void setMaxResourcesUsageCollectedSize(unsigned int maxResourcesUsageCollectedSize){
          _maxResourcesUsageCollectedSize = maxResourcesUsageCollectedSize;
        }

      private:
        std::vector<ClientWorkerItem>* _queue;
        unsigned long long _refreshNotFoundCounter;
        unsigned int _purgeMode;
        unsigned long long _purgeCounter;
        unsigned int _heartbeatTimeout;

        unsigned int _maxResourcesUsageCollectedSize;
        HCE::handlers::ResourceUsageManager* _resourceUsageManager;
    };
  }
}

#endif /* __HCE_CLIENTS_QUEUE_MANAGER_INCLUDED__ */