AbstractWS.h

Go to the documentation of this file.

 
#ifndef _WINDOWSLIDER_ABSTRACTWS_H_
#define _WINDOWSLIDER_ABSTRACTWS_H_
#include <cstdint>
#include <vector>
#include <Common/Types.h>
#include <Utils/SPSCQueue.hpp>
#include <Utils/UtilityFunctions.hpp>
#include <time.h>
#include <numeric>
using namespace INTELLI;
using namespace std;
namespace INTELLI {
class AbstractWS {
 
 protected:
 
  INTELLI::join_type_t myType = INTELLI::CNT_BASED;
  size_t countS, countR;
  size_t windowLen = 0;
  size_t slideLen = 1;
  //partition
  std::vector<size_t> partitionWeight;
  std::vector<size_t> partitionSizeFinal;
  size_t threads;
  size_t sLen, rLen;
  bool runTimeScheduling = false;
  bool timeBased = false;
  bool isRunning = false;
  struct timeval timeSys;  
  TuplePtrQueue TuplePtrQueueInS;
  TuplePtrQueue TuplePtrQueueInR;
  string nameTag;
  size_t oldestWindowBelong(size_t ts) {
    if (ts < windowLen) {
      return 0;
    }
    return ((ts - windowLen) / slideLen) + 1;
  }
 public:
 
  void setTimeBased(bool ts) {
    timeBased = ts;
  }
  bool isTimeBased() {
    return timeBased;
  }
 
  void setRunTimeScheduling(bool r) {
    runTimeScheduling = r;
  }
  bool isRunTimeScheduling() {
    return runTimeScheduling;
  }
  void setWindowLen(size_t wl) {
    windowLen = wl;
  }
  void setSlideLen(size_t sli) {
    slideLen = sli;
  }
  size_t getSlideLen() {
    return slideLen;
  }
  void setParallelSMP(size_t threads) {
    partitionWeight = std::vector<size_t>(threads);
    for (size_t i = 0; i < threads; i++) {
      partitionWeight[i] = 1;
    }
  }
  void reset() {
    countS = 0;
    countR = 0;
 
    gettimeofday(&timeSys, NULL);
  }
 
  AbstractWS() {
    reset();
  }
  size_t getTimeStamp() {
 
    return UtilityFunctions::timeLastUs(timeSys) / TIME_STEP;
  }
  //init with length of queue
  AbstractWS(size_t sLen, size_t rLen);
  //feed the tuple S
  virtual void feedTupleS(TuplePtr ts) {
    //assert(ts);
    TuplePtrQueueInS->push(ts);
  }
  //feed the tuple R
  virtual void feedTupleR(TuplePtr tr) {
    //assert(tr);
    TuplePtrQueueInR->push(tr);
  }
  ~AbstractWS();
 
  //init the join processors
  virtual void initJoinProcessors() {
 
  }
  virtual void terminateJoinProcessors() {
 
  }
  virtual void waitAckFromJoinProcessors() {
 
  }
  //get the join result
  virtual size_t getJoinResult() {
    return 0;
  }
  struct timeval getSysTime() {
    return timeSys;
  }
  void setSysTime(struct timeval tv) {
    timeSys = tv;
  }
  string getName() {
    return nameTag;
  }
};
}
#endif