lp-reliability.cpp

Go to the documentation of this file.

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/*
 * Copyright (c) 2014-2024,  Regents of the University of California,
 *                           Arizona Board of Regents,
 *                           Colorado State University,
 *                           University Pierre & Marie Curie, Sorbonne University,
 *                           Washington University in St. Louis,
 *                           Beijing Institute of Technology,
 *                           The University of Memphis.
 *
 * This file is part of NFD (Named Data Networking Forwarding Daemon).
 * See AUTHORS.md for complete list of NFD authors and contributors.
 *
 * NFD is free software: you can redistribute it and/or modify it under the terms
 * of the GNU General Public License as published by the Free Software Foundation,
 * either version 3 of the License, or (at your option) any later version.
 *
 * NFD is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.  See the GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * NFD, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
 */
 
#include "lp-reliability.hpp"
#include "common/global.hpp"
#include "generic-link-service.hpp"
#include "transport.hpp"
 
#include <ndn-cxx/lp/fields.hpp>
 
#include <set>
 
namespace nfd::face {
 
NFD_LOG_INIT(LpReliability);
 
LpReliability::LpReliability(const LpReliability::Options& options, GenericLinkService* linkService)
  : m_options(options)
  , m_linkService(linkService)
  , m_firstUnackedFrag(m_unackedFrags.begin())
  , m_lastTxSeqNo(-1) // set to "-1" to start TxSequence numbers at 0
{
  BOOST_ASSERT(m_linkService != nullptr);
  BOOST_ASSERT(m_options.idleAckTimerPeriod > 0_ns);
}
 
void
LpReliability::setOptions(const Options& options)
{
  BOOST_ASSERT(options.idleAckTimerPeriod > 0_ns);
 
  if (m_options.isEnabled && !options.isEnabled) {
    m_idleAckTimer.cancel();
  }
 
  m_options = options;
}
 
void
LpReliability::handleOutgoing(std::vector<lp::Packet>& frags, lp::Packet&& pkt, bool isInterest)
{
  BOOST_ASSERT(m_options.isEnabled);
 
  auto unackedFragsIt = m_unackedFrags.begin();
  auto sendTime = time::steady_clock::now();
 
  auto netPkt = make_shared<NetPkt>(std::move(pkt), isInterest);
  netPkt->unackedFrags.reserve(frags.size());
 
  for (lp::Packet& frag : frags) {
    // Non-IDLE packets are required to have assigned Sequence numbers with LpReliability enabled
    BOOST_ASSERT(frag.has<lp::SequenceField>());
 
    // Assign TxSequence number
    lp::Sequence txSeq = assignTxSequence(frag);
 
    // Store LpPacket for future retransmissions
    unackedFragsIt = m_unackedFrags.try_emplace(unackedFragsIt, txSeq, frag);
    unackedFragsIt->second.sendTime = sendTime;
    auto rto = m_rttEst.getEstimatedRto();
    lp::Sequence seq = frag.get<lp::SequenceField>();
    NFD_LOG_FACE_TRACE("transmitting seq=" << seq << ", txseq=" << txSeq << ", rto=" <<
                       time::duration_cast<time::milliseconds>(rto).count() << "ms");
    unackedFragsIt->second.rtoTimer = getScheduler().schedule(rto, [=] {
      onLpPacketLost(txSeq, true);
    });
    unackedFragsIt->second.netPkt = netPkt;
 
    if (m_unackedFrags.size() == 1) {
      m_firstUnackedFrag = m_unackedFrags.begin();
    }
 
    // Add to associated NetPkt
    netPkt->unackedFrags.push_back(unackedFragsIt);
  }
}
 
bool
LpReliability::processIncomingPacket(const lp::Packet& pkt)
{
  BOOST_ASSERT(m_options.isEnabled);
 
  bool isDuplicate = false;
  auto now = time::steady_clock::now();
 
  // Extract and parse Acks
  for (lp::Sequence ackTxSeq : pkt.list<lp::AckField>()) {
    auto fragIt = m_unackedFrags.find(ackTxSeq);
    if (fragIt == m_unackedFrags.end()) {
      // Ignore an Ack for an unknown TxSequence number
      NFD_LOG_FACE_DEBUG("received ack for unknown txseq=" << ackTxSeq);
      continue;
    }
    auto& frag = fragIt->second;
 
    // Cancel the RTO timer for the acknowledged fragment
    frag.rtoTimer.cancel();
 
    if (frag.retxCount == 0) {
      NFD_LOG_FACE_TRACE("received ack for seq=" << frag.pkt.get<lp::SequenceField>() << ", txseq=" <<
                         ackTxSeq << ", retx=0, rtt=" <<
                         time::duration_cast<time::milliseconds>(now - frag.sendTime).count() << "ms");
      // This sequence had no retransmissions, so use it to estimate the RTO
      m_rttEst.addMeasurement(now - frag.sendTime);
    }
    else {
      NFD_LOG_FACE_TRACE("received ack for seq=" << frag.pkt.get<lp::SequenceField>() << ", txseq=" <<
                         ackTxSeq << ", retx=" << frag.retxCount);
    }
 
    // Look for frags with TxSequence numbers < ackTxSeq (allowing for wraparound) and consider
    // them lost if a configurable number of Acks containing greater TxSequence numbers have been
    // received.
    auto lostLpPackets = findLostLpPackets(fragIt);
 
    // Remove the fragment from the map of unacknowledged fragments and from its associated network
    // packet. Potentially increment the start of the window.
    onLpPacketAcknowledged(fragIt);
 
    // This set contains TxSequences that have been removed by onLpPacketLost below because they
    // were part of a network packet that was removed due to a fragment exceeding retx, as well as
    // any other TxSequences removed by onLpPacketLost. This prevents onLpPacketLost from being
    // called later for an invalid iterator.
    std::set<lp::Sequence> removedLpPackets;
 
    // Resend or fail fragments considered lost. Potentially increment the start of the window.
    for (lp::Sequence txSeq : lostLpPackets) {
      if (removedLpPackets.find(txSeq) == removedLpPackets.end()) {
        auto removedTxSeqs = onLpPacketLost(txSeq, false);
        for (auto removedTxSeq : removedTxSeqs) {
          removedLpPackets.insert(removedTxSeq);
        }
      }
    }
  }
 
  // If packet has Fragment and TxSequence fields, extract TxSequence and add to AckQueue
  if (pkt.has<lp::FragmentField>() && pkt.has<lp::TxSequenceField>()) {
    NFD_LOG_FACE_TRACE("queueing ack for remote txseq=" << pkt.get<lp::TxSequenceField>());
    m_ackQueue.push(pkt.get<lp::TxSequenceField>());
 
    // Check for received frames with duplicate Sequences
    if (pkt.has<lp::SequenceField>()) {
      lp::Sequence pktSequence = pkt.get<lp::SequenceField>();
      isDuplicate = m_recentRecvSeqs.count(pktSequence) > 0;
      // Check for recent received Sequences to remove
      auto now = time::steady_clock::now();
      auto rto = m_rttEst.getEstimatedRto();
      while (!m_recentRecvSeqsQueue.empty() &&
             now > m_recentRecvSeqs[m_recentRecvSeqsQueue.front()] + rto) {
        m_recentRecvSeqs.erase(m_recentRecvSeqsQueue.front());
        m_recentRecvSeqsQueue.pop();
      }
      m_recentRecvSeqs.try_emplace(pktSequence, now);
      m_recentRecvSeqsQueue.push(pktSequence);
    }
 
    startIdleAckTimer();
  }
 
  return !isDuplicate;
}
 
void
LpReliability::piggyback(lp::Packet& pkt, ssize_t mtu)
{
  BOOST_ASSERT(m_options.isEnabled);
  BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
 
  // up to 2 extra octets reserved for potential TLV-LENGTH size increases
  ssize_t pktSize = pkt.wireEncode().size();
  ssize_t reservedSpace = tlv::sizeOfVarNumber(ndn::MAX_NDN_PACKET_SIZE) -
                          tlv::sizeOfVarNumber(pktSize);
  ssize_t remainingSpace = (mtu == MTU_UNLIMITED ? ndn::MAX_NDN_PACKET_SIZE : mtu) - reservedSpace;
  remainingSpace -= pktSize;
 
  while (!m_ackQueue.empty()) {
    lp::Sequence ackTxSeq = m_ackQueue.front();
    // Ack size = Ack TLV-TYPE (3 octets) + TLV-LENGTH (1 octet) + lp::Sequence (8 octets)
    const ssize_t ackSize = tlv::sizeOfVarNumber(lp::tlv::Ack) +
                            tlv::sizeOfVarNumber(sizeof(lp::Sequence)) +
                            sizeof(lp::Sequence);
 
    if (ackSize > remainingSpace) {
      break;
    }
 
    NFD_LOG_FACE_TRACE("piggybacking ack for remote txseq=" << ackTxSeq);
 
    pkt.add<lp::AckField>(ackTxSeq);
    m_ackQueue.pop();
    remainingSpace -= ackSize;
  }
}
 
lp::Sequence
LpReliability::assignTxSequence(lp::Packet& frag)
{
  lp::Sequence txSeq = ++m_lastTxSeqNo;
  frag.set<lp::TxSequenceField>(txSeq);
  if (!m_unackedFrags.empty() && m_lastTxSeqNo == m_firstUnackedFrag->first) {
    NDN_THROW(std::length_error("TxSequence range exceeded"));
  }
  return m_lastTxSeqNo;
}
 
void
LpReliability::startIdleAckTimer()
{
  if (m_idleAckTimer) {
    // timer is already running, do nothing
    return;
  }
 
  m_idleAckTimer = getScheduler().schedule(m_options.idleAckTimerPeriod, [this] {
    while (!m_ackQueue.empty()) {
      m_linkService->requestIdlePacket();
    }
  });
}
 
std::vector<lp::Sequence>
LpReliability::findLostLpPackets(LpReliability::UnackedFrags::iterator ackIt)
{
  std::vector<lp::Sequence> lostLpPackets;
 
  for (auto it = m_firstUnackedFrag; ; ++it) {
    if (it == m_unackedFrags.end()) {
      it = m_unackedFrags.begin();
    }
 
    if (it->first == ackIt->first) {
      break;
    }
 
    auto& unackedFrag = it->second;
    unackedFrag.nGreaterSeqAcks++;
    NFD_LOG_FACE_TRACE("received ack=" << ackIt->first << " before=" << it->first <<
                       ", before count=" << unackedFrag.nGreaterSeqAcks);
 
    if (unackedFrag.nGreaterSeqAcks >= m_options.seqNumLossThreshold) {
      lostLpPackets.push_back(it->first);
    }
  }
 
  return lostLpPackets;
}
 
std::vector<lp::Sequence>
LpReliability::onLpPacketLost(lp::Sequence txSeq, bool isTimeout)
{
  BOOST_ASSERT(m_unackedFrags.count(txSeq) > 0);
  auto txSeqIt = m_unackedFrags.find(txSeq);
 
  auto& txFrag = txSeqIt->second;
  txFrag.rtoTimer.cancel();
  auto netPkt = txFrag.netPkt;
  std::vector<lp::Sequence> removedThisTxSeq;
  lp::Sequence seq = txFrag.pkt.get<lp::SequenceField>();
 
  if (isTimeout) {
    NFD_LOG_FACE_TRACE("rto timer expired for seq=" << seq << ", txseq=" << txSeq);
  }
  else { // lost due to out-of-order TxSeqs
    NFD_LOG_FACE_TRACE("seq=" << seq << ", txseq=" << txSeq <<
                       " considered lost from acks for more recent txseqs");
  }
 
  // Check if maximum number of retransmissions exceeded
  if (txFrag.retxCount >= m_options.maxRetx) {
    NFD_LOG_FACE_DEBUG("seq=" << seq << " exceeded allowed retransmissions: DROP");
    // Delete all LpPackets of NetPkt from m_unackedFrags (except this one)
    for (size_t i = 0; i < netPkt->unackedFrags.size(); i++) {
      if (netPkt->unackedFrags[i] != txSeqIt) {
        removedThisTxSeq.push_back(netPkt->unackedFrags[i]->first);
        deleteUnackedFrag(netPkt->unackedFrags[i]);
      }
    }
 
    ++m_linkService->nRetxExhausted;
 
    // Notify strategy of dropped Interest (if any)
    if (netPkt->isInterest) {
      BOOST_ASSERT(netPkt->pkt.has<lp::FragmentField>());
      auto frag = netPkt->pkt.get<lp::FragmentField>();
      onDroppedInterest(Interest(Block({frag.first, frag.second})));
    }
 
    // Delete this LpPacket from m_unackedFrags
    removedThisTxSeq.push_back(txSeqIt->first);
    deleteUnackedFrag(txSeqIt);
  }
  else {
    // Assign new TxSequence
    lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
    netPkt->didRetx = true;
 
    // Move fragment to new TxSequence mapping
    auto hint = m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
                ? m_firstUnackedFrag
                : m_unackedFrags.end();
    auto newTxFragIt = m_unackedFrags.try_emplace(hint, newTxSeq, txFrag.pkt);
    auto& newTxFrag = newTxFragIt->second;
    newTxFrag.retxCount = txFrag.retxCount + 1;
    newTxFrag.netPkt = netPkt;
 
    // Update associated NetPkt
    auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
    BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
    *fragInNetPkt = newTxFragIt;
 
    removedThisTxSeq.push_back(txSeqIt->first);
    deleteUnackedFrag(txSeqIt);
 
    // Retransmit fragment
    m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt));
 
    auto rto = m_rttEst.getEstimatedRto();
    NFD_LOG_FACE_TRACE("retransmitting seq=" << seq << ", txseq=" << newTxSeq << ", retx=" <<
                       txFrag.retxCount << ", rto=" <<
                       time::duration_cast<time::milliseconds>(rto).count() << "ms");
 
    // Start RTO timer for this sequence
    newTxFrag.rtoTimer = getScheduler().schedule(rto, [=] {
      onLpPacketLost(newTxSeq, true);
    });
  }
 
  return removedThisTxSeq;
}
 
void
LpReliability::onLpPacketAcknowledged(UnackedFrags::iterator fragIt)
{
  auto netPkt = fragIt->second.netPkt;
 
  // Remove from NetPkt unacked fragment list
  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), fragIt);
  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
  *fragInNetPkt = netPkt->unackedFrags.back();
  netPkt->unackedFrags.pop_back();
 
  // Check if network-layer packet completely received. If so, increment counters
  if (netPkt->unackedFrags.empty()) {
    if (netPkt->didRetx) {
      ++m_linkService->nRetransmitted;
    }
    else {
      ++m_linkService->nAcknowledged;
    }
  }
 
  deleteUnackedFrag(fragIt);
}
 
void
LpReliability::deleteUnackedFrag(UnackedFrags::iterator fragIt)
{
  lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
  lp::Sequence currentTxSeq = fragIt->first;
  auto nextFragIt = m_unackedFrags.erase(fragIt);
 
  if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
    // If "first" fragment in send window (allowing for wraparound), increment window begin
    if (nextFragIt == m_unackedFrags.end()) {
      m_firstUnackedFrag = m_unackedFrags.begin();
    }
    else {
      m_firstUnackedFrag = nextFragIt;
    }
  }
  else if (m_unackedFrags.empty()) {
    m_firstUnackedFrag = m_unackedFrags.end();
  }
}
 
std::ostream&
operator<<(std::ostream& os, const FaceLogHelper<LpReliability>& flh)
{
  if (flh.obj.getLinkService() == nullptr) {
    os << "[id=0,local=unknown,remote=unknown] ";
  }
  else {
    os << FaceLogHelper<LinkService>(*flh.obj.getLinkService());
  }
  return os;
}
 
} // namespace nfd::face