DTP.cc

Go to the documentation of this file.

// The MIT License (MIT)
//
// Copyright (c) 2014-2016 Brno University of Technology, PRISTINE project
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

#include "DTP.h"

const char * SIG_STAT_DTP_RTT = "DTP_RTT";
const char * SIG_STAT_DTP_CLOSED_WIN_Q = "DTP_CLOSED_WIN_Q";
const char * DTP_SEQ_NUM_RCVD = "DTP_SEQ_NUM_RCVD";
const char * DTP_SEQ_NUM_SENT = "DTP_SEQ_NUM_SENT";

Define_Module(DTP);

DTP::DTP()
{
  deletePdu = 0;

  rcvrInactivityTimer = nullptr;
  senderInactivityTimer = nullptr;


  rcvrInactivityPolicy = nullptr;
  senderInactivityPolicy = nullptr;
  initialSeqNumPolicy = nullptr;
  rttEstimatorPolicy = nullptr;

  state = nullptr;

  pduDroppingEnabled = false;

  sduSeqNum = 0;
  //intentionally left out callInitialize

}
DTP::~DTP()
{

  flushAllQueuesAndPrepareToDie();
//  delete state;
  if (state->isDtcpPresent())
  {

  }

}

cModule* DTP::createPolicyModule(const char* prefix, const char* name)
{
  if (std::string(par(name).stringValue()).empty())
  {
    return nullptr;
  }
  else
  {
    std::stringstream moduleName;
    moduleName << prefix << par(name).stringValue();
    cModuleType* policyType = cModuleType::get(moduleName.str().c_str());
    return policyType->createScheduleInit(name, getParentModule());
  }
}

void DTP::runRTTEstimatorPolicy()
{
  Enter_Method
  ("RTTEstimatorPolicy");
  rttEstimatorPolicy->call(state, dtcp->getDTCPState());

  emit(sigStatDTPRTT, state->getRtt());

}

void DTP::runCongestionNotificationPolicy()
{
  dtcp->runECNSlowDownPolicy(state);
}

void DTP::handleQueueInfoFromSocket(QueueInfo* queueInfo)
{
  handleQueueInfo(queueInfo);
}

void DTP::handleQueueInfo(QueueInfo* queueInfo)
{
}

void DTP::initSignalsAndListeners()
{
  //Signals that this module is emmiting
  sigEFCPStahpSending = registerSignal(SIG_EFCP_StopSending);
  sigEFCPStartSending = registerSignal(SIG_EFCP_StartSending);
  sigStatDTPRTT = registerSignal(SIG_STAT_DTP_RTT);
  sigStatDTPClosedWinQ = registerSignal(SIG_STAT_DTP_CLOSED_WIN_Q);
//  sigStatDTPRxCount   = registerSignal(SIG_STAT_DTP_RX_SENT);
  sigStatDTPRecvSeqNum    = registerSignal(DTP_SEQ_NUM_RCVD);
  sigStatDTPSentSeqNum    = registerSignal(DTP_SEQ_NUM_SENT);
}

void DTP::initialize(int step)
{

  if (step == 0)
  {

    initGates();
    initialSeqNumPolicy = getRINAModule<InitialSeqNumPolicyBase*>(this, 1, {INITIAL_SEQ_NUM_POLICY_NAME});
    rcvrInactivityPolicy = getRINAModule<RcvrInactivityPolicyBase*>(this, 1, {RCVR_INACTIVITY_POLICY_NAME});
    senderInactivityPolicy = getRINAModule<SenderInactivityPolicyBase*>(this, 1, {SENDER_INACTIVITY_POLICY_NAME});
    rttEstimatorPolicy = getRINAModule<RTTEstimatorPolicyBase*>(this, 1, {RTT_ESTIMATOR_POLICY_NAME});

    senderInactivityTimer = new SenderInactivityTimer();
    rcvrInactivityTimer = new RcvrInactivityTimer();


    rendezvousEnabled = getModuleByPath(".^.^.efcp")->par("rendezvousEnabled").boolValue();

  }
  else if (step == 1)
  {

    initSignalsAndListeners();

//    if(state->isDtcpPresent()){

//    }else{
//      senderInactivityTimer = nullptr;
//      rcvrInactivityTimer = nullptr;
//    }

//  par(RCVR_INACTIVITY_POLICY_NAME).setStringValue(getModuleByPath((std::string(".^.^.") + std::string(MOD_EFCP)).c_str())->par(RCVR_INACTIVITY_POLICY_NAME).stringValue());
//  par(SENDER_INACTIVITY_POLICY_NAME).setStringValue(getModuleByPath((std::string(".^.^.") + std::string(MOD_EFCP)).c_str())->par(SENDER_INACTIVITY_POLICY_NAME).stringValue());
//  par(INITIAL_SEQ_NUM_POLICY_NAME).setStringValue(getModuleByPath((std::string(".^.^.") + std::string(MOD_EFCP)).c_str())->par(INITIAL_SEQ_NUM_POLICY_NAME).stringValue());
//  par(RTT_ESTIMATOR_POLICY_NAME).setStringValue(getModuleByPath((std::string(".^.^.") + std::string(MOD_EFCP)).c_str())->par(RTT_ESTIMATOR_POLICY_NAME).stringValue());

//    rcvrInactivityPolicy = (RcvrInactivityPolicyBase*) createPolicyModule(RCVR_INACTIVITY_POLICY_PREFIX,
//        RCVR_INACTIVITY_POLICY_NAME);
//    senderInactivityPolicy = (SenderInactivityPolicyBase*) createPolicyModule(SENDER_INACTIVITY_POLICY_PREFIX,
//        SENDER_INACTIVITY_POLICY_NAME);
//    initialSeqNumPolicy = (InitialSeqNumPolicyBase*) createPolicyModule(INITIAL_SEQ_NUM_POLICY_PREFIX,
//        INITIAL_SEQ_NUM_POLICY_NAME);
//    rttEstimatorPolicy = (RTTEstimatorPolicyBase*) createPolicyModule(RTT_ESTIMATOR_POLICY_PREFIX,
//        RTT_ESTIMATOR_POLICY_NAME);

//    runInitialSeqNumPolicy();


    bool interrupter = false;
    interrupter = getModuleByPath(".^.^.")->par("interrupter").boolValue();
    if(interrupter){

      TheInterrupterTimer* startR = new TheInterrupterTimer();
      startR->setAction(INTERRUPTER_ZERO_RCV_CREDIT);
      scheduleAt(120, startR);

      TheInterrupterTimer* stopR = new TheInterrupterTimer();
      stopR->setAction(INTERRUPTER_INFLATE_RCV_CREDIT);
      scheduleAt(140, stopR);

    }

    //set initial value for RTO based on RTT from .ned
    rttEstimatorPolicy->setRto(state->getRtt());

  }

}

void DTP::initGates()
{
  northI = this->gateHalf(GATE_DTP_NORTHIO, cGate::INPUT);
  northO = this->gateHalf(GATE_DTP_NORTHIO, cGate::OUTPUT);
  southI = this->gateHalf(GATE_DTP_SOUTHIO, cGate::INPUT);
  southO = this->gateHalf(GATE_DTP_SOUTHIO, cGate::OUTPUT);
}
const QoSCube* DTP::getQoSCube() const
{
  return state->getQoSCube();
}

void DTP::setQoSCube(const QoSCube* qosCube)
{
  //TODO A2 Make copy; ... Why?
  state->setQoSCube(qosCube);
}

void DTP::setPduDroppingEnabled(bool pduDroppingEnabled)
{
  this->pduDroppingEnabled = pduDroppingEnabled;
}

void DTP::flushAllQueuesAndPrepareToDie()
{
  cancelAndDelete(senderInactivityTimer);
  senderInactivityTimer = nullptr;
  cancelAndDelete(rcvrInactivityTimer);
  rcvrInactivityTimer = nullptr;

  for(auto it = aTimerQ.begin(); it != aTimerQ.end(); )
  {
    cancelAndDelete((*it));
    it = aTimerQ.erase(it);
  }

}

void DTP::redrawGUI()
{
  if (!getEnvir()->isGUI())
  {
    return;
  }

  if (state->isDtcpPresent())
  {
    dtcp->redrawGUI();
  }

  cDisplayString& disp = getDisplayString();
  disp.setTagArg("t", 1, "r");
  std::ostringstream desc;
  desc << "nextSeqNum: " << state->getNextSeqNumToSendWithoutIncrement() << "\n";

  desc << "rLWE: " << state->getRcvLeftWinEdge() << "\n";

  desc << "maxSeqNumRcvd: " << state->getMaxSeqNumRcvd() << "\n";

  std::vector<DataTransferPDU*>::iterator it;
  std::vector<DataTransferPDU*>* pduQ;

  pduQ = state->getReassemblyPDUQ();
  if (pduQ->empty())
  {
    desc << "reassemblyQ: empty" << "\n";
  }
  else
  {
    desc << "reassemblyQ: ";
    std::vector<DataTransferPDU*>::iterator it;
    for (it = pduQ->begin(); it != pduQ->end(); ++it)
    {
      desc << (*it)->getSeqNum() << " | ";
    }
    desc << "\n";
  }

  desc << "droppedPDU: " << state->getDropDup() << "\n";
  desc << "rtt: " << state->getRtt() << "\n";

  disp.setTagArg("t", 0, desc.str().c_str());

}

void DTP::setConnId(const ConnectionId& connId)
{
  this->connId = connId;
}

void DTP::handleRendezvousTimer(DTCPRendezvousTimer* timer)
{
//  timer->setCounter(timer->getCounter() + 1);
  //TODO A2 increment counter to indicate retransmission of RendezvousPDU
  rendezvousCondition();

}

void DTP::changeInBuffers()
{
  dtcp->updateRcvRtWinEdge(state);

  if(state->isDtcpPresent()){
    if(dtcp->dtcpState->isRcvRendez() && (dtcp->dtcpState->getRcvRightWinEdge() >= dtcp->dtcpState->getRendezSeqNum())){
      sendReliableControlPDU();
    }
  }

}

void DTP::sendReliableControlPDU()
{
  sendAckFlowPDU();
  dtcp->startReliableCPDUTimer();
}

void DTP::handleInterrupterTimer(TheInterrupterTimer* interrupter)
{
  switch(interrupter->getAction()){
    case INTERRUPTER_ZERO_RCV_CREDIT:{
      dtcp->getDTCPState()->setRcvCredit(0);
      break;
    }
    case INTERRUPTER_INFLATE_RCV_CREDIT:{
      dtcp->getDTCPState()->setRcvCredit(10);
      changeInBuffers();


    }
  }

  delete interrupter;
}

void DTP::handleMessage(cMessage *msg)
{
  if (msg->isSelfMessage())
  {
    /* Timers */
    DTPTimers* timer = static_cast<DTPTimers*>(msg);
    switch (timer->getType())
    {
      case (DTP_RCVR_INACTIVITY_TIMER): {
        handleDTPRcvrInactivityTimer(static_cast<RcvrInactivityTimer*>(timer));
        break;
      }

      case (DTP_SENDER_INACTIVITY_TIMER): {
        handleDTPSenderInactivityTimer(static_cast<SenderInactivityTimer*>(timer));
        break;
      }

      case (DTP_A_TIMER): {
        handleDTPATimer(static_cast<ATimer*>(timer));
//        delete msg;
        break;
      }

      case (DTP_INTERRUPTER_TIMER): {
        handleInterrupterTimer(static_cast<TheInterrupterTimer*>(msg));
        break;

      }

    }
  }
  else
  {

    /* Either PDUs from RMT or UserDataField from Delimiting */
    if (msg->arrivedOn(northI->getId()))
    {

      handleMsgFromUp(static_cast<UserDataField*>(msg));

    }
    else if (msg->arrivedOn(southI->getId()))
    {
      handleMsgFromRMT((PDU*) msg);
    }
  }

  if (state->isDtcpPresent())
  {
    emit(sigStatDTPClosedWinQ, dtcp->dtcpState->getClosedWinQueLen());
//      emit(sigStatDTPRxCount, dtcp->dtcpState->getRxSent());
  }

  redrawGUI();
}

void DTP::setPDUHeader(PDU* pdu)
{
  pdu->setConnId(this->flow->getConId());
  pdu->setSrcAddr(this->flow->getSrcAddr());
  pdu->setDstAddr(this->flow->getDstAddr());
  pdu->setSrcApn(this->flow->getDstApni().getApn());
  pdu->setDstApn(this->flow->getSrcApni().getApn());
}


void DTP::generateDTPDU(UserDataField* userDataField)
{
  DataTransferPDU* dataPDU = new DataTransferPDU();
  setPDUHeader(dataPDU);
  dataPDU->setSeqNum(state->getNextSeqNumToSend());
  /* Set DRF flag in PDU */
  if (setDRFInPDU(false))
  {
    dataPDU->setFlags(dataPDU->getFlags() | DRF_FLAG);
  }
  dataPDU->encapsulate(userDataField);
  state->pushBackToGeneratedPDUQ(dataPDU);
}

void DTP::handleMsgFromUp(UserDataField* userDataField)
{
  cancelEvent(senderInactivityTimer);

//  delimit(sduData);
  userDataFieldQOut.push_back(userDataField);

  generatePDUsnew();

//  generateDTPDU(userDataField);

  trySendGenPDUs(state->getGeneratedPDUQ());

  schedule(senderInactivityTimer);
}

void DTP::addPDUToReassemblyQ(DataTransferPDU* pdu)
{
  state->addPDUToReassemblyQ(pdu);
}

void DTP::delimitFromRMT()
{
  Enter_Method_Silent();
  PDUQ_t* pduQ = state->getReassemblyPDUQ();

  for (auto it = pduQ->begin();
          it != pduQ->end() && (*it)->getSeqNum() <= state->getRcvLeftWinEdge();)
  {/* DO NOT FORGET TO PUT '++it' in all cases where we DO NOT erase PDUs from queue */
    send((*it)->decapsulate(), northO);
    delete (*it);
    it = pduQ->erase(it);
  }

}

bool DTP::commonRcvControl(ControlPDU* pdu)
{
  if (pdu->getSeqNum() <= dtcp->getLastCtrlSeqNumRcv())/* Duplicate ControlPDU */
  {
    if ((pdu->getType() & PDU_ACK_BIT) == PDU_ACK_BIT)
    {
      dtcp->incDupAcks();
    }
    if ((pdu->getType() & PDU_FC_BIT) == PDU_FC_BIT)
    {
      dtcp->incDupFC();
    }
    delete pdu;
    return false;
  }
  else if (pdu->getSeqNum() > dtcp->getLastCtrlSeqNumRcv() + 1)/* Out of order */
  {
    /* LostControlPDU Policy */
    dtcp->runLostControlPDUPolicy(state);
  }
  else
  {
    dtcp->setLastCtrlSeqnumRcvd(pdu->getSeqNum());
  }

  return true;
}

void DTP::sendFCOnlyPDU()
{
  Enter_Method_Silent
  ();

  FlowControlOnlyPDU* fcOnlyPdu = new FlowControlOnlyPDU();

  setPDUHeader(fcOnlyPdu);

  fcOnlyPdu->setSeqNum(dtcp->getNextSndCtrlSeqNum());

  fillFlowControlPDU(fcOnlyPdu);

  sendToRMT(fcOnlyPdu);
}

void DTP::fillFlowControlPDU(FlowControlPDU* flowControlPdu)
{

  flowControlPdu->setRcvRightWinEdge(dtcp->getRcvRightWinEdge());

  flowControlPdu->setRcvRate(dtcp->getRcvrRate());

  flowControlPdu->setTimeUnit(dtcp->getSendingTimeUnit());

  flowControlPdu->setSndLeftWinEdge(dtcp->getSndLeftWinEdge());

  flowControlPdu->setSndRightWinEdge(dtcp->getSndRtWinEdge());

  flowControlPdu->setSndRate(dtcp->getSendingRate());

  dtcp->getDTCPState()->setRcvRightWinEdgeSent(dtcp->getDTCPState()->getRcvRightWinEdge());
}

void DTP::sendAckFlowPDU(unsigned int seqNum, bool seqNumValid)
{
  Enter_Method_Silent();

  if(!state->isDtcpPresent()){
    return;
  }

  if (!seqNumValid)
  {
    seqNum = state->getRcvLeftWinEdge();
  }


  if (dtcp->dtcpState->isFCPresent() && dtcp->dtcpState->isRxPresent())
  {
    // Send Ack/Flow Control PDU with LWE and RWE
    AckFlowPDU* ackFlowPdu = new AckFlowPDU();
    setPDUHeader(ackFlowPdu);
    ackFlowPdu->setSeqNum(dtcp->getNextSndCtrlSeqNum());

    ackFlowPdu->setAckNackSeqNum(seqNum);

    fillFlowControlPDU(ackFlowPdu);
    sendToRMT(ackFlowPdu);
  }
  else if (dtcp->dtcpState->isFCPresent())
  {
    sendFCOnlyPDU();
  }
  else if (dtcp->dtcpState->isRxPresent())
  {
    sendAckOnlyPDU(seqNum);
  }

}

void DTP::handleControlPDUFromRMT(ControlPDU* pdu)
{

  if (!commonRcvControl(pdu))
  {
    //The Control PDU has been deleted because of duplicate
    return;
  }

  if (pdu->getType() == CONTROL_ACK_PDU)
  {
    dtcp->runRcvrControlAckPolicy(state);
    if(dtcp->getDTCPState()->isSndRendez()){
      ControlAckPDU* controlAckPDU = static_cast<ControlAckPDU*>(pdu);
      if(dtcp->getDTCPState()->getRendezvousTimer()->getSeqNum() == controlAckPDU->getLastCtrlSeqNumRcv()){
        cancelAndDelete(dtcp->getDTCPState()->getRendezvousTimer());
        dtcp->getDTCPState()->setRendezvousTimer(nullptr);
      }


    }

  }
  else if (pdu->getType() == RENDEZVOUS_PDU)
  {
    RendezvousPDU* rendezPDU = (RendezvousPDU*) pdu;
    dtcp->dtcpState->setRendezSeqNum(rendezPDU->getRendezSeqNum());
//    if(rendezPDU->getLastCtrlSeqNumRcv() == dtcp->dtcpState->getLastControlSeqNumSent()){
      /* The sender sent this RendezvousPDU based on up-to-date information
       * so this C-PDU can be assumed valid.
       */
      dtcp->dtcpState->setRcvRendez(true);
      sendControlAckPDU();
//    }
  }
  else
  {

    if (pdu->getType() & PDU_SEL_BIT)
    {
      /*
        SELECT_ACK_PDU        = 0xC9;
        SELECT_NACK_PDU       = 0xCA;
        SELECT_ACK_FLOW_PDU   = 0xCD;
        SELECT_NACK_FLOW_PDU  = 0xCE;
       */

      //RTT estimator
      runRTTEstimatorPolicy();

      if (pdu->getType() & PDU_NACK_BIT)
      {
        SelectiveNackPDU* selNackPdu = (SelectiveNackPDU*) pdu;

        for (unsigned int i = 0; i < selNackPdu->getNackListLen(); i++)
        {
          unsigned int startSeqNum = selNackPdu->getNackList(i * 2);
          unsigned int endSeqNum = selNackPdu->getNackList((i * 2) + 1);

          dtcp->nackPDU(startSeqNum, endSeqNum);

        }
      }
      else if (pdu->getType() & PDU_ACK_BIT)
      {

        //TODO B1 Shoudn't this be a policy?
        SelectiveAckPDU* selAckPdu = (SelectiveAckPDU*) pdu;
        unsigned int tempSLWE = 0;
        for (unsigned int i = 0; i < selAckPdu->getNackListLen(); i++)
        {
          unsigned int startSeqNum = selAckPdu->getNackList(i * 2);
          unsigned int endSeqNum = selAckPdu->getNackList((i * 2) + 1);

          tempSLWE = std::max(endSeqNum, tempSLWE);

          dtcp->ackPDU(startSeqNum, endSeqNum);

          // min = dtcp->getSmallestSeqNumFromRxQOrNextSeqNumToSend - 1
          //state->setSenderLeftWinEdge(std::min(min, state->getSndLeftWinEdge);

          //            state->setSenderLeftWinEdge(tempSLWE);
          //TODO B2 O'really? Shouldn't it always be nextSeqNum -1?
          dtcp->updateSenderLWE(tempSLWE + 1);

        }
      }

    }
    else

    if (pdu->getType() & (PDU_ACK_BIT | PDU_NACK_BIT))
    {

      //Retrieve the Time of this Ack - RTT estimator policy
      runRTTEstimatorPolicy();

      if (pdu->getType() & PDU_NACK_BIT)
      {

        NackOnlyPDU *nackPdu = (NackOnlyPDU*) pdu;

        dtcp->nackPDU(nackPdu->getAckNackSeqNum());

      }
      else if (pdu->getType() & PDU_ACK_BIT)
      {

        AckOnlyPDU *ackPdu = (AckOnlyPDU*) pdu;
        EV << getFullPath() << ": PDU number: " << ackPdu->getAckNackSeqNum() << " Acked" << endl;

        dtcp->runSenderAckPolicy(state);

      }
      /* End of Ack/Nack */

    }

    if ((pdu->getType() & PDU_FC_BIT))
    {
      FlowControlOnlyPDU *flowPdu = (FlowControlOnlyPDU*) pdu;

      //Update RightWindowEdge and SendingRate.
      dtcp->setSndRtWinEdge(flowPdu->getRcvRightWinEdge());
      dtcp->setSendingRate(flowPdu->getRcvRate());

      if (dtcp->getDTCPState()->getClosedWinQueLen() > 0)
      {
        /* Note: The ClosedWindow flag could get set back to true immediately in trySendGenPDUs */
        dtcp->getDTCPState()->setClosedWindow(false);
        trySendGenPDUs(dtcp->getDTCPState()->getClosedWindowQ());

        if (dtcp->dtcpState->isWinBased())
        {
          if (!dtcp->isClosedWinQClosed())
          {
            notifyStartSending();
          }

        }
      }

      rendezvousCondition();
    }

  }
  delete pdu;
}

void DTP::handleMsgFromRMT(PDU* msg)
{

  state->setCurrentPdu(msg);

  if (dynamic_cast<DataTransferPDU*>(msg))
  {
    DataTransferPDU* pdu = (DataTransferPDU*) msg;

    handleDataTransferPDUFromRMT(pdu);

  }
  else if (dynamic_cast<ControlPDU*>(msg))
  {
    ControlPDU* pdu = (ControlPDU*) msg;
    EV << getFullPath() << ": Control PDU number: " << pdu->getSeqNum() << " received" << endl;

    //Putting it before ControlAckPDU might be an issue

    handleControlPDUFromRMT(pdu);
  }
  else
  {

    throw cRuntimeError("Unexptected PDU Type");
  }

  state->setCurrentPdu(nullptr);
}



bool DTP::isATimerQEmpty()
{

  return aTimerQ.empty();
}


void DTP::cancelATimer(unsigned int seqNum)
{
  Enter_Method_Silent();
  for (auto it = aTimerQ.begin(); it != aTimerQ.end() && (*it)->getSeqNum() <= seqNum;)
  {
    cancelAndDelete((*it));
    it = aTimerQ.erase(it);
  }


}

void DTP::startATimer(unsigned int seqNum)
{

  if(!(state->getQoSCube()->getATime() > 0.0))
  {
    return;
  }
  //TODO A1 Make a vector in DTPState of all active ATimers.
  Enter_Method_Silent();
  ATimer* aTimer = new ATimer();
  aTimer->setSeqNum(seqNum);
  schedule(aTimer);



  auto it = aTimerQ.begin();
  for(; it != aTimerQ.end(); ++it)
  {
    if((*it)->getSeqNum() > seqNum)
    {
      break;
    }
  }
  aTimerQ.insert(it, aTimer);


}

bool DTP::isDuplicate(unsigned int seqNum)
{
  /* Not a true duplicate. (Might be a duplicate among the gaps) */
  bool dup = false;
  PDUQ_t::iterator it;
  PDUQ_t* pduQ = state->getReassemblyPDUQ();
  for (it = pduQ->begin(); it != pduQ->end(); ++it)
  {
    if ((*it)->getSeqNum() == seqNum)
    {
      dup = true;
      break;
    }
  }
  return dup;
}

double DTP::getATime()
{
  return state->getQoSCube()->getATime();
}

void DTP::handleDataTransferPDUFromRMT(DataTransferPDU* pdu)
{

  EV << getFullPath() << ": PDU number: " << pdu->getSeqNum() << " received" << endl;

  emit(sigStatDTPRecvSeqNum, pdu->getSeqNum());

  cancelEvent(rcvrInactivityTimer);
  /*
   * What to do when you want to discard the incomming PDU?
   * How to handle both cases - you want to keep it for further processing/ you want to drop it.
   */
  if (state->isDtcpPresent())
  {
    if (dtcp->dtcpState->isWinBased())
    {
      if (pdu->getSeqNum() > dtcp->getRcvRightWinEdge())
      {
        schedule(rcvrInactivityTimer);
        dtcp->runRcvFCOverrunPolicy(state);
        if (state->getCurrentPdu() == nullptr)
        {
          return;
        }
      }
    }
    else if (dtcp->dtcpState->isRateBased())
    {
      if (dtcp->getPdusRcvdInTimeUnit() + 1 > dtcp->getRcvrRate())
      {
        schedule(rcvrInactivityTimer);
        dtcp->runRcvFCOverrunPolicy(state);
        if (state->getCurrentPdu() == nullptr)
        {
          return;
        }
      }
    }

    if (dtcp->dtcpState->isFCPresent())
    {
      //    dtcp->resetWindowTimer();

      /* Run ECN policy */
      dtcp->runECNPolicy(state);

    }
  }

  // if PDU.DRF == true
  if (pdu->getFlags() & DRF_FLAG)
  {
    bubble("Received PDU with DRF set");
    /* Case 1) DRF is set - either first PDU or new run */

    delimitFromRMT();

    //Flush the PDUReassemblyQueue
    flushReassemblyPDUQ();

    state->setMaxSeqNumRcvd(pdu->getSeqNum());
    state->setRcvLeftWinEdge(state->getMaxSeqNumRcvd());

    //Put PDU on ReassemblyQ
    addPDUToReassemblyQ(pdu);

    /* I guess i did not think this through (will be removed) */
//    if (getATime() > 0)
//    {
//      startATimer(pdu->getSeqNum());
//    }
//    else
//    {
//      svUpdate(pdu->getSeqNum());
//    }

    if (state->isDtcpPresent())
    {
      svUpdate(pdu->getSeqNum());

    }
    else
    {
      state->setRcvLeftWinEdge(pdu->getSeqNum());
      //TODO A2: No A-Timer?
    }
    delimitFromRMT();

    schedule(rcvrInactivityTimer);

  }
  else
  {

    /* Not the start of a run */
    if (pdu->getSeqNum() <= state->getRcvLeftWinEdge() || pdu->getSeqNum() == state->getMaxSeqNumRcvd())
    {
      bubble("Dropping duplicate PDU");
      EV << getFullPath() << ":Duplicated PDU number: " << pdu->getSeqNum() << " received - DROPPING!" << endl;
      /* Case 2) A Real Duplicate */
      //Discard PDU and increment counter of dropped duplicates PDU
      delete pdu;

      state->incDropDup();

      sendAckFlowPDU();

      return;
    }
    if (state->getRcvLeftWinEdge() < pdu->getSeqNum() && pdu->getSeqNum() < state->getMaxSeqNumRcvd())
    {
      /* Not a true duplicate. (Might be a duplicate among the gaps) */
      if (isDuplicate(pdu->getSeqNum()))
      {
        /* Case 3) Duplicate Among gaps */
        EV << getFullPath() << ":Duplicated PDU number: " << pdu->getSeqNum() << " received - DROPPING!" << endl;
        //Discard PDU and increment counter of dropped duplicates PDU
        delete pdu;
        //increment counter of dropped duplicates PDU
        state->incDropDup();

        //send an Ack/Flow Control PDU with current window values
        sendAckFlowPDU();
        return;
      }
      else
      {
        /* Case 3) This goes in a gap */
        /* Put at least the User-Data of the PDU with its Sequence Number on PDUReassemblyQueue in Sequence Number order */
        addPDUToReassemblyQ(pdu);

//        svUpdate(state->getMaxSeqNumRcvd()); /* Update left edge, etc */

        if (state->isDtcpPresent())
        {
          svUpdate(state->getMaxSeqNumRcvd()); /* Update left edge, etc */
        }
        else
        {
          state->setRcvLeftWinEdge(state->getMaxSeqNumRcvd());
          /* No A-Timer necessary, already running */
        }

        delimitFromRMT();
        schedule(rcvrInactivityTimer);
        return;
      }
    }
    /* Case 4) This is in order */
    if (pdu->getSeqNum() == state->getMaxSeqNumRcvd() + 1)
//    if (pdu->getSeqNum() == state->getRcvLeftWinEdge())
    {
      state->incMaxSeqNumRcvd();
      addPDUToReassemblyQ(pdu);

//      if (getATime() > 0)
//      {
//        startATimer(pdu->getSeqNum());
//      }
//      else
//      {
//        svUpdate(state->getMaxSeqNumRcvd());
//      }

      if (state->isDtcpPresent())
      {
        svUpdate(state->getMaxSeqNumRcvd()); /* Update Left Edge, etc. */
      }
      else
      {
        state->setRcvLeftWinEdge(state->getMaxSeqNumRcvd());
        //start A-Timer (for this PDU)
        startATimer(pdu->getSeqNum());
      }

      delimitFromRMT();/* Create as many whole SDUs as possible */

      schedule(rcvrInactivityTimer);

    }
    else
    {
      /* Case 5) it is out of order */
      if (pdu->getSeqNum() > state->getMaxSeqNumRcvd() + 1)
      {
        state->setMaxSeqNumRcvd(pdu->getSeqNum());
        addPDUToReassemblyQ(pdu);

//        if (getATime() > 0)
//        {
//          startATimer(state->getMaxSeqNumRcvd());
//        }
//        else
//        {
//          svUpdate(state->getMaxSeqNumRcvd());
//        }

        if (state->isDtcpPresent())
        {
          svUpdate(state->getMaxSeqNumRcvd()); /* Update Left Edge, etc. */
        }
        else
        {
          startATimer(state->getMaxSeqNumRcvd());
        }

        delimitFromRMT();
        schedule(rcvrInactivityTimer);

      }
//  /* Backstop timer */
//  schedule(rcvrInactivityTimer);
    }

    //TODO C1 DIF.integrity
    /* If we are encrypting, we can't let PDU sequence numbers roll over */

    //If DIF.Integrity and PDU.SeqNum > SequenceNumberRollOverThreshhold Then
    //RequestFAICreateNewConnection( PDU.FlowID )
    //Fi
  }

}

void DTP::handleDTPRcvrInactivityTimer(RcvrInactivityTimer* timer)
{

  runRcvrInactivityTimerPolicy();

}

void DTP::handleDTPSenderInactivityTimer(SenderInactivityTimer* timer)
{
  runSenderInactivityTimerPolicy();

}

void DTP::resetSenderInactivTimer()
{
  Enter_Method_Silent();
  cancelEvent(senderInactivityTimer);
  schedule(senderInactivityTimer);
}

void DTP::handleDTPATimer(ATimer* timer)
{

  state->setTmpAtimer(timer);

  if (state->isDtcpPresent())
  {
    //    runSendingAckPolicy(timer);
    dtcp->runSendingAckPolicy(state, timer);
  }
  else
  {

    //Update RcvLeftWindowEdge
    PDUQ_t* pduQ = state->getReassemblyPDUQ();

    // Advance RLWE as much as possible
    for (auto it = pduQ->begin(); it != pduQ->end() && (*it)->getSeqNum() <= timer->getSeqNum(); ++it)
    {
      if((state->getRcvLeftWinEdge() + 1) + state->getQoSCube()->getMaxAllowGap() >= (*it)->getSeqNum())
      {
        state->setRcvLeftWinEdge((*it)->getSeqNum());
      }
      else
      {
        break;
      }
    }
    //Discard any PDUs that are bigger than RLWE but smaller than A-Timer seqNum
    for (auto it = pduQ->begin(); it != pduQ->end() && (*it)->getSeqNum() <= timer->getSeqNum();)
    {
      if((*it)->getSeqNum() > state->getRcvLeftWinEdge())
      {
        delete (*it);
        it = pduQ->erase(it);
      }

    }

    if (state->getRcvLeftWinEdge() > timer->getSeqNum())
    {
      bubble("RcvLeftWindowEdge SHOULD not be bigger than seqNum in A-Timer, right?");
      //      throw cRuntimeError("RcvLeftWindowEdge SHOULD not be bigger than seqNum in A-Timer, right?");
    }


    state->setRcvLeftWinEdge(timer->getSeqNum());

    cancelATimer(timer->getSeqNum());


    //Invoke delimiting
    delimitFromRMT();


    //reset SenderInactivity timer
    //    resetSenderInactivTimer();


  }

  state->setTmpAtimer(nullptr);
}

unsigned int DTP::delimit(SDUData* sduData)
{


  //TODO A1 update return statement
return 1;

//  //TODO B2 Does PDU header counts to MaxFlowPDUSize? edit: yes, it does!
//  if (sdu->getSize() > state->getMaxFlowPduSize())
//  {
//    unsigned int size = state->getMaxFlowPduSize();
//    //SDU is bigger than Max Allowed PDU
//    //S/DUFrag *frag = new SDUFrag
//
//    SDU* tmp;
//    unsigned int i;
//    for (i = 0; i * size < sdu->getSize(); i++)
//    {
//      tmp = sdu->genFragment(size, i, i * size);
//      dataQ.push_back(tmp);
//
//    }
//    return i;
//  }
//  else
//  {
//    dataQ.push_back(sdu);
//    return 1;
//  }

}

unsigned int DTP::delimitFromRMT(PDU *pdu, unsigned int len)
{
  unsigned int counter = 0;

//  if()

  return counter;
}

bool DTP::setDRFInPDU(bool override)
{
  if (state->isSetDrfFlag() || override)
  {
    state->setSetDrfFlag(false);
    return true;
  }
  else
  {
    return false;
  }
}

void DTP::generatePDUsnew()
{
  DataTransferPDU* baseDataPDU = new DataTransferPDU();
  setPDUHeader(baseDataPDU);

//  //invoke SDU protection so we don't have to bother with it afterwards; EDIT: sduQ is not used anymore!!!
//  for (std::vector<SDU*>::iterator it = dataQ.begin(); it != dataQ.end(); ++it)
//  {
//    sduProtection(*it);
//  }

  while (!userDataFieldQOut.empty())
  {
    DataTransferPDU* genPDU = baseDataPDU->dup();
    genPDU->setSeqNum(state->getNextSeqNumToSend());

    /* Set DRF flag in PDU */
    if (setDRFInPDU(false))
    {
      genPDU->setFlags(genPDU->getFlags() | DRF_FLAG);
    }

    genPDU->encapsulate(userDataFieldQOut.front());
    userDataFieldQOut.erase(userDataFieldQOut.begin());

    state->pushBackToGeneratedPDUQ(genPDU);
  }

  //HOTFIX
  if(!state->isDtcpPresent()){
    state->setSetDrfFlag(true);
  }

  delete baseDataPDU;
}

void DTP::fillRendezvousPDU(RendezvousPDU* rendezPDU)
{
  /* Fill in Rendezvous PDU */
  fillControlAckPDU(rendezPDU);
  if(dtcp->dtcpState->isRxPresent()){
    rendezPDU->setRendezSeqNum(dtcp->dtcpState->getSndLeftWinEdge() + 1);
  }else{
    rendezPDU->setRendezSeqNum(state->getNextSeqNumToSendWithoutIncrement());
  }
}

void DTP::sendRendezvousPDU()
{
  /* Send Rendezvous PDU */
  RendezvousPDU* rendezPDU = new RendezvousPDU();
  setPDUHeader(rendezPDU);
  /* Fill in Rendezvous PDU */
  fillRendezvousPDU(rendezPDU);
  sendToRMT(rendezPDU);
  //TODO A! restart SenderInactivity Timer
}

void DTP::rendezvousCondition()
{
  Enter_Method_Silent();

  //TODO A1 Remove after sucessfull testing
  if(!rendezvousEnabled){
    return;
  }

  /* Rendezvous condition */
  if (!dtcp->getDTCPState()->isSndRendez())
  {
//    if (((state->isDtcpPresent() && dtcp->getDTCPState()->isRxPresent() && dtcp->getDTCPState()->getRxQLen() == 0) || (!dtcp->getDTCPState()->isRxPresent())) && (dtcp->getDTCPState()->getClosedWinQueLen() != 0))
    if ((state->isDtcpPresent() && (!dtcp->getDTCPState()->isRxPresent() || dtcp->getDTCPState()->getRxQLen() == 0)) && (dtcp->getDTCPState()->getClosedWinQueLen() != 0))
    {//condition is satisfied if all DT-PDUs are Acked or Retransmission is not present; AND there is DT-PDU on closedWindowQ (we have something to send)


      if (dtcp->getDTCPState()->getSndRightWinEdge() < state->getNextSeqNumToSendWithoutIncrement())
      {
        /* Send Rendezvous PDU */
        sendRendezvousPDU();
        dtcp->getDTCPState()->setSndRendez(true);

        dtcp->startRendezvousTimer();

      }
    }
  }else{
    /* Sender is in rendez_at_sender state */
    if(dtcp->getDTCPState()->getSndRightWinEdge() > state->getLastSeqNumSent()){
      dtcp->getDTCPState()->setSndRendez(false);
      cancelAndDelete(dtcp->getDTCPState()->getRendezvousTimer());
      dtcp->getDTCPState()->setRendezvousTimer(nullptr);
    }
  }
}

void DTP::trySendGenPDUs(std::vector<DataTransferPDU*>* pduQ)
{

  if (state->isDtcpPresent())
  {
    //postablePDUs = empty;

    //if flowControl present
    if (dtcp->dtcpState->isWinBased() || dtcp->dtcpState->isRateBased())
    {
      std::vector<DataTransferPDU*>::iterator it;
      for (it = pduQ->begin(); it != pduQ->end(); it = pduQ->begin())
      {
        if (dtcp->dtcpState->isWinBased())
        {
          if (((*it)->getSeqNum() <= dtcp->getSndRtWinEdge()) && !dtcp->getDTCPState()->isClosedWindow())
          {
            /* The Window is Open. */
            dtcp->runTxControlPolicy(state, pduQ);
            /* Watchout because the current 'it' could be freed */
          }
          else
          {
            /* The Window is Closed */
            dtcp->getDTCPState()->setClosedWindow(true);

            if (pduQ == dtcp->getDTCPState()->getClosedWindowQ())
            {
              /* It indicates that trySendGenPDUs method has been called for closedWindowQ
               * So no point of trying to put the DT-PDU on closedWindowQ (again).
               */
              break;
            }

            if (dtcp->getDTCPState()->getClosedWinQueLen() < dtcp->getDTCPState()->getMaxClosedWinQueLen() - 1)
            {
              /* Put PDU on the closedWindowQueue */
              dtcp->getDTCPState()->pushBackToClosedWinQ((*it));
              /* I know it is nasty to access it directly */
              state->getGeneratedPDUQ()->erase(it);

            }
            else
            {
              state->setCurrentPdu(*it);
              state->getGeneratedPDUQ()->erase(it);

              dtcp->runSndFCOverrunPolicy(state);

            }

            /* Rendezvous condition */
            rendezvousCondition();
          }
        }    // end of Window based

        if (dtcp->dtcpState->isRateBased())
        {
          if (dtcp->getPdusSentInTimeUnit() < dtcp->getSendingRate())
          {
            dtcp->runNoRateSlowDownPolicy(state);
          }
          else
          {
            /* Exceeding Sending Rate */

            dtcp->runNoOverridePeakPolicy(state);
          }
        }    // end of RateBased

        if ((dtcp->getDTCPState()->isRateBased() && dtcp->getDTCPState()->isWinBased()) && (dtcp->getDTCPState()->isClosedWindow() ^ dtcp->isSendingRateFullfilled()))
        {
          dtcp->runReconcileFCPolicy(state);
        }

      }    //end of for

    }
    else
    {
      /* FlowControl is not present */
      std::vector<DataTransferPDU*>::iterator it;
      PDUQ_t* pduQ = state->getGeneratedPDUQ();
      for (it = pduQ->begin(); it != pduQ->end();)
      {
        state->pushBackToPostablePDUQ((*it));
        it = pduQ->erase(it);
      }
    }

    /* Iterate over postablePDUs and give them to the RMT */
    if (dtcp->dtcpState->isRxPresent())
    {
      std::vector<DataTransferPDU*>::iterator it;
      PDUQ_t* pduQ = state->getPostablePDUQ();
      for (it = pduQ->begin(); it != pduQ->end();)
      {
        /* Put a copy of each PDU in the RetransmissionQueue */
        dtcp->pushBackToRxQ((*it)->dup());

        sendToRMT((*it));

        it = pduQ->erase(it);
      }

    }
    else
    {
      /*No Retransmission Control is present, but FlowControl */
      /* Post all postablePDUs to RMT */
      std::vector<DataTransferPDU*>::iterator it;
      PDUQ_t* pduQ = state->getPostablePDUQ();
      for (it = pduQ->begin(); it != pduQ->end();)
      {

        sendToRMT((*it));
        it = pduQ->erase(it);
      }

    }

  }
  else
  {
    /* DTCP is not present */
    /* Post all generatedPDUs to RMT */
    std::vector<DataTransferPDU*>::iterator it;
    PDUQ_t* pduQ = state->getGeneratedPDUQ();
    for (it = pduQ->begin(); it != pduQ->end();)
    {
      sendToRMT((*it));
      it = pduQ->erase(it);
    }
  }
}

void DTP::sduProtection(SDU *sdu)
{
  //TODO C1

}

bool DTP::runInitialSeqNumPolicy()
{
  Enter_Method("InitialSeqNumPolicy");

//  if(initialSeqNumPolicy->run(state, dtcp->getDTCPState())){
//    initialSeqNumPolicy->defaultAction(state, dtcp->getDTCPState());
//  }

  if (state->isDtcpPresent())
  {
    initialSeqNumPolicy->call(state, dtcp->getDTCPState());
  }
  else
  {
    initialSeqNumPolicy->call(state, nullptr);
  }

  return false;
}

void DTP::sendAckOnlyPDU(unsigned int seqNum)
{

  //send an Ack/FlowControlPDU
  AckOnlyPDU* ackPDU = new AckOnlyPDU();
  setPDUHeader(ackPDU);
  ackPDU->setSeqNum(dtcp->getNextSndCtrlSeqNum());
  ackPDU->setAckNackSeqNum(seqNum);
  EV << getFullPath() << ": Sending Ack for PDU number: " << seqNum << endl;
  //    send(ackPDU, southO);
  sendToRMT(ackPDU);
}

void DTP::fillControlAckPDU(ControlAckPDU* ctrlAckPdu)
{
  ctrlAckPdu->setSeqNum(dtcp->getNextSndCtrlSeqNum());
  ctrlAckPdu->setLastCtrlSeqNumRcv(dtcp->getLastCtrlSeqNumRcv());
  ctrlAckPdu->setRcvLeftWinEdge(dtcp->getSndLeftWinEdge());
  ctrlAckPdu->setRcvRightWinEdge(dtcp->getSndRtWinEdge());
  ctrlAckPdu->setSndLeftWinEdge(state->getRcvLeftWinEdge());
  ctrlAckPdu->setSndRightWinEdge(dtcp->getRcvRightWinEdge());
  ctrlAckPdu->setRcvRate(dtcp->getRcvrRate());
}

void DTP::sendControlAckPDU()
{
  Enter_Method_Silent();

  ControlAckPDU* ctrlAckPdu = new ControlAckPDU();
  setPDUHeader(ctrlAckPdu);
  fillControlAckPDU(ctrlAckPdu);

  sendToRMT(ctrlAckPdu);
}

void DTP::sendEmptyDTPDU()
{
  Enter_Method_Silent
  ();

  //Send Transfer PDU With Zero length
  DataTransferPDU* dataPdu = new DataTransferPDU();
  setPDUHeader(dataPdu);
  unsigned int seqNum = state->getNextSeqNumToSend();
  dataPdu->setSeqNum(seqNum);

  if (setDRFInPDU(false))
  {
    dataPdu->setFlags(dataPdu->getFlags() | DRF_FLAG);
  }
  Data* data = new Data();
  data->setDataType(DATA_SDU_COMPLETE);

  PDUData* pduData = new PDUData();
  pduData->encapsulate(data);


  UserDataField* userData = new UserDataField();
  userData->encapsulate(pduData);
  userData->setCompleteSDU(true);

  dataPdu->encapsulate(userData);

  if (dtcp->dtcpState->isRxPresent())
  {
//    RxExpiryTimer* rxExpTimer = new RxExpiryTimer("RxExpiryTimer");
//    rxExpTimer->setPdu(dataPdu->dup());
//    rxQ.push_back(rxExpTimer);
//    schedule(rxExpTimer);
    dtcp->pushBackToRxQ(dataPdu->dup());
  }

  sendToRMT(dataPdu);
}

void DTP::notifyAboutUnableMaintain()
{
  //FIX A2 - activate when CDAP Splitter is ready
  return;
  // Notify User Flow that we were unable to maintain the QoS for this connection
  //CDAPMessage* uMaintainMsg = new CDAP_M_Unable_Maintain;
  //SDU* sdu = new SDU();
  //sdu->addUserData(uMaintainMsg);
  //send(sdu, northO);
}

void DTP::notifyStartSending()
{
  if (state->isBlockingPort())
  {
    emit(sigEFCPStartSending, flow);
    state->setBlockingPort(false);
  }
}

void DTP::notifyStopSending()
{

  // Notify User Flow to Stop sending due to closed window and full closedWindowQ.

  if (!state->isBlockingPort())
  {
    emit(sigEFCPStahpSending, flow);
    state->setBlockingPort(true);
  }

}

void DTP::notifyAboutInactivity()
{
  //FIX A2 - activate when CDAP Splitter is ready
  return;
  // Notify User Flow there has been no activity for awhile.
  //CDAPMessage* inactivMsg = new CDAP_M_Inactiv();
  //SDU* sdu = new SDU();
  //sdu->addUserData(inactivMsg);
  //send(sdu, northO);

}

void DTP::runRcvrInactivityTimerPolicy()
{
  Enter_Method
  ("RcvrInactivityPolicy");
  if (state->isDtcpPresent())
  {
    rcvrInactivityPolicy->call(state, dtcp->getDTCPState());
  }
  else
  {
    rcvrInactivityPolicy->call(state, nullptr);
  }
}

void DTP::runSenderInactivityTimerPolicy()
{
  Enter_Method
  ("SenderInactivityPolicy");

  if (state->isDtcpPresent())
  {
    senderInactivityPolicy->call(state, dtcp->getDTCPState());
  }
  else
  {
    senderInactivityPolicy->call(state, nullptr);
  }

}

void DTP::sendToRMT(PDU* pdu)
{
  Enter_Method_Silent("SendToRMT");
  take(pdu);
  if (pdu->getType() == DATA_TRANSFER_PDU)
  {
      emit(sigStatDTPSentSeqNum, pdu->getSeqNum());
    //TODO B1 What to do in case of retransmission?
    state->setLastSeqNumSent(pdu->getSeqNum());
    EV << getFullPath() << ": PDU number: " << pdu->getSeqNum() << " sent in time: " << simTime() << endl;
  }
  else
  {
    //This should be controlPDU so do not have to increment LastSeqNumSent
    dtcp->dtcpState->setLastControlSeqNumSent(pdu->getSeqNum());
    EV << getFullPath() << ": Control PDU number: " << pdu->getSeqNum() << " sent in time: " << simTime() << endl;
  }

  send(pdu, southO);
}

double DTP::getRxTime()
{
  //TODO A2 Epsilon
  // RTT + A + epsilon

  return state->getRtt() + getQoSCube()->getATime() / 1000;
}

unsigned int DTP::getAllowableGap()
{

  getQoSCube()->getMaxAllowGap();
  return 4;
}

//TODO B! When to call it?
void DTP::rcvrBufferStateChange()
{
  if (dtcp->dtcpState->isRateBased())
  {
    dtcp->runRateReductionPolicy(state);
  }
}

void DTP::svUpdate(unsigned int seqNum)
{


  if(dtcp->dtcpState->isRcvRendez()){
    dtcp->stopReliableCPDUTimer();
    dtcp->dtcpState->setRcvRendez(false);
  }
  //update RcvLeftWindoEdge
   state->updateRcvLWE(seqNum);

  if (dtcp->dtcpState->isFCPresent())
  {
    if (dtcp->dtcpState->isWinBased())
    {
      dtcp->runRcvrFCPolicy(state);
    }
  }

  if (dtcp->dtcpState->isRxPresent())
  {
    dtcp->runRcvrAckPolicy(state);
  }

  if (dtcp->dtcpState->isFCPresent() && !dtcp->dtcpState->isRxPresent())
  {

    dtcp->runReceivingFCPolicy(state);
  }

}

void DTP::flushReassemblyPDUQ()
{
  state->clearReassemblyPDUQ();
}

void DTP::clearRxQ()
{
  dtcp->clearRxQ();
}

void DTP::schedule(DTPTimers *timer, double time)
{
  //TODO A1 Can't have MPL = 0
  double MPL = (state->getMPL() > 0) ? state->getMPL() : 0;

//    (state->isDtcpPresent() && dtcp->dtcpState->isRxPresent() )? dtcp->getDataReXmitMax() : 0;
  unsigned int rxCount = (state->isDtcpPresent() && dtcp->dtcpState->isRxPresent()) ? dtcp->getDataReXmitMax() : 0;
  double R = (state->getRtt() > 0 && rxCount > 0) ? state->getRtt() * rxCount : 0;
  double A = (state->getQoSCube()->getATime() > 0) ? state->getQoSCube()->getATime() / 1000 : 0;

  switch (timer->getType())
  {

    case (DTP_SENDER_INACTIVITY_TIMER):
      //3(MPL+R+A)
      scheduleAt(simTime() + 3 * (MPL + R + A), timer);
      break;
    case (DTP_RCVR_INACTIVITY_TIMER):
      //2(MPL+R+A)
      scheduleAt(simTime() + 2 * (MPL + R + A), timer);
      break;
    case (DTP_A_TIMER):

      scheduleAt(simTime() + A, timer);
      break;
//    case (DTP_DELIMITING_TIMER):
//        if(!timer->isScheduled()){
//          scheduleAt(simTime() + state->getDelimitDelay(), timer);
//        }
//     break;
  }
}

void DTP::setFlow(const Flow* flow)
{
  this->flow = flow;
}

void DTP::setDTCP(DTCP* dtcp)
{
  this->dtcp = dtcp;
  if (dtcp != nullptr)
  {
    dtcp->setDTP(this);
  }
}

void DTP::setState(DTPState* state)
{
  this->state = state;
}

void DTP::handleQueueInfoFromRMT(QueueInfo* queueInfo)
{
  handleQueueInfo(queueInfo);
}