diff --git a/daemon/fw/PCONStrategy.cpp b/daemon/fw/PCONStrategy.cpp new file mode 100644 index 00000000..6c069228 --- /dev/null +++ b/daemon/fw/PCONStrategy.cpp @@ -0,0 +1,174 @@ +#include "PCONStrategy.hpp" +#include + +namespace nfd +{ + namespace fw + { + NFD_REGISTER_STRATEGY(PCONStrategy); + NFD_LOG_INIT(PCONStrategy); + ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + //// PCONStrategy + ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + const time::milliseconds PCONStrategy::RETX_SUPPRESSION_INITIAL(10); + const time::milliseconds PCONStrategy::RETX_SUPPRESSION_MAX(250); + + PCONStrategy::PCONStrategy(nfd::Forwarder &forwarder, const ndn::Name &name) + : Strategy(forwarder), ProcessNackTraits(this), m_retxSuppression(RETX_SUPPRESSION_INITIAL, + RetxSuppressionExponential::DEFAULT_MULTIPLIER, + RETX_SUPPRESSION_MAX) + { + this->setInstanceName(makeInstanceName(name, getStrategyName())); + } + + const Name & + PCONStrategy::getStrategyName() + { + static Name strategyName("/localhost/nfd/strategy/PCON/%FD%01"); + return strategyName; + } + + void + PCONStrategy::afterReceiveInterest(const nfd::FaceEndpoint &ingress, const ndn::Interest &interest, + const std::shared_ptr &pitEntry) + { + const fib::Entry &fibEntry = this->lookupFib(*pitEntry); + const fib::NextHopList &nexthops = fibEntry.getNextHops(); + MtForwardingInfo *measurementInfo = this->getPrefixMeasurements(fibEntry); + + if (measurementInfo == nullptr) + { + measurementInfo = this->addPrefixMeasurements(fibEntry); + const Name prefixName = fibEntry.getPrefix(); + measurementInfo->setPrefix(prefixName.toUri()); + initializeForwMap(measurementInfo, nexthops); + } + + RetxSuppressionResult suppression = m_retxSuppression.decidePerPitEntry(*pitEntry); + if (suppression == RetxSuppressionResult::SUPPRESS) + { + NFD_LOG_DEBUG(interest << " from=" << ingress << " suppressed"); + return; + } + + Face *outFace = nullptr; + if (ingress.face.getScope() != ndn::nfd::FACE_SCOPE_NON_LOCAL) + { + auto selected = std::find_if(nexthops.begin(), nexthops.end(), [&](const auto &nexthop) { + return isNextHopEligible(ingress.face, interest, nexthop, pitEntry); + }); + if (selected != nexthops.end()) + { + outFace = &((*selected).getFace()); + } + } + else + { + // Random number between 0 and 1. + double r = ((double)rand() / (RAND_MAX)); + + double percSum = 0; + + // Add all eligbile faces to list (excludes current downstream) + std::vector eligbleFaces; + for (auto &n : fibEntry.getNextHops()) + { + if (isNextHopEligible(ingress.face, interest, n, pitEntry)) + { + // Add up percentage Sum. + percSum += measurementInfo->getforwPerc(n.getFace().getId()); + eligbleFaces.push_back(&n.getFace()); + } + } + + if (eligbleFaces.size() < 1) + { + return; + } + else if (eligbleFaces.size() == 1) + { + outFace = eligbleFaces.front(); + } + else + { + // Choose face according to current forwarding percentage: + double forwPerc = 0; + for (auto face : eligbleFaces) + { + forwPerc += measurementInfo->getforwPerc(face->getId()) / percSum; + if (r < forwPerc) + { + outFace = face; + break; + } + } + } + } + + if (outFace == nullptr) + { + lp::NackHeader nackHeader; + nackHeader.setReason(lp::NackReason::NO_ROUTE); + this->sendNack(pitEntry, ingress.face, nackHeader); + this->rejectPendingInterest(pitEntry); + return; + } + else + { + this->sendInterest(pitEntry, *outFace, interest); + } + + } + + void + PCONStrategy::beforeSatisfyInterest(const shared_ptr &pitEntry, + const FaceEndpoint &ingress, const Data &data) + { + Name currentPrefix; + MtForwardingInfo *measurementInfo; + std::tie(currentPrefix, measurementInfo) = this->findPrefixMeasurementsLPM( + *pitEntry); + + if (data.getCongestionMark() > 0 && ingress.face.getId() > 256) + { + // std::cout << ">>2: " << data.getCongestionMark() << std::endl; + double fwPerc = measurementInfo->getforwPerc(ingress.face.getId()); + double change_perc = CHANGE_PER_MARK * fwPerc; + this->reduceFwPerc(measurementInfo, ingress.face.getId(), change_perc); + } + } + + void PCONStrategy::afterReceiveNack(const nfd::FaceEndpoint &ingress, const ndn::lp::Nack &nack, + const std::shared_ptr &pitEntry) + { + this->processNack(ingress.face, nack, pitEntry); + } + + void + PCONStrategy::initializeForwMap(MtForwardingInfo *measurementInfo, + const fib::NextHopList &nextHops) + { + int lowestId = std::numeric_limits::max(); + int minCost = std::numeric_limits::max(); + + for (auto &n : nextHops) + { + if (n.getCost() < minCost) + { + minCost = n.getCost(); + lowestId = n.getFace().getId(); + } + } + + for (auto &n : nextHops) + { + double perc = 0.0; + if (n.getFace().getId() == lowestId) + { + perc = 1.0; + } + measurementInfo->setforwPerc(n.getFace().getId(), perc); + } + } + } +} \ No newline at end of file diff --git a/daemon/fw/PCONStrategy.hpp b/daemon/fw/PCONStrategy.hpp new file mode 100644 index 00000000..dd3f570b --- /dev/null +++ b/daemon/fw/PCONStrategy.hpp @@ -0,0 +1,150 @@ +#ifndef NFD_DAEMON_FW_PCON_STRATEGY_HPP +#define NFD_DAEMON_FW_PCON_STRATEGY_HPP + +#include +#include "face/face.hpp" +#include "fw/strategy.hpp" +#include "fw/retx-suppression-exponential.hpp" +#include "fw/algorithm.hpp" +#include "fw/process-nack-traits.hpp" +#include +#include +#include "mt-forward-info.hpp" + +namespace nfd +{ + namespace fw + { + class PCONStrategy : public Strategy, public ProcessNackTraits + { + public: + explicit PCONStrategy(Forwarder &forwarder, const Name &name = getStrategyName()); + + ~PCONStrategy() override = default; + + void + afterReceiveInterest(const FaceEndpoint &ingress, const Interest &interest, + const shared_ptr &pitEntry) override; + + void + beforeSatisfyInterest(const shared_ptr &pitEntry, + const FaceEndpoint &ingress, const Data &data) override; + + void + afterReceiveNack(const FaceEndpoint &ingress, const lp::Nack &nack, + const shared_ptr &pitEntry) override; + + static const Name & + getStrategyName(); + + protected: + friend ProcessNackTraits; + + private: + void + initializeForwMap(MtForwardingInfo *measurementInfo, + const fib::NextHopList &nextHops); + + MtForwardingInfo * + getPrefixMeasurements(const fib::Entry &fibEntry) + { + std::string key = fibEntry.getPrefix().getPrefix(1).toUri(); + if (this->innerMap.count(key) == 0) + { + return nullptr; + } + else + { + return this->innerMap[key].get(); + } + } + + MtForwardingInfo * + addPrefixMeasurements(const fib::Entry &fibEntry) + { + std::string key = fibEntry.getPrefix().getPrefix(1).toUri(); + if (this->innerMap.count(key) == 0) + { + this->innerMap[key] = make_unique(); + } + return this->innerMap[key].get(); + // measurements::Entry* me = measurements.get(fibEntry); + // // std::cout << fibEntry.getPrefix().toUri() << std::endl; + // measurements.extendLifetime(*me, 8_s); + // return me->insertStrategyInfo().first; + } + + std::tuple + findPrefixMeasurementsLPM(const pit::Entry &pitEntry) + { + + std::string key = pitEntry.getName().getPrefix(1).toUri(); + if (this->innerMap.count(key) == 0) { + return std::forward_as_tuple(Name(), nullptr); + } + + return std::forward_as_tuple(pitEntry.getName().getPrefix(1), this->innerMap[key].get()); + // measurements::Entry *me = measurements.findLongestPrefixMatch(pitEntry); + // if (me == nullptr) + // { + // std::cout << "Name " << pitEntry.getName().toUri() << " not found!\n"; + // return std::forward_as_tuple(Name(), nullptr); + // } + // return std::forward_as_tuple(me->getName(), me->getStrategyInfo()); + } + + static void + reduceFwPerc(MtForwardingInfo *forwInfo, + const FaceId reducedFaceId, + const double change) + { + if (forwInfo->getFaceCount() == 1) + { + // std::cout << "reduce1" << std::endl; + return; + } + // Reduction is at most the current forwarding percentage of the face that is reduced. + double changeRate = 0 - std::min(change, forwInfo->getforwPerc(reducedFaceId)); + + // std::cout << "reduce2: " << changeRate << std::endl; + // Decrease fw percentage of the given face: + forwInfo->increaseforwPerc(reducedFaceId, changeRate); + double sumFWPerc = 0; + sumFWPerc += forwInfo->getforwPerc(reducedFaceId); + const auto forwMap = forwInfo->getForwPercMap(); + + // std::cout << "\n"; + for (auto f : forwMap) + { + auto tempChangeRate = changeRate; + auto &faceId = f.first; + if (faceId == reducedFaceId) + { // Do nothing. Percentage has already been added. + } + else + { + // Increase forwarding percentage of all other faces by and equal amount. + tempChangeRate = std::abs(changeRate / (double)(forwMap.size() - 1)); + forwInfo->increaseforwPerc((faceId), tempChangeRate); + sumFWPerc += forwInfo->getforwPerc(faceId); + } + } + + // if (sumFWPerc < 0.999 || sumFWPerc > 1.001) + // { + // std::cout << StrHelper::getTime() << "ERROR! Sum of fw perc out of range: " << sumFWPerc + // << "\n"; + // } + } + + private: + static const time::milliseconds RETX_SUPPRESSION_INITIAL; + static const time::milliseconds RETX_SUPPRESSION_MAX; + double CHANGE_PER_MARK = 0.1; + std::unordered_map> innerMap; + RetxSuppressionExponential m_retxSuppression; + }; + } +} + +#endif \ No newline at end of file diff --git a/daemon/fw/mt-forward-info.hpp b/daemon/fw/mt-forward-info.hpp new file mode 100644 index 00000000..d5558e5f --- /dev/null +++ b/daemon/fw/mt-forward-info.hpp @@ -0,0 +1,115 @@ +#ifndef NFD_DAEMON_FW_MT_FORWARDING_INFO_HPP +#define NFD_DAEMON_FW_MT_FORWARDING_INFO_HPP + +#include "face/face.hpp" +#include "fw/strategy-info.hpp" + +namespace nfd +{ + namespace fw + { + + /** + * Measurement information that can be saved and retrieved per-name-prefix. + */ + class MtForwardingInfo : public StrategyInfo + { + public: + static constexpr int + getTypeId() + { + return 1051; + } + + MtForwardingInfo() + : m_ownPrefix("null") + { + } + + double + getforwPerc(FaceId face) + { + // if (m_forwPercMap.find(face) == m_forwPercMap.end()) + // { + // std::cout << time::steady_clock::now().time_since_epoch().count() / 1000 * 1000 + // << " ms, couldn't find face " << face << "\n"; + // } + // assert(m_forwPercMap.find(face) != m_forwPercMap.end()); + double forwPerc = m_forwPercMap.at(face); + // assert(forwPerc >= 0 && forwPerc <= 1); + + return forwPerc; + } + + void + setforwPerc(FaceId faceId, double perc) + { + m_forwPercMap[faceId] = perc; + } + + void + increaseforwPerc(FaceId faceId, double changeRate) + { + m_forwPercMap[faceId] += changeRate; + } + + const std::map + getForwPercMap() const + { + return m_forwPercMap; + } + + int + getFaceCount() + { + std::vector faceIdList; + for (auto faceInfo : m_forwPercMap) + { + faceIdList.push_back(faceInfo.first); + } + return faceIdList.size(); + } + + void + setPrefix(std::string prefix) + { + m_ownPrefix = prefix; + } + + std::string + getPrefix() const + { + return m_ownPrefix; + } + + private: + /** These Functions were used to disable faces in the "highly congested" state:*/ + // void + // enableFace(FaceId faceId) + // { + // m_disabledFaces.erase(faceId); + // } + // + // void + // disableFace(FaceId faceId) + // { + // m_disabledFaces.emplace(faceId); + //// std::cout << "Disabling face " << faceId << "\n"; + // } + // + // bool + // isFaceEnabled(FaceId faceId) + // { + // bool disabled = (m_disabledFaces.find(faceId) != m_disabledFaces.end()); + // return !disabled; + // } + std::string m_ownPrefix; + std::map m_forwPercMap; + + std::unordered_set m_disabledFaces; + }; + + } //fw +} //nfd + +#endif \ No newline at end of file