/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ /** * Copyright (C) 2014 Named Data Networking Project * See COPYING for copyright and distribution information. */ #include "forwarder.hpp" #include "core/logger.hpp" #include "best-route-strategy.hpp" namespace nfd { NFD_LOG_INIT("Forwarder"); const Name Forwarder::s_localhostName("ndn:/localhost"); Forwarder::Forwarder(boost::asio::io_service& ioService) : m_scheduler(ioService) , m_lastFaceId(0) { m_strategy = make_shared(boost::ref(*this)); } void Forwarder::addFace(shared_ptr face) { FaceId faceId = ++m_lastFaceId; face->setId(faceId); m_faces[faceId] = face; NFD_LOG_INFO("addFace id=" << faceId); face->onReceiveInterest += bind(&Forwarder::onInterest, this, boost::ref(*face), _1); face->onReceiveData += bind(&Forwarder::onData, this, boost::ref(*face), _1); } void Forwarder::removeFace(shared_ptr face) { FaceId faceId = face->getId(); m_faces.erase(faceId); face->setId(INVALID_FACEID); NFD_LOG_INFO("removeFace id=" << faceId); // XXX This clears all subscriptions, because EventEmitter // does not support only removing Forwarder's subscription face->onReceiveInterest.clear(); face->onReceiveData .clear(); m_fib.removeNextHopFromAllEntries(face); } shared_ptr Forwarder::getFace(FaceId id) { std::map >::iterator i = m_faces.find(id); return (i == m_faces.end()) ? (shared_ptr()) : (i->second); } void Forwarder::onInterest(Face& face, const Interest& interest) { this->onIncomingInterest(face, interest); } void Forwarder::onData(Face& face, const Data& data) { this->onIncomingData(face, data); } void Forwarder::onIncomingInterest(Face& inFace, const Interest& interest) { // receive Interest NFD_LOG_DEBUG("onIncomingInterest face=" << inFace.getId() << " interest=" << interest.getName()); const_cast(interest).setIncomingFaceId(inFace.getId()); // /localhost scope control bool violatesLocalhost = !inFace.isLocal() && s_localhostName.isPrefixOf(interest.getName()); if (violatesLocalhost) { NFD_LOG_DEBUG("onIncomingInterest face=" << inFace.getId() << " interest=" << interest.getName() << " violates /localhost"); return; } // PIT insert std::pair, bool> pitInsertResult = m_pit.insert(interest); shared_ptr pitEntry = pitInsertResult.first; // detect loop and record Nonce bool isLoop = ! pitEntry->addNonce(interest.getNonce()); if (isLoop) { // goto Interest loop pipeline this->onInterestLoop(inFace, interest, pitEntry); return; } // cancel unsatisfy & straggler timer this->cancelUnsatisfyAndStragglerTimer(pitEntry); const pit::InRecordCollection& inRecords = pitEntry->getInRecords(); bool isPending = inRecords.begin() == inRecords.end(); if (!isPending) { // CS lookup const Data* csMatch = m_cs.find(interest); if (csMatch != 0) { // XXX should we lookup PIT for other Interests that also match csMatch? // goto outgoing Data pipeline this->onOutgoingData(*csMatch, inFace); return; } } // insert InRecord pit::InRecordCollection::iterator inRecordIt = pitEntry->insertOrUpdateInRecord(inFace.shared_from_this(), interest); // app chosen nexthops bool isAppChosenNexthops = false; // TODO get from local control header if (isAppChosenNexthops) { // TODO foreach chosen nexthop: goto outgoing Interest pipeline return; } // FIB lookup shared_ptr fibEntry = m_fib.findLongestPrefixMatch(interest.getName()); // TODO use Fib::findParent(pitEntry) // dispatch to strategy this->dispatchToStrategy(inFace, interest, fibEntry, pitEntry); } void Forwarder::onInterestLoop(Face& inFace, const Interest& interest, shared_ptr pitEntry) { NFD_LOG_DEBUG("onInterestLoop face=" << inFace.getId() << " interest=" << interest.getName()); // do nothing, which means Interest is dropped } void Forwarder::onOutgoingInterest(shared_ptr pitEntry, Face& outFace) { NFD_LOG_DEBUG("onOutgoingInterest face=" << outFace.getId() << " interest=" << pitEntry->getName()); // pick Interest const Interest& interest = pitEntry->getInterest(); // TODO pick the last incoming Interest // insert OutRecord pit::OutRecordCollection::iterator outRecordIt = pitEntry->insertOrUpdateOutRecord(outFace.shared_from_this(), interest); // set PIT unsatisfy timer this->setUnsatisfyTimer(pitEntry); // send Interest outFace.sendInterest(interest); } void Forwarder::onInterestRebuff(shared_ptr pitEntry) { NFD_LOG_DEBUG("onInterestRebuff interest=" << pitEntry->getName()); // set PIT straggler timer this->setStragglerTimer(pitEntry); } void Forwarder::onInterestUnsatisfied(shared_ptr pitEntry) { NFD_LOG_DEBUG("onInterestUnsatisfied interest=" << pitEntry->getName()); // invoke PIT unsatisfied callback // TODO // PIT delete m_pit.remove(pitEntry); } void Forwarder::onIncomingData(Face& inFace, const Data& data) { // receive Data NFD_LOG_DEBUG("onIncomingData face=" << inFace.getId() << " data=" << data.getName()); const_cast(data).setIncomingFaceId(inFace.getId()); // /localhost scope control bool violatesLocalhost = !inFace.isLocal() && s_localhostName.isPrefixOf(data.getName()); if (violatesLocalhost) { NFD_LOG_DEBUG("onIncomingData face=" << inFace.getId() << " interest=" << data.getName() << " violates /localhost"); return; } // PIT match shared_ptr pitMatches = m_pit.findAllDataMatches(data); if (pitMatches->begin() == pitMatches->end()) { // goto Data unsolicited pipeline this->onDataUnsolicited(inFace, data); return; } // CS insert m_cs.insert(data); std::set > pendingDownstreams; // foreach PitEntry for (pit::DataMatchResult::iterator it = pitMatches->begin(); it != pitMatches->end(); ++it) { shared_ptr pitEntry = *it; NFD_LOG_DEBUG("onIncomingData matching=" << pitEntry->getName()); // cancel unsatisfy & straggler timer this->cancelUnsatisfyAndStragglerTimer(pitEntry); // remember pending downstreams const pit::InRecordCollection& inRecords = pitEntry->getInRecords(); for (pit::InRecordCollection::const_iterator it = inRecords.begin(); it != inRecords.end(); ++it) { if (it->getExpiry() > time::now()) { pendingDownstreams.insert(it->getFace()); } } // mark PIT satisfied pitEntry->deleteInRecords(); pitEntry->deleteOutRecord(inFace.shared_from_this()); // set PIT straggler timer this->setStragglerTimer(pitEntry); // invoke PIT satisfy callback // TODO } // foreach pending downstream for (std::set >::iterator it = pendingDownstreams.begin(); it != pendingDownstreams.end(); ++it) { // goto outgoing Data pipeline this->onOutgoingData(data, **it); } } void Forwarder::onDataUnsolicited(Face& inFace, const Data& data) { // accept to cache? bool acceptToCache = false;// TODO decision if (acceptToCache) { // CS insert m_cs.insert(data); } NFD_LOG_DEBUG("onDataUnsolicited face=" << inFace.getId() << " data=" << data.getName() << " acceptToCache=" << acceptToCache); } void Forwarder::onOutgoingData(const Data& data, Face& outFace) { NFD_LOG_DEBUG("onOutgoingData face=" << outFace.getId() << " data=" << data.getName()); // traffic manager // pass through // send Data outFace.sendData(data); } static inline bool compare_InRecord_expiry(const pit::InRecord& a, const pit::InRecord& b) { return a.getExpiry() < b.getExpiry(); } void Forwarder::setUnsatisfyTimer(shared_ptr pitEntry) { const pit::InRecordCollection& inRecords = pitEntry->getInRecords(); pit::InRecordCollection::const_iterator lastExpiring = std::max_element(inRecords.begin(), inRecords.end(), &compare_InRecord_expiry); time::Point lastExpiry = lastExpiring->getExpiry(); time::Duration lastExpiryFromNow = lastExpiry - time::now(); if (lastExpiryFromNow <= time::seconds(0)) { // TODO all InRecords are already expired; will this happen? } pitEntry->m_unsatisfyTimer = m_scheduler.scheduleEvent(lastExpiryFromNow, bind(&Forwarder::onInterestUnsatisfied, this, pitEntry)); } void Forwarder::setStragglerTimer(shared_ptr pitEntry) { time::Duration stragglerTime = time::milliseconds(100); pitEntry->m_stragglerTimer = m_scheduler.scheduleEvent(stragglerTime, bind(&Pit::remove, &m_pit, pitEntry)); } void Forwarder::cancelUnsatisfyAndStragglerTimer(shared_ptr pitEntry) { m_scheduler.cancelEvent(pitEntry->m_unsatisfyTimer); m_scheduler.cancelEvent(pitEntry->m_stragglerTimer); } void Forwarder::dispatchToStrategy(const Face& inFace, const Interest& interest, shared_ptr fibEntry, shared_ptr pitEntry) { m_strategy->afterReceiveInterest(inFace, interest, fibEntry, pitEntry); // TODO dispatch according to fibEntry } } // namespace nfd