AbstractJP.h

 
//
// Created by tony on 12/03/22.
//
 
#ifndef _JOINPROCESSOR_ABSTRACTJP_H_
#define _JOINPROCESSOR_ABSTRACTJP_H_
#include <Utils/UtilityFunctions.hpp>
#include <Common/Types.h>
#include <barrier>
#include <Utils/AbstractC20Thread.h>
#include <Utils/C20Buffers.hpp>
#include <JoinAlgo/JoinAlgoTable.h>
#include <memory>
namespace INTELLI {
class AbstractLazyJP;
class AbstractJP : public AbstractC20Thread {
 protected:
  TuplePtrQueue TuplePtrQueueInS;
  TuplePtrQueue TuplePtrQueueInR;
  // cmd linked to window slider
  CmdQueuePtr cmdQueueIn;
  CmdQueuePtr cmdQueueOut;
  size_t sysId;
  struct timeval timeSys;  
  bool timeBased = false;
  size_t windowLen = 0;
  size_t windowLenGlobal = 0;
  size_t slideLenGlobal = 0;
  size_t joinedResult = 0;
  JoinAlgoTablePtr myAlgo;
  //response of my self
  void sendResponseCmd(join_cmd_t cmd) {
    cmdQueueOut->push(cmd);
  }
  void sendAck() {
    sendResponseCmd(CMD_ACK);
  }
  bool testCmd(join_cmd_t cmd) {
    if (!cmdQueueIn->empty()) {
      join_cmd_t cmdIn = *cmdQueueIn->front();
      cmdQueueIn->pop();
      if (cmdIn == cmd) {
        return true;
      }
    }
    return false;
  }
  virtual void inlineMain() {
 
  }
  size_t oldestWindowBelong(size_t ts) {
    if (ts < windowLenGlobal) {
      return 0;
    }
    return ((ts - windowLenGlobal) / slideLenGlobal) + 1;
  }
 public:
  AbstractJP() {}
  ~ AbstractJP() {}
  void inputCmd(join_cmd_t cmd) {
    cmdQueueIn->push(cmd);
  }
  join_cmd_t waitResponse() {
    while (cmdQueueOut->empty()) {}
    join_cmd_t ru = *cmdQueueOut->front();
    cmdQueueOut->pop();
    return ru;
  }
  virtual void init(size_t sLen, size_t rLen, size_t _sysId) {
    TuplePtrQueueInS = newTuplePtrQueue(sLen);
    TuplePtrQueueInR = newTuplePtrQueue(rLen);
    cmdQueueIn = newCmdQueue(1);
    cmdQueueOut = newCmdQueue(1);
    sysId = _sysId;
    myAlgo = newJoinAlgoTable();
  }
  void setTimeBased(bool ts) {
    timeBased = ts;
  }
  bool isTimeBased() {
    return timeBased;
  }
  void setWindowLen(size_t wl) {
    windowLen = wl;
  }
  void setGlobalWindow(size_t wlen, size_t sli) {
    windowLenGlobal = wlen;
    slideLenGlobal = sli;
  }
  virtual void feedTupleS(TuplePtr ts) {
    TuplePtrQueueInS->push(ts);
  }
  virtual void feedTupleR(TuplePtr tr) {
    TuplePtrQueueInR->push(tr);
  }
  void setTimeVal(struct timeval tv) {
    timeSys = tv;
  }
  size_t getTimeStamp() {
    return UtilityFunctions::timeLastUs(timeSys) / TIME_STEP;
  }
  size_t getJoinedResult() {
    return joinedResult;
  }
  int cpuBind = -1;
  void setCore(int id) {
    cpuBind = id;
  }
};
 
}
#endif //ALIANCEDB_SRC_JOINPROCESSOR_ABSTRACTJP_H_