NS-3 based Named Data Networking (NDN) simulator
ndnSIM 2.5: NDN, CCN, CCNx, content centric networks
API Documentation
lp-reliability.cpp
Go to the documentation of this file.
1 /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
2 /*
3  * Copyright (c) 2014-2017, Regents of the University of California,
4  * Arizona Board of Regents,
5  * Colorado State University,
6  * University Pierre & Marie Curie, Sorbonne University,
7  * Washington University in St. Louis,
8  * Beijing Institute of Technology,
9  * The University of Memphis.
10  *
11  * This file is part of NFD (Named Data Networking Forwarding Daemon).
12  * See AUTHORS.md for complete list of NFD authors and contributors.
13  *
14  * NFD is free software: you can redistribute it and/or modify it under the terms
15  * of the GNU General Public License as published by the Free Software Foundation,
16  * either version 3 of the License, or (at your option) any later version.
17  *
18  * NFD is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
19  * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
20  * PURPOSE. See the GNU General Public License for more details.
21  *
22  * You should have received a copy of the GNU General Public License along with
23  * NFD, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
24  */
25 
26 #include "lp-reliability.hpp"
27 #include "generic-link-service.hpp"
28 #include "transport.hpp"
29 
30 namespace nfd {
31 namespace face {
32 
34  : m_options(options)
35  , m_linkService(linkService)
36  , m_firstUnackedFrag(m_unackedFrags.begin())
37  , m_lastTxSeqNo(-1) // set to "-1" to start TxSequence numbers at 0
38  , m_isIdleAckTimerRunning(false)
39 {
40  BOOST_ASSERT(m_linkService != nullptr);
41 
42  BOOST_ASSERT(m_options.idleAckTimerPeriod > time::nanoseconds::zero());
43 }
44 
45 void
47 {
48  BOOST_ASSERT(options.idleAckTimerPeriod > time::nanoseconds::zero());
49 
50  if (m_options.isEnabled && !options.isEnabled) {
51  this->stopIdleAckTimer();
52  }
53 
54  m_options = options;
55 }
56 
57 const GenericLinkService*
59 {
60  return m_linkService;
61 }
62 
63 void
64 LpReliability::handleOutgoing(std::vector<lp::Packet>& frags, lp::Packet&& pkt, bool isInterest)
65 {
66  BOOST_ASSERT(m_options.isEnabled);
67 
68  auto unackedFragsIt = m_unackedFrags.begin();
69  auto sendTime = time::steady_clock::now();
70 
71  auto netPkt = make_shared<NetPkt>(std::move(pkt), isInterest);
72  netPkt->unackedFrags.reserve(frags.size());
73 
74  for (lp::Packet& frag : frags) {
75  // Assign TxSequence number
76  lp::Sequence txSeq = assignTxSequence(frag);
77 
78  // Store LpPacket for future retransmissions
79  unackedFragsIt = m_unackedFrags.emplace_hint(unackedFragsIt,
80  std::piecewise_construct,
81  std::forward_as_tuple(txSeq),
82  std::forward_as_tuple(frag));
83  unackedFragsIt->second.sendTime = sendTime;
84  unackedFragsIt->second.rtoTimer =
85  scheduler::schedule(m_rto.computeRto(), bind(&LpReliability::onLpPacketLost, this, unackedFragsIt));
86  unackedFragsIt->second.netPkt = netPkt;
87 
88  if (m_unackedFrags.size() == 1) {
89  m_firstUnackedFrag = m_unackedFrags.begin();
90  }
91 
92  // Add to associated NetPkt
93  netPkt->unackedFrags.push_back(unackedFragsIt);
94  }
95 }
96 
97 void
99 {
100  BOOST_ASSERT(m_options.isEnabled);
101 
102  auto now = time::steady_clock::now();
103 
104  // Extract and parse Acks
105  for (lp::Sequence ackSeq : pkt.list<lp::AckField>()) {
106  auto fragIt = m_unackedFrags.find(ackSeq);
107  if (fragIt == m_unackedFrags.end()) {
108  // Ignore an Ack for an unknown TxSequence number
109  continue;
110  }
111  auto& frag = fragIt->second;
112 
113  // Cancel the RTO timer for the acknowledged fragment
114  frag.rtoTimer.cancel();
115 
116  if (frag.retxCount == 0) {
117  // This sequence had no retransmissions, so use it to calculate the RTO
118  m_rto.addMeasurement(time::duration_cast<RttEstimator::Duration>(now - frag.sendTime));
119  }
120 
121  // Look for frags with TxSequence numbers < ackSeq (allowing for wraparound) and consider them
122  // lost if a configurable number of Acks containing greater TxSequence numbers have been
123  // received.
124  auto lostLpPackets = findLostLpPackets(fragIt);
125 
126  // Remove the fragment from the map of unacknowledged fragments and from its associated network
127  // packet. Potentially increment the start of the window.
128  onLpPacketAcknowledged(fragIt);
129 
130  // Resend or fail fragments considered lost. Potentially increment the start of the window.
131  for (UnackedFrags::iterator txSeqIt : lostLpPackets) {
132  this->onLpPacketLost(txSeqIt);
133  }
134  }
135 
136  // If packet has Fragment and TxSequence fields, extract TxSequence and add to AckQueue
137  if (pkt.has<lp::FragmentField>() && pkt.has<lp::TxSequenceField>()) {
138  m_ackQueue.push(pkt.get<lp::TxSequenceField>());
139  if (!m_isIdleAckTimerRunning) {
140  this->startIdleAckTimer();
141  }
142  }
143 }
144 
145 void
147 {
148  BOOST_ASSERT(m_options.isEnabled);
149  BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
150 
151  // up to 2 extra octets reserved for potential TLV-LENGTH size increases
152  ssize_t pktSize = pkt.wireEncode().size();
153  ssize_t reservedSpace = tlv::sizeOfVarNumber(ndn::MAX_NDN_PACKET_SIZE) -
154  tlv::sizeOfVarNumber(pktSize);
155  ssize_t remainingSpace = (mtu == MTU_UNLIMITED ? ndn::MAX_NDN_PACKET_SIZE : mtu) - reservedSpace;
156  remainingSpace -= pktSize;
157 
158  while (!m_ackQueue.empty()) {
159  lp::Sequence ackSeq = m_ackQueue.front();
160  // Ack Size = Ack Type (3 octets) + Ack Length (1 octet) + Value (1, 2, 4, or 8 octets)
161  ssize_t ackSize = tlv::sizeOfVarNumber(lp::tlv::Ack) +
163  tlv::sizeOfNonNegativeInteger(std::numeric_limits<lp::Sequence>::max())) +
165 
166  if (ackSize > remainingSpace) {
167  break;
168  }
169 
170  pkt.add<lp::AckField>(ackSeq);
171  m_ackQueue.pop();
172  remainingSpace -= ackSize;
173  }
174 }
175 
177 LpReliability::assignTxSequence(lp::Packet& frag)
178 {
179  lp::Sequence txSeq = ++m_lastTxSeqNo;
180  frag.set<lp::TxSequenceField>(txSeq);
181  if (m_unackedFrags.size() > 0 && m_lastTxSeqNo == m_firstUnackedFrag->first) {
182  BOOST_THROW_EXCEPTION(std::length_error("TxSequence range exceeded"));
183  }
184  return m_lastTxSeqNo;
185 }
186 
187 void
188 LpReliability::startIdleAckTimer()
189 {
190  BOOST_ASSERT(!m_isIdleAckTimerRunning);
191  m_isIdleAckTimerRunning = true;
192 
193  m_idleAckTimer = scheduler::schedule(m_options.idleAckTimerPeriod, [this] {
194  while (!m_ackQueue.empty()) {
195  m_linkService->requestIdlePacket();
196  }
197 
198  m_isIdleAckTimerRunning = false;
199  });
200 }
201 
202 void
203 LpReliability::stopIdleAckTimer()
204 {
205  m_idleAckTimer.cancel();
206  m_isIdleAckTimerRunning = false;
207 }
208 
209 std::vector<LpReliability::UnackedFrags::iterator>
210 LpReliability::findLostLpPackets(LpReliability::UnackedFrags::iterator ackIt)
211 {
212  std::vector<UnackedFrags::iterator> lostLpPackets;
213 
214  for (auto it = m_firstUnackedFrag; ; ++it) {
215  if (it == m_unackedFrags.end()) {
216  it = m_unackedFrags.begin();
217  }
218 
219  if (it->first == ackIt->first) {
220  break;
221  }
222 
223  auto& unackedFrag = it->second;
224  unackedFrag.nGreaterSeqAcks++;
225 
226  if (unackedFrag.nGreaterSeqAcks >= m_options.seqNumLossThreshold) {
227  lostLpPackets.push_back(it);
228  }
229  }
230 
231  return lostLpPackets;
232 }
233 
234 void
235 LpReliability::onLpPacketLost(UnackedFrags::iterator txSeqIt)
236 {
237  BOOST_ASSERT(m_unackedFrags.count(txSeqIt->first) > 0);
238 
239  auto& txFrag = txSeqIt->second;
240  txFrag.rtoTimer.cancel();
241  auto netPkt = txFrag.netPkt;
242 
243  // Check if maximum number of retransmissions exceeded
244  if (txFrag.retxCount >= m_options.maxRetx) {
245  // Delete all LpPackets of NetPkt from m_unackedFrags (except this one)
246  for (size_t i = 0; i < netPkt->unackedFrags.size(); i++) {
247  if (netPkt->unackedFrags[i] != txSeqIt) {
248  deleteUnackedFrag(netPkt->unackedFrags[i]);
249  }
250  }
251 
252  ++m_linkService->nRetxExhausted;
253 
254  // Notify strategy of dropped Interest (if any)
255  if (netPkt->isInterest) {
256  BOOST_ASSERT(netPkt->pkt.has<lp::FragmentField>());
257  ndn::Buffer::const_iterator fragBegin, fragEnd;
258  std::tie(fragBegin, fragEnd) = netPkt->pkt.get<lp::FragmentField>();
259  Block frag(&*fragBegin, std::distance(fragBegin, fragEnd));
260  onDroppedInterest(Interest(frag));
261  }
262 
263  deleteUnackedFrag(txSeqIt);
264  }
265  else {
266  // Assign new TxSequence
267  lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
268  netPkt->didRetx = true;
269 
270  // Move fragment to new TxSequence mapping
271  auto newTxFragIt = m_unackedFrags.emplace_hint(
272  m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
273  ? m_firstUnackedFrag
274  : m_unackedFrags.end(),
275  std::piecewise_construct,
276  std::forward_as_tuple(newTxSeq),
277  std::forward_as_tuple(txFrag.pkt));
278  auto& newTxFrag = newTxFragIt->second;
279  newTxFrag.retxCount = txFrag.retxCount + 1;
280  newTxFrag.netPkt = netPkt;
281 
282  // Update associated NetPkt
283  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
284  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
285  *fragInNetPkt = newTxFragIt;
286 
287  deleteUnackedFrag(txSeqIt);
288 
289  // Retransmit fragment
290  m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt));
291 
292  // Start RTO timer for this sequence
293  newTxFrag.rtoTimer = scheduler::schedule(m_rto.computeRto(),
294  bind(&LpReliability::onLpPacketLost, this, newTxFragIt));
295  }
296 }
297 
298 void
299 LpReliability::onLpPacketAcknowledged(UnackedFrags::iterator fragIt)
300 {
301  auto netPkt = fragIt->second.netPkt;
302 
303  // Remove from NetPkt unacked fragment list
304  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), fragIt);
305  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
306  *fragInNetPkt = netPkt->unackedFrags.back();
307  netPkt->unackedFrags.pop_back();
308 
309  // Check if network-layer packet completely received. If so, increment counters
310  if (netPkt->unackedFrags.empty()) {
311  if (netPkt->didRetx) {
312  ++m_linkService->nRetransmitted;
313  }
314  else {
315  ++m_linkService->nAcknowledged;
316  }
317  }
318 
319  deleteUnackedFrag(fragIt);
320 }
321 
322 void
323 LpReliability::deleteUnackedFrag(UnackedFrags::iterator fragIt)
324 {
325  lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
326  lp::Sequence currentTxSeq = fragIt->first;
327  auto nextFragIt = m_unackedFrags.erase(fragIt);
328 
329  if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
330  // If "first" fragment in send window (allowing for wraparound), increment window begin
331  if (nextFragIt == m_unackedFrags.end()) {
332  m_firstUnackedFrag = m_unackedFrags.begin();
333  }
334  else {
335  m_firstUnackedFrag = nextFragIt;
336  }
337  }
338  else if (m_unackedFrags.empty()) {
339  m_firstUnackedFrag = m_unackedFrags.end();
340  }
341 }
342 
343 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
344  : pkt(std::move(pkt))
345  , sendTime(time::steady_clock::now())
346  , retxCount(0)
347  , nGreaterSeqAcks(0)
348 {
349 }
350 
351 LpReliability::NetPkt::NetPkt(lp::Packet&& pkt, bool isInterest)
352  : pkt(std::move(pkt))
353  , isInterest(isInterest)
354  , didRetx(false)
355 {
356 }
357 
358 } // namespace face
359 } // namespace nfd
Duration computeRto() const
std::vector< typename FIELD::ValueType > list() const
Definition: packet.hpp:121
void setOptions(const Options &options)
set options for reliability
GenericLinkService is a LinkService that implements the NDNLPv2 protocol.
void processIncomingPacket(const lp::Packet &pkt)
extract and parse all Acks and add Ack for contained Fragment (if any) to AckQueue ...
Packet & set(const typename FIELD::ValueType &value)
remove all occurrences of FIELD, and add a FIELD with value
Definition: packet.hpp:141
const ssize_t MTU_UNLIMITED
indicates the transport has no limit on payload size
Definition: transport.hpp:96
static time_point now() noexcept
Definition: time.cpp:80
void piggyback(lp::Packet &pkt, ssize_t mtu)
called by GenericLinkService to attach Acks onto an outgoing LpPacket
Packet & add(const typename FIELD::ValueType &value)
add a FIELD with value
Definition: packet.hpp:153
void handleOutgoing(std::vector< lp::Packet > &frags, lp::Packet &&pkt, bool isInterest)
observe outgoing fragment(s) of a network packet and store for potential retransmission ...
STL namespace.
bool has() const
Definition: packet.hpp:78
bool isEnabled
enables link-layer reliability
uint64_t Sequence
represents a sequence number
Definition: sequence.hpp:35
Table::const_iterator iterator
Definition: cs-internal.hpp:41
void addMeasurement(Duration measure)
size_t size() const
Get size of encoded wire, including Type-Length-Value.
Definition: block.cpp:300
FIELD::ValueType get(size_t index=0) const
Definition: packet.hpp:101
Copyright (c) 2011-2015 Regents of the University of California.
Definition: ndn-common.hpp:40
constexpr size_t sizeOfNonNegativeInteger(uint64_t integer)
Get number of bytes necessary to hold value of nonNegativeInteger.
Definition: tlv.hpp:467
const GenericLinkService * getLinkService() const
LpReliability(const Options &options, GenericLinkService *linkService)
EventId schedule(time::nanoseconds after, const EventCallback &event)
schedule an event
Definition: scheduler.cpp:47
Block wireEncode() const
encode packet into wire format
Definition: packet.cpp:131
time::nanoseconds idleAckTimerPeriod
period between sending pending Acks in an IDLE packet
constexpr size_t sizeOfVarNumber(uint64_t number)
Get number of bytes necessary to hold value of VAR-NUMBER.
Definition: tlv.hpp:414
uint32_t type() const
Get TLV-TYPE.
Definition: block.hpp:235
const size_t MAX_NDN_PACKET_SIZE
practical limit of network layer packet size
Definition: tlv.hpp:39