fix: Do calculateRate by trigger not by scheduler

daemon/face/TokenBucket.cpp

@@ -1,6 +1,7 @@
 #include "TokenBucket.hpp"
 #include "common/global.hpp"
 #include "common/logger.hpp"
+#include <iostream>
 #include <functional>
 namespace nfd {
     namespace face {
@@ -27,6 +28,26 @@ namespace nfd {
             }
             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;

daemon/face/TokenBucket.hpp

@@ -15,6 +15,12 @@ namespace nfd {
              * 周期性增加token
              **/
             void generateToken();
+
+            /**
+             * 根据当前时间生成token
+            */
+            void generateTokenByCurrentTime(long currentTime);
+
             /**
              * 尝试消费n个token
              * @param tokenNum
@@ -28,6 +34,7 @@ namespace nfd {
             uint64_t bucketSize;    // 令牌桶容量
             std::string name;
             bool isStart; // 标识是否启动
+            long lastGenerateTokenTime = -1;
         };
     }
 }

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

@@ -49,10 +49,12 @@ GenericLinkService::GenericLinkService(const GenericLinkService::Options& option
   , 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;
@@ -69,12 +71,14 @@ GenericLinkService::GenericLinkService(const GenericLinkService::Options& option
         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)(newTaskPtr->waitNsTime)),
                                     std::bind(this->sendPacketCallback, newTaskPtr));
@@ -404,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());
   }
 }
 
@@ -462,6 +466,7 @@ GenericLinkService::decodeInterest(const Block& netPkt, const lp::Packet& firstP
   }
 
   this->globalSp.receiveInterest(*interest);
+  this->globalSp.triggerCalObjectRate();
   this->receiveInterest(*interest, endpointId);
 }
 
@@ -521,6 +526,8 @@ GenericLinkService::decodeData(const Block& netPkt, const lp::Packet& firstPkt,
       // TR = min(TR, OR)
       this->globalSp.assignObjectRate(*data);
   }
+
+  this->globalSp.triggerCalObjectRate();
   this->receiveData(*data, endpointId);
 }
 

daemon/face/generic-link-service.hpp

@@ -166,7 +166,7 @@ public:
      */
     ssize_t overrideMtu = std::numeric_limits<ssize_t>::max();
     
-    uint64_t C = 10000000;
+    uint64_t C = 1000000;
     size_t vqSize = 20;
   };
 

daemon/face/queues.cpp

@@ -36,7 +36,7 @@ namespace nfd {
             if (rateLimit > 0) {
                 this->needRateLimit = true;
                 this->tokenBucket = std::make_shared<TokenBucket>(rateLimit, 20 * 1000 * 8, name);
-                this->tokenBucket->start();
+                // this->tokenBucket->start();
             } else {
                 this->needRateLimit = false;
             }
@@ -48,7 +48,6 @@ namespace nfd {
         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);
@@ -127,11 +126,16 @@ namespace nfd {
             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++;
@@ -172,6 +176,9 @@ namespace nfd {
                 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;
             }
@@ -186,6 +193,8 @@ namespace nfd {
                     }
                     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;
                 }
             }
             
@@ -300,8 +309,11 @@ namespace nfd {
             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);
@@ -310,6 +322,7 @@ namespace nfd {
                     }
                 }
             } 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(
@@ -373,22 +386,25 @@ namespace nfd {
             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); });
+            // 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);
         }
@@ -398,6 +414,7 @@ namespace nfd {
             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) {

daemon/face/queues.hpp

@@ -158,6 +158,13 @@ namespace nfd {
                 }
                 return this->virtualQueues[key]->objectRate;
             }
+
+            void triggerByCurrentTime(long currentTime) {
+                if (this->needRateLimit) {
+                    this->tokenBucket->generateTokenByCurrentTime(currentTime);
+                }
+            }
+
         public:
             double originAvaliableBandwidth = 0;
         private:
@@ -200,6 +207,11 @@ namespace nfd {
             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;
@@ -262,15 +274,27 @@ namespace nfd {
                     delaySensitiveServiceSp(availableBandwidth, vqSize),
                     wfqQueue(availableBandwidth, vqSize),
                     availableBandwidth(availableBandwidth) {
-                
-                getScheduler().schedule(time::nanoseconds(1000000 * this->calObjectRateInterval),
-                                        [this] { this->calObjectRate(false); });
+                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;
@@ -278,6 +302,7 @@ namespace nfd {
             int calObjectRateInterval = 100;
             uint32_t availableBandwidth;
             size_t interestBytes = 0;
+            long lastTriggerTime = 0;
         };
     }
 }

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/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) &&

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,