// // Created by 阙建明 on 2021/6/27. // #include "DataRateMerger.hpp" #include #include #include #include "daemon/common/logger.hpp" NFD_LOG_INIT(DataRateMerger); namespace nfd { namespace fw { std::time_t getTimeStamp() { std::chrono::time_point tp = std::chrono::time_point_cast( std::chrono::system_clock::now()); auto tmp = std::chrono::duration_cast(tp.time_since_epoch()); std::time_t timestamp = tmp.count(); //std::time_t timestamp = std::chrono::system_clock::to_time_t(tp); return timestamp; } void DataRateCollection::addRate(int id, uint64_t rate) { if (this->dateRates.count(id) == 0) { this->dateRates[id] = make_shared(); this->items.push_back(this->dateRates[id]); //NFD_LOG_DEBUG("add: " << id); } this->dateRates[id]->recordTime = getTimeStamp(); this->dateRates[id]->dataRate = rate; // 调用和个方法计算比率 this->getMergeDataRate(); } uint64_t DataRateCollection::getMergeDataRate() { //auto nowTime = getTimeStamp(); uint64_t totalRate = 0; for (auto &item : this->dateRates) { //if (nowTime - item.second->recordTime > 100) { // // 超过100ms重置 // item.second->recordTime = 0; //} if (item.second->recordTime > 0) { totalRate += item.second->dataRate; } } //// 计算比率 //for (auto &item : this->dateRates) { // if (item.second->recordTime > 0) { // item.second->rato = item.second->dataRate * 1.0 / totalRate; // } //} //NFD_LOG_DEBUG("totalRate: " << totalRate); return totalRate; } fib::NextHopList::const_iterator DataRateCollection::selectFaceToForward(const FaceEndpoint &ingress, const fib::Entry& fibEntry) { fib::NextHopList::const_iterator selected; const fib::NextHopList& nexthops = fibEntry.getNextHops(); if (!this->isInitial) { // 第一次初始化,给所有可能的下一跳设置一个初始的转发权重 for (selected = nexthops.begin(); selected != nexthops.end(); ++selected) { this->addRate(selected->getFace().getId(), 100000); } this->isInitial = true; return nexthops.begin(); } else { std::vector v; std::vector ids; for (auto &item : this->dateRates) { if (item.first != ingress.face.getId()) { v.emplace_back(item.second->dataRate); ids.emplace_back(item.first); } } boost::random::discrete_distribution<> prob_dis(v.begin(), v.end()); const size_t randomIndex = prob_dis(this->m_randomGenerator); uint64_t currentId = ids[randomIndex]; //NFD_LOG_DEBUG("SECOND: " << v.size() << " => " << currentId); for (selected = nexthops.begin(); selected != nexthops.end() && currentId != selected->getFace().getId(); ++selected) { } return selected; } } void DataRateMerger::beforeSatisfyInterest(const shared_ptr &pitEntry, const FaceEndpoint &ingress, const Data &data) { auto fairRateTag = data.getTag(); if (fairRateTag != nullptr) { auto flowKey = data.getName().getPrefix(1).toUri(); if (this->flowMap.count(flowKey) == 0) { this->flowMap[flowKey] = make_shared(); } //NFD_LOG_DEBUG("THIRD: " << ingress.face.getId() << " => " << data.getName().toUri()); if (ingress.face.getId() != 254) { this->flowMap[flowKey]->addRate(ingress.face.getId(), fairRateTag->get()); } data.setTag(make_shared(this->flowMap[flowKey]->getMergeDataRate())); //NFD_LOG_DEBUG(flowKey); } } void DataRateMerger::afterContentStoreHit(const shared_ptr& pitEntry, const FaceEndpoint& ingress, const Data& data) { auto flowKey = data.getName().getPrefix(1).toUri(); if (this->flowMap.count(flowKey) != 0) { data.setTag(make_shared(this->flowMap[flowKey]->getMergeDataRate())); } else { } } fib::NextHopList::const_iterator DataRateMerger::afterReceiveInterest(const FaceEndpoint &ingress, const fib::Entry& fibEntry, const Interest& interest) { auto flowKey = interest.getName().getPrefix(1).toUri(); if (this->flowMap.count(flowKey) == 0) { this->flowMap[flowKey] = make_shared(); } return this->flowMap[flowKey]->selectFaceToForward(ingress, fibEntry); } } }