2022-05-23

字数统计: 3.7k字 | 阅读时长: 16分

NDN代码分析(5) 兴趣包和数据包代码

从consumer发兴趣包到producer返回data包的全过程

一图明白整个发包、收包流程

黄颜色可以理解为预处理流程，发端执行「黄色+蓝色」的流程
中间的节点执行「绿色+蓝色」的流程
最后的节点（生产者或消费者）执行「绿色+蓝色+紫色」的流程

生产者发包

上篇文章中，我们知道了：整个ns3软件的仿真流程就是：添加事件->仿真器执行所有事件。

那么， Consumer 发兴趣包的事件是如何添加的呢？答案是：我们设置了消费者的兴趣频率是10，所以每秒发10个包。具体而言，消费者的类型是 ConsumerCbr ，而其中发包则是由函数 ScheduleNextPacket 控制，核心代码如下——

void
ConsumerCbr::ScheduleNextPacket()
{
  if (m_firstTime) {
    m_sendEvent = Simulator::Schedule(Seconds(0.0), &Consumer::SendPacket, this);
    m_firstTime = false;
  }
  else if (!m_sendEvent.IsRunning())
    m_sendEvent = Simulator::Schedule((m_random == 0) ? Seconds(1.0 / m_frequency)
                                                      : Seconds(m_random->GetValue()),
                                      &Consumer::SendPacket, this);
}

该函数的事件请求调用基类的 Consumer::SendPacket 函数，该函数造出一个兴趣包，然后用 m_appLink->onReceiveInterest 发出去.

void
Consumer::SendPacket()
{
  // ...省略前面的seq处理过程，下面是核心逻辑

  // shared_ptr<Interest> interest = make_shared<Interest> ();
  shared_ptr<Interest> interest = make_shared<Interest>();//shared_ptr是智能指针，make_shared动态分配类型为Interest的对象，()为初始化参数
  interest->setNonce(m_rand->GetValue(0, std::numeric_limits<uint32_t>::max()));
  interest->setName(*nameWithSequence);
  interest->setCanBePrefix(false);
  time::milliseconds interestLifeTime(m_interestLifeTime.GetMilliSeconds());
  interest->setInterestLifetime(interestLifeTime);

  WillSendOutInterest(seq);

  m_transmittedInterests(interest, this, m_face);
  m_appLink->onReceiveInterest(*interest);  // 核心逻辑

  ScheduleNextPacket();
}

AppLinkService::onReceiveInterest 则是调用了 LinkService::receiveInterest 函数，该函数触发了 LinkService 的 afterReceiveInterest 信号。

等下，有没有感觉这个信号很熟悉？回一下 Forwarder 的构造函数，里面的 faceTable 将里面每个 face 的 afterReceiveInterest 信号都连接到了 Forwarder::startProcessInterest 上。

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);
    });
});

但好像还是差点意思，因为前面触发的是 LinkService 的 afterReceiveInterest 信号，这里是 face 的 afterReceiveInterest 信号。它们之间是不是有什么联系呢？我们看看 LinkService 的头文件定义，再结合 Face 的构造函数，发现里面有玄机。

// 链路服务类
class LinkService : protected virtual LinkServiceCounters, noncopyable
{
// 前面的函数省略，关键在里面的成员函数
private:
  Face* m_face;  
  Transport* m_transport;
};

// ndnStack在InstallAll时，调用了本函数，为节点生成了链路、物理传输助手，并把它们绑定到face上
shared_ptr<Face>
StackHelper::PointToPointNetDeviceCallback(Ptr<Node> node, Ptr<L3Protocol> ndn,
                                           Ptr<NetDevice> device) const
{
  // 前面省略，这里创建的链路助手和物理传输层助手，并交给这个face管理
  auto linkService = make_unique<::nfd::face::GenericLinkService>(opts);

  auto transport = make_unique<NetDeviceTransport>(node, netDevice,
                                                   constructFaceUri(netDevice),
                                                   constructFaceUri(remoteNetDevice));

  auto face = std::make_shared<Face>(std::move(linkService), std::move(transport));
  face->setMetric(1);

  ndn->addFace(face);
  NS_LOG_LOGIC("Node " << node->GetId() << ": added Face as face #"
                       << face->getLocalUri());

  return face;
}

// Face的构造函数，构造Face时将会和链路服务、物理传输层服务捆绑到一起
Face::Face(unique_ptr<LinkService> service, unique_ptr<Transport> transport)
  : afterReceiveInterest(service->afterReceiveInterest)
  , afterReceiveData(service->afterReceiveData)
  , afterReceiveNack(service->afterReceiveNack)
  , onDroppedInterest(service->onDroppedInterest)
  , afterStateChange(transport->afterStateChange)
  , m_id(INVALID_FACEID)
  , m_service(std::move(service))
  , m_transport(std::move(transport))
  , m_counters(m_service->getCounters(), m_transport->getCounters())
  , m_metric(0)
{
  m_service->setFaceAndTransport(*this, *m_transport);
  m_transport->setFaceAndLinkService(*this, *m_service);
}

注意到 Signal 继承了 noncopyable ，而 Face 的构造函数中 afterReceiveData(service->afterReceiveData) 又把自己的信号和链路服务的信号捆绑到了一起，所以 LinkService 的 afterReceiveInterest 通过 Face 连接到了 Forwarder::startProcessInterest 上。

也就是说， LinkService 的 afterReceiveInterest 信号等价于 Forwarder::startProcessInterest 函数！

而 Forwarder::startProcessInterest 移交给了 Forwarder::onIncomingInterest 函数，详见文章ndnSIM学习（七）——转发处理forwarder.cpp、forwarder.hpp，该函数流程图如下

总之就是 Forwarder 对兴趣包进行处理。如果找到了，就会执行 Strategy::sendData 。否则就是 onOutgoingInterest 向下一跳继续往后找。这里我们假设没找到，继续向下一跳找。

这里执行 face.sendInterest 函数，又执行 LinkService::sendInterest ，然后 GenericLinkService::doSendInterest 到 sendNetPacket 给链路层进行 Frag 分片，分完片执行 sendLpPacket ，再通过 sendPacket 交给物理传输层执行 Transport::send ，再交给 NetDeviceTransport::doSend ，移交给 PointToPointNetDevice::Send ，其中执行了 TransmitStart ，又踢皮球给了 PointToPointChannel::TransmitStart 。终于，经过无数踢皮球后，我们终于看到了这一句话——

1
2
3

Simulator::ScheduleWithContext (m_link[wire].m_dst->GetNode ()->GetId (),
                                txTime + m_delay, &PointToPointNetDevice::Receive,
                                m_link[wire].m_dst, p->Copy ());

也就是说，如果向下一跳继续往后找的话，最终就是给目标节点 dst 添加一个 PointToPointNetDevice::Receive 事件。我个人比较好奇：每个包发了后，会让仿真系统推迟一段时间 txTime ，这是如何实现的。其实是在这里实现的——

bool
PointToPointNetDevice::TransmitStart (Ptr<Packet> p)
{
  // 前面的设置略

  Time txTime = m_bps.CalculateBytesTxTime (p->GetSize ());  // 就是这里，用m_bps结合p->GetSize()算出来的
  Time txCompleteTime = txTime + m_tInterframeGap;

  NS_LOG_LOGIC ("Schedule TransmitCompleteEvent in " << txCompleteTime.GetSeconds () << "sec");
  Simulator::Schedule (txCompleteTime, &PointToPointNetDevice::TransmitComplete, this);

  bool result = m_channel->TransmitStart (p, this, txTime);
  if (result == false)
    {
      m_phyTxDropTrace (p);
    }
  return result;
}

PointToPointNetDevice::Receive 事件触发后，会执行 m_promiscCallback 回调函数。妈呀，看到回调函数就头疼了。经过我一个小时的探案，终于发现这个回调函数绑定的是 Node::PromiscReceiveFromDevice 函数，这个函数踢皮球给 ReceiveFromDevice ，它执行该节点的 m_handlers ，而这个 ProtocolHandler 。。。我吐了，怎么又是回调函数啊！还是回调函数族。

还好，这个比较好找，这个回调函数是由 Node::RegisterProtocolHandler 执行 m_handlers.push_back 。这次运气很好，我眼睛很尖，一眼就看到了 NetDeviceTransport 的构造函数里有这样的代码

1
2
3

m_node->RegisterProtocolHandler(MakeCallback(&NetDeviceTransport::receiveFromNetDevice, this),
                                L3Protocol::ETHERNET_FRAME_TYPE, m_netDevice,
                                true /*promiscuous mode*/);

也就是说，Node::ReceiveFromDevice 把皮球踢给了 NetDeviceTransport::receiveFromNetDevice ，这个函数去掉报头扔给 Transport::receive 。这个函数终于把数据转交给了链路层函数 LinkService::receivePacket ，这个函数一脚踢给 GenericLinkService::doReceivePacket 。

void
GenericLinkService::doReceivePacket(const Block& packet, const EndpointId& endpoint)
{
  try {
    // 省略前面的代码

    bool isReassembled = false;
    Block netPkt;
    lp::Packet firstPkt;
    std::tie(isReassembled, netPkt, firstPkt) = m_reassembler.receiveFragment(endpoint, pkt);
    if (isReassembled) {
      this->decodeNetPacket(netPkt, firstPkt, endpoint);
    }
  }
  catch (const tlv::Error& e) {
    // 省略这部分代码
  }
}

也就是说，这个函数负责接收每个 Fragment ，合并它们，并检查其完整性。如果完整，那么就可以执行 decodeNetPacket 解码函数了。具体内容贴代码就行了，这里不细讲了。

void
GenericLinkService::decodeNetPacket(const Block& netPkt, const lp::Packet& firstPkt,
                                    const EndpointId& endpointId)
{
  try {
    switch (netPkt.type()) {
      case tlv::Interest:
        if (firstPkt.has<lp::NackField>()) {
          this->decodeNack(netPkt, firstPkt, endpointId);
        }
        else {
          this->decodeInterest(netPkt, firstPkt, endpointId);
        }
        break;
      case tlv::Data:
        this->decodeData(netPkt, firstPkt, endpointId);
        break;
      default:
        ++this->nInNetInvalid;
        NFD_LOG_FACE_WARN("unrecognized network-layer packet TLV-TYPE " << netPkt.type() << ": DROP");
        return;
    }
  }
  catch (const tlv::Error& e) {
    ++this->nInNetInvalid;
    NFD_LOG_FACE_WARN("packet parse error (" << e.what() << "): DROP");
  }
}

因为我们这里收到的是兴趣包，所以执行 GenericLinkService::decodeInterest 。在这个函数将 Block 类型的 netPkt 输入强制转化为了 Interest 类型，然后按照 lp::Packet& firstPkt 给兴趣包打上各种标签，最后执行 LinkService::receiveInterest ，火箭发射，芜湖！

LinkService::receiveInterest 通过 afterReceiveInterest 信号，移交给 Forwarder 层的 Forwarder::startProcessInterest 再到 Forwarder::onIncomingInterest 。什么？你这里看不懂？前面已经分析过一遍了，不用再讲了吧，看不懂再回去看吧。

这里执行 face.sendInterest 函数，又执行 LinkService::sendInterest ，然后这里因为已经到了生产者了，而生产者由 AppLinkService 管理，所以不是 GenericLinkService::doSendInterest ，而是 AppLinkService::doSendInterest 进行。该函数添加了事件 App::OnInterest ，而实际上根据其虚特性，真正执行了 Producer::OnInterest

消费者回包

消费者用 Producer::OnInterest 收到了兴趣包后，返回一个数据包，调用 AppLinkService::onReceiveData ，再调用 LinkService::receiveData ，触发 afterReceiveData 。还是前面分析的，信号会连接到 Forwarder 层的 Forwarder::startProcessData 再到 Forwarder::onIncomingData ，详见文章ndnSIM学习（七）——转发处理forwarder.cpp、forwarder.hpp，该函数流程图如下

最后执行 onOutgoingData ，跳到 Face::sendData ，再跳到 LinkService::sendData ，继续跳到 GenericLinkService::doSendData ，跳到 GenericLinkService::sendNetPacket ，跳到 GenericLinkService::sendLpPacket ，跳到 LinkService::sendPacket ，终于该到物理传输层了。

交给物理传输层执行 Transport::send ，再交给 NetDeviceTransport::doSend ，移交给 PointToPointNetDevice::Send ，其中执行了 TransmitStart ，又踢皮球给了 PointToPointChannel::TransmitStart 。给目标节点 dst 添加一个 PointToPointNetDevice::Receive 事件，执行 m_promiscCallback 回调函数（这个回调函数绑定的是 Node::PromiscReceiveFromDevice 函数）。

这个函数踢皮球给 ReceiveFromDevice ，它执行该节点的 m_handlers ，通过回调函数执行 NetDeviceTransport::receiveFromNetDevice ，这个函数去掉报头扔给 Transport::receive 。这个函数终于把数据转交给了链路层函数 LinkService::receivePacket ，这个函数一脚踢给 GenericLinkService::doReceivePacket ，执行 decodeNetPacket 解码函数了。

好了！终于到和前面不一样的地方了！因为这里是数据包，执行 decodeData ，转到 LinkService::receiveData ，就到了 afterReceiveData 信号了。信号会连接到 Forwarder 层的 Forwarder::startProcessData 再到 Forwarder::onIncomingData 。

假设现在已经到了消费者的地方了。执行 onOutgoingData ，跳到 Face::sendData ，再跳到 LinkService::sendData 。因为已经到了消费者了，而消费者是由 AppLinkService 管理的，所以是由AppLinkService::doSendData 进行。该函数添加了事件 App::Data ，而实际上根据其虚特性，真正执行了 Consumer::Data 。

奈斯！全部搞定了！

总结

虽然过程实在是很晕，但其实整个代码的逻辑还是很清晰的。将代码分为网络层、链路层、物理传输层，每层都干自己该干的事情，组合起来就是从consumer发兴趣包到producer返回data包的全过程了。

apps之ndn-producer.cpp和ndn-consumer.cpp源码分析

~/ndnSIM/ns-3/src/ndnSIM/apps/ndn-producer.hpp 文件中定义了ndnSIM中生产者(producer)类的定义，代码如下：

class Producer : public App {
public:
  // 接口，返回关于本类对象的TypeId信息，方便外部配置
  static TypeId GetTypeId(void);

  // 构造函数，每当构造一个对象时，日志里输出相关信息
  // e.g. 输出[+0.000000000s 2 ndn.Producer:Producer()]
  Producer();  

  // 收到interest包后，装载一个data包（内部有编码为block的成员），最后把任务交接给到m_appLink->onReceiveData
  // 后续详见下面的流程图
  virtual void OnInterest(shared_ptr<const Interest> interest); //后续跳转到onReceiveData

protected:
  // 调用App创建相关的App管理接口，如AppLinkService这种东西，并把管理链路层和传输层的接口放到Ndn Stack里
  virtual void StartApplication(); // Called at time specified by Start

  // m_active = false并关闭m_face
  virtual void StopApplication(); // Called at time specified by Stop

private:
  Name m_prefix;  // 前缀
  Name m_postfix;  // 后缀
  uint32_t m_virtualPayloadSize;  // 一个包装载数据byte数
  Time m_freshness;  // Fresh time

  uint32_t m_signature;  // SignatureValue
  Name m_keyLocator;
};

~/ndnSIM/ns-3/src/ndnSIM/apps/ndn-consumer.hpp 文件中定义了ndnSIM中消费者(consumer)类的定义，代码如下：

class Consumer : public App {
public:
  // 接口，返回关于本类对象的TypeId信息
  static TypeId GetTypeId();

  // 构造函数以及虚析构函数（便于派生）
  Consumer();
  virtual ~Consumer(){};

  // 收到消费者返回的Data包，执行相应的处理与回显，并更新RTT(Round-Trip Time)
  virtual void OnData(shared_ptr<const Data> contentObject);

  // 如果Not ACK，显示对应的信息与原因
  virtual void OnNack(shared_ptr<const lp::Nack> nack);

  // 超时就重传（RTT翻倍）
  virtual void OnTimeout(uint32_t sequenceNumber);

  // 发interest包，该函数由ScheduleNextPacket设置事件触发
  // 该函数给interest包填充信息后把任务交接给m_appLink->onReceiveInterest准备向自己的链路层发interest
  void SendPacket();

  // 在兴趣包实际发送之前的准备工作
  virtual void WillSendOutInterest(uint32_t sequenceNumber);

public:
  typedef void (*LastRetransmittedInterestDataDelayCallback)(Ptr<App> app, uint32_t seqno, Time delay, int32_t hopCount);  // 函数指针类型
  typedef void (*FirstInterestDataDelayCallback)(Ptr<App> app, uint32_t seqno, Time delay, uint32_t retxCount, int32_t hopCount);  // 函数指针类型

protected:
  // 调用App创建相关的App管理接口，如AppLinkService这种东西，并把管理链路层和传输层的接口放到Ndn Stack里
  // 创建完执行ScheduleNextPacket，调用派生类的发包控制函数
  virtual void StartApplication();

  // m_active = false并关闭m_face
  virtual void StopApplication();

  // 发包的调度策略，如Cbr就是以恒定速率发包（纯虚函数，由子类实现）
  virtual void ScheduleNextPacket() = 0;

  // 检查数据包是否需要超时重传
  void CheckRetxTimeout();

  // 设置检查重传超时的频率
  void SetRetxTimer(Time retxTimer);

  // 获取当前检查重传超时的频率
  Time GetRetxTimer() const;

protected:
  Ptr<UniformRandomVariable> m_rand; // 随机数nonce生成器

  uint32_t m_seq;      // 标志目前的请求序列数值seq
  uint32_t m_seqMax;   // 最多请求多少个seq
  EventId m_sendEvent; // 发包对应的事件Id，在子类的发包事件中会返回一个发包事件Id，当App结束时，将会取消掉该事件
  Time m_retxTimer;    // 重传时间
  EventId m_retxEvent; // m_retxEvent对应的事件Id

  Ptr<RttEstimator> m_rtt; // RTT估计器

  Time m_offTime;          ///< \brief Time interval between packets
  Name m_interestName;     // 就是ndn-simple.cpp设置的Prefix
  Time m_interestLifeTime; // 就是ndn-simple.cpp设置的LifeTime

  // 省略部分定义

  RetxSeqsContainer m_retxSeqs; // 重传序列，每当超时的时候，就将目标序列号加入该重传序列
  
  SeqTimeoutsContainer m_seqTimeouts; // 超时序列号+时间

  SeqTimeoutsContainer m_seqLastDelay;// 
  SeqTimeoutsContainer m_seqFullDelay;
  std::map<uint32_t, uint32_t> m_seqRetxCounts;

  TracedCallback<Ptr<App> /* app */, uint32_t /* seqno */, Time /* delay */, int32_t /*hop count*/>
    m_lastRetransmittedInterestDataDelay;
  TracedCallback<Ptr<App> /* app */, uint32_t /* seqno */, Time /* delay */,
                 uint32_t /*retx count*/, int32_t /*hop count*/> m_firstInterestDataDelay;
};

ndn-simple.cpp 里的消费者是 ConsumerCbr ，其中我推测 Cbr 的含义是 Constant BitRate 的缩写，所以 ConsumerCbr 其实就是以常数速率发包的消费者。

class ConsumerCbr : public Consumer {
public:
  static TypeId GetTypeId();

  ConsumerCbr();
  virtual ~ConsumerCbr();

protected:
  // 发包规则：以常数频率Frequence发interest
  virtual void ScheduleNextPacket();

  // 设置随机性规则，可以选择均匀分布、泊松分布
  void SetRandomize(const std::string& value);

  // 获取随机规则
  std::string GetRandomize() const;

protected:
  double m_frequency;  // 发包频率
  bool m_firstTime;  // 标志是不是第一次请求
  Ptr<RandomVariableStream> m_random;  // 随机数生成器
  std::string m_randomType;  // 随机的规则
};

其他消费者

因为基类 Consumer 的函数 ScheduleNextPacket 是纯虚函数，所以消费者一定得用其派生类，比如前面的 ConsumerCbr 就是一个派生类。这里列举一下ndnSIM的消费者（我没细看，可能有错误，仅供参考）

ConsumerBatches 用于批量管理消费者，下面管理了一个消费者的节点链表。一旦 ConsumerBatches::StartApplication 被执行，整个类就会为它管理的所有节点执行 ConsumerBatches::ScheduleNextPacket 。其中每过 m_rtt->RetransmitTimeout() 就会触发该消费者的 Consumer::SendPacket
ConsumerCbr 是以恒定速率发送兴趣包请求的消费者。
ConsumerZipfMandelbrot 继承了 ConsumerCbr ，发包服从Zipf-Mandelbrot律，我也不知道是什么玩意。
ConsumerWindow 似乎是以滑动窗口的方式执行拥塞控制的消费者，我没有仔细看里面的代码，所以并不确定。代码里提到本类是高度实验性的，要小心使用
ConsumerPcon 继承了 ConsumerWindow 类，参考了论文https://dl.acm.org/citation.cfm?id=2984369