[exprimental] Add strategies more control over satisfying Data
With this patch, strategies now selectively decide not to satisfy some
(all) of the incoming interests listed in PIT. However, if there are
multiple PIT entries being satisfied with the same Data packet, the
decision may not be honored (e.g., when another PIT entry/strategy
decided to satisfy a specific downstream).
Change-Id: I2edf19ae5bdeab0b69c57c7af7bb292db05885bb
diff --git a/daemon/fw/forwarder.cpp b/daemon/fw/forwarder.cpp
index 43fc0c6..6ce928e 100644
--- a/daemon/fw/forwarder.cpp
+++ b/daemon/fw/forwarder.cpp
@@ -334,74 +334,73 @@
// CS insert
m_cs.insert(data);
- // when only one PIT entry is matched, trigger strategy: after receive Data
- if (pitMatches.size() == 1) {
- auto& pitEntry = pitMatches.front();
+ std::set<std::pair<Face*, EndpointId>> satisfiedDownstreams;
+ std::multimap<std::pair<Face*, EndpointId>, std::shared_ptr<pit::Entry>> unsatisfiedPitEntries;
+ for (const auto& pitEntry : pitMatches) {
NFD_LOG_DEBUG("onIncomingData matching=" << pitEntry->getName());
- // set PIT expiry timer to now
- this->setExpiryTimer(pitEntry, 0_ms);
-
+ // invoke PIT satisfy callback
beforeSatisfyInterest(*pitEntry, ingress.face, data);
- // trigger strategy: after receive Data
- m_strategyChoice.findEffectiveStrategy(*pitEntry).afterReceiveData(data, ingress, pitEntry);
- // mark PIT satisfied
- pitEntry->isSatisfied = true;
- pitEntry->dataFreshnessPeriod = data.getFreshnessPeriod();
+ std::set<std::pair<Face*, EndpointId>> unsatisfiedDownstreams;
+ this->dispatchToStrategy(*pitEntry,
+ [&] (fw::Strategy& strategy) {
+ strategy.satisfyInterest(pitEntry, ingress, data,
+ satisfiedDownstreams, unsatisfiedDownstreams);
+ });
+ for (const auto& endpoint : unsatisfiedDownstreams) {
+ unsatisfiedPitEntries.emplace(endpoint, pitEntry);
+ }
- // Dead Nonce List insert if necessary (for out-record of ingress face)
- this->insertDeadNonceList(*pitEntry, &ingress.face);
-
- // delete PIT entry's out-record
- pitEntry->deleteOutRecord(ingress.face);
- }
- // when more than one PIT entry is matched, trigger strategy: before satisfy Interest,
- // and send Data to all matched out faces
- else {
- std::set<Face*> pendingDownstreams;
- auto now = time::steady_clock::now();
-
- for (const auto& pitEntry : pitMatches) {
- NFD_LOG_DEBUG("onIncomingData matching=" << pitEntry->getName());
-
- // remember pending downstreams
- for (const pit::InRecord& inRecord : pitEntry->getInRecords()) {
- if (inRecord.getExpiry() > now) {
- pendingDownstreams.insert(&inRecord.getFace());
- }
- }
-
+ if (unsatisfiedDownstreams.empty()) {
// set PIT expiry timer to now
this->setExpiryTimer(pitEntry, 0_ms);
- // invoke PIT satisfy callback
- beforeSatisfyInterest(*pitEntry, ingress.face, data);
- m_strategyChoice.findEffectiveStrategy(*pitEntry).beforeSatisfyInterest(data, ingress, pitEntry);
-
// mark PIT satisfied
pitEntry->isSatisfied = true;
- pitEntry->dataFreshnessPeriod = data.getFreshnessPeriod();
-
- // Dead Nonce List insert if necessary (for out-record of ingress face)
- this->insertDeadNonceList(*pitEntry, &ingress.face);
-
- // clear PIT entry's in and out records
- pitEntry->clearInRecords();
- pitEntry->deleteOutRecord(ingress.face);
}
- // foreach pending downstream
- for (const auto& pendingDownstream : pendingDownstreams) {
- if (pendingDownstream->getId() == ingress.face.getId() &&
- pendingDownstream->getLinkType() != ndn::nfd::LINK_TYPE_AD_HOC) {
- continue;
+ // Dead Nonce List insert if necessary (for out-record of inFace)
+ this->insertDeadNonceList(*pitEntry, &ingress.face);
+
+ pitEntry->dataFreshnessPeriod = data.getFreshnessPeriod();
+
+ // clear PIT entry's in and out records
+ for (const auto& endpoint : satisfiedDownstreams) {
+ pitEntry->deleteInRecord(*endpoint.first);
+ }
+ pitEntry->deleteOutRecord(ingress.face);
+ }
+
+ // now check all unsatisfied entries against to be satisfied downstreams, in case there is
+ // intersect, and those PIT entries will be actually satisfied regardless strategy's choice
+ for (const auto& unsatisfied : unsatisfiedPitEntries) {
+ auto downstreamIt = satisfiedDownstreams.find(unsatisfied.first);
+ if (downstreamIt != satisfiedDownstreams.end()) {
+ auto pitEntry = unsatisfied.second;
+ pitEntry->deleteInRecord(*unsatisfied.first.first);
+
+ if (pitEntry->getInRecords().empty()) { // if nothing left, "closing down" the entry
+ // set PIT expiry timer to now
+ this->setExpiryTimer(pitEntry, 0_ms);
+
+ // mark PIT satisfied
+ pitEntry->isSatisfied = true;
}
- // goto outgoing Data pipeline
- this->onOutgoingData(data, *pendingDownstream);
}
}
+
+ // foreach pending downstream
+ for (const auto& downstream : satisfiedDownstreams) {
+ if (downstream.first->getId() == ingress.face.getId() &&
+ downstream.second == ingress.endpoint &&
+ downstream.first->getLinkType() != ndn::nfd::LINK_TYPE_AD_HOC) {
+ continue;
+ }
+
+ this->onOutgoingData(data, FaceEndpoint(*downstream.first, downstream.second));
+ }
}
void