fix: Disable rate limit in compare scenario

daemon/face/TokenBucket.cpp

@@ -0,0 +1,62 @@
+#include "TokenBucket.hpp"
+#include "common/global.hpp"
+#include "common/logger.hpp"
+#include <iostream>
+#include <functional>
+namespace nfd {
+    namespace face {
+        NFD_LOG_INIT(QSCCPTokenBucket);
+        TokenBucket::TokenBucket(uint64_t rateLimit, uint64_t bucketSize, const std::string& name): rateLimit(rateLimit), bucketSize(bucketSize), name(name) {
+            this->eachStep = this->rateLimit / 250.0;
+            this->currentTokenSize = 0;
+            this->isStart = false;
+            // NFD_LOG_DEBUG("Create tokenBucket: " << name << ", eachStep: " << this->eachStep << ", rateLimit: " << this->rateLimit);
+        }
+        void TokenBucket::start() {
+            if (this->isStart) {
+                return;
+            }
+            // 每 4ms 增加一次令牌
+            getScheduler().schedule(time::milliseconds(4), std::bind(&TokenBucket::generateToken, this));
+        }
+        void TokenBucket::generateToken() {
+            // NFD_LOG_DEBUG(this->name << "=> do generateToken => " << this->currentTokenSize);
+            if (this->currentTokenSize + this->eachStep > this->bucketSize) {
+                this->currentTokenSize = this->bucketSize;
+            } else {
+                this->currentTokenSize += this->eachStep;
+            }
+            getScheduler().schedule(time::milliseconds(4), std::bind(&TokenBucket::generateToken, this));
+        }
+
+        void TokenBucket::generateTokenByCurrentTime(long currentTime) {
+            if (this->lastGenerateTokenTime == -1) {
+                this->lastGenerateTokenTime = currentTime;
+                return;
+            }
+            if (currentTime - this->lastGenerateTokenTime < 1) {
+                return;
+            }
+            long currentStep = (currentTime - this->lastGenerateTokenTime) * this->rateLimit / 1000.0;
+            // std::cout << "currentStep: " << currentStep << "currentTime: " << currentTime << "lastGenerateTokenTime: " << this->lastGenerateTokenTime << std::endl;
+            // NFD_LOG_DEBUG(this->name << "=> do generateToken => " << this->currentTokenSize);
+            this->lastGenerateTokenTime = currentTime;
+            if (this->currentTokenSize + currentStep > this->bucketSize) {
+                this->currentTokenSize = this->bucketSize;
+            } else {
+                this->currentTokenSize += currentStep;
+            }
+        }
+
+        bool TokenBucket::use(uint64_t tokenNum) {
+            if(this->currentTokenSize > tokenNum) {
+                this->currentTokenSize -= tokenNum;
+                // NFD_LOG_DEBUG(this->name << "=> use success => " << this->currentTokenSize);
+                return true;
+            } else {
+                // NFD_LOG_DEBUG(this->name << "=> use failed => " << this->currentTokenSize);
+                return false;
+            }
+        }
+    }
+}

daemon/face/TokenBucket.hpp

@@ -0,0 +1,41 @@
+//
+// Created by gdcni21 on 3/25/22.
+//
+#ifndef NS_TOKENBUCKET_H
+#define NS_TOKENBUCKET_H
+#include <memory>
+#include <string>
+namespace nfd {
+    namespace face {
+        class TokenBucket {
+        public:
+            explicit TokenBucket(uint64_t rateLimit, uint64_t bucketSize, const std::string& name);
+            void start();
+            /**
+             * 周期性增加token
+             **/
+            void generateToken();
+
+            /**
+             * 根据当前时间生成token
+            */
+            void generateTokenByCurrentTime(long currentTime);
+
+            /**
+             * 尝试消费n个token
+             * @param tokenNum
+             * @return
+             */
+            bool use(uint64_t tokenNum);
+        public:
+            double eachStep; // 每个周期增加的令牌
+            double currentTokenSize; // 当前令牌数
+            uint64_t rateLimit;     // 令牌生成速率
+            uint64_t bucketSize;    // 令牌桶容量
+            std::string name;
+            bool isStart; // 标识是否启动
+            long lastGenerateTokenTime = -1;
+        };
+    }
+}
+#endif //NS_TOKENBUCKET_H

daemon/face/ethernet-channel.cpp

@@ -204,6 +204,9 @@ EthernetChannel::createFace(const ethernet::Address& remoteEndpoint,
   if (params.mtu) {
     options.overrideMtu = *params.mtu;
   }
+  if (params.bandwidth) {
+    options.C = *params.bandwidth;
+  }
 
   auto linkService = make_unique<GenericLinkService>(options);
   auto transport = make_unique<UnicastEthernetTransport>(*m_localEndpoint, remoteEndpoint,

daemon/face/face-common.hpp

@@ -81,6 +81,7 @@ struct FaceParams
   optional<time::nanoseconds> baseCongestionMarkingInterval;
   optional<uint64_t> defaultCongestionThreshold;
   optional<ssize_t> mtu;
+  optional<uint64_t> bandwidth;
   bool wantLocalFields = false;
   bool wantLpReliability = false;
   boost::logic::tribool wantCongestionMarking = boost::logic::indeterminate;

daemon/face/generic-link-service.cpp

@@ -24,7 +24,7 @@
  */
 
 #include "generic-link-service.hpp"
-
+#include <iostream>
 #include <ndn-cxx/lp/pit-token.hpp>
 #include <ndn-cxx/lp/tags.hpp>
 
@@ -47,10 +47,44 @@ GenericLinkService::GenericLinkService(const GenericLinkService::Options& option
   , m_lastSeqNo(-2)
   , m_nextMarkTime(time::steady_clock::TimePoint::max())
   , m_nMarkedSinceInMarkingState(0)
+  , globalSp(options.C * 0.99, options.vqSize)
 {
+  std::cout << "C: " << options.C << std::endl;
   m_reassembler.beforeTimeout.connect([this] (auto...) { ++this->nReassemblyTimeouts; });
   m_reliability.onDroppedInterest.connect([this] (const auto& i) { this->notifyDroppedInterest(i); });
   nReassembling.observe(&m_reassembler);
+  this->sendPacketCallback = [this](std::shared_ptr<VirtualQueueItem> taskPtr) {
+        this->globalSp.triggerCalObjectRate();
+        if (taskPtr != nullptr) {
+            const Block &block = taskPtr->item->wireEncode();
+            // std::cout << "interest wireEncode size: " << block.size() << std::endl;
+            lp::Packet lpPacket(block);
+            // std::cout << "Interest lppacket size before encode lp fields: " << lpPacket.wireEncode().size() << std::endl;
+            encodeLpFields(*(taskPtr->item), lpPacket);
+            // std::cout << "Interest lppacket size after encode lp fields: " << lpPacket.wireEncode().size() << std::endl;
+            this->sendNetPacket(std::move(lpPacket), true);
+            // lp::Packet lpPacket(taskPtr->item->wireEncode());
+            // encodeLpFields(*(taskPtr->item), lpPacket);
+            // this->sendNetPacket(std::move(lpPacket), taskPtr->endpointId, true);
+        }
+        // 得到下一个需要调度的包
+        auto newTaskPtr = this->globalSp.scheduleNext();
+        if (newTaskPtr == nullptr) {
+            //NFD_LOG_ERROR("FUCK");
+            // std::cout << "FUCK" << std::endl;
+            // 如果没有可调度的包,则 1 ms 之后再调度
+            this->canScheduler = true;
+//            getScheduler().schedule(time::milliseconds(1), std::bind(this->sendPacketCallback, nullptr));
+        } else {
+            // 等待足够的 token 再发送
+            this->canScheduler = false;
+            // std::cout << "Route waitTime: " << newTaskPtr->waitNsTime << std::endl;
+            // NFD_LOG_DEBUG("Route waitTime: " << newTaskPtr->waitNsTime);
+            getScheduler().schedule(time::nanoseconds((uint32_t)(0)),
+                                    std::bind(this->sendPacketCallback, newTaskPtr));
+        }
+    };
+    // getScheduler().schedule(time::nanoseconds(0), std::bind(this->sendPacketCallback, nullptr));
 }
 
 void
@@ -115,11 +149,34 @@ GenericLinkService::sendLpPacket(lp::Packet&& pkt)
 void
 GenericLinkService::doSendInterest(const Interest& interest)
 {
-  lp::Packet lpPacket(interest.wireEncode());
+  if (interest.getName().getPrefix(1).toUri() == "/localhost") {
+        const Block &block = interest.wireEncode();
+        // // std::cout << "interest wireEncode size: " << block.size() << std::endl;
+        lp::Packet lpPacket(block);
+        // // std::cout << "lppacket size before encode lp fields: " << lpPacket.wireEncode().size() << std::endl;
+        encodeLpFields(interest, lpPacket);
+        // // std::cout << "lppacket size after encode lp fields: " << lpPacket.wireEncode().size() << std::endl;
+        this->sendNetPacket(std::move(lpPacket), true);
+    } else {
+        this->globalSp.appendInterest(interest);
+        if (this->canScheduler) {
+            this->sendPacketCallback(nullptr);
+        }
+        //if (!success) {
+        //    // send Nack
+        //    lp::Nack nack(interest);
+        //    nack.setReason(lp::NackReason::CONGESTION);
+        //    if (this->nwfq.canMark(nack)) {
+        //        // 拥塞标记
+        //        nack.getInterest().setTag(make_shared<lp::CongestionMarkTag>(1));
+        //    }
+        //    this->receiveNack(nack, endpointId);
+        //}
+    }
 
-  encodeLpFields(interest, lpPacket);
+  // encodeLpFields(interest, lpPacket);
 
-  this->sendNetPacket(std::move(lpPacket), true);
+  // this->sendNetPacket(std::move(lpPacket), true);
 }
 
 void
@@ -351,7 +408,7 @@ GenericLinkService::decodeNetPacket(const Block& netPkt, const lp::Packet& first
   }
   catch (const tlv::Error& e) {
     ++this->nInNetInvalid;
-    NFD_LOG_FACE_WARN("packet parse error (" << e.what() << "): DROP");
+    NFD_LOG_FACE_WARN("packet parse error (" << e.what() << "): DROP" << netPkt.type());
   }
 }
 
@@ -408,6 +465,8 @@ GenericLinkService::decodeInterest(const Block& netPkt, const lp::Packet& firstP
     interest->setTag(make_shared<lp::PitToken>(firstPkt.get<lp::PitTokenField>()));
   }
 
+  this->globalSp.receiveInterest(*interest);
+  this->globalSp.triggerCalObjectRate();
   this->receiveInterest(*interest, endpointId);
 }
 
@@ -462,6 +521,13 @@ GenericLinkService::decodeData(const Block& netPkt, const lp::Packet& firstPkt,
     }
   }
 
+  // 不处理本地命令通信
+  if (data->getName().getPrefix(1).toUri() != "/localhost") {
+      // TR = min(TR, OR)
+      this->globalSp.assignObjectRate(*data);
+  }
+
+  this->globalSp.triggerCalObjectRate();
   this->receiveData(*data, endpointId);
 }
 

daemon/face/generic-link-service.hpp

@@ -30,6 +30,7 @@
 #include "lp-fragmenter.hpp"
 #include "lp-reassembler.hpp"
 #include "lp-reliability.hpp"
+#include "queues.hpp"
 
 namespace nfd {
 namespace face {
@@ -164,6 +165,9 @@ public:
      *  being set with canOverrideMtuTo().
      */
     ssize_t overrideMtu = std::numeric_limits<ssize_t>::max();
+    
+    uint64_t C = 100000000; // 100Mbps
+    size_t vqSize = 20;
   };
 
   /** \brief counters provided by GenericLinkService
@@ -310,6 +314,9 @@ PUBLIC_WITH_TESTS_ELSE_PRIVATE:
   /// number of marked packets in the current incident of congestion
   size_t m_nMarkedSinceInMarkingState;
 
+  GlobalSP globalSp;
+  bool canScheduler = true;
+  std::function<void(std::shared_ptr<VirtualQueueItem>)> sendPacketCallback;
   friend class LpReliability;
 };
 

daemon/face/queues.cpp

@@ -0,0 +1,534 @@
+#include "queues.hpp"
+#include "common/global.hpp"
+#include <ndn-cxx/encoding/estimator.hpp>
+namespace nfd
+{
+    namespace face
+    {
+        NFD_LOG_INIT(QSCCPQueue);
+        bool VirtualQueue::push(int seq, const std::shared_ptr<const Interest> &interest,
+                                uint64_t dataSize, bool needInsert)
+        {
+            this->recvBits += dataSize;
+            if (!needInsert)
+            {
+                return false;
+            }
+            if (this->currentSize >= this->size)
+            {
+
+                return false;
+            }
+            this->lastSeq = seq;
+            VirtualQueueItem vqi;
+            vqi.seq = seq;
+            vqi.item = interest;
+            vqi.queue = shared_from_this();
+            vqi.dataSize = dataSize;
+            this->innerQueue.push_back(vqi);
+            this->currentSize++;
+            auto serviceClass = interest->getServiceClass();
+            if (serviceClass && *serviceClass > 7)
+            {
+                this->penaltyFactor = 0.5;
+            }
+            return true;
+        }
+        WDRRQueue::WDRRQueue(uint32_t rateLimit, uint32_t availableBandwidth, double weight, const std::string &name,
+                             uint32_t tos, size_t vqSize) : availableBandwidth(availableBandwidth),
+                                                            weight(weight), name(name), tos(tos), vqSize(vqSize)
+        {
+
+            if (rateLimit > 0)
+            {
+                this->needRateLimit = true;
+                this->tokenBucket = std::make_shared<TokenBucket>(rateLimit, 20 * 1000 * 8, name);
+                // this->tokenBucket->start();
+            }
+            else
+            {
+                this->needRateLimit = false;
+            }
+        }
+        void WDRRQueue::updateAvailableBandwidth(uint32_t C)
+        {
+            this->availableBandwidth = C;
+        }
+        bool WDRRQueue::appendInterest(const Interest &interest)
+        {
+            std::string key = interest.getName().getPrefix(1).toUri();
+            if (this->virtualQueues.count(key) == 0)
+            {
+                this->virtualQueues[key] = std::make_shared<VirtualQueue>(1, this->vqSize);
+                this->virtualQueues[key]->tos = this->tos;
+                this->flows.push_back(key);
+                this->totalWeight += 1;
+                for (auto &item : this->virtualQueues)
+                {
+                    item.second->objectRate = this->availableBandwidth * item.second->weight / this->totalWeight;
+                    item.second->objectRate = item.second->objectRate * item.second->penaltyFactor;
+                }
+            }
+            bool needInsert = true;
+            if (this->cdt > 0 && this->currentPacketNum >= this->cdt)
+            {
+
+                needInsert = false;
+            }
+            uint64_t dataSize = 9000;
+            auto dsz = interest.getDsz();
+            if (dsz)
+            {
+                dataSize = *dsz;
+            }
+            this->recvBits += dataSize;
+            auto success = this->virtualQueues[key]->push(0, interest.shared_from_this(), dataSize, needInsert);
+            if (success)
+            {
+                this->currentPacketNum++;
+            }
+            return success;
+        }
+        std::shared_ptr<VirtualQueueItem> WDRRQueue::scheduleNext()
+        {
+            if (this->nextScheduleQueue >= this->flows.size())
+            {
+                for (auto &key : this->flows)
+                {
+                    if (this->virtualQueues.count(key) == 0)
+                    {
+                        continue;
+                    }
+                    auto &item = this->virtualQueues[key];
+                    item->balance += item->weight * item->penaltyFactor * this->MTU_QUANTA;
+                }
+                this->nextScheduleQueue = 0;
+            }
+            std::shared_ptr<VirtualQueue> selectedVQ = nullptr;
+            for (; this->nextScheduleQueue < this->flows.size(); this->nextScheduleQueue++)
+            {
+                auto &item = this->virtualQueues[this->flows[this->nextScheduleQueue]];
+                if (item->empty())
+                {
+                    item->balance = 0;
+                    continue;
+                }
+                if (item->front().dataSize < item->balance)
+                {
+                    selectedVQ = item;
+                    break;
+                }
+                if (item->len() > 10)
+                {
+                }
+            }
+            if (selectedVQ == nullptr && this->currentPacketNum > 0)
+            {
+                return scheduleNext();
+            }
+            if (selectedVQ != nullptr)
+            {
+                if (needRateLimit && !this->tokenBucket->use(selectedVQ->front().dataSize))
+                {
+                    return nullptr;
+                }
+                auto item = std::make_shared<VirtualQueueItem>(selectedVQ->front());
+                selectedVQ->balance -= item->dataSize;
+                selectedVQ->pop();
+                if (selectedVQ->penaltyFactor < 0.9)
+                {
+                    item->item->setServiceClass(selectedVQ->tos + 8);
+                }
+                this->sendBits += item->dataSize;
+                this->currentPacketNum--;
+                return item;
+            }
+            return nullptr;
+        }
+        uint32_t WDRRQueue::calObjectRate(bool getCurrentRateOnly)
+        {
+            uint32_t currentRate = this->recvBits * 1000 / this->calObjectRateInterval;
+            uint32_t currentSendRate = this->sendBits * 1000 / this->calObjectRateInterval;
+            if (getCurrentRateOnly)
+            {
+                return currentRate;
+            }
+            // if (currentRate > 0) {
+            //     std::cout << "currentRate: " << currentRate << ", " << this->recvBits << "," << this->calObjectRateInterval << ", AvaliableBandwidth: " << this->availableBandwidth << std::endl;
+            // }
+            bool hasCleanFlow = false;
+            for (auto it = this->virtualQueues.begin(); it != this->virtualQueues.end();)
+            {
+                // std::cout << "key: " << it->first << ", recvBits" << it->second->recvBits << std::endl;
+                if (it->second->recvBits == 0 && it->second->empty())
+                {
+                    this->totalWeight -= it->second->weight;
+                    this->virtualQueues.erase(it++); // here is the key
+                    // std::cout << "do clean: " << std::endl;
+                    hasCleanFlow = true;
+                }
+                else
+                {
+                    it++;
+                }
+            }
+            if (hasCleanFlow)
+            {
+                this->flows.resize(this->virtualQueues.size());
+                int idx = 0;
+                for (auto it = this->virtualQueues.begin(); it != this->virtualQueues.end(); it++, idx++)
+                {
+                    this->flows[idx] = it->first;
+                }
+            }
+            double totalHightSpeedFlowWeight = 0;
+            int hightSppedFlowNum = 0;
+            uint32_t consumeRate = 0;
+            double totalWeightThisTerm = 0;
+            for (auto &item : this->virtualQueues)
+            {
+                totalWeightThisTerm += (item.second->weight * item.second->penaltyFactor);
+            }
+
+            for (auto &item : this->virtualQueues)
+            {
+                item.second->objectRate = this->availableBandwidth * item.second->weight * item.second->penaltyFactor / totalWeightThisTerm;
+                item.second->currentFlowRate = item.second->recvBits * 1000 / this->calObjectRateInterval;
+                item.second->currentRatio = item.second->currentFlowRate * 1.0 / item.second->objectRate;
+
+                auto targetHighFlowRate = (item.second->objectRate - 500000) * 1.0 / item.second->objectRate;
+                if (item.second->currentRatio > targetHighFlowRate)
+                {
+
+                    if (item.second->penaltyFactor > 0.99)
+                    {
+                        hightSppedFlowNum++;
+                        totalHightSpeedFlowWeight += item.second->weight;
+                    }
+                    consumeRate += item.second->objectRate;
+                }
+                else
+                {
+                    consumeRate += item.second->currentFlowRate;
+                }
+                NFD_LOG_DEBUG(this->name << " => currentRate: " << item.second->currentFlowRate << ", consumerRate: " << consumeRate << ", objectRate: " << item.second->objectRate << ", availableBandwidth" << this->availableBandwidth << ", weight: " << item.second->weight << ", totalWeight: " << this->totalWeight << ", currentSendRate: " << currentSendRate);
+                // std::cout << this->name << " => currentRate: " << item.second->currentFlowRate << ", consumerRate: " << consumeRate << ", objectRate: " << item.second->objectRate <<
+                //     ", availableBandwidth" << this->availableBandwidth << ", weight: " << item.second->weight << ", totalWeight: " <<
+                //     this->totalWeight << ", currentSendRate: " << currentSendRate << std::endl;
+
+                item.second->recvBits = 0;
+            }
+
+            if (consumeRate < this->availableBandwidth && hightSppedFlowNum > 0)
+            {
+
+                auto remainRate = this->availableBandwidth - consumeRate;
+                for (auto &item : this->virtualQueues)
+                {
+                    if (item.second->penaltyFactor > 0.99 && item.second->currentRatio > 0.8)
+                    {
+                        item.second->objectRate += (remainRate * item.second->weight / totalHightSpeedFlowWeight);
+                    }
+                    NFD_LOG_DEBUG(this->name << " => objectRate: " << item.second->objectRate << ", remainRate: " << remainRate << ", item.weight: " << item.second->weight << ", totalWeight: " << totalHightSpeedFlowWeight);
+                    // std::cout << this->name << " => objectRate: " << item.second->objectRate << ", remainRate: " <<
+                    // remainRate << ", item.weight: " << item.second->weight << ", totalWeight: " << totalHightSpeedFlowWeight << std::endl;
+                }
+            }
+
+            for (auto &item : this->virtualQueues)
+            {
+
+                if (item.second->currentFlowRate * 1.0 / item.second->objectRate > item.second->gredyRatio)
+                {
+                    if (item.second->gredyTimes < item.second->gredyThreshold)
+                    {
+                        item.second->gredyTimes++;
+                    }
+                    if (item.second->gredyTimes == item.second->gredyThreshold)
+                    {
+                        // std::cout << item.first << ": PenaltyFactor => " << item.second->gredyTimes << ", currentFlowRate: " << item.second->currentFlowRate << ", objectRate: " << item.second->objectRate << std::endl;
+                        item.second->penaltyFactor = 0.5;
+                        item.second->size = 20;
+                    }
+                }
+                else
+                {
+                    if (item.second->gredyTimes > 0)
+                    {
+                        item.second->gredyTimes--;
+                    }
+                }
+            }
+            // if (this->virtualQueues.size() > 0)
+            // {
+            //     std::cout << "\n\n=====================: " << this->availableBandwidth << std::endl;
+            // }
+            for (auto &item : this->virtualQueues)
+            {
+                if (item.second->penaltyFactor > 0.99)
+                {
+                    auto rate = this->calDiscountFactor(item.second->currentSize * 1.0 / item.second->size);
+                    item.second->objectRate *= rate;
+                }
+                // std::cout << item.first << ", currentFlowRate: " << item.second->currentFlowRate
+                //           << ", objectRate: " << item.second->objectRate << ", size: " << item.second->currentSize
+                //           << ", weight: " << item.second->weight << ", penaltyFactor: " << item.second->penaltyFactor
+                //           << std::endl;
+            }
+            // if (this->virtualQueues.size() > 0)
+            // {
+            //     std::cout << "\n----------------------" << std::endl;
+            // }
+            this->sendBits = 0;
+            this->recvBits = 0;
+            return currentRate;
+        }
+        uint32_t WDRRQueue::scheduleOnePacketForWFQ(uint32_t totalLastSeq)
+        {
+            if (this->cachePacketForWFQ != nullptr)
+            {
+                return this->cachePacketForWFQ->seq;
+            }
+            if (this->currentPacketNum == 0)
+            {
+                return UINT32_MAX;
+            }
+            this->cachePacketForWFQ = this->scheduleNext();
+            if (this->cachePacketForWFQ == nullptr)
+            {
+
+                return UINT32_MAX;
+            }
+            if (this->lastSeq == -1)
+            {
+                this->lastSeq = totalLastSeq;
+            }
+            this->cachePacketForWFQ->seq = this->lastSeq + 1.0 * this->cachePacketForWFQ->dataSize / this->weight;
+            if (this->currentPacketNum == 0)
+            {
+                this->lastSeq = -1;
+            }
+            else
+            {
+                this->lastSeq = this->cachePacketForWFQ->seq;
+            }
+
+            return this->cachePacketForWFQ->seq;
+        }
+        std::shared_ptr<VirtualQueueItem> DelaySensitiveServiceSP::scheduleNext()
+        {
+            auto res = this->CS7Queue->scheduleNext();
+            if (res == nullptr)
+            {
+                res = this->CS6Queue->scheduleNext();
+            }
+            if (res == nullptr)
+            {
+                res = this->EFQueue->scheduleNext();
+            }
+            return res;
+        }
+        uint32_t DelaySensitiveServiceSP::calObjectRate(bool getCurrentRateOnly)
+        {
+            uint32_t res = this->CS7Queue->calObjectRate(getCurrentRateOnly);
+            res += this->CS6Queue->calObjectRate(getCurrentRateOnly);
+            res += this->EFQueue->calObjectRate(getCurrentRateOnly);
+            return res;
+        }
+        std::shared_ptr<VirtualQueueItem> WFQQueue::scheduleNext()
+        {
+            std::shared_ptr<WDRRQueue> selectedQueue = nullptr;
+            uint32_t currentMinSeq = UINT32_MAX;
+            for (auto &queue : this->queues)
+            {
+                uint32_t seq = queue->scheduleOnePacketForWFQ(this->lastSeq);
+                if (seq < currentMinSeq)
+                {
+                    currentMinSeq = seq;
+                    selectedQueue = queue;
+                }
+            }
+            if (selectedQueue != nullptr)
+            {
+                this->lastSeq = currentMinSeq;
+                auto selectPacket = selectedQueue->frontForWFQ(this->lastSeq);
+                selectedQueue->popForWFQ();
+
+                return selectPacket;
+            }
+            return nullptr;
+        }
+        uint32_t WFQQueue::calObjectRate(uint32_t availableBandwidth)
+        {
+            int totalRate = 0;
+            double totalHightSpeedFlowWeight = 0;
+            int hightSppedFlowNum = 0;
+            double totalHasFlowQueueWeight = 0;
+            int hasFlowQueueNum = 0;
+            for (auto &queue : this->queues)
+            {
+                auto recvRate = queue->calObjectRate(true);
+                queue->originAvaliableBandwidth = (queue->getWeight() / totalWeight) * availableBandwidth;
+                auto remainRate = queue->getRemainBandwidthRate(queue->originAvaliableBandwidth);
+                if (remainRate < 0.2)
+                {
+                    hightSppedFlowNum++;
+                    totalHightSpeedFlowWeight += queue->getWeight();
+                    totalRate += queue->originAvaliableBandwidth;
+                }
+                else if (remainRate < 0.95)
+                {
+                    hasFlowQueueNum++;
+                    totalHasFlowQueueWeight += queue->getWeight();
+                    totalRate += recvRate;
+                }
+                else
+                {
+                    totalRate += recvRate;
+                }
+            }
+            this->availableBandwidth = availableBandwidth;
+            int remain = this->availableBandwidth - totalRate;
+            if (hightSppedFlowNum > 0)
+            {
+                // std::cout << "high speed flow num: " << hightSppedFlowNum << ", totalHightSpeedFlowWeight: " << totalHightSpeedFlowWeight << std::endl;
+                for (auto &queue : this->queues)
+                {
+                    if (queue->getRemainBandwidthRate(queue->originAvaliableBandwidth) < 0.2)
+                    {
+                        // std::cout << "!!!" << queue->getWeight() * remain * 1.0 / totalHightSpeedFlowWeight +
+                        // queue->originAvaliableBandwidth << std::endl;
+                        queue->updateAvailableBandwidth(
+                            queue->getWeight() * remain * 1.0 / totalHightSpeedFlowWeight +
+                            queue->originAvaliableBandwidth);
+                    }
+                    else
+                    {
+                        queue->updateAvailableBandwidth(queue->originAvaliableBandwidth);
+                    }
+                }
+            }
+            else
+            {
+                // std::cout << "has flow queue num: " << hasFlowQueueNum << ", totalHasFlowQueueWeight: " << totalHasFlowQueueWeight << std::endl;
+                for (auto &queue : this->queues)
+                {
+                    if (queue->getRemainBandwidthRate(queue->originAvaliableBandwidth) < 0.95)
+                    {
+                        queue->updateAvailableBandwidth(
+                            queue->getWeight() * remain * 1.0 / totalHasFlowQueueWeight +
+                            queue->originAvaliableBandwidth);
+                    }
+                    else
+                    {
+                        queue->updateAvailableBandwidth(queue->originAvaliableBandwidth);
+                    }
+                }
+            }
+            uint32_t currentTotalRate = 0;
+
+            for (auto &queue : this->queues)
+            {
+                currentTotalRate += queue->calObjectRate(false);
+            }
+            return currentTotalRate;
+        }
+        std::shared_ptr<VirtualQueueItem> GlobalSP::scheduleNext()
+        {
+            std::shared_ptr<VirtualQueueItem> selectedPacket = this->delaySensitiveServiceSp.scheduleNext();
+            if (selectedPacket == nullptr)
+            {
+                selectedPacket = wfqQueue.scheduleNext();
+            }
+            if (selectedPacket != nullptr)
+            {
+                selectedPacket->waitNsTime = selectedPacket->dataSize * 1000000000.0 / this->availableBandwidth + 1;
+                // 计算 DR => DR'(/p, j) = min(OR(/p, j), RR(/p, j) / sum(RR(/p)) * sum(AR(/p)))
+                auto currentObjectRate = selectedPacket->queue->objectRate;
+                auto downstreamRate = selectedPacket->item->getDownstreamRate();
+                if (downstreamRate)
+                {
+                    if (*downstreamRate > currentObjectRate)
+                    {
+                        selectedPacket->item->setDownstreamRate(currentObjectRate);
+                    }
+                }
+                else
+                {
+                    selectedPacket->item->setDownstreamRate(currentObjectRate);
+                }
+            }
+            return selectedPacket;
+        }
+        std::shared_ptr<WDRRQueue> GlobalSP::getQueueByTos(int tosValue)
+        {
+            switch (tosValue)
+            {
+            case 0:
+                return this->delaySensitiveServiceSp.getCS7Queue();
+            case 1:
+                return this->delaySensitiveServiceSp.getCS6Queue();
+            case 2:
+                return this->delaySensitiveServiceSp.getEFQueue();
+            case 3:
+                return this->wfqQueue.getAF4Queue();
+            case 4:
+                return this->wfqQueue.getAF3Queue();
+            case 5:
+                return this->wfqQueue.getAF2Queue();
+            case 6:
+                return this->wfqQueue.getAF1Queue();
+            case 7:
+            default:
+                return this->wfqQueue.getBEQueue();
+            }
+        }
+        bool GlobalSP::appendInterest(const Interest &interest)
+        {
+            int serviceClass = 7;
+            if (interest.getServiceClass())
+            {
+                serviceClass = (*interest.getServiceClass()) % 8;
+            }
+            // std::cout << "appendInterest => serviceClass: " << serviceClass << std::endl;
+            return this->getQueueByTos(serviceClass)->appendInterest(interest);
+        }
+        void GlobalSP::receiveInterest(const Interest &interest)
+        {
+            ndn::encoding::EncodingEstimator estimator;
+            auto size = interest.wireEncode(estimator) + 9;
+            this->interestBytes += size;
+            // std::cout << this << "receiveInterest => size: " << size << ", interestBytes: " << this->interestBytes << std::endl;
+        }
+
+        void GlobalSP::calObjectRate(bool getCurrentRateOnly)
+        {
+            // getScheduler().schedule(time::nanoseconds(1000000 * this->calObjectRateInterval),
+            //                             [this] { this->calObjectRate(false); });
+            uint32_t remain = this->availableBandwidth;
+            remain -= this->interestBytes * 8000 / this->calObjectRateInterval;
+            this->interestBytes = 0;
+            remain -= this->delaySensitiveServiceSp.calObjectRate(false);
+
+            // std::cout << "AvaliableBandwidth: " << this->availableBandwidth << ", remain: " << remain << ", interestBytes" << this->interestBytes << std::endl;
+
+            this->wfqQueue.calObjectRate(remain);
+        }
+        void GlobalSP::assignObjectRate(const Data &data)
+        {
+            std::string key = data.getName().getPrefix(1).toUri();
+            int serviceClass = 7;
+            if (data.getServiceClass())
+            {
+                serviceClass = (*data.getServiceClass()) % 8;
+            }
+            // std::cout << "assignObjectRate => serviceClass: " << serviceClass << std::endl;
+            auto objectRate = this->getQueueByTos(serviceClass)->getObjectRateByFlowKey(key);
+            auto targetRate = data.getTargetRate();
+            if (!targetRate || *targetRate > objectRate)
+            {
+                data.setTargetRate(objectRate);
+            }
+        }
+    }
+}

daemon/face/queues.hpp

@@ -0,0 +1,309 @@
+#ifndef NS_QUEUES_HPP
+#define NS_QUEUES_HPP
+#include <deque>
+#include <unordered_map>
+#include <memory>
+#include <iostream>
+#include "common/global.hpp"
+#include "link-service.hpp"
+#include <ndn-cxx/lp/tags.hpp>
+#include "TokenBucket.hpp"
+namespace nfd {
+    namespace face {
+        class VirtualQueue;
+        struct VirtualQueueItem {
+            int seq;
+            std::shared_ptr<const Interest> item;
+            uint32_t dataSize;
+            uint32_t waitNsTime;
+            std::shared_ptr<VirtualQueue> queue;
+        };
+        class VirtualQueue : public std::enable_shared_from_this<VirtualQueue>{
+        public:
+            explicit VirtualQueue(double weight, int size) :
+                    std::enable_shared_from_this<VirtualQueue>(),
+                    penaltyFactor(1.0),
+                    weight(weight), size(size), currentSize(0), lastSeq(0),
+                    C(0) {
+            }
+            void enableTokenBucket(uint32_t limitRate) {
+                this->C = limitRate;
+            }
+            bool push(int seq, const std::shared_ptr<const Interest>& interest,
+                      uint64_t dataSize, bool needInsert);
+            bool empty() {
+                return this->innerQueue.empty();
+            }
+            uint32_t frontSeq() {
+                if (this->innerQueue.empty()) {
+                    return this->lastSeq;
+                }
+                return this->innerQueue.front().seq;
+            }
+            VirtualQueueItem front() {
+                return this->innerQueue.front();
+            }
+            uint32_t backSeq() {
+                return this->lastSeq;
+            }
+            VirtualQueueItem back() {
+                return this->innerQueue.back();
+            }
+            void pop() {
+                if (this->currentSize <= 0) {
+                    return;
+                }
+                this->currentSize--;
+                this->innerQueue.pop_front();
+            }
+            void popBack() {
+                if (this->currentSize <= 0) {
+                    return;
+                }
+                this->currentSize--;
+                this->innerQueue.pop_back();
+            }
+            int len() {
+                return this->innerQueue.size();
+            }
+            void setLastSeq(uint32_t seq) {
+                this->lastSeq = seq;
+            }
+            uint32_t getLastSeq() {
+                return lastSeq;
+            }
+            void setWeight(double weight) {
+                this->weight = weight;
+            }
+            double getWeight() {
+                return this->weight;
+            }
+        public:
+            uint32_t objectRate;
+            double penaltyFactor = 1.0;
+            uint32_t gredyTimes = 0;
+            double gredyRatio = 1.2;
+            double gredyThreshold = 10;
+            double weight = 1.0;
+            uint32_t tos = 5;
+            double balance = 0;
+            int size = 0;
+            int currentSize = 0;
+            uint32_t recvBits = 0;
+            uint32_t currentFlowRate = 0;
+            double currentRatio = 0.0;
+        private:
+            uint32_t lastSeq;
+            int C;
+            std::deque<VirtualQueueItem> innerQueue;
+        };
+        class WDRRQueue {
+        public:
+            explicit WDRRQueue(uint32_t rateLimit, uint32_t availableBandwidth, double weight, const std::string& name,
+                                uint32_t tos = 5, size_t vqSize = 20);
+            void updateAvailableBandwidth(uint32_t C);
+            bool appendInterest(const Interest &interest);
+            std::shared_ptr<VirtualQueueItem> scheduleNext();
+            uint32_t calObjectRate(bool getCurrentRateOnly);
+            void setCDTLimit(int cdt) {
+                this->cdt = cdt;
+            }
+            uint32_t scheduleOnePacketForWFQ(uint32_t totalLastSeq);
+            std::shared_ptr<VirtualQueueItem> frontForWFQ(uint32_t totalLastSeq) {
+                if (this->cachePacketForWFQ == nullptr) {
+                    this->scheduleOnePacketForWFQ(totalLastSeq);
+                }
+                return this->cachePacketForWFQ;
+            }
+            void popForWFQ() {
+                this->cachePacketForWFQ = nullptr;
+            }
+            double getRemainBandwidthRate(double curAvailableBandWidth) {
+                auto currentRate = this->calObjectRate(true);
+                if (currentRate > curAvailableBandWidth) {
+                    return 0;
+                }
+                return (curAvailableBandWidth - currentRate) * 1.0 / curAvailableBandWidth;
+            }
+            uint32_t getRemainBandwidth() {
+                return this->availableBandwidth - this->calObjectRate(true);
+            }
+            double getWeight() {
+                return this->weight;
+            }
+            uint32_t getAvailableBandwidth() {
+                return this->availableBandwidth;
+            }
+            /**
+             * @brief 
+             * 
+             * \sigma=\frac{1}{3}\arccot\funcapply(8\delta-4)
+             * @param x 
+             * @return double 
+             */
+            double calDiscountFactor(double x) {
+                return 1.0 / 3 * (M_PI / 2 - atan(8 * x - 4)) + 0.034;
+            }
+            /**
+             * @brief Get the Object Rate By Flow Key object
+             * 
+             * @param key 
+             * @return uint32_t 
+             */
+            uint32_t getObjectRateByFlowKey(const std::string& key) {
+                if (this->virtualQueues.count(key) == 0) {
+                    std::cout << "FUCK: " << key << ", count: " << this->virtualQueues.size() << 
+                    ", flow: " << this->flows.size() << "availbleRate: " << this->availableBandwidth << std::endl;
+                    return 0;
+                }
+                return this->virtualQueues[key]->objectRate;
+            }
+
+            void triggerByCurrentTime(long currentTime) {
+                if (this->needRateLimit) {
+                    this->tokenBucket->generateTokenByCurrentTime(currentTime);
+                }
+            }
+
+        public:
+            double originAvaliableBandwidth = 0;
+        private:
+            std::unordered_map<std::string, std::shared_ptr<VirtualQueue>> virtualQueues;
+            std::vector<std::string> flows;
+            size_t nextScheduleQueue = 0;
+            std::shared_ptr<TokenBucket> tokenBucket;
+            uint32_t availableBandwidth;
+            // uint32_t extraBandwidth;
+            double weight;
+            std::string name;
+            uint32_t tos = 5;
+            bool needRateLimit;
+            double totalWeight;
+            uint32_t sendBits = 0;
+            uint32_t recvBits = 0;
+            int vqSize = 20;
+            int cdt = -1;
+            int currentPacketNum = 0;
+            int MTU_QUANTA = 2 * 1000 * 8;
+            int calObjectRateInterval = 100;
+            int64_t lastSeq = -1;
+            std::shared_ptr<VirtualQueueItem> cachePacketForWFQ = nullptr;
+        };
+        class DelaySensitiveServiceSP {
+        public:
+            explicit DelaySensitiveServiceSP(uint32_t availableBandwidth, size_t vqSize = 20) {
+                this->CS7Queue = std::make_shared<WDRRQueue>((uint32_t) (0.2 * availableBandwidth), (uint32_t) (0.2 * availableBandwidth), 2, "CS7", 2, vqSize);
+                this->CS6Queue = std::make_shared<WDRRQueue>((uint32_t) (0.2 * availableBandwidth), (uint32_t) (0.2 * availableBandwidth), 2, "CS6", 1, vqSize);
+                this->EFQueue = std::make_shared<WDRRQueue>((uint32_t) (0.3 * availableBandwidth), (uint32_t) (0.3 * availableBandwidth), 3, "EF", 0, vqSize);
+            }
+            std::shared_ptr<VirtualQueueItem> scheduleNext();
+            uint32_t calObjectRate(bool getCurrentRateOnly);
+            std::shared_ptr<WDRRQueue> getCS7Queue() {
+                return this->CS7Queue;
+            }
+            std::shared_ptr<WDRRQueue> getCS6Queue() {
+                return this->CS6Queue;
+            }
+            std::shared_ptr<WDRRQueue> getEFQueue() {
+                return this->EFQueue;
+            }
+            void triggerByCurrentTime(long currentTime) {
+                this->CS7Queue->triggerByCurrentTime(currentTime);
+                this->CS6Queue->triggerByCurrentTime(currentTime);
+                this->EFQueue->triggerByCurrentTime(currentTime);
+            }
+        private:
+            std::shared_ptr<WDRRQueue> CS7Queue;
+            std::shared_ptr<WDRRQueue> CS6Queue;
+            std::shared_ptr<WDRRQueue> EFQueue;
+        };
+        class WFQQueue {
+        public:
+            explicit WFQQueue(uint32_t availableBandwidth, size_t vqSize) :
+                    AF4Weight(5),
+                    AF3Weight(4),
+                    AF2Weight(3),
+                    AF1Weight(2),
+                    BEWeight(1),
+                    totalWeight(AF4Weight + AF3Weight + AF2Weight + AF1Weight + BEWeight),
+                    queues(5),
+                    lastSeq(0),
+                    availableBandwidth(availableBandwidth) {
+                this->queues[0] = std::make_shared<WDRRQueue>(0, availableBandwidth * AF4Weight / totalWeight, 5, "AF4", 3, vqSize);
+                this->queues[1] = std::make_shared<WDRRQueue>(0, availableBandwidth * AF3Weight / totalWeight, 4, "AF3", 4, vqSize);
+                this->queues[2] = std::make_shared<WDRRQueue>(0, availableBandwidth * AF2Weight / totalWeight, 3, "AF2", 5, vqSize);
+                this->queues[3] = std::make_shared<WDRRQueue>(0, availableBandwidth * AF1Weight / totalWeight, 2, "AF1", 6, vqSize);
+                this->queues[4] = std::make_shared<WDRRQueue>(0, availableBandwidth * BEWeight / totalWeight, 1, "BE", 7, vqSize);
+                for (auto &queue: this->queues) {
+                    queue->setCDTLimit(this->cdt);
+                }
+            }
+            std::shared_ptr<VirtualQueueItem> scheduleNext();
+            uint32_t calObjectRate(uint32_t availableBandwidth);
+            std::shared_ptr<WDRRQueue> getAF4Queue() {
+                return this->queues[0];
+            }
+            std::shared_ptr<WDRRQueue> getAF3Queue() {
+                return this->queues[1];
+            }
+            std::shared_ptr<WDRRQueue> getAF2Queue() {
+                return this->queues[2];
+            }
+            std::shared_ptr<WDRRQueue> getAF1Queue() {
+                return this->queues[3];
+            }
+            std::shared_ptr<WDRRQueue> getBEQueue() {
+                return this->queues[4];
+            }
+        private:
+            double AF4Weight = 5;
+            double AF3Weight = 4;
+            double AF2Weight = 3;
+            double AF1Weight = 2;
+            double BEWeight = 1;
+            double totalWeight = 15;
+            std::vector<std::shared_ptr<WDRRQueue>> queues;
+            uint32_t lastSeq;
+            u_int32_t availableBandwidth;
+            int cdt = 200;
+            int hql;
+        };
+        class GlobalSP {
+        public:
+            explicit GlobalSP(uint32_t availableBandwidth, size_t vqSize = 20) :
+                    delaySensitiveServiceSp(availableBandwidth, vqSize),
+                    wfqQueue(availableBandwidth, vqSize),
+                    availableBandwidth(availableBandwidth) {
+                lastTriggerTime = getCurrentTime();
+                // getScheduler().schedule(time::nanoseconds(1000000 * this->calObjectRateInterval),
+                //                         [this] { this->calObjectRate(false); });
+            }
+            std::shared_ptr<VirtualQueueItem> scheduleNext();
+            bool appendInterest(const Interest &interest);
+            void receiveInterest(const Interest &interest);
+            void calObjectRate(bool getCurrentRateOnly);
+            std::shared_ptr<WDRRQueue> getQueueByTos(int tosValue);
+            long getCurrentTime() {
+                return std::chrono::duration_cast<std::chrono::milliseconds>(
+                        std::chrono::system_clock::now().time_since_epoch()).count();
+            }
+            void triggerCalObjectRate() {
+                auto currentTime = getCurrentTime();
+                this->delaySensitiveServiceSp.triggerByCurrentTime(currentTime);
+                if (currentTime - lastTriggerTime > calObjectRateInterval) {
+                    this->calObjectRate(false);
+                    lastTriggerTime = currentTime;
+                }
+            }
+            void assignObjectRate(const Data &data);
+        private:
+            DelaySensitiveServiceSP delaySensitiveServiceSp;
+            WFQQueue wfqQueue;
+            int calObjectRateInterval = 100;
+            uint32_t availableBandwidth;
+            size_t interestBytes = 0;
+            long lastTriggerTime = 0;
+        };
+    }
+}
+#endif //NS_QUEUES_HPP

daemon/face/tcp-channel.cpp

@@ -123,6 +123,9 @@ TcpChannel::createFace(ip::tcp::socket&& socket,
     if (params.defaultCongestionThreshold) {
       options.defaultCongestionThreshold = *params.defaultCongestionThreshold;
     }
+    if (params.bandwidth) {
+      options.C = *params.bandwidth;
+    }
 
     auto linkService = make_unique<GenericLinkService>(options);
     auto faceScope = m_determineFaceScope(socket.local_endpoint().address(),

daemon/face/udp-channel.cpp

@@ -184,6 +184,10 @@ UdpChannel::createFace(const udp::Endpoint& remoteEndpoint,
     options.overrideMtu = *params.mtu;
   }
 
+  if (params.bandwidth) {
+    options.C = *params.bandwidth;
+  }
+
   auto linkService = make_unique<GenericLinkService>(options);
   auto transport = make_unique<UnicastUdpTransport>(std::move(socket), params.persistency,
                                                     m_idleFaceTimeout);

daemon/fw/PCONStrategy.cpp

@@ -1,174 +0,0 @@
-#include "PCONStrategy.hpp"
-#include <ndn-cxx/lp/tags.hpp>
-
-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<PCONStrategy>(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<nfd::pit::Entry> &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<Face *> 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<pit::Entry> &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<nfd::pit::Entry> &pitEntry)
-        {
-            this->processNack(ingress.face, nack, pitEntry);
-        }
-
-        void
-        PCONStrategy::initializeForwMap(MtForwardingInfo *measurementInfo,
-                                        const fib::NextHopList &nextHops)
-        {
-            int lowestId = std::numeric_limits<int>::max();
-            int minCost = std::numeric_limits<int>::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);
-            }
-        }
-    }
-}

daemon/fw/PCONStrategy.hpp

@@ -1,150 +0,0 @@
-#ifndef NFD_DAEMON_FW_PCON_STRATEGY_HPP
-#define NFD_DAEMON_FW_PCON_STRATEGY_HPP
-
-#include <boost/random/mersenne_twister.hpp>
-#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 <unordered_map>
-#include <limits>
-#include "mt-forward-info.hpp"
-
-namespace nfd
-{
-    namespace fw
-    {
-        class PCONStrategy : public Strategy, public ProcessNackTraits<PCONStrategy>
-        {
-        public:
-            explicit PCONStrategy(Forwarder &forwarder, const Name &name = getStrategyName());
-
-            ~PCONStrategy() override = default;
-
-            void
-            afterReceiveInterest(const FaceEndpoint &ingress, const Interest &interest,
-                                 const shared_ptr<pit::Entry> &pitEntry) override;
-
-            void
-            beforeSatisfyInterest(const shared_ptr<pit::Entry> &pitEntry,
-                                  const FaceEndpoint &ingress, const Data &data) override;
-
-            void
-            afterReceiveNack(const FaceEndpoint &ingress, const lp::Nack &nack,
-                             const shared_ptr<pit::Entry> &pitEntry) override;
-
-            static const Name &
-            getStrategyName();
-
-        protected:
-            friend ProcessNackTraits<PCONStrategy>;
-
-        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<MtForwardingInfo>();
-                }
-                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<MtForwardingInfo>().first;
-            }
-
-            std::tuple<Name, MtForwardingInfo *>
-            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<MtForwardingInfo>());
-            }
-
-            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<std::string, unique_ptr<MtForwardingInfo>> innerMap;
-            RetxSuppressionExponential m_retxSuppression;
-        };
-    }
-}
-
-#endif

daemon/fw/mt-forward-info.hpp

@@ -1,115 +0,0 @@
-#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<FaceId, double>
-            getForwPercMap() const
-            {
-                return m_forwPercMap;
-            }
-
-            int
-            getFaceCount()
-            {
-                std::vector<FaceId> 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<FaceId, double> m_forwPercMap;
-
-            std::unordered_set<FaceId> m_disabledFaces;
-        };
-
-    } //fw
-} //nfd
-
-#endif

daemon/fw/qsccp-strategy.cpp

@@ -0,0 +1,527 @@
+#include "qsccp-strategy.hpp"
+#include <ndn-cxx/lp/tags.hpp>
+#include <boost/random/discrete_distribution.hpp>
+#include <boost/random/uniform_real_distribution.hpp>
+#include "common/global.hpp"
+
+namespace nfd
+{
+    namespace fw
+    {
+
+        NFD_LOG_INIT(QSCCPStrategy);
+        NFD_REGISTER_STRATEGY(QSCCPStrategy);
+        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+        //// QSCCPStrategy
+        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+        const time::milliseconds QSCCPStrategy::RETX_SUPPRESSION_INITIAL(10);
+        const time::milliseconds QSCCPStrategy::RETX_SUPPRESSION_MAX(250);
+
+        QSCCPStrategy::QSCCPStrategy(nfd::Forwarder &forwarder, const ndn::Name &name)
+            : Strategy(forwarder), ProcessNackTraits<QSCCPStrategy>(this)
+        {
+            this->setInstanceName(makeInstanceName(name, getStrategyName()));
+        }
+
+        void
+        QSCCPStrategy::afterReceiveInterest(const nfd::FaceEndpoint &ingress, const ndn::Interest &interest,
+                                            const std::shared_ptr<nfd::pit::Entry> &pitEntry)
+        {
+            RetxSuppressionResult suppression = m_retxSuppression.decidePerPitEntry(*pitEntry);
+            if (suppression == RetxSuppressionResult::SUPPRESS)
+            {
+                NFD_LOG_DEBUG(interest << " from=" << ingress << " suppressed");
+                return;
+            }
+
+            const fib::Entry &fibEntry = this->lookupFib(*pitEntry);
+            const fib::NextHopList &nexthops = fibEntry.getNextHops();
+            auto selected = nexthops.end();
+
+            if (ingress.face.getScope() != ndn::nfd::FACE_SCOPE_NON_LOCAL)
+            {
+                selected = std::find_if(nexthops.begin(), nexthops.end(), [&](const auto &nexthop)
+                                        { return isNextHopEligible(ingress.face, interest, nexthop, pitEntry); });
+            }
+            else
+            {
+                auto selectedFaceId = rs.afterReceiveInterest(ingress, interest, pitEntry, fibEntry);
+                if (selectedFaceId == ingress.face.getId() || selectedFaceId == 0)
+                {
+                    selected = nexthops.end();
+                }
+                else
+                {
+                    for (auto it = nexthops.begin(); it != nexthops.end(); it++)
+                    {
+                        if (it->getFace().getId() == selectedFaceId)
+                        {
+                            selected = it;
+                            break;
+                        }
+                    }
+                }
+            }
+
+            if (selected == nexthops.end())
+            {
+                lp::NackHeader nackHeader;
+                nackHeader.setReason(lp::NackReason::NO_ROUTE);
+                this->sendNack(pitEntry, ingress.face, nackHeader);
+                this->rejectPendingInterest(pitEntry);
+                return;
+            }
+            else
+            {
+                this->sendInterest(pitEntry, selected->getFace(), interest);
+            }
+        }
+
+        void QSCCPStrategy::afterReceiveNack(const nfd::FaceEndpoint &ingress, const ndn::lp::Nack &nack,
+                                             const std::shared_ptr<nfd::pit::Entry> &pitEntry)
+        {
+            rs.afterReceiveNack(ingress, nack, pitEntry);
+            this->processNack(ingress.face, nack, pitEntry);
+        }
+
+        void QSCCPStrategy::beforeSatisfyInterest(const std::shared_ptr<nfd::pit::Entry> &pitEntry,
+                                                  const nfd::FaceEndpoint &ingress, const ndn::Data &data)
+        {
+            if (ingress.face.getId() <= 256)
+            {
+                return;
+            }
+            rs.beforeSatisfyInterest(pitEntry, ingress, data);
+            NFD_LOG_DEBUG("BeforeSatisfyInterest");
+        }
+
+        const ndn::Name &QSCCPStrategy::getStrategyName()
+        {
+            static Name strategyName("/localhost/nfd/strategy/QSCCP/%FD%01");
+            return strategyName;
+        }
+
+        void QSCCPStrategy::afterContentStoreHit(const shared_ptr<pit::Entry> &pitEntry,
+                                                 const FaceEndpoint &ingress, const Data &data)
+        {
+            NFD_LOG_DEBUG("After Content Store Hit: " << pitEntry->getName().toUri());
+        }
+
+        void QSCCPStrategy::afterReceiveData(const shared_ptr<pit::Entry> &pitEntry, const FaceEndpoint &ingress,
+                                             const Data &data)
+        {
+
+            if (ingress.face.getScope() != ndn::nfd::FACE_SCOPE_NON_LOCAL)
+            {
+                this->sendDataToAll(pitEntry, ingress.face, data);
+                return;
+            }
+
+            rs.beforeSatisfyInterest(pitEntry, ingress, data);
+            //
+            // this->beforeSatisfyInterest(pitEntry, ingress, data);
+
+            // Override sendDataToAll
+            std::set<Face *> pendingDownstreams;
+            auto now = time::steady_clock::now();
+
+            // remember pending downstreams
+            for (const pit::InRecord &inRecord : pitEntry->getInRecords())
+            {
+                if (inRecord.getExpiry() > now)
+                {
+                    if (inRecord.getFace().getId() == ingress.face.getId() &&
+                        inRecord.getFace().getLinkType() != ndn::nfd::LINK_TYPE_AD_HOC)
+                    {
+                        continue;
+                    }
+                    pendingDownstreams.emplace(&inRecord.getFace());
+                }
+            }
+
+            // int downStreamNum = 0;
+            for (const auto &pendingDownstream : pendingDownstreams)
+            {
+                FaceEndpoint egress(*pendingDownstream, 0);
+                if (ingress.face.getId() > 256)
+                {
+                    this->rs.beforeSendData(pitEntry, egress, data);
+                }
+                this->sendData(pitEntry, data, egress.face);
+                // downStreamNum++;
+            }
+            // NFD_LOG_DEBUG("DownStreamNum: " << downStreamNum);
+        }
+
+        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+        //// RateCollection
+        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+        void RateCollection::doPeriodicCleaning(bool needErase, uint32_t defaultValue)
+        {
+            // std::cout << "\n\n===================== doPeriodicCleaning ==========================" << std::endl;
+            bool hasClean = false;
+            for (auto it = this->rates.begin(); it != this->rates.end();)
+            {
+                if (it->second->recvCount == 0)
+                {
+                    if (needErase)
+                    {
+                        this->totalRate -= it->second->rateValue;
+                        // std::cout << "hasClean: " << it->second->recvCount << ", id: " << it->second->id << std::endl;
+                        this->rates.erase(it++); // here is the key
+                    }
+                    else
+                    {
+                        this->addOrUpdateRate(it->second->id, defaultValue);
+                    }
+
+                    hasClean = true;
+                }
+                else
+                {
+                    // std::cout << "recvCount: " << it->second->recvCount << ", id: " << it->second->id << std::endl;
+                    it->second->recvCount = 0;
+                    it++;
+                }
+            }
+
+            if (needErase && hasClean)
+            {
+                this->items.resize(this->rates.size());
+                int idx = 0;
+                for (auto it = this->rates.begin(); it != this->rates.end(); it++, idx++)
+                {
+                    this->items[idx] = it->second;
+                }
+            }
+        }
+
+        bool RateCollection::addOrUpdateRate(face::FaceId id, uint32_t rate)
+        {
+            bool needUpdate = false;
+            bool newFace = false;
+            // NFD_LOG_DEBUG("addOrUpdateRate: " << this->name << ", " << id << ", " << rate);
+            if (this->rates.count(id) == 0)
+            {
+                needUpdate = true;
+                this->rates[id] = std::make_shared<RateItem>(rate, id);
+                this->rates[id]->recvCount = 1;
+                this->items.push_back(this->rates[id]);
+                this->totalRate += rate;
+                newFace = true;
+            }
+            else
+            {
+                if (rate != this->rates[id]->rateValue)
+                {
+                    needUpdate = true;
+                }
+                this->rates[id]->recvCount += 1;
+                this->totalRate = this->totalRate - this->rates[id]->rateValue + rate;
+                this->rates[id]->rateValue = rate;
+            }
+
+            if (needUpdate)
+            {
+                for (size_t i = 0; i < this->items.size(); i++)
+                {
+                    if (this->totalRate < 1)
+                    {
+                        this->items[i]->ratio = 1 / this->items.size();
+                    }
+                    else
+                    {
+                        this->items[i]->ratio = this->items[i]->rateValue * 1.0 / this->totalRate;
+                    }
+                }
+            }
+            return newFace;
+        }
+
+        face::FaceId RateCollection::selectNextFace(uint64_t dataSize, face::FaceId exceptId)
+        {
+            if (this->nextForwardFace >= this->items.size())
+            {
+                for (auto &item : this->items)
+                {
+                    if (item->balance < 100000)
+                    {
+                        item->balance += (item->ratio * this->MTU_QUANTA);
+                        // std::cout << "banlance+: " << item->balance << ", ratio: " << item->ratio << std::endl;
+                    }
+                    else
+                    {
+                        // std::cout << "banlance-: " << item->balance << ", ratio: " << item->ratio << std::endl;
+                    }
+                    // NFD_LOG_DEBUG("Begin select next face: " << item->balance << ", " << item->ratio);
+                }
+                this->nextForwardFace = 0;
+            }
+
+            face::FaceId selectedId = 0;
+            size_t exceptCount = 0;
+            for (; this->nextForwardFace < this->items.size(); this->nextForwardFace++)
+            {
+                if (this->items[this->nextForwardFace]->balance > dataSize &&
+                    this->items[this->nextForwardFace]->id != exceptId)
+                {
+                    selectedId = this->items[this->nextForwardFace]->id;
+                    this->items[this->nextForwardFace]->balance -= dataSize;
+                    break;
+                }
+                else if (this->items[this->nextForwardFace]->id == exceptId || this->items[this->nextForwardFace]->ratio < 10e-6)
+                {
+                    exceptCount++;
+                }
+            }
+
+            if (selectedId == 0 && (this->items.size() - exceptCount) > 0)
+            {
+                // std::cout << "select: " << this->items.size() << ", count: " << exceptCount << std::endl;
+                return selectNextFace(dataSize, exceptId);
+            }
+            else
+            {
+                return selectedId;
+            }
+        }
+
+        face::FaceId RateCollection::selectFaceByRatio(const FaceEndpoint &ingress, const fib::Entry &fibEntry, const ndn::Interest &interest)
+        {
+            if (this->items.empty())
+            {
+                const fib::NextHopList &nexthops = fibEntry.getNextHops();
+                for (auto nexthop : nexthops)
+                {
+                    this->addOrUpdateRate(nexthop.getFace().getId(), 200000);
+                }
+
+                if (this->items.empty())
+                {
+                    return 0;
+                }
+            }
+
+            auto inId = ingress.face.getId();
+
+            if (this->items.size() == 1)
+            {
+                // NFD_LOG_DEBUG("ONE ratio => " << ", " << this->items[0]->rateValue);
+                if (this->items[0]->id == inId)
+                {
+                    std::cout << "1" << std::endl;
+                    return 0;
+                }
+                return this->items[0]->id;
+            }
+            else
+            {
+                // auto res = ingress.face.getId();
+
+                uint64_t dataSize = 9000;
+                auto dsz = interest.getDsz();
+                if (dsz)
+                {
+                    dataSize = *dsz;
+                }
+                // std::cout << "nexthop num: " << fibEntry.getNextHops().size() << ", item size: " << this->items.size() << std::endl;
+                return this->selectNextFace(dataSize, inId);
+            }
+        }
+
+        double RateCollection::getRatioByFace(face::FaceId id)
+        {
+            if (totalRate == 0 && this->getFaceNum() > 0)
+            {
+                return 1.0 / this->getFaceNum();
+            }
+            else if (rates.count(id) == 0)
+            {
+                return 0.01;
+            }
+            else
+            {
+                return this->rates[id]->rateValue * 1.0 / this->totalRate;
+            }
+        }
+
+        void RateCollection::calSplitRateByRate()
+        {
+            if (this->totalAvaliableRate == 0 || this->items.size() == 0)
+            {
+                return;
+            }
+            if (this->items.size() == 1)
+            {
+                this->items[0]->allocateRate = this->items[0]->rateValue < this->totalAvaliableRate ? this->items[0]->rateValue : this->totalAvaliableRate;
+                return;
+            }
+            std::vector<shared_ptr<RateItem>> sortedItems;
+            sortedItems.assign(this->items.begin(), this->items.end());
+            std::sort(sortedItems.begin(), sortedItems.end(), [](shared_ptr<RateItem> &ri1, shared_ptr<RateItem> &ri2)
+                      { return ri1->rateValue < ri2->rateValue; });
+
+            uint32_t remainRate = this->totalAvaliableRate;
+            int remainCount = sortedItems.size();
+            // std::cout << "\n\n=========== begin ===========" << std::endl;
+            for (auto &rateItem : sortedItems)
+            {
+                int thisTurn = remainRate / remainCount;
+                if (thisTurn <= rateItem->rateValue)
+                {
+                    remainRate -= thisTurn;
+                    rateItem->allocateRate = thisTurn;
+                    // std::cout << "assign1: " << thisTurn << ", rateValue: " << rateItem->rateValue << ", totalReaminRate: " << this->totalAvaliableRate << std::endl;
+                }
+                else
+                {
+                    remainRate -= rateItem->rateValue;
+                    rateItem->allocateRate = rateItem->rateValue;
+                    // std::cout << "assign2: " << rateItem->rateValue << ", rateValue: " << rateItem->rateValue << std::endl;
+                }
+                remainCount--;
+            }
+            // std::cout << "=========== end ===========\n\n" << std::endl;
+        }
+
+        uint32_t RateCollection::getAllocateRate(face::FaceId id)
+        {
+            if (rates.count(id) == 0)
+            {
+                return 0;
+            }
+            else
+            {
+                return this->rates[id]->allocateRate;
+            }
+        }
+
+        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+        //// RateStore
+        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+        RateStore::RateStore()
+        {
+            getScheduler().schedule(time::milliseconds(100), std::bind(&RateStore::periodicCleaning, this));
+        }
+
+        void
+        RateStore::periodicCleaning()
+        {
+            getScheduler().schedule(time::milliseconds(100), std::bind(&RateStore::periodicCleaning, this));
+            if (turn % 1 == 0)
+            {
+                for (auto it = this->ART.begin(); it != this->ART.end(); it++)
+                {
+                    it->second->doPeriodicCleaning(true);
+                }
+                for (auto it = this->RRT.begin(); it != this->RRT.end(); it++)
+                {
+                    it->second->doPeriodicCleaning(false, 200000);
+                }
+            }
+            // for (auto it = this->ART.begin(); it != this->ART.end(); it++) {
+            //     it->second->calSplitRateByRate();
+            // }
+            turn++;
+        }
+
+        face::FaceId RateStore::afterReceiveInterest(const nfd::FaceEndpoint &ingress, const ndn::Interest &interest,
+                                                     const std::shared_ptr<nfd::pit::Entry> &pitEntry,
+                                                     const fib::Entry &fibEntry)
+        {
+            auto flowKey = interest.getName().getPrefix(1).toUri();
+
+            auto downstreamRate = interest.getDownstreamRate();
+            if (downstreamRate)
+            {
+                if (this->ART.count(flowKey) == 0)
+                {
+                    this->ART[flowKey] = make_shared<RateCollection>();
+                    this->ART[flowKey]->setName("ART");
+                }
+                auto &rc = this->ART[flowKey];
+                if (rc->addOrUpdateRate(ingress.face.getId(), *downstreamRate))
+                {
+                    rc->calSplitRateByRate();
+                }
+            }
+
+            if (this->RRT.count(flowKey) == 0)
+            {
+                this->RRT[flowKey] = make_shared<RateCollection>();
+                this->RRT[flowKey]->setName("RRT");
+            }
+
+            auto selectedFaceId = this->RRT[flowKey]->selectFaceByRatio(ingress, fibEntry, interest);
+
+            if (selectedFaceId > 0)
+            {
+                // DR(/p, j) = min(OR(/p, j), RR(/p, j) / sum(RR(/p)) * sum(AR(/p)))
+                // 这边只计算了 RR(/p, j) / sum(RR(/p)) * sum(AR(/p)) ，和 OR 的对比见 ns3/src/ndnSIM/NFD/damon/face/queues.cpp
+                auto newDR = this->RRT[flowKey]->getRatioByFace(selectedFaceId) * this->ART[flowKey]->getMergeRate();
+                interest.setDownstreamRate(newDR);
+            }
+            return selectedFaceId;
+        }
+
+        void RateStore::afterReceiveNack(const nfd::FaceEndpoint &ingress, const ndn::lp::Nack &nack,
+                                         const std::shared_ptr<nfd::pit::Entry> &pitEntry)
+        {
+        }
+
+        void RateStore::beforeSatisfyInterest(const std::shared_ptr<nfd::pit::Entry> &pitEntry,
+                                              const nfd::FaceEndpoint &ingress, const ndn::Data &data)
+        {
+            auto targetRate = data.getTargetRate();
+            auto flowKey = data.getName().getPrefix(1).toUri();
+            if (this->RRT.count(flowKey) == 0)
+            {
+                this->RRT[flowKey] = make_shared<RateCollection>();
+                this->RRT[flowKey]->setName("RRT");
+            }
+            if (targetRate)
+            {
+                // 保存 Data 中携带的 Target Rate
+                this->RRT[flowKey]->addOrUpdateRate(ingress.face.getId(), *targetRate);
+            }
+            else
+            {
+                this->RRT[flowKey]->addOrUpdateRate(ingress.face.getId(), QSCCPStrategy::getMaxTR());
+            }
+        }
+
+        void RateStore::beforeSendData(const shared_ptr<pit::Entry> &pitEntry, const FaceEndpoint &egress,
+                                       const Data &data)
+        {
+            auto flowKey = data.getName().getPrefix(1).toUri();
+
+            auto mergeRate = this->RRT[flowKey]->getMergeRate();
+            if (this->ART.count(flowKey) == 0)
+            {
+                this->ART[flowKey] = make_shared<RateCollection>();
+                this->ART[flowKey]->setName("ART");
+            }
+
+            // if (flowKey == "/C") {
+            //     std::cout << "mergeRate: " << mergeRate << std::endl;
+            //     this->RRT[flowKey]->printItems();
+            // }
+
+            auto rc = this->ART[flowKey];
+            rc->updateTotalAvaliableRate(mergeRate);
+            // rc->calSplitRateByRate();
+
+            if (rc->getFaceNum() <= 1)
+            {
+                data.setTargetRate(mergeRate);
+            }
+            else
+            {
+                data.setTargetRate(static_cast<uint64_t>(rc->getRatioByFace(egress.face.getId()) * mergeRate));
+            }
+        }
+    }
+}

daemon/fw/qsccp-strategy.hpp

@@ -0,0 +1,171 @@
+//
+// Created by gdcni21 on 3/24/22.
+//
+#include <boost/random/mersenne_twister.hpp>
+#include "face/face.hpp"
+#include "fw/strategy.hpp"
+#include "fw/retx-suppression-exponential.hpp"
+#include "fw/retx-suppression-fixed.hpp"
+#include "fw/algorithm.hpp"
+#include "fw/process-nack-traits.hpp"
+#include <unordered_map>
+#include <limits>
+#include <iostream>
+
+namespace nfd
+{
+    namespace fw
+    {
+        struct RateItem
+        {
+            uint32_t rateValue;
+            face::FaceId id;
+            double ratio;
+            double balance;
+            uint32_t allocateRate;
+            uint32_t recvCount;
+            explicit RateItem(uint32_t value = 0, face::FaceId faceId = 0) : 
+                            rateValue(value), id(faceId), ratio(0), 
+                            balance(0.0), allocateRate(0), recvCount(0) {}
+            
+        };
+
+        class RateCollection
+        {
+        public:
+            bool addOrUpdateRate(face::FaceId id, uint32_t rate);
+
+            size_t getFaceNum()
+            {
+                return this->rates.size();
+            }
+
+            uint32_t getMergeRate() const
+            {
+                return this->totalRate;
+            }
+
+            void printItems() {
+                for (auto &rateItem : this->items) {
+                    std::cout << "id: " << rateItem->id << ", rate: " << rateItem->rateValue << ", ratio: " << rateItem->ratio << std::endl;
+                }
+            }
+
+            face::FaceId selectNextFace(uint64_t dataSize, face::FaceId exceptId);
+
+            face::FaceId selectFaceByRatio(const FaceEndpoint &ingress, const fib::Entry &fibEntry, const ndn::Interest &interest);
+
+            double getRatioByFace(face::FaceId id);
+
+            void calSplitRateByRate();
+
+            uint32_t getAllocateRate(face::FaceId id);
+
+            void updateTotalAvaliableRate(uint32_t totalAvailableRate) {
+                if (totalAvailableRate > 1.2 * this->totalAvaliableRate || totalAvailableRate < 0.8 * this->totalAvaliableRate) {
+                    this->totalAvaliableRate = totalAvailableRate;
+                    this->calSplitRateByRate();
+                } else {
+                    this->totalAvaliableRate = totalAvailableRate;
+                }
+                // std::cout << "updateRate: " << totalAvailableRate << std::endl;
+            }
+
+            void setName(const std::string &name) { this->name = name; }
+
+            std::string getName() { return this->name; }
+
+            void doPeriodicCleaning(bool needErase = true, uint32_t defaultValue = 0);
+
+        private:
+            std::unordered_map<face::FaceId, shared_ptr<RateItem>> rates;
+            std::vector<shared_ptr<RateItem>> items;
+            uint32_t totalRate;
+            boost::random::mt19937 m_randomGenerator;
+
+            float randomThreshold = 0.99;
+            int nextForwardFace;
+            int MTU_QUANTA = 2 * 1000 * 8;
+            std::string name = "";
+            uint32_t totalAvaliableRate = 0;
+        };
+
+        class RateStore
+        {
+        public:
+            explicit RateStore();
+
+            face::FaceId
+            afterReceiveInterest(const FaceEndpoint &ingress, const Interest &interest,
+                                 const shared_ptr<pit::Entry> &pitEntry, const fib::Entry &fibEntry);
+
+            void
+            afterReceiveNack(const FaceEndpoint &ingress, const lp::Nack &nack,
+                             const shared_ptr<pit::Entry> &pitEntry);
+
+            void
+            beforeSatisfyInterest(const shared_ptr<pit::Entry> &pitEntry,
+                                  const FaceEndpoint &ingress, const Data &data);
+
+            void
+            beforeSendData(const shared_ptr<pit::Entry> &pitEntry,
+                           const FaceEndpoint &egress, const Data &data);
+
+            void 
+            periodicCleaning();
+
+        private:
+            // ART => Available Rate Table =>
+            std::unordered_map<std::string, shared_ptr<RateCollection>> ART;
+            // RRT => Recommend Rate Table => 
+            std::unordered_map<std::string, shared_ptr<RateCollection>> RRT;
+            uint32_t turn = 0;
+        };
+
+        class QSCCPStrategy : public Strategy, public ProcessNackTraits<QSCCPStrategy>
+        {
+        public:
+            explicit QSCCPStrategy(Forwarder &forwarder, const Name &name = getStrategyName());
+
+            ~QSCCPStrategy() override = default;
+
+            void
+            afterReceiveInterest(const FaceEndpoint &ingress, const Interest &interest,
+                                 const shared_ptr<pit::Entry> &pitEntry) override;
+
+            void
+            afterReceiveNack(const FaceEndpoint &ingress, const lp::Nack &nack,
+                             const shared_ptr<pit::Entry> &pitEntry) override;
+
+            void
+            beforeSatisfyInterest(const shared_ptr<pit::Entry> &pitEntry,
+                                  const FaceEndpoint &ingress, const Data &data) override;
+
+            void
+            afterReceiveData(const shared_ptr<pit::Entry> &pitEntry,
+                             const FaceEndpoint &ingress, const Data &data) override;
+
+            void
+            afterContentStoreHit(const shared_ptr<pit::Entry> &pitEntry,
+                                 const FaceEndpoint &ingress, const Data &data) override;
+
+            static uint64_t
+            getMaxTR()
+            {
+                return std::numeric_limits<uint32_t>::max();
+            }
+
+            static const Name &
+            getStrategyName();
+
+        protected:
+            friend ProcessNackTraits<QSCCPStrategy>;
+
+        private:
+            static const time::milliseconds RETX_SUPPRESSION_INITIAL;
+            static const time::milliseconds RETX_SUPPRESSION_MAX;
+            RetxSuppressionFixed m_retxSuppression;
+            RateStore rs;
+        };
+    }
+}

daemon/mgmt/face-manager.cpp

@@ -120,6 +120,9 @@ FaceManager::createFace(const ControlParameters& parameters,
     // The face system limits MTUs to ssize_t, but the management protocol uses uint64_t
     faceParams.mtu = std::min<uint64_t>(std::numeric_limits<ssize_t>::max(), parameters.getMtu());
   }
+  if (parameters.hasBandwidth()) {
+    faceParams.bandwidth = parameters.getBandwidth();
+  }
   faceParams.wantLocalFields = parameters.hasFlagBit(ndn::nfd::BIT_LOCAL_FIELDS_ENABLED) &&
                                parameters.getFlagBit(ndn::nfd::BIT_LOCAL_FIELDS_ENABLED);
   faceParams.wantLpReliability = parameters.hasFlagBit(ndn::nfd::BIT_LP_RELIABILITY_ENABLED) &&

nfd.conf.sample.in

@@ -47,7 +47,7 @@ tables
 {
   ; ContentStore size limit in number of packets
   ; default is 65536, about 500MB with 8KB packet size
-  cs_max_packets 65536
+  cs_max_packets 0
 
   ; Set the CS replacement policy.
   ; Available policies are: priority_fifo, lru

tools/nfdc/face-module.cpp

@@ -59,7 +59,8 @@ FaceModule::registerCommands(CommandParser& parser)
     .addArg("congestion-marking", ArgValueType::BOOLEAN, Required::NO, Positional::NO)
     .addArg("congestion-marking-interval", ArgValueType::UNSIGNED, Required::NO, Positional::NO)
     .addArg("default-congestion-threshold", ArgValueType::UNSIGNED, Required::NO, Positional::NO)
-    .addArg("mtu", ArgValueType::STRING, Required::NO, Positional::NO);
+    .addArg("mtu", ArgValueType::STRING, Required::NO, Positional::NO)
+    .addArg("bandwidth", ArgValueType::UNSIGNED, Required::NO, Positional::NO);
   parser.addCommand(defFaceCreate, &FaceModule::create);
 
   CommandDefinition defFaceDestroy("face", "destroy");
@@ -162,6 +163,7 @@ FaceModule::create(ExecuteContext& ctx)
   auto baseCongestionMarkingIntervalMs = ctx.args.getOptional<uint64_t>("congestion-marking-interval");
   auto defaultCongestionThreshold = ctx.args.getOptional<uint64_t>("default-congestion-threshold");
   auto mtuArg = ctx.args.getOptional<std::string>("mtu");
+  auto bandwidth = ctx.args.getOptional<uint64_t>("bandwidth");
 
   // MTU is nominally a uint64_t, but can be the string value 'auto' to unset an override MTU
   optional<uint64_t> mtu;
@@ -270,6 +272,9 @@ FaceModule::create(ExecuteContext& ctx)
     if (mtu) {
       params.setMtu(*mtu);
     }
+    if (bandwidth) {
+      params.setBandwidth(*bandwidth);
+    }
 
     ctx.controller.start<ndn::nfd::FaceCreateCommand>(
       params,