daemon/face/generic-link-service.cpp - NFD - Gitiles

 /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
 /*
  * Copyright (c) 2014-2019,  Regents of the University of California,
  *                           Arizona Board of Regents,
  *                           Colorado State University,
  *                           University Pierre & Marie Curie, Sorbonne University,
  *                           Washington University in St. Louis,
  *                           Beijing Institute of Technology,
  *                           The University of Memphis.
  *
  * This file is part of NFD (Named Data Networking Forwarding Daemon).
  * See AUTHORS.md for complete list of NFD authors and contributors.
  *
  * NFD is free software: you can redistribute it and/or modify it under the terms
  * of the GNU General Public License as published by the Free Software Foundation,
  * either version 3 of the License, or (at your option) any later version.
  *
  * NFD is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
  * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
  * PURPOSE.  See the GNU General Public License for more details.
  *
  * You should have received a copy of the GNU General Public License along with
  * NFD, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
  */

 #include "generic-link-service.hpp"

 #include <ndn-cxx/lp/tags.hpp>

 #include <cmath>

 namespace nfd {
 namespace face {

 NFD_LOG_INIT(GenericLinkService);

 constexpr size_t CONGESTION_MARK_SIZE = tlv::sizeOfVarNumber(lp::tlv::CongestionMark) + // type
                                         tlv::sizeOfVarNumber(sizeof(uint64_t)) +        // length
                                         tlv::sizeOfNonNegativeInteger(UINT64_MAX);      // value

 constexpr uint32_t DEFAULT_CONGESTION_THRESHOLD_DIVISOR = 2;

 GenericLinkService::GenericLinkService(const GenericLinkService::Options& options)
   : m_options(options)
   , m_fragmenter(m_options.fragmenterOptions, this)
   , m_reassembler(m_options.reassemblerOptions, this)
   , m_reliability(m_options.reliabilityOptions, this)
   , m_lastSeqNo(-2)
   , m_nextMarkTime(time::steady_clock::TimePoint::max())
   , m_lastMarkTime(time::steady_clock::TimePoint::min())
   , m_nMarkedSinceInMarkingState(0)
 {
   m_reassembler.beforeTimeout.connect([this] (auto...) { ++this->nReassemblyTimeouts; });
   m_reliability.onDroppedInterest.connect([this] (const auto& i) { this->notifyDroppedInterest(i); });
   nReassembling.observe(&m_reassembler);
 }

 void
 GenericLinkService::setOptions(const GenericLinkService::Options& options)
 {
   m_options = options;
   m_fragmenter.setOptions(m_options.fragmenterOptions);
   m_reassembler.setOptions(m_options.reassemblerOptions);
   m_reliability.setOptions(m_options.reliabilityOptions);
 }

 void
 GenericLinkService::requestIdlePacket(const EndpointId& endpointId)
 {
   // No need to request Acks to attach to this packet from LpReliability, as they are already
   // attached in sendLpPacket
   this->sendLpPacket({}, endpointId);
 }

 void
 GenericLinkService::sendLpPacket(lp::Packet&& pkt, const EndpointId& endpointId)
 {
   const ssize_t mtu = this->getTransport()->getMtu();

   if (m_options.reliabilityOptions.isEnabled) {
     m_reliability.piggyback(pkt, mtu);
   }

   if (m_options.allowCongestionMarking) {
     checkCongestionLevel(pkt);
   }

   auto block = pkt.wireEncode();
   if (mtu != MTU_UNLIMITED && block.size() > static_cast<size_t>(mtu)) {
     ++this->nOutOverMtu;
     NFD_LOG_FACE_WARN("attempted to send packet over MTU limit");
     return;
   }
   this->sendPacket(block, endpointId);
 }

 void
 GenericLinkService::doSendInterest(const Interest& interest, const EndpointId& endpointId)
 {
   lp::Packet lpPacket(interest.wireEncode());

   encodeLpFields(interest, lpPacket);

   this->sendNetPacket(std::move(lpPacket), endpointId, true);
 }

 void
 GenericLinkService::doSendData(const Data& data, const EndpointId& endpointId)
 {
   lp::Packet lpPacket(data.wireEncode());

   encodeLpFields(data, lpPacket);

   this->sendNetPacket(std::move(lpPacket), endpointId, false);
 }

 void
 GenericLinkService::doSendNack(const lp::Nack& nack, const EndpointId& endpointId)
 {
   lp::Packet lpPacket(nack.getInterest().wireEncode());
   lpPacket.add<lp::NackField>(nack.getHeader());

   encodeLpFields(nack, lpPacket);

   this->sendNetPacket(std::move(lpPacket), endpointId, false);
 }

 void
 GenericLinkService::encodeLpFields(const ndn::PacketBase& netPkt, lp::Packet& lpPacket)
 {
   if (m_options.allowLocalFields) {
     shared_ptr<lp::IncomingFaceIdTag> incomingFaceIdTag = netPkt.getTag<lp::IncomingFaceIdTag>();
     if (incomingFaceIdTag != nullptr) {
       lpPacket.add<lp::IncomingFaceIdField>(*incomingFaceIdTag);
     }
   }

   shared_ptr<lp::CongestionMarkTag> congestionMarkTag = netPkt.getTag<lp::CongestionMarkTag>();
   if (congestionMarkTag != nullptr) {
     lpPacket.add<lp::CongestionMarkField>(*congestionMarkTag);
   }

   if (m_options.allowSelfLearning) {
     shared_ptr<lp::NonDiscoveryTag> nonDiscoveryTag = netPkt.getTag<lp::NonDiscoveryTag>();
     if (nonDiscoveryTag != nullptr) {
       lpPacket.add<lp::NonDiscoveryField>(*nonDiscoveryTag);
     }

     shared_ptr<lp::PrefixAnnouncementTag> prefixAnnouncementTag = netPkt.getTag<lp::PrefixAnnouncementTag>();
     if (prefixAnnouncementTag != nullptr) {
       lpPacket.add<lp::PrefixAnnouncementField>(*prefixAnnouncementTag);
     }
   }
 }

 void
 GenericLinkService::sendNetPacket(lp::Packet&& pkt, const EndpointId& endpointId, bool isInterest)
 {
   std::vector<lp::Packet> frags;
   ssize_t mtu = this->getTransport()->getMtu();

   // Make space for feature fields in fragments
   if (m_options.reliabilityOptions.isEnabled && mtu != MTU_UNLIMITED) {
     mtu -= LpReliability::RESERVED_HEADER_SPACE;
   }

   if (m_options.allowCongestionMarking && mtu != MTU_UNLIMITED) {
     mtu -= CONGESTION_MARK_SIZE;
   }

   BOOST_ASSERT(mtu == MTU_UNLIMITED || mtu > 0);

   if (m_options.allowFragmentation && mtu != MTU_UNLIMITED) {
     bool isOk = false;
     std::tie(isOk, frags) = m_fragmenter.fragmentPacket(pkt, mtu);
     if (!isOk) {
       // fragmentation failed (warning is logged by LpFragmenter)
       ++this->nFragmentationErrors;
       return;
     }
   }
   else {
     if (m_options.reliabilityOptions.isEnabled) {
       frags.push_back(pkt);
     }
     else {
       frags.push_back(std::move(pkt));
     }
   }

   if (frags.size() == 1) {
     // even if indexed fragmentation is enabled, the fragmenter should not
     // fragment the packet if it can fit in MTU
     BOOST_ASSERT(!frags.front().has<lp::FragIndexField>());
     BOOST_ASSERT(!frags.front().has<lp::FragCountField>());
   }

   // Only assign sequences to fragments if packet contains more than 1 fragment
   if (frags.size() > 1) {
     // Assign sequences to all fragments
     this->assignSequences(frags);
   }

   if (m_options.reliabilityOptions.isEnabled && frags.front().has<lp::FragmentField>()) {
     m_reliability.handleOutgoing(frags, std::move(pkt), isInterest);
   }

   for (lp::Packet& frag : frags) {
     this->sendLpPacket(std::move(frag), endpointId);
   }
 }

 void
 GenericLinkService::assignSequence(lp::Packet& pkt)
 {
   pkt.set<lp::SequenceField>(++m_lastSeqNo);
 }

 void
 GenericLinkService::assignSequences(std::vector<lp::Packet>& pkts)
 {
   std::for_each(pkts.begin(), pkts.end(), [this] (auto& pkt) { this->assignSequence(pkt); });
 }

 void
 GenericLinkService::checkCongestionLevel(lp::Packet& pkt)
 {
   ssize_t sendQueueLength = getTransport()->getSendQueueLength();
   // This operation requires that the transport supports retrieving current send queue length
   if (sendQueueLength < 0) {
     return;
   }

   // To avoid overflowing the queue, set the congestion threshold to at least half of the send
   // queue capacity.
   size_t congestionThreshold = m_options.defaultCongestionThreshold;
   if (getTransport()->getSendQueueCapacity() >= 0) {
     congestionThreshold = std::min(congestionThreshold,
                                    static_cast<size_t>(getTransport()->getSendQueueCapacity()) /
                                                        DEFAULT_CONGESTION_THRESHOLD_DIVISOR);
   }

   if (sendQueueLength > 0) {
     NFD_LOG_FACE_TRACE("txqlen=" << sendQueueLength << " threshold=" << congestionThreshold <<
                        " capacity=" << getTransport()->getSendQueueCapacity());
   }

   if (static_cast<size_t>(sendQueueLength) > congestionThreshold) { // Send queue is congested
     const auto now = time::steady_clock::now();
     if (now >= m_nextMarkTime || now >= m_lastMarkTime + m_options.baseCongestionMarkingInterval) {
       // Mark at most one initial packet per baseCongestionMarkingInterval
       if (m_nMarkedSinceInMarkingState == 0) {
         m_nextMarkTime = now;
       }

       // Time to mark packet
       pkt.set<lp::CongestionMarkField>(1);
       ++nCongestionMarked;
       NFD_LOG_FACE_DEBUG("LpPacket was marked as congested");

       ++m_nMarkedSinceInMarkingState;
       // Decrease the marking interval by the inverse of the square root of the number of packets
       // marked in this incident of congestion
       m_nextMarkTime += time::nanoseconds(static_cast<time::nanoseconds::rep>(
                                             m_options.baseCongestionMarkingInterval.count() /
                                             std::sqrt(m_nMarkedSinceInMarkingState)));
       m_lastMarkTime = now;
     }
   }
   else if (m_nextMarkTime != time::steady_clock::TimePoint::max()) {
     // Congestion incident has ended, so reset
     NFD_LOG_FACE_DEBUG("Send queue length dropped below congestion threshold");
     m_nextMarkTime = time::steady_clock::TimePoint::max();
     m_nMarkedSinceInMarkingState = 0;
   }
 }

 void
 GenericLinkService::doReceivePacket(const Block& packet, const EndpointId& endpoint)
 {
   try {
     lp::Packet pkt(packet);

     if (m_options.reliabilityOptions.isEnabled) {
       m_reliability.processIncomingPacket(pkt);
     }

     if (!pkt.has<lp::FragmentField>()) {
       NFD_LOG_FACE_TRACE("received IDLE packet: DROP");
       return;
     }

     if ((pkt.has<lp::FragIndexField>() || pkt.has<lp::FragCountField>()) &&
         !m_options.allowReassembly) {
       NFD_LOG_FACE_WARN("received fragment, but reassembly disabled: DROP");
       return;
     }

     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) {
     ++this->nInLpInvalid;
     NFD_LOG_FACE_WARN("packet parse error (" << e.what() << "): DROP");
   }
 }

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

 void
 GenericLinkService::decodeInterest(const Block& netPkt, const lp::Packet& firstPkt,
                                    const EndpointId& endpointId)
 {
   BOOST_ASSERT(netPkt.type() == tlv::Interest);
   BOOST_ASSERT(!firstPkt.has<lp::NackField>());

   // forwarding expects Interest to be created with make_shared
   auto interest = make_shared<Interest>(netPkt);

   if (firstPkt.has<lp::NextHopFaceIdField>()) {
     if (m_options.allowLocalFields) {
       interest->setTag(make_shared<lp::NextHopFaceIdTag>(firstPkt.get<lp::NextHopFaceIdField>()));
     }
     else {
       NFD_LOG_FACE_WARN("received NextHopFaceId, but local fields disabled: DROP");
       return;
     }
   }

   if (firstPkt.has<lp::CachePolicyField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received CachePolicy with Interest: DROP");
     return;
   }

   if (firstPkt.has<lp::IncomingFaceIdField>()) {
     NFD_LOG_FACE_WARN("received IncomingFaceId: IGNORE");
   }

   if (firstPkt.has<lp::CongestionMarkField>()) {
     interest->setTag(make_shared<lp::CongestionMarkTag>(firstPkt.get<lp::CongestionMarkField>()));
   }

   if (firstPkt.has<lp::NonDiscoveryField>()) {
     if (m_options.allowSelfLearning) {
       interest->setTag(make_shared<lp::NonDiscoveryTag>(firstPkt.get<lp::NonDiscoveryField>()));
     }
     else {
       NFD_LOG_FACE_WARN("received NonDiscovery, but self-learning disabled: IGNORE");
     }
   }

   if (firstPkt.has<lp::PrefixAnnouncementField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received PrefixAnnouncement with Interest: DROP");
     return;
   }

   this->receiveInterest(*interest, endpointId);
 }

 void
 GenericLinkService::decodeData(const Block& netPkt, const lp::Packet& firstPkt,
                                const EndpointId& endpointId)
 {
   BOOST_ASSERT(netPkt.type() == tlv::Data);

   // forwarding expects Data to be created with make_shared
   auto data = make_shared<Data>(netPkt);

   if (firstPkt.has<lp::NackField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received Nack with Data: DROP");
     return;
   }

   if (firstPkt.has<lp::NextHopFaceIdField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received NextHopFaceId with Data: DROP");
     return;
   }

   if (firstPkt.has<lp::CachePolicyField>()) {
     // CachePolicy is unprivileged and does not require allowLocalFields option.
     // In case of an invalid CachePolicyType, get<lp::CachePolicyField> will throw,
     // so it's unnecessary to check here.
     data->setTag(make_shared<lp::CachePolicyTag>(firstPkt.get<lp::CachePolicyField>()));
   }

   if (firstPkt.has<lp::IncomingFaceIdField>()) {
     NFD_LOG_FACE_WARN("received IncomingFaceId: IGNORE");
   }

   if (firstPkt.has<lp::CongestionMarkField>()) {
     data->setTag(make_shared<lp::CongestionMarkTag>(firstPkt.get<lp::CongestionMarkField>()));
   }

   if (firstPkt.has<lp::NonDiscoveryField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received NonDiscovery with Data: DROP");
     return;
   }

   if (firstPkt.has<lp::PrefixAnnouncementField>()) {
     if (m_options.allowSelfLearning) {
       data->setTag(make_shared<lp::PrefixAnnouncementTag>(firstPkt.get<lp::PrefixAnnouncementField>()));
     }
     else {
       NFD_LOG_FACE_WARN("received PrefixAnnouncement, but self-learning disabled: IGNORE");
     }
   }

   this->receiveData(*data, endpointId);
 }

 void
 GenericLinkService::decodeNack(const Block& netPkt, const lp::Packet& firstPkt,
                                const EndpointId& endpointId)
 {
   BOOST_ASSERT(netPkt.type() == tlv::Interest);
   BOOST_ASSERT(firstPkt.has<lp::NackField>());

   lp::Nack nack((Interest(netPkt)));
   nack.setHeader(firstPkt.get<lp::NackField>());

   if (firstPkt.has<lp::NextHopFaceIdField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received NextHopFaceId with Nack: DROP");
     return;
   }

   if (firstPkt.has<lp::CachePolicyField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received CachePolicy with Nack: DROP");
     return;
   }

   if (firstPkt.has<lp::IncomingFaceIdField>()) {
     NFD_LOG_FACE_WARN("received IncomingFaceId: IGNORE");
   }

   if (firstPkt.has<lp::CongestionMarkField>()) {
     nack.setTag(make_shared<lp::CongestionMarkTag>(firstPkt.get<lp::CongestionMarkField>()));
   }

   if (firstPkt.has<lp::NonDiscoveryField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received NonDiscovery with Nack: DROP");
     return;
   }

   if (firstPkt.has<lp::PrefixAnnouncementField>()) {
     ++this->nInNetInvalid;
     NFD_LOG_FACE_WARN("received PrefixAnnouncement with Nack: DROP");
     return;
   }

   this->receiveNack(nack, endpointId);
 }

 } // namespace face
 } // namespace nfd
	/* -- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -- */
	/*
	* Copyright (c) 2014-2019, Regents of the University of California,
	* Arizona Board of Regents,
	* Colorado State University,
	* University Pierre & Marie Curie, Sorbonne University,
	* Washington University in St. Louis,
	* Beijing Institute of Technology,
	* The University of Memphis.
	*
	* This file is part of NFD (Named Data Networking Forwarding Daemon).
	* See AUTHORS.md for complete list of NFD authors and contributors.
	*
	* NFD is free software: you can redistribute it and/or modify it under the terms
	* of the GNU General Public License as published by the Free Software Foundation,
	* either version 3 of the License, or (at your option) any later version.
	*
	* NFD is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
	* without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
	* PURPOSE. See the GNU General Public License for more details.
	*
	* You should have received a copy of the GNU General Public License along with
	* NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
	*/

	#include "generic-link-service.hpp"

	#include <ndn-cxx/lp/tags.hpp>

	#include <cmath>

	namespace nfd {
	namespace face {

	NFD_LOG_INIT(GenericLinkService);

	constexpr size_t CONGESTION_MARK_SIZE = tlv::sizeOfVarNumber(lp::tlv::CongestionMark) + // type
	tlv::sizeOfVarNumber(sizeof(uint64_t)) + // length
	tlv::sizeOfNonNegativeInteger(UINT64_MAX); // value

	constexpr uint32_t DEFAULT_CONGESTION_THRESHOLD_DIVISOR = 2;

	GenericLinkService::GenericLinkService(const GenericLinkService::Options& options)
	: m_options(options)
	, m_fragmenter(m_options.fragmenterOptions, this)
	, m_reassembler(m_options.reassemblerOptions, this)
	, m_reliability(m_options.reliabilityOptions, this)
	, m_lastSeqNo(-2)
	, m_nextMarkTime(time::steady_clock::TimePoint::max())
	, m_lastMarkTime(time::steady_clock::TimePoint::min())
	, m_nMarkedSinceInMarkingState(0)
	{
	m_reassembler.beforeTimeout.connect([this] (auto...) { ++this->nReassemblyTimeouts; });
	m_reliability.onDroppedInterest.connect([this] (const auto& i) { this->notifyDroppedInterest(i); });
	nReassembling.observe(&m_reassembler);
	}

	void
	GenericLinkService::setOptions(const GenericLinkService::Options& options)
	{
	m_options = options;
	m_fragmenter.setOptions(m_options.fragmenterOptions);
	m_reassembler.setOptions(m_options.reassemblerOptions);
	m_reliability.setOptions(m_options.reliabilityOptions);
	}

	void
	GenericLinkService::requestIdlePacket(const EndpointId& endpointId)
	{
	// No need to request Acks to attach to this packet from LpReliability, as they are already
	// attached in sendLpPacket
	this->sendLpPacket({}, endpointId);
	}

	void
	GenericLinkService::sendLpPacket(lp::Packet&& pkt, const EndpointId& endpointId)
	{
	const ssize_t mtu = this->getTransport()->getMtu();

	if (m_options.reliabilityOptions.isEnabled) {
	m_reliability.piggyback(pkt, mtu);
	}

	if (m_options.allowCongestionMarking) {
	checkCongestionLevel(pkt);
	}

	auto block = pkt.wireEncode();
	if (mtu != MTU_UNLIMITED && block.size() > static_cast<size_t>(mtu)) {
	++this->nOutOverMtu;
	NFD_LOG_FACE_WARN("attempted to send packet over MTU limit");
	return;
	}
	this->sendPacket(block, endpointId);
	}

	void
	GenericLinkService::doSendInterest(const Interest& interest, const EndpointId& endpointId)
	{
	lp::Packet lpPacket(interest.wireEncode());

	encodeLpFields(interest, lpPacket);

	this->sendNetPacket(std::move(lpPacket), endpointId, true);
	}

	void
	GenericLinkService::doSendData(const Data& data, const EndpointId& endpointId)
	{
	lp::Packet lpPacket(data.wireEncode());

	encodeLpFields(data, lpPacket);

	this->sendNetPacket(std::move(lpPacket), endpointId, false);
	}

	void
	GenericLinkService::doSendNack(const lp::Nack& nack, const EndpointId& endpointId)
	{
	lp::Packet lpPacket(nack.getInterest().wireEncode());
	lpPacket.add<lp::NackField>(nack.getHeader());

	encodeLpFields(nack, lpPacket);

	this->sendNetPacket(std::move(lpPacket), endpointId, false);
	}

	void
	GenericLinkService::encodeLpFields(const ndn::PacketBase& netPkt, lp::Packet& lpPacket)
	{
	if (m_options.allowLocalFields) {
	shared_ptr<lp::IncomingFaceIdTag> incomingFaceIdTag = netPkt.getTag<lp::IncomingFaceIdTag>();
	if (incomingFaceIdTag != nullptr) {
	lpPacket.add<lp::IncomingFaceIdField>(*incomingFaceIdTag);
	}
	}

	shared_ptr<lp::CongestionMarkTag> congestionMarkTag = netPkt.getTag<lp::CongestionMarkTag>();
	if (congestionMarkTag != nullptr) {
	lpPacket.add<lp::CongestionMarkField>(*congestionMarkTag);
	}

	if (m_options.allowSelfLearning) {
	shared_ptr<lp::NonDiscoveryTag> nonDiscoveryTag = netPkt.getTag<lp::NonDiscoveryTag>();
	if (nonDiscoveryTag != nullptr) {
	lpPacket.add<lp::NonDiscoveryField>(*nonDiscoveryTag);
	}

	shared_ptr<lp::PrefixAnnouncementTag> prefixAnnouncementTag = netPkt.getTag<lp::PrefixAnnouncementTag>();
	if (prefixAnnouncementTag != nullptr) {
	lpPacket.add<lp::PrefixAnnouncementField>(*prefixAnnouncementTag);
	}
	}
	}

	void
	GenericLinkService::sendNetPacket(lp::Packet&& pkt, const EndpointId& endpointId, bool isInterest)
	{
	std::vector<lp::Packet> frags;
	ssize_t mtu = this->getTransport()->getMtu();

	// Make space for feature fields in fragments
	if (m_options.reliabilityOptions.isEnabled && mtu != MTU_UNLIMITED) {
	mtu -= LpReliability::RESERVED_HEADER_SPACE;
	}

	if (m_options.allowCongestionMarking && mtu != MTU_UNLIMITED) {
	mtu -= CONGESTION_MARK_SIZE;
	}

	BOOST_ASSERT(mtu == MTU_UNLIMITED \|\| mtu > 0);

	if (m_options.allowFragmentation && mtu != MTU_UNLIMITED) {
	bool isOk = false;
	std::tie(isOk, frags) = m_fragmenter.fragmentPacket(pkt, mtu);
	if (!isOk) {
	// fragmentation failed (warning is logged by LpFragmenter)
	++this->nFragmentationErrors;
	return;
	}
	}
	else {
	if (m_options.reliabilityOptions.isEnabled) {
	frags.push_back(pkt);
	}
	else {
	frags.push_back(std::move(pkt));
	}
	}

	if (frags.size() == 1) {
	// even if indexed fragmentation is enabled, the fragmenter should not
	// fragment the packet if it can fit in MTU
	BOOST_ASSERT(!frags.front().has<lp::FragIndexField>());
	BOOST_ASSERT(!frags.front().has<lp::FragCountField>());
	}

	// Only assign sequences to fragments if packet contains more than 1 fragment
	if (frags.size() > 1) {
	// Assign sequences to all fragments
	this->assignSequences(frags);
	}

	if (m_options.reliabilityOptions.isEnabled && frags.front().has<lp::FragmentField>()) {
	m_reliability.handleOutgoing(frags, std::move(pkt), isInterest);
	}

	for (lp::Packet& frag : frags) {
	this->sendLpPacket(std::move(frag), endpointId);
	}
	}

	void
	GenericLinkService::assignSequence(lp::Packet& pkt)
	{
	pkt.set<lp::SequenceField>(++m_lastSeqNo);
	}

	void
	GenericLinkService::assignSequences(std::vector<lp::Packet>& pkts)
	{
	std::for_each(pkts.begin(), pkts.end(), [this] (auto& pkt) { this->assignSequence(pkt); });
	}

	void
	GenericLinkService::checkCongestionLevel(lp::Packet& pkt)
	{
	ssize_t sendQueueLength = getTransport()->getSendQueueLength();
	// This operation requires that the transport supports retrieving current send queue length
	if (sendQueueLength < 0) {
	return;
	}

	// To avoid overflowing the queue, set the congestion threshold to at least half of the send
	// queue capacity.
	size_t congestionThreshold = m_options.defaultCongestionThreshold;
	if (getTransport()->getSendQueueCapacity() >= 0) {
	congestionThreshold = std::min(congestionThreshold,
	static_cast<size_t>(getTransport()->getSendQueueCapacity()) /
	DEFAULT_CONGESTION_THRESHOLD_DIVISOR);
	}

	if (sendQueueLength > 0) {
	NFD_LOG_FACE_TRACE("txqlen=" << sendQueueLength << " threshold=" << congestionThreshold <<
	" capacity=" << getTransport()->getSendQueueCapacity());
	}

	if (static_cast<size_t>(sendQueueLength) > congestionThreshold) { // Send queue is congested
	const auto now = time::steady_clock::now();
	if (now >= m_nextMarkTime \|\| now >= m_lastMarkTime + m_options.baseCongestionMarkingInterval) {
	// Mark at most one initial packet per baseCongestionMarkingInterval
	if (m_nMarkedSinceInMarkingState == 0) {
	m_nextMarkTime = now;
	}

	// Time to mark packet
	pkt.set<lp::CongestionMarkField>(1);
	++nCongestionMarked;
	NFD_LOG_FACE_DEBUG("LpPacket was marked as congested");

	++m_nMarkedSinceInMarkingState;
	// Decrease the marking interval by the inverse of the square root of the number of packets
	// marked in this incident of congestion
	m_nextMarkTime += time::nanoseconds(static_cast<time::nanoseconds::rep>(
	m_options.baseCongestionMarkingInterval.count() /
	std::sqrt(m_nMarkedSinceInMarkingState)));
	m_lastMarkTime = now;
	}
	}
	else if (m_nextMarkTime != time::steady_clock::TimePoint::max()) {
	// Congestion incident has ended, so reset
	NFD_LOG_FACE_DEBUG("Send queue length dropped below congestion threshold");
	m_nextMarkTime = time::steady_clock::TimePoint::max();
	m_nMarkedSinceInMarkingState = 0;
	}
	}

	void
	GenericLinkService::doReceivePacket(const Block& packet, const EndpointId& endpoint)
	{
	try {
	lp::Packet pkt(packet);

	if (m_options.reliabilityOptions.isEnabled) {
	m_reliability.processIncomingPacket(pkt);
	}

	if (!pkt.has<lp::FragmentField>()) {
	NFD_LOG_FACE_TRACE("received IDLE packet: DROP");
	return;
	}

	if ((pkt.has<lp::FragIndexField>() \|\| pkt.has<lp::FragCountField>()) &&
	!m_options.allowReassembly) {
	NFD_LOG_FACE_WARN("received fragment, but reassembly disabled: DROP");
	return;
	}

	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) {
	++this->nInLpInvalid;
	NFD_LOG_FACE_WARN("packet parse error (" << e.what() << "): DROP");
	}
	}

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

	void
	GenericLinkService::decodeInterest(const Block& netPkt, const lp::Packet& firstPkt,
	const EndpointId& endpointId)
	{
	BOOST_ASSERT(netPkt.type() == tlv::Interest);
	BOOST_ASSERT(!firstPkt.has<lp::NackField>());

	// forwarding expects Interest to be created with make_shared
	auto interest = make_shared<Interest>(netPkt);

	if (firstPkt.has<lp::NextHopFaceIdField>()) {
	if (m_options.allowLocalFields) {
	interest->setTag(make_shared<lp::NextHopFaceIdTag>(firstPkt.get<lp::NextHopFaceIdField>()));
	}
	else {
	NFD_LOG_FACE_WARN("received NextHopFaceId, but local fields disabled: DROP");
	return;
	}
	}

	if (firstPkt.has<lp::CachePolicyField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received CachePolicy with Interest: DROP");
	return;
	}

	if (firstPkt.has<lp::IncomingFaceIdField>()) {
	NFD_LOG_FACE_WARN("received IncomingFaceId: IGNORE");
	}

	if (firstPkt.has<lp::CongestionMarkField>()) {
	interest->setTag(make_shared<lp::CongestionMarkTag>(firstPkt.get<lp::CongestionMarkField>()));
	}

	if (firstPkt.has<lp::NonDiscoveryField>()) {
	if (m_options.allowSelfLearning) {
	interest->setTag(make_shared<lp::NonDiscoveryTag>(firstPkt.get<lp::NonDiscoveryField>()));
	}
	else {
	NFD_LOG_FACE_WARN("received NonDiscovery, but self-learning disabled: IGNORE");
	}
	}

	if (firstPkt.has<lp::PrefixAnnouncementField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received PrefixAnnouncement with Interest: DROP");
	return;
	}

	this->receiveInterest(*interest, endpointId);
	}

	void
	GenericLinkService::decodeData(const Block& netPkt, const lp::Packet& firstPkt,
	const EndpointId& endpointId)
	{
	BOOST_ASSERT(netPkt.type() == tlv::Data);

	// forwarding expects Data to be created with make_shared
	auto data = make_shared<Data>(netPkt);

	if (firstPkt.has<lp::NackField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received Nack with Data: DROP");
	return;
	}

	if (firstPkt.has<lp::NextHopFaceIdField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received NextHopFaceId with Data: DROP");
	return;
	}

	if (firstPkt.has<lp::CachePolicyField>()) {
	// CachePolicy is unprivileged and does not require allowLocalFields option.
	// In case of an invalid CachePolicyType, get<lp::CachePolicyField> will throw,
	// so it's unnecessary to check here.
	data->setTag(make_shared<lp::CachePolicyTag>(firstPkt.get<lp::CachePolicyField>()));
	}

	if (firstPkt.has<lp::IncomingFaceIdField>()) {
	NFD_LOG_FACE_WARN("received IncomingFaceId: IGNORE");
	}

	if (firstPkt.has<lp::CongestionMarkField>()) {
	data->setTag(make_shared<lp::CongestionMarkTag>(firstPkt.get<lp::CongestionMarkField>()));
	}

	if (firstPkt.has<lp::NonDiscoveryField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received NonDiscovery with Data: DROP");
	return;
	}

	if (firstPkt.has<lp::PrefixAnnouncementField>()) {
	if (m_options.allowSelfLearning) {
	data->setTag(make_shared<lp::PrefixAnnouncementTag>(firstPkt.get<lp::PrefixAnnouncementField>()));
	}
	else {
	NFD_LOG_FACE_WARN("received PrefixAnnouncement, but self-learning disabled: IGNORE");
	}
	}

	this->receiveData(*data, endpointId);
	}

	void
	GenericLinkService::decodeNack(const Block& netPkt, const lp::Packet& firstPkt,
	const EndpointId& endpointId)
	{
	BOOST_ASSERT(netPkt.type() == tlv::Interest);
	BOOST_ASSERT(firstPkt.has<lp::NackField>());

	lp::Nack nack((Interest(netPkt)));
	nack.setHeader(firstPkt.get<lp::NackField>());

	if (firstPkt.has<lp::NextHopFaceIdField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received NextHopFaceId with Nack: DROP");
	return;
	}

	if (firstPkt.has<lp::CachePolicyField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received CachePolicy with Nack: DROP");
	return;
	}

	if (firstPkt.has<lp::IncomingFaceIdField>()) {
	NFD_LOG_FACE_WARN("received IncomingFaceId: IGNORE");
	}

	if (firstPkt.has<lp::CongestionMarkField>()) {
	nack.setTag(make_shared<lp::CongestionMarkTag>(firstPkt.get<lp::CongestionMarkField>()));
	}

	if (firstPkt.has<lp::NonDiscoveryField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received NonDiscovery with Nack: DROP");
	return;
	}

	if (firstPkt.has<lp::PrefixAnnouncementField>()) {
	++this->nInNetInvalid;
	NFD_LOG_FACE_WARN("received PrefixAnnouncement with Nack: DROP");
	return;
	}

	this->receiveNack(nack, endpointId);
	}

	} // namespace face
	} // namespace nfd