字数统计: 6.7k字   |   阅读时长: 31分

NDN代码分析(3) forwarder模块代码

相比起之前的代码， forwarder.cpp 是一个大工程，头铁硬碰显然是不合适的。在看代码之前，我们应当对 Forwarder 类大致在干什么有一些初步的了解。实际上， Forwarder 类主要控制前向转发过程，我们可以把整个 Forwarder 类看作一个贴心的管家，通过 nameTree 管理手下的 fib 、 pit 、 cs ，还通过 m_faceTable 管理了手下的所有的 face 接口，并同时管理着其余的各项东西。

既然是是一个管家，那么涉及的内容肯定是盘根错节的。但幸运的是，我们能在官网上查到NFD的开发者手册，该手册截至目前最新的版本为第11版，下载链接为NFD Developer’s Guide (Latest Revision 11)

在本手册的第四章，对于转发部分有着详细的讲解.

整体转发工作流程

引用手册里的图，整个转发流程其实就很清晰了。整个转发分为三个大板块： onIncomingInterest 、 onIncomingData 、 onIncomingNack 。这里我对这些流程大方向进行一些简单的梳理：

  • 如果来了兴趣包，执行onIncomingInterest
    • 如果重复的 Nonce ，则说明发生了循环，执行 onInterestLoop
    • 如果CS缓存未命中，则执行 onContentStoreMiss 函数，执行相应的策略
    • 如果CS缓存命中，则执行 onContentStoreHit 函数，执行相应的策略
    • 无论CS命中还是CS非命中，只要最后是决定要转发，都会执行 onOutgoingInterest
  • 如果来了数据包，执行onIncomingData
    • 如果查PIT查不到请求，那么数据包是不请自来的，执行 onDataUnsolicited
    • 查有几个匹配的PIT条目，对所有条目的下游执行 onOutgoingData

部分名词说明：

  • in-record和out-record我没有进行翻译。
  • downstream我翻译为了“下游”
  • upstream我翻译为了“上游”
  • 其中in-record代表interest的下游接口，是内容的请求者。out-record代表interest的上游接口，是潜在的内容源。interest从下游到上游，data从上游到下游。

onIncomingInterest工作流程

触发方式： Forwarder::onIncomingInterest 函数从 Forwarder::startProcessInterest 方法进入，该方法由接口 Face::afterReceiveData 信号触发。

引用手册里的图，转发流程如下图所示。对于整个流程，简要地说：

  • 首先以 /localhost 为前缀是本地通信保留的，所以以这个为前缀的将会被直接扔掉。
  • 然后检测当前随机数有没有在死亡随机数列表里，如果在，说明兴趣包循环了，执行 onInterestLoop
  • 再看有没有到生产者的区域，如果到了，就可以剥掉转发前缀了。
  • 接着查PIT（虽然我们在ndn原理里认为先查CS再查PIT，然而查CS开销可能会很大。所以可以先查PIT，如果PIT里有人请求过了，说明该兴趣如今还未决，CS肯定没有，就可以节省开销了），看看PIT里有没有重复随机数。如果没有就取消掉PIT的到期定时器，为PIT续1s
  • 最后查CS（其实先查CS再查PIT是等价的，只是先查PIT的话，如果PIT有就不用查CS了，节省开销），决定执行 onContentStoreHit 或者 onContentStoreMiss 。

onInterestLoop工作流程

上面 onIncomingInterest 说了，如果检测到相同的随机数，就说明发生了循环，这时就会执行 onInterestLoop 函数。这个时候分两种情况讨论：

  • 如果是点对点，那么发生循环相当于Not ACK，于是将原因标记为Nack。
  • 如果是多播的话，直接丢弃就行，不需要Nack。

onContentStoreHit工作流程

如果缓存命中，就设置好到期定时器，并执行相应的 afterContentStoreHit 策略，如下图所示：

​ 此管道在 Forwarder::onContentStoreHit 方法中实现，并在执行内容存储查找并找到匹配项后进入传入的兴趣管道。 此管道的输入参数包括兴趣包、传入的 Face、PIT 条目和匹配的数据包。如图 9 所示，此管道首先将到期计时器设置为现在在 Interest 上，然后在 Content Store 命中所选策略的触发器后调用。

onContentStoreMiss工作流程

如果缓存未命中，就给对应条目的PIT插入一个in-record，到期时间设置为最后一个in-record的时间,默认是4秒。如果兴趣包内钦定好了下一跳的接口id，那就对该接口执行 onOutgoingInterest ，不然就执行策略 afterReceiveInterest ，通过FIB路由表查出下一跳（如果查到并决定转发，那么对查到的下一跳执行 onOutgoingInterest ）。若PIT in-record到期计时器到期，则执行 Interest Finalize 管道（第 4.2.6 节）。如下图所示：

onOutgoingInterest工作流程

首先插入一个out-record（如果已经存在相同的接口就更新），记录下最后一个包的随机数和到期时间，然后执行 sendInterest 函数。如下图所示：

onInterestFinalize工作流程

一些善后工作，如插入死亡随机数列表、删除对应的PIT条目等。

onIncomingData工作流程

触发方式： Forwarder::onIncomingData 函数从 Forwarder::startProcessData 方法进入，该方法由接口 Face::afterReceiveData 信号触发。

简要地说：

  • 首先以 /localhost 为前缀是本地通信保留的，所以以这个为前缀的将会被直接扔掉。
  • 然后查PIT，如果PIT没有，但来了这个数据包，说明数据包是不请自来的（unsolicited）。如果PIT有，那么尝试将数据包插入CS。
  • 再检测有几个匹配的PIT条目（一般来说应该只有一个，因为多个PIT意味着多个转发策略，可能导致转发策略之间的冲突）
    • 如果单个匹配，将PIT到期时间设置为现在，并调用 afterReceiveData 策略，标记该PIT条目已被满足，删掉out-record，尝试插入死亡随机数列表。
    • 如果多个匹配，对于每个匹配项目，管道会先记住其pending的下一跳们（最后汇总一起执行 onOutgoingData ），其余操作与单个匹配类似，只不过变成了调用 beforeSatisfyInterest 策略，并清除条目的out-record。

onDataUnsolicited工作流程

如果数据是不请自来的，则根据 m_unsolicitedDataPolicy 的决定来判断。如果认为应当缓存就尝试缓存，否则啥都不干。

onOutgoingData工作流程

首先检查下 /localhost ，（然后下个版本计划进行流量管理，这个版本还没实现，）最后通过 sendData 函数传出去。

onIncomingNack函数

Nack(Not ACK)在NFD里会分为两部分处理： Incoming Nack 和 Outgoing Nack 。这部分讲的是Incoming Nack( processing of incoming Nacks)。

触发方式： Forwarder::onIncomingNack 函数从 Forwarder::startProcessNack 方法进入，该方法由接口 Face::afterReceiveNack 信号触发。

  • 首先，如果传入接口不是点对点的，那么直接丢弃（因为Nack只在点对点链路上被定义）。
  • 如果和out-record的最后一个记录的随机数相同，相当于产生环路，那么该record将会被标记为Nack。如果所有out-record都Nack了，将到期时间设置到期时间为0，并执行策略 afterReceiveNack 。
  • 如果随机数不同，那么就不执行操作。

onOutgoingNack函数

触发方式： Forwarder::onOutgoingNack 函数从 Strategy::sendNack 方法进入。

  • 首先，查in-record，如果查不到就不执行操作（不知道哪个interest的问题，不知道给谁发Nack）
  • 如果下一跳不是点对点的，那么直接丢弃（因为Nack只在点对点链路上被定义）。
  • 如果没毛病的话，就给in-record的interest贴上Nack，通过传出接口让链路层执行 sendNack

前置补充内容：信号类

在进入代码分析之前，我想先向大家介绍一下信号(Signal)类。为什么要介绍它呢？因为它在整个 Forwarder 类、甚至整个ndnSIM代码里都是大量出现的。因此，我们单独开了一个小章节来讲解信号类。其实信号类很简单，也很好理解，说白了，一个信号里面储存了一堆操作（函数），每当这个信号被触发的时候，这些操作就会被执行一遍。或者对于基础好一点的同学来说：一个信号里面储存了函数指针链表，并且将运算符 () 重载为了执行函数指针链表内的所有函数：

  • 具体而言，信号类的定义中最重要的部分如下所示。信号类重载了 () ，使得信号类的调用方式和函数一致。而且 Owner 作为了友元类，可以随意获取信号的私有成员。每当一个函数被 connect 到信号上，就会被添加到 m_slot 这个链表中。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
template<typename Owner, typename ...TArgs>
class Signal : noncopyable
{
public:
  typedef function<void(const TArgs&...)> Handler;
  // 连接信号
  Connection connect(Handler handler);
  // 重载()，使得能用()直接传递函数参数，让信号看起来跟函数一样
  void operator()(const TArgs&... args);
  // 定义开头的模板参数Owner为友元类
  friend Owner;
  // 其余代码省略

private:
  SlotList m_slots;
  // 其余代码省略
};

信号类有什么好处呢？信号类可以让一个类调用另一个类的方法，比如 Today 这个函数就只用关心相对抽象层次高的内容，比如每天只需要干 BeforeDoSomething 、 Dosomething 、 AfterDoSomething ，至于具体 BeforeDoSomething 是什么，可以由其他类来往里面添加(connect)事项。这样就可以用一个“管家”方便地管理下面的子类了。

forwarder.hpp

从整个 forwarder.hpp 代码中的 Forwarder 类，结合前面的信号类，我们能够看到整个 forwarder.cpp 想做的事情。总的来说，就是 startProcessInterest 、 startProcessData 、 startProcessNack 这三个函数衍生出来一串函数，并且单独为FIB搞了个 startProcessNewNextHop ，用于添加新的路由。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class Forwarder
{
public:
  // 构造函数，用于给成员的各种信号连接到本类的管道上
  explicit Forwarder(FaceTable& faceTable);
  VIRTUAL_WITH_TESTS ~Forwarder();

  // Get和Set方法
  const ForwarderCounters& getCounters() const
      { return m_counters; }
  fw::UnsolicitedDataPolicy& getUnsolicitedDataPolicy() const
      { return *m_unsolicitedDataPolicy; }
  void setUnsolicitedDataPolicy(unique_ptr<fw::UnsolicitedDataPolicy> policy)
      { BOOST_ASSERT(policy != nullptr);
        m_unsolicitedDataPolicy = std::move(policy); }

public: // 这些函数通过目标类的信号，使得目标类可以通过信号使用本类的管道
  void startProcessInterest(const FaceEndpoint& ingress, const Interest& interest)
      { this->onIncomingInterest(ingress, interest); }
  void startProcessData(const FaceEndpoint& ingress, const Data& data)
      { this->onIncomingData(ingress, data); }
  void startProcessNack(const FaceEndpoint& ingress, const lp::Nack& nack)
      { this->onIncomingNack(ingress, nack); }
  void startProcessNewNextHop(const Name& prefix, const fib::NextHop& nextHop)
      { this->onNewNextHop(prefix, nextHop); }

  // Get方法
  NameTree& getNameTree()
      { return m_nameTree; }
  Fib& getFib()
      { return m_fib; }
  Pit& getPit()
      { return m_pit; }
  Cs& getCs()
      { return m_cs; }
  Measurements& getMeasurements()
      { return m_measurements; }
  StrategyChoice& getStrategyChoice()
      { return m_strategyChoice; }
  DeadNonceList& getDeadNonceList()
      { return m_deadNonceList; }
  NetworkRegionTable& getNetworkRegionTable()
      { return m_networkRegionTable; }

public:  // 信号(Signal)
  signal::Signal<Forwarder, pit::Entry, Face, Data> beforeSatisfyInterest;
  signal::Signal<Forwarder, pit::Entry> beforeExpirePendingInterest;
  signal::Signal<Forwarder, Interest, Data> afterCsHit;
  signal::Signal<Forwarder, Interest> afterCsMiss;

PUBLIC_WITH_TESTS_ELSE_PRIVATE: // 管道(pipelines)
  VIRTUAL_WITH_TESTS void
  onIncomingInterest(const FaceEndpoint& ingress, const Interest& interest);
  VIRTUAL_WITH_TESTS void
  onInterestLoop(const FaceEndpoint& ingress, const Interest& interest);
  VIRTUAL_WITH_TESTS void
  onContentStoreMiss(const FaceEndpoint& ingress,
                     const shared_ptr<pit::Entry>& pitEntry, const Interest& interest);
  VIRTUAL_WITH_TESTS void
  onContentStoreHit(const FaceEndpoint& ingress, const shared_ptr<pit::Entry>& pitEntry,
                    const Interest& interest, const Data& data);
  VIRTUAL_WITH_TESTS void
  onOutgoingInterest(const shared_ptr<pit::Entry>& pitEntry,
                     const FaceEndpoint& egress, const Interest& interest);
  VIRTUAL_WITH_TESTS void
  onInterestFinalize(const shared_ptr<pit::Entry>& pitEntry);
  VIRTUAL_WITH_TESTS void
  onIncomingData(const FaceEndpoint& ingress, const Data& data);
  VIRTUAL_WITH_TESTS void
  onDataUnsolicited(const FaceEndpoint& ingress, const Data& data);
  VIRTUAL_WITH_TESTS void
  onOutgoingData(const Data& data, const FaceEndpoint& egress);
  VIRTUAL_WITH_TESTS void
  onIncomingNack(const FaceEndpoint& ingress, const lp::Nack& nack);
  VIRTUAL_WITH_TESTS void
  onOutgoingNack(const shared_ptr<pit::Entry>& pitEntry,
                 const FaceEndpoint& egress, const lp::NackHeader& nack);
  VIRTUAL_WITH_TESTS void
  onDroppedInterest(const FaceEndpoint& egress, const Interest& interest);
  VIRTUAL_WITH_TESTS void
  onNewNextHop(const Name& prefix, const fib::NextHop& nextHop);

PROTECTED_WITH_TESTS_ELSE_PRIVATE:
  void
  setExpiryTimer(const shared_ptr<pit::Entry>& pitEntry, time::milliseconds duration);
  VIRTUAL_WITH_TESTS void
  insertDeadNonceList(pit::Entry& pitEntry, Face* upstream);
#ifdef WITH_TESTS
  virtual void
  dispatchToStrategy(pit::Entry& pitEntry, std::function<void(fw::Strategy&)> trigger)
#else
  template<class Function>
  void
  dispatchToStrategy(pit::Entry& pitEntry, Function trigger)
#endif
  {
    trigger(m_strategyChoice.findEffectiveStrategy(pitEntry));
  }

private:
  ForwarderCounters m_counters;

  FaceTable& m_faceTable;
  unique_ptr<fw::UnsolicitedDataPolicy> m_unsolicitedDataPolicy;

  NameTree           m_nameTree;
  Fib                m_fib;
  Pit                m_pit;
  Cs                 m_cs;
  Measurements       m_measurements;
  StrategyChoice     m_strategyChoice;
  DeadNonceList      m_deadNonceList;
  NetworkRegionTable m_networkRegionTable;
  shared_ptr<Face>   m_csFace;

  // allow Strategy (base class) to enter pipelines
  friend class fw::Strategy;
};

forwarder.cpp的构造函数

整个构造函数主要是在连接各种信号，具体而言，干了这些事——

  • 给 m_faceTable 添加保留一个接口( face ) contentstore:// （旧版本0.27代码没有这一行）
  • 让m_faceTable的 afterAdd信号装配操作（使得每次新添加的face的各信号能连接到对应的Forwarder的函数）：
    • 让每个 face.afterReceiveInterest 连接到 this->startProcessInterest （处理兴趣包）
    • 让每个 face.afterReceiveData 连接到 this->startProcessData （处理数据包）
    • 让每个 face.afterReceiveNack 连接到 this->startProcessNack （处理Nack）
    • 让每个 face.onDroppedInterest 连接到 this->onDroppedInterest （往FIB添加新路由时）
  • 给 m_faceTable 的 beforeRemove 信号装配操作： cleanupOnFaceRemoval （每次移除 face 前先 cleanup ）
  • 给 m_fib 的 afterNewNextHop 信号装配操作： this->startProcessNewNextHop
  • 设置 m_strategyChoice 为默认的 fw::BestRouteStrategy2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
static NamegetDefaultStrategyName()
    { return fw::BestRouteStrategy2::getStrategyName(); }

Forwarder::Forwarder(FaceTable& faceTable)
  : m_faceTable(faceTable)
  , m_unsolicitedDataPolicy(make_unique<fw::DefaultUnsolicitedDataPolicy>())
  , m_fib(m_nameTree)
  , m_pit(m_nameTree)
  , m_measurements(m_nameTree)
  , m_strategyChoice(*this)
  , m_csFace(face::makeNullFace(FaceUri("contentstore://")))
{
  m_faceTable.addReserved(m_csFace, face::FACEID_CONTENT_STORE);

  m_faceTable.afterAdd.connect([this] (const Face& face) {
    face.afterReceiveInterest.connect(
      [this, &face] (const Interest& interest, const EndpointId& endpointId) {
        this->startProcessInterest(FaceEndpoint(face, endpointId), interest);
      });
    face.afterReceiveData.connect(
      [this, &face] (const Data& data, const EndpointId& endpointId) {
        this->startProcessData(FaceEndpoint(face, endpointId), data);
      });
    face.afterReceiveNack.connect(
      [this, &face] (const lp::Nack& nack, const EndpointId& endpointId) {
        this->startProcessNack(FaceEndpoint(face, endpointId), nack);
      });
    face.onDroppedInterest.connect(
      [this, &face] (const Interest& interest) {
        this->onDroppedInterest(FaceEndpoint(face, 0), interest);
      });
  });

  m_faceTable.beforeRemove.connect([this] (const Face& face) {
    cleanupOnFaceRemoval(m_nameTree, m_fib, m_pit, face);
  });

  m_fib.afterNewNextHop.connect([&] (const Name& prefix, const fib::NextHop& nextHop) {
    this->startProcessNewNextHop(prefix, nextHop);
  });

  m_strategyChoice.setDefaultStrategy(getDefaultStrategyName());
}

回顾下 forwarder.hpp 的函数的18~25行，构造函数里信号连接的 startProcessXXX 函数转到了 Forwarder 的 onIncomingXXX 了。于是让我们再来关注这几个函数

1
2
3
4
5
6
7
8
void startProcessInterest(const FaceEndpoint& ingress, const Interest& interest)
    { this->onIncomingInterest(ingress, interest); }
void startProcessData(const FaceEndpoint& ingress, const Data& data)
    { this->onIncomingData(ingress, data); }
void startProcessNack(const FaceEndpoint& ingress, const lp::Nack& nack)
    { this->onIncomingNack(ingress, nack); }
void startProcessNewNextHop(const Name& prefix, const fib::NextHop& nextHop)
    { this->onNewNextHop(prefix, nextHop); }

前三个函数就是手册里讲的那一长串了。

onIncomingInterest函数及其后续函数

onIncomingInterest函数

触发方式： Forwarder::onIncomingInterest 函数从 Forwarder::startProcessInterest 方法进入，该方法由接口 Face::afterReceiveData 信号触发。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
void
Forwarder::onIncomingInterest(const FaceEndpoint& ingress, const Interest& interest)
{
  // 给interest包打上IncomingFaceId标签
  NFD_LOG_DEBUG("onIncomingInterest in=" << ingress << " interest=" << interest.getName());//可以理解为interest的下游接口
  interest.setTag(make_shared<lp::IncomingFaceIdTag>(ingress.face.getId()));
  ++m_counters.nInInterests;

  // 如果face不是local，同时interest的Name的前缀又是"localhost"，则drop
  bool isViolatingLocalhost = ingress.face.getScope() == ndn::nfd::FACE_SCOPE_NON_LOCAL &&
                              scope_prefix::LOCALHOST.isPrefixOf(interest.getName());
  if (isViolatingLocalhost) {
    NFD_LOG_DEBUG("onIncomingInterest in=" << ingress
                  << " interest=" << interest.getName() << " violates /localhost");
    // (drop)
    return;
  }

  // 检查m_deadNonceList里有没有与当前包一样的内容，如果有，说明interest循环了，执行this->onInterestLoop
  bool hasDuplicateNonceInDnl = m_deadNonceList.has(interest.getName(), interest.getNonce());
  if (hasDuplicateNonceInDnl) {
    // goto Interest loop pipeline
    this->onInterestLoop(ingress, interest);
    return;
  }

  // 如果interest传到了Producer的地方，就剥掉forwardingHint
  if (!interest.getForwardingHint().empty() &&
      m_networkRegionTable.isInProducerRegion(interest.getForwardingHint())) {
    NFD_LOG_DEBUG("onIncomingInterest in=" << ingress
                  << " interest=" << interest.getName() << " reaching-producer-region");
    const_cast<Interest&>(interest).setForwardingHint({});
  }

  // 尝试将interest插入PIT表（如果在PIT找到了该interest，自然就不必插入了）
  shared_ptr<pit::Entry> pitEntry = m_pit.insert(interest).first;

  // 检测PIT里有没有相同的Nonce
  int dnw = fw::findDuplicateNonce(*pitEntry, interest.getNonce(), ingress.face);
  bool hasDuplicateNonceInPit = dnw != fw::DUPLICATE_NONCE_NONE;
  if (ingress.face.getLinkType() == ndn::nfd::LINK_TYPE_POINT_TO_POINT) {
    // p2p face的duplicate Nonce不被看作Loop
    hasDuplicateNonceInPit = hasDuplicateNonceInPit && !(dnw & fw::DUPLICATE_NONCE_IN_SAME);
  }
  // 如果重复了，说明interest循环了，执行this->onInterestLoop
  if (hasDuplicateNonceInPit) {
    // goto Interest loop pipeline
    this->onInterestLoop(ingress, interest);
    this->dispatchToStrategy(*pitEntry,
      [&] (fw::Strategy& strategy) { strategy.afterReceiveLoopedInterest(ingress, interest, *pitEntry); });
    return;
  }

  // 查CS
  if (!pitEntry->hasInRecords()) {
    m_cs.find(interest,
              bind(&Forwarder::onContentStoreHit, this, ingress, pitEntry, _1, _2),
              bind(&Forwarder::onContentStoreMiss, this, ingress, pitEntry, _1));
  }
  else {  // 如果都有人请求过了，那肯定说明CS里没有，就不用浪费时间查询了
    this->onContentStoreMiss(ingress, pitEntry, interest);
  }
}

onInterestLoop函数

onInterestLoop 函数就是执行一下 interest 包陷入循环后的善后操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
void
Forwarder::onInterestLoop(const FaceEndpoint& ingress, const Interest& interest)
{
  // 如果是广播或者ad-hoc就不特殊处理，直接return
  if (ingress.face.getLinkType() != ndn::nfd::LINK_TYPE_POINT_TO_POINT) {
    NFD_LOG_DEBUG("onInterestLoop in=" << ingress
                  << " interest=" << interest.getName() << " drop");
    return;
  }

  NFD_LOG_DEBUG("onInterestLoop in=" << ingress << " interest=" << interest.getName()
                << " send-Nack-duplicate");

  // 如果是p2p，interest都循环了，相当于Not ACK，执行Nack操作
  lp::Nack nack(interest);
  nack.setReason(lp::NackReason::DUPLICATE);
  ingress.face.sendNack(nack, ingress.endpoint);
}

onContentStoreMiss函数

如果CS中没有，执行 onContentStoreMiss 函数，给PIT加上输入记录，更新到期时间。

然后看看 interest 包里有没有钦定的下一跳路由的标签（如 ndn-simple.cpp 的例子就是没有），如果有，就让钦定好的下一跳路由进行 onOutgoingInterest 。如果找不到就根据 pitEntry 的状态决定是否转发（如果要转发，就是根据FIB的查出最佳的下一跳，然后执行对应的 onOutgoingInterest 函数）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
void
Forwarder::onContentStoreMiss(const FaceEndpoint& ingress,
                              const shared_ptr<pit::Entry>& pitEntry, const Interest& interest)
{
  NFD_LOG_DEBUG("onContentStoreMiss interest=" << interest.getName());
  ++m_counters.nCsMisses;
  afterCsMiss(interest);  // 似乎该信号是空的

  // 给pitEntry加上InRecord
  pitEntry->insertOrUpdateInRecord(ingress.face, interest);

  // 更新到期时间
  auto lastExpiring = std::max_element(pitEntry->in_begin(), pitEntry->in_end(),
                                       [] (const auto& a, const auto& b) {
                                         return a.getExpiry() < b.getExpiry();
                                       });
  auto lastExpiryFromNow = lastExpiring->getExpiry() - time::steady_clock::now();
  this->setExpiryTimer(pitEntry, time::duration_cast<time::milliseconds>(lastExpiryFromNow));

  // 如果钦定好了下一跳路由
  auto nextHopTag = interest.getTag<lp::NextHopFaceIdTag>();
  if (nextHopTag != nullptr) {
    // 并且能找到下一跳的face
    Face* nextHopFace = m_faceTable.get(*nextHopTag);
    if (nextHopFace != nullptr) {
      NFD_LOG_DEBUG("onContentStoreMiss interest=" << interest.getName()
                    << " nexthop-faceid=" << nextHopFace->getId());
      // 那么就交给下一跳路由进行Outgoing
      this->onOutgoingInterest(pitEntry, FaceEndpoint(*nextHopFace, 0), interest);
    }
    return;
  }

  // 根据pitEntry的状态信息，选择收到interest后的转发/不转发策略
  this->dispatchToStrategy(*pitEntry,
    [&] (fw::Strategy& strategy) {
      strategy.afterReceiveInterest(FaceEndpoint(ingress.face, 0), interest, pitEntry);
    });
}

onContentStoreHit函数

CS缓存命中将会触发，不过以 ndn-simple.cpp 为例，因为包里面一个Payload的 1024B 里是随机的垃圾数据，所以基本上不可能出现 CS 命中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
void
Forwarder::onContentStoreHit(const FaceEndpoint& ingress, const shared_ptr<pit::Entry>& pitEntry,
                             const Interest& interest, const Data& data)
{
  NFD_LOG_DEBUG("onContentStoreHit interest=" << interest.getName());
  ++m_counters.nCsHits;
  afterCsHit(interest, data);

  // 给数据打标签
  data.setTag(make_shared<lp::IncomingFaceIdTag>(face::FACEID_CONTENT_STORE));

  pitEntry->isSatisfied = true;  // 把该Entry标记为Satisfied
  pitEntry->dataFreshnessPeriod = data.getFreshnessPeriod();  //  更新dataFreshnessPeriod

  // 将到期时间设置为现在，到时候执行onInterestFinalize
  this->setExpiryTimer(pitEntry, 0_ms);

  beforeSatisfyInterest(*pitEntry, *m_csFace, data);  // 该信号由strategy、asf-strategy等设置
  this->dispatchToStrategy(*pitEntry,
    [&] (fw::Strategy& strategy) { strategy.beforeSatisfyInterest(pitEntry, FaceEndpoint(*m_csFace, 0), data); });

  // 执行Strategy::sendData->Forwarder::onOutgoingData
  this->dispatchToStrategy(*pitEntry,
    [&] (fw::Strategy& strategy) { strategy.afterContentStoreHit(pitEntry, ingress, data); });
}

onOutgoingInterest函数

如果决定转发 interest ，则会调用本函数

1
2
3
4
5
6
7
8
9
10
11
12
13
void
Forwarder::onOutgoingInterest(const shared_ptr<pit::Entry>& pitEntry,
                              const FaceEndpoint& egress, const Interest& interest)
{
  NFD_LOG_DEBUG("onOutgoingInterest out=" << egress << " interest=" << pitEntry->getName());

  // 给pitEntry加上OutRecord
  pitEntry->insertOrUpdateOutRecord(egress.face, interest);

  // 交给链路层执行doSendInterest
  egress.face.sendInterest(interest, egress.endpoint);
  ++m_counters.nOutInterests;
}

onInterestFinalize函数

一些善后工作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
void
Forwarder::onInterestFinalize(const shared_ptr<pit::Entry>& pitEntry)
{
  NFD_LOG_DEBUG("onInterestFinalize interest=" << pitEntry->getName()
                << (pitEntry->isSatisfied ? " satisfied" : " unsatisfied"));

  if (!pitEntry->isSatisfied) {
    beforeExpirePendingInterest(*pitEntry);
  }

  // 如果MustBeFresh，且FreshnessPeriod小于LifeTime，则需要插入DeadNonceList
  this->insertDeadNonceList(*pitEntry, nullptr);

  // Increment satisfied/unsatisfied Interests counter
  if (pitEntry->isSatisfied) {
    ++m_counters.nSatisfiedInterests;
  }
  else {
    ++m_counters.nUnsatisfiedInterests;
  }

  // 删掉PIT表里对应条目 
  pitEntry->expiryTimer.cancel();
  m_pit.erase(pitEntry.get());
}

onIncomingData函数及其后续函数

onIncomingData函数

拉来了 data 包后，执行该函数，给每个in-record都发包。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
void
Forwarder::onIncomingData(const FaceEndpoint& ingress, const Data& data)
{
  // 给data包打上OncomingFaceId标签
  NFD_LOG_DEBUG("onIncomingData in=" << ingress << " data=" << data.getName());
  data.setTag(make_shared<lp::IncomingFaceIdTag>(ingress.face.getId()));
  ++m_counters.nInData;

  // 如果face不是local，同时interest的Name的前缀又是"localhost"，则drop
  bool isViolatingLocalhost = ingress.face.getScope() == ndn::nfd::FACE_SCOPE_NON_LOCAL &&
                              scope_prefix::LOCALHOST.isPrefixOf(data.getName());
  if (isViolatingLocalhost) {
    NFD_LOG_DEBUG("onIncomingData in=" << ingress << " data=" << data.getName() << " violates /localhost");
    // (drop)
    return;
  }

  // 如果PIT表里没有对应条目，说明data是不请自来的，执行onDataUnsolicited
  pit::DataMatchResult pitMatches = m_pit.findAllDataMatches(data);
  if (pitMatches.size() == 0) {
    // goto Data unsolicited pipeline
    this->onDataUnsolicited(ingress, data);
    return;
  }

  // 尝试往CS里插入data
  m_cs.insert(data);

  // 如果是单个PIT命中
  if (pitMatches.size() == 1) {
    auto& pitEntry = pitMatches.front();

    NFD_LOG_DEBUG("onIncomingData matching=" << pitEntry->getName());

    // 设置PIT到期，准备onOutgoingData
    this->setExpiryTimer(pitEntry, 0_ms);

    beforeSatisfyInterest(*pitEntry, ingress.face, data);
    // 调用afterReceiveData，这将执行Strategy::sendDataToAll（最终调用到onOutgoingData）
    this->dispatchToStrategy(*pitEntry,
      [&] (fw::Strategy& strategy) { strategy.afterReceiveData(pitEntry, ingress, data); });

    // 设置pitEntry为已满足（从而有可能添加到DeadNonceList）
    pitEntry->isSatisfied = true;
    pitEntry->dataFreshnessPeriod = data.getFreshnessPeriod();

    // 如果MustBeFresh，且FreshnessPeriod小于LifeTime，则需要插入DeadNonceList
    this->insertDeadNonceList(*pitEntry, &ingress.face);

    // 删掉该条目的out-record
    pitEntry->deleteOutRecord(ingress.face);
  }
  // 如果是多个PIT命中
  else {
    std::set<std::pair<Face*, EndpointId>> pendingDownstreams;
    auto now = time::steady_clock::now();

    // 把每个的in-record添加到“下一跳们”（后续集中处理）
    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.emplace(&inRecord.getFace(), 0);
        }
      }

      // 设置PIT到期，准备onOutgoingData
      this->setExpiryTimer(pitEntry, 0_ms);

      // invoke PIT satisfy callback
      beforeSatisfyInterest(*pitEntry, ingress.face, data);
      this->dispatchToStrategy(*pitEntry,
        [&] (fw::Strategy& strategy) { strategy.beforeSatisfyInterest(pitEntry, ingress, data); });

      // 设置pitEntry为已满足（从而有可能添加到DeadNonceList）
      pitEntry->isSatisfied = true;
      pitEntry->dataFreshnessPeriod = data.getFreshnessPeriod();

      // 如果MustBeFresh，且FreshnessPeriod小于LifeTime，则需要插入DeadNonceList
      this->insertDeadNonceList(*pitEntry, &ingress.face);

      // 清除掉in-record
      pitEntry->clearInRecords();
      pitEntry->deleteOutRecord(ingress.face);
    }

    // 对于“下一跳们”的每一个，都执行onOutgoingData
    for (const auto& pendingDownstream : pendingDownstreams) {
      if (pendingDownstream.first->getId() == ingress.face.getId() &&
          pendingDownstream.second == ingress.endpoint &&
          pendingDownstream.first->getLinkType() != ndn::nfd::LINK_TYPE_AD_HOC) {
        continue;
      }
      // goto outgoing Data pipeline
      this->onOutgoingData(data, FaceEndpoint(*pendingDownstream.first, pendingDownstream.second));
    }
  }
}

onDataUnsolicited函数

如果数据是不请自来的，执行本函数。看看对于不请自来的数据的策略是啥，如果策略是缓存，那么就尝试插入CS。否则啥都不干。

1
2
3
4
5
6
7
8
9
10
11
12
void
Forwarder::onDataUnsolicited(const FaceEndpoint& ingress, const Data& data)
{
  // 查m_unsolicitedDataPolicy，看看对于不请自来的数据的策略是啥
  fw::UnsolicitedDataDecision decision = m_unsolicitedDataPolicy->decide(ingress.face, data);
  if (decision == fw::UnsolicitedDataDecision::CACHE) {  // 如果策略是缓存
    // 尝试插入CS
    m_cs.insert(data, true);
  }

  NFD_LOG_DEBUG("onDataUnsolicited in=" << ingress << " data=" << data.getName() << " decision=" << decision);
}

onOutgoingData函数

做完一些检查工作后，准备发data包

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
void
Forwarder::onOutgoingData(const Data& data, const FaceEndpoint& egress)
{
  if (egress.face.getId() == face::INVALID_FACEID) {  // 接收数据方失效，return
    NFD_LOG_WARN("onOutgoingData out=(invalid) data=" << data.getName());
    return;
  }
  NFD_LOG_DEBUG("onOutgoingData out=" << egress << " data=" << data.getName());

  // 如果face不是local，同时interest的Name的前缀又是"localhost"，则drop
  bool isViolatingLocalhost = egress.face.getScope() == ndn::nfd::FACE_SCOPE_NON_LOCAL &&
                              scope_prefix::LOCALHOST.isPrefixOf(data.getName());
  if (isViolatingLocalhost) {
    NFD_LOG_DEBUG("onOutgoingData out=" << egress << " data=" << data.getName() << " violates /localhost");
    // (drop)
    return;
  }

  // TODO: 下个版本计划进行流量管理，然而ndnSIM2.8这个版本还没实现

  // 发data包
  egress.face.sendData(data, egress.endpoint);
  ++m_counters.nOutData;
}

onIncomingNack函数

如果Incoming时Nack了，则执行该函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
void
Forwarder::onIncomingNack(const FaceEndpoint& ingress, const lp::Nack& nack)
{
  // 给nack包打上IncomingFaceId标签
  nack.setTag(make_shared<lp::IncomingFaceIdTag>(ingress.face.getId()));
  ++m_counters.nInNacks;

  // 如果不是点对点通信，那么根本都没定义Nack，直接扔掉
  if (ingress.face.getLinkType() != ndn::nfd::LINK_TYPE_POINT_TO_POINT) {
    NFD_LOG_DEBUG("onIncomingNack in=" << ingress
                  << " nack=" << nack.getInterest().getName() << "~" << nack.getReason()
                  << " link-type=" << ingress.face.getLinkType());
    return;
  }

  // 查PIT
  shared_ptr<pit::Entry> pitEntry = m_pit.find(nack.getInterest());
  // 如果PIT查不到就扔掉
  if (pitEntry == nullptr) {
    NFD_LOG_DEBUG("onIncomingNack in=" << ingress << " nack=" << nack.getInterest().getName()
                  << "~" << nack.getReason() << " no-PIT-entry");
    return;
  }

  // 查out-record
  auto outRecord = pitEntry->getOutRecord(ingress.face);
  // 如果out-record查不到，扔掉
  if (outRecord == pitEntry->out_end()) {
    NFD_LOG_DEBUG("onIncomingNack in=" << ingress << " nack=" << nack.getInterest().getName()
                  << "~" << nack.getReason() << " no-out-record");
    return;
  }

  // 和out-record的最后一个记录的随机数不同，扔掉
  if (nack.getInterest().getNonce() != outRecord->getLastNonce()) {
    NFD_LOG_DEBUG("onIncomingNack in=" << ingress << " nack=" << nack.getInterest().getName()
                  << "~" << nack.getReason() << " wrong-Nonce " << nack.getInterest().getNonce()
                  << "!=" << outRecord->getLastNonce());
    return;
  }

  NFD_LOG_DEBUG("onIncomingNack in=" << ingress << " nack=" << nack.getInterest().getName()
                << "~" << nack.getReason() << " OK");

  // 标记out-record为Nack
  outRecord->setIncomingNack(nack);

  // 如果所有out-record都Nack了，将到期时间设置为现在
  if (!fw::hasPendingOutRecords(*pitEntry)) {
    this->setExpiryTimer(pitEntry, 0_ms);
  }

  // 执行策略afterReceiveNack
  this->dispatchToStrategy(*pitEntry,
    [&] (fw::Strategy& strategy) { strategy.afterReceiveNack(ingress, nack, pitEntry); });
}

onOutgoingNack函数

如果Outgoing时Nack了，则执行该函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
void
Forwarder::onOutgoingNack(const shared_ptr<pit::Entry>& pitEntry,
                          const FaceEndpoint& egress, const lp::NackHeader& nack)
{
  if (egress.face.getId() == face::INVALID_FACEID) {
    NFD_LOG_WARN("onOutgoingNack out=(invalid)"
                 << " nack=" << pitEntry->getInterest().getName() << "~" << nack.getReason());
    return;
  }

  // 检查in-record
  auto inRecord = pitEntry->getInRecord(egress.face);

  // 如果找不到in-record，就丢掉（都不知道哪个interest的问题，不发Nack啊）
  if (inRecord == pitEntry->in_end()) {
    NFD_LOG_DEBUG("onOutgoingNack out=" << egress
                  << " nack=" << pitEntry->getInterest().getName()
                  << "~" << nack.getReason() << " no-in-record");
    return;
  }

  // 如果不是点对点通信，那么根本都没定义Nack，直接扔掉
  if (egress.face.getLinkType() != ndn::nfd::LINK_TYPE_POINT_TO_POINT) {
    NFD_LOG_DEBUG("onOutgoingNack out=" << egress
                  << " nack=" << pitEntry->getInterest().getName() << "~" << nack.getReason()
                  << " link-type=" << egress.face.getLinkType());
    return;
  }

  NFD_LOG_DEBUG("onOutgoingNack out=" << egress
                << " nack=" << pitEntry->getInterest().getName()
                << "~" << nack.getReason() << " OK");

  // 准备好Nack包
  lp::Nack nackPkt(inRecord->getInterest());
  nackPkt.setHeader(nack);

  // 擦掉in-record
  pitEntry->deleteInRecord(egress.face);

  // 给interest发Nack
  egress.face.sendNack(nackPkt, egress.endpoint);
  ++m_counters.nOutNacks;
}

总的来说， Forwarder 类的功能很清晰，主体就是处理interest、data、Nack这三个玩意，其余具体的函数不过是它的实现罢了。比如处理interest，每次发包都要查PIT、CS，找下一跳查FIB。这也是 Forwarder 要管理NameTree的原因，因为转发功能就要用到这三个玩意，而它们通用的接口则是由FaceTable统一管理。具体的转发策略则是由 Strategy 及其派生类实现，具体的内部操作就要看CS、PIT、FIB里面的实现了，它们对于Forwarder而言只暴露出了接口，相当于透明的。

而 onIncomingInterest 、 onIncomingData 、 onIncomingNack 函数则是由信号 Face::afterReceiveInterest 、 Face::afterReceiveData 、 Face::afterReceiveNack 触发，这也降低了各模块的耦合程度。

由 Forwarder 类发散开， Strategy 是具体的转发策略， Cs 、 Pit 、 Fib 是三个基本功能的实现， Producer 和 Consumer 是生产者和消费者， Interest 和 Data 是兴趣包和数据包的类型。这也呼应了ndnSIM学习（三）——ndnSIM源码阅读计划为什么要看这些内容了。

扫一扫，分享到微信

微信分享二维码
本站总访问量次 |

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

这一定是一个<br><br>很长很长的一生啊