RuleManager.hpp

Go to the documentation of this file.

//
// Created by tjb3 on 2/22/17.
//

#ifndef HTGS_RULEMANAGER_HPP
#define HTGS_RULEMANAGER_HPP

#include <htgs/api/Bookkeeper.hpp>
#include <htgs/api/IRule.hpp>
#include <htgs/core/graph/Connector.hpp>
#include <htgs/core/rules/AnyRuleManagerInOnly.hpp>

namespace htgs {

template<class V, class W>
class IRule;

template<class T, class U>
class RuleManager : public AnyRuleManagerInOnly<T> {
  static_assert(std::is_base_of<IData, T>::value, "T must derive from IData");
  static_assert(std::is_base_of<IData, U>::value, "U must derive from IData");

 public:

  RuleManager(std::shared_ptr<htgs::IRule<T, U>> rule)// TODO: Delete or Add #ifdef, TaskGraphCommunicator *communicator)
      : rule(rule), /*TODO: Delete or Add #ifdef communicator(communicator),*/ pipelineId(0), numPipelines(1), terminated(false) {}

  virtual ~RuleManager() override {}

  void executeTask(std::shared_ptr<T> data) override {

    if (this->rule->canUseLocks()) {
      this->rule->getMutex().lock();
    }

    // Check if the rule is expecting data or not
    checkRuleTermination();

    HTGS_DEBUG_VERBOSE("Rule: " << rule->getName() << " consuming data: " << data);
    auto result = rule->applyRuleFunction(data, pipelineId);

    if (result != nullptr && result->size() > 0) {
      if (this->connector != nullptr) {
#ifdef WS_PROFILE
        sendWSProfileUpdate(this, StatusCode::ACTIVATE_EDGE);
#endif
        this->connector->produceData(result);
      }
    }


    // Check if the rule is ready to be terminated after processing data (in case no more data
    checkRuleTermination();

    if (this->rule->canUseLocks()) {
      this->rule->getMutex().unlock();
    }
  }

  RuleManager<T, U> *copy() override {
    return new RuleManager<T, U>(this->rule); // TODO: Delete or Add #ifdef , this->communicator);
  }

  std::string getName(int flags = 0) override {
    std::string inOutLabel = (((flags & DOTGEN_FLAG_SHOW_IN_OUT_TYPES) != 0) ? ("\\nout: " + this->outTypeName()) : "");

    return this->rule->getName() + inOutLabel;
  }

  void debug() override {
    HTGS_DEBUG(this->getName() << " output connector: " << this->connector);
  }

  std::shared_ptr<AnyConnector> getConnector() override {
    return this->connector;
  }

  void initialize(size_t pipelineId, size_t numPipelines, std::string address) override {
    HTGS_DEBUG_VERBOSE("Initialized " << this->getName() << " pipeline id: " << pipelineId);
    this->pipelineId = pipelineId;
    this->numPipelines = numPipelines;
    this->address = address;
  }

  void shutdown() override {
    HTGS_DEBUG("Shutting down " << this->getName() << " pipeline id: " << pipelineId);

    // Check if the rule manager was terminated by it's rule
    if (!this->terminated) {

      // Close any active connections
      HTGS_DEBUG("Waking up connector");
      this->connector->producerFinished();
      this->connector->wakeupConsumer();

#ifdef WS_PROFILE
      sendWSProfileUpdate(this->connector.get(), StatusCode::DECREMENT);
#endif
    }

    // Shutdown the rule's pipeline ID
    rule->shutdownRule(this->pipelineId);
  }

  bool isTerminated() override {
    return terminated;
  }

  void setOutputConnector(std::shared_ptr<AnyConnector> connector) override {
    this->connector = std::static_pointer_cast<Connector<U>>(connector);
    HTGS_DEBUG_VERBOSE("Connector " << this->connector << " adding producer: " << this->getName() << " " << this <<
                               " to connector " << this->connector);
  }

  void checkRuleTermination() override {
    if (!terminated) {
      // Check if the rule is ready to be terminated before and after processing data
      if (rule->canTerminateRule(pipelineId)) {
        terminated = true;
        this->connector->producerFinished();
        if (this->connector->isInputTerminated()) {
          this->connector->wakeupConsumer();
        }
#ifdef WS_PROFILE
        sendWSProfileUpdate(this->connector.get(), StatusCode::DECREMENT);
#endif
      }
    }
  }

 private:

#ifdef WS_PROFILE
  void sendWSProfileUpdate(void *addr, StatusCode code)
  {
    if (this->getName() == "WebSocketProfiler")
      return;
    std::shared_ptr<ProfileData> updateStatus(new ChangeStatusProfile(addr, code));
    std::shared_ptr<DataPacket> dataPacket(new DataPacket(this->getName(), "", "WebSocketProfiler", "0", updateStatus));
    this->communicator->produceDataPacket(dataPacket);
  }
#endif


  std::string inTypeName() {
#if defined( __GLIBCXX__ ) || defined( __GLIBCPP__ )
    int status;
    char *realName = abi::__cxa_demangle(typeid(T).name(), 0, 0, &status);
    std::string ret(realName);

    free(realName);

    return ret;
#else
    return typeid(T).name();
#endif

  }

  std::string outTypeName() {
#if defined( __GLIBCXX__ ) || defined( __GLIBCPP__ )
    int status;
    char *realName = abi::__cxa_demangle(typeid(U).name(), 0, 0, &status);
    std::string ret(realName);

    free(realName);

    return ret;
#else
    return typeid(U).name();
#endif

  }


  std::shared_ptr<IRule<T, U>> rule; 
  // TODO: Delete or Add #ifdef
  // TaskGraphCommunicator *communicator; //!< The task graph communicator
  size_t pipelineId; 
  size_t numPipelines; 
  std::string address; 
  std::shared_ptr<htgs::Connector<U>> connector; 
  volatile bool terminated; 

};

}

#endif //HTGS_RULEMANAGER_HPP