SplitJoinJP.h

Go to the documentation of this file.

 
//
// Created by tony on 18/03/22.
//
 
#ifndef _JOINPROCESSOR_SPLITJOINJP_H_
#define _JOINPROCESSOR_SPLITJOINJP_H_
#include <thread>
#include <Common/Types.h>
#include <Utils/UtilityFunctions.hpp>
#include <barrier>
 
#include <JoinProcessor/AbstractJP.h>
using namespace INTELLI;
using namespace std;
namespace INTELLI {
class SplitJoinJP : public AbstractJP {
 protected:
  virtual void inlineMain();
  TuplePtrQueue TuplePtrQueueLocalS;
  TuplePtrQueue TuplePtrQueueLocalR;
  /*
   *  @brief A 'current window' copy of TuplePtrQueueLocalS when join a tuple
   */
  //C20Buffer<TuplePtr> windowS;
  /*
   *  @brief A 'current window' copy of TuplePtrQueueLocalR when join a tuple
   */
  // C20Buffer<TuplePtr> windowR;
  size_t sCnt = 0;
  size_t sMax = 0;
  void expireS(size_t cond);
  void expireR(size_t cond);
  void joinS(TuplePtr ts);
  void joinR(TuplePtr tr);
 public:
  SplitJoinJP() {
 
  }
  ~SplitJoinJP() {
 
  }
  void init(size_t sLen, size_t rLen, size_t _sysId) {
    AbstractJP::init(sLen, rLen, _sysId);
    TuplePtrQueueLocalS = newTuplePtrQueue(sLen);
    TuplePtrQueueLocalR = newTuplePtrQueue(rLen);
    //windowS = C20Buffer<TuplePtr>(sLen);
    // windowR = C20Buffer<TuplePtr>(rLen);
    sCnt = 0;
  }
  void setMaxSCnt(size_t ms) {
    sMax = ms;
  }
  /*
   * @brief feed a tuple s into the s input queue
   * @param ts The tuple
   *//*
  virtual void feedTupleS(TuplePtr ts) {
    sCnt++;
    if (sCnt == sysId + 1) //should process this S
    {
      TuplePtrQueueInS->push(ts);
    }
    if (sCnt == sMax) {
      sCnt = 0;
    }
  }*/
};
typedef std::shared_ptr<SplitJoinJP> SplitJoinJPPtr;
}
#endif //ALIANCEDB_INCLUDE_JOINPROCESSOR_SPLITJP_H_