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-2019, 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 #include "common/global.hpp"
30 
31 namespace nfd {
32 namespace face {
33 
35  : m_options(options)
36  , m_linkService(linkService)
37  , m_firstUnackedFrag(m_unackedFrags.begin())
38  , m_lastTxSeqNo(-1) // set to "-1" to start TxSequence numbers at 0
39 {
40  BOOST_ASSERT(m_linkService != nullptr);
41  BOOST_ASSERT(m_options.idleAckTimerPeriod > 0_ns);
42 }
43 
44 void
46 {
47  BOOST_ASSERT(options.idleAckTimerPeriod > 0_ns);
48 
49  if (m_options.isEnabled && !options.isEnabled) {
50  m_idleAckTimer.cancel();
51  }
52 
53  m_options = options;
54 }
55 
56 const GenericLinkService*
58 {
59  return m_linkService;
60 }
61 
62 void
63 LpReliability::handleOutgoing(std::vector<lp::Packet>& frags, lp::Packet&& pkt, bool isInterest)
64 {
65  BOOST_ASSERT(m_options.isEnabled);
66 
67  auto unackedFragsIt = m_unackedFrags.begin();
68  auto sendTime = time::steady_clock::now();
69 
70  auto netPkt = make_shared<NetPkt>(std::move(pkt), isInterest);
71  netPkt->unackedFrags.reserve(frags.size());
72 
73  for (lp::Packet& frag : frags) {
74  // Assign TxSequence number
75  lp::Sequence txSeq = assignTxSequence(frag);
76 
77  // Store LpPacket for future retransmissions
78  unackedFragsIt = m_unackedFrags.emplace_hint(unackedFragsIt,
79  std::piecewise_construct,
80  std::forward_as_tuple(txSeq),
81  std::forward_as_tuple(frag));
82  unackedFragsIt->second.sendTime = sendTime;
83  unackedFragsIt->second.rtoTimer = getScheduler().schedule(m_rttEst.getEstimatedRto(),
84  [=] { onLpPacketLost(txSeq); });
85  unackedFragsIt->second.netPkt = netPkt;
86 
87  if (m_unackedFrags.size() == 1) {
88  m_firstUnackedFrag = m_unackedFrags.begin();
89  }
90 
91  // Add to associated NetPkt
92  netPkt->unackedFrags.push_back(unackedFragsIt);
93  }
94 }
95 
96 void
98 {
99  BOOST_ASSERT(m_options.isEnabled);
100 
101  auto now = time::steady_clock::now();
102 
103  // Extract and parse Acks
104  for (lp::Sequence ackSeq : pkt.list<lp::AckField>()) {
105  auto fragIt = m_unackedFrags.find(ackSeq);
106  if (fragIt == m_unackedFrags.end()) {
107  // Ignore an Ack for an unknown TxSequence number
108  continue;
109  }
110  auto& frag = fragIt->second;
111 
112  // Cancel the RTO timer for the acknowledged fragment
113  frag.rtoTimer.cancel();
114 
115  if (frag.retxCount == 0) {
116  // This sequence had no retransmissions, so use it to estimate the RTO
117  m_rttEst.addMeasurement(now - frag.sendTime);
118  }
119 
120  // Look for frags with TxSequence numbers < ackSeq (allowing for wraparound) and consider them
121  // lost if a configurable number of Acks containing greater TxSequence numbers have been
122  // received.
123  auto lostLpPackets = findLostLpPackets(fragIt);
124 
125  // Remove the fragment from the map of unacknowledged fragments and from its associated network
126  // packet. Potentially increment the start of the window.
127  onLpPacketAcknowledged(fragIt);
128 
129  // This set contains TxSequences that have been removed by onLpPacketLost below because they
130  // were part of a network packet that was removed due to a fragment exceeding retx, as well as
131  // any other TxSequences removed by onLpPacketLost. This prevents onLpPacketLost from being
132  // called later for an invalid iterator.
133  std::set<lp::Sequence> removedLpPackets;
134 
135  // Resend or fail fragments considered lost. Potentially increment the start of the window.
136  for (lp::Sequence txSeq : lostLpPackets) {
137  if (removedLpPackets.find(txSeq) == removedLpPackets.end()) {
138  auto removedThisTxSeq = onLpPacketLost(txSeq);
139  for (auto removedTxSeq : removedThisTxSeq) {
140  removedLpPackets.insert(removedTxSeq);
141  }
142  }
143  }
144  }
145 
146  // If packet has Fragment and TxSequence fields, extract TxSequence and add to AckQueue
147  if (pkt.has<lp::FragmentField>() && pkt.has<lp::TxSequenceField>()) {
148  m_ackQueue.push(pkt.get<lp::TxSequenceField>());
149  startIdleAckTimer();
150  }
151 }
152 
153 void
154 LpReliability::piggyback(lp::Packet& pkt, ssize_t mtu)
155 {
156  BOOST_ASSERT(m_options.isEnabled);
157  BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
158 
159  // up to 2 extra octets reserved for potential TLV-LENGTH size increases
160  ssize_t pktSize = pkt.wireEncode().size();
161  ssize_t reservedSpace = tlv::sizeOfVarNumber(ndn::MAX_NDN_PACKET_SIZE) -
162  tlv::sizeOfVarNumber(pktSize);
163  ssize_t remainingSpace = (mtu == MTU_UNLIMITED ? ndn::MAX_NDN_PACKET_SIZE : mtu) - reservedSpace;
164  remainingSpace -= pktSize;
165 
166  while (!m_ackQueue.empty()) {
167  lp::Sequence ackSeq = m_ackQueue.front();
168  // Ack size = Ack TLV-TYPE (3 octets) + TLV-LENGTH (1 octet) + lp::Sequence (8 octets)
169  const ssize_t ackSize = tlv::sizeOfVarNumber(lp::tlv::Ack) +
170  tlv::sizeOfVarNumber(sizeof(lp::Sequence)) +
171  sizeof(lp::Sequence);
172 
173  if (ackSize > remainingSpace) {
174  break;
175  }
176 
177  pkt.add<lp::AckField>(ackSeq);
178  m_ackQueue.pop();
179  remainingSpace -= ackSize;
180  }
181 }
182 
183 lp::Sequence
184 LpReliability::assignTxSequence(lp::Packet& frag)
185 {
186  lp::Sequence txSeq = ++m_lastTxSeqNo;
187  frag.set<lp::TxSequenceField>(txSeq);
188  if (m_unackedFrags.size() > 0 && m_lastTxSeqNo == m_firstUnackedFrag->first) {
189  NDN_THROW(std::length_error("TxSequence range exceeded"));
190  }
191  return m_lastTxSeqNo;
192 }
193 
194 void
195 LpReliability::startIdleAckTimer()
196 {
197  if (m_idleAckTimer) {
198  // timer is already running, do nothing
199  return;
200  }
201 
202  m_idleAckTimer = getScheduler().schedule(m_options.idleAckTimerPeriod, [this] {
203  while (!m_ackQueue.empty()) {
204  m_linkService->requestIdlePacket(0);
205  }
206  });
207 }
208 
209 std::vector<lp::Sequence>
210 LpReliability::findLostLpPackets(LpReliability::UnackedFrags::iterator ackIt)
211 {
212  std::vector<lp::Sequence> 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->first);
228  }
229  }
230 
231  return lostLpPackets;
232 }
233 
234 std::vector<lp::Sequence>
235 LpReliability::onLpPacketLost(lp::Sequence txSeq)
236 {
237  BOOST_ASSERT(m_unackedFrags.count(txSeq) > 0);
238  auto txSeqIt = m_unackedFrags.find(txSeq);
239 
240  auto& txFrag = txSeqIt->second;
241  txFrag.rtoTimer.cancel();
242  auto netPkt = txFrag.netPkt;
243  std::vector<lp::Sequence> removedThisTxSeq;
244 
245  // Check if maximum number of retransmissions exceeded
246  if (txFrag.retxCount >= m_options.maxRetx) {
247  // Delete all LpPackets of NetPkt from m_unackedFrags (except this one)
248  for (size_t i = 0; i < netPkt->unackedFrags.size(); i++) {
249  if (netPkt->unackedFrags[i] != txSeqIt) {
250  removedThisTxSeq.push_back(netPkt->unackedFrags[i]->first);
251  deleteUnackedFrag(netPkt->unackedFrags[i]);
252  }
253  }
254 
255  ++m_linkService->nRetxExhausted;
256 
257  // Notify strategy of dropped Interest (if any)
258  if (netPkt->isInterest) {
259  BOOST_ASSERT(netPkt->pkt.has<lp::FragmentField>());
260  ndn::Buffer::const_iterator fragBegin, fragEnd;
261  std::tie(fragBegin, fragEnd) = netPkt->pkt.get<lp::FragmentField>();
262  Block frag(&*fragBegin, std::distance(fragBegin, fragEnd));
263  onDroppedInterest(Interest(frag));
264  }
265 
266  removedThisTxSeq.push_back(txSeqIt->first);
267  deleteUnackedFrag(txSeqIt);
268  }
269  else {
270  // Assign new TxSequence
271  lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
272  netPkt->didRetx = true;
273 
274  // Move fragment to new TxSequence mapping
275  auto newTxFragIt = m_unackedFrags.emplace_hint(
276  m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
277  ? m_firstUnackedFrag
278  : m_unackedFrags.end(),
279  std::piecewise_construct,
280  std::forward_as_tuple(newTxSeq),
281  std::forward_as_tuple(txFrag.pkt));
282  auto& newTxFrag = newTxFragIt->second;
283  newTxFrag.retxCount = txFrag.retxCount + 1;
284  newTxFrag.netPkt = netPkt;
285 
286  // Update associated NetPkt
287  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
288  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
289  *fragInNetPkt = newTxFragIt;
290 
291  removedThisTxSeq.push_back(txSeqIt->first);
292  deleteUnackedFrag(txSeqIt);
293 
294  // Retransmit fragment
295  m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt), 0);
296 
297  // Start RTO timer for this sequence
298  newTxFrag.rtoTimer = getScheduler().schedule(m_rttEst.getEstimatedRto(),
299  [=] { onLpPacketLost(newTxSeq); });
300  }
301 
302  return removedThisTxSeq;
303 }
304 
305 void
306 LpReliability::onLpPacketAcknowledged(UnackedFrags::iterator fragIt)
307 {
308  auto netPkt = fragIt->second.netPkt;
309 
310  // Remove from NetPkt unacked fragment list
311  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), fragIt);
312  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
313  *fragInNetPkt = netPkt->unackedFrags.back();
314  netPkt->unackedFrags.pop_back();
315 
316  // Check if network-layer packet completely received. If so, increment counters
317  if (netPkt->unackedFrags.empty()) {
318  if (netPkt->didRetx) {
319  ++m_linkService->nRetransmitted;
320  }
321  else {
322  ++m_linkService->nAcknowledged;
323  }
324  }
325 
326  deleteUnackedFrag(fragIt);
327 }
328 
329 void
330 LpReliability::deleteUnackedFrag(UnackedFrags::iterator fragIt)
331 {
332  lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
333  lp::Sequence currentTxSeq = fragIt->first;
334  auto nextFragIt = m_unackedFrags.erase(fragIt);
335 
336  if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
337  // If "first" fragment in send window (allowing for wraparound), increment window begin
338  if (nextFragIt == m_unackedFrags.end()) {
339  m_firstUnackedFrag = m_unackedFrags.begin();
340  }
341  else {
342  m_firstUnackedFrag = nextFragIt;
343  }
344  }
345  else if (m_unackedFrags.empty()) {
346  m_firstUnackedFrag = m_unackedFrags.end();
347  }
348 }
349 
350 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
351  : pkt(std::move(pkt))
352  , sendTime(time::steady_clock::now())
353  , retxCount(0)
354  , nGreaterSeqAcks(0)
355 {
356 }
357 
358 LpReliability::NetPkt::NetPkt(lp::Packet&& pkt, bool isInterest)
359  : pkt(std::move(pkt))
360  , isInterest(isInterest)
361  , didRetx(false)
362 {
363 }
364 
365 } // namespace face
366 } // namespace nfd
void processIncomingPacket(const lp::Packet &pkt)
extract and parse all Acks and add Ack for contained Fragment (if any) to AckQueue ...
const GenericLinkService * getLinkService() const
const ssize_t MTU_UNLIMITED
indicates the transport has no limit on payload size
Definition: transport.hpp:91
void piggyback(lp::Packet &pkt, ssize_t mtu)
called by GenericLinkService to attach Acks onto an outgoing LpPacket
bool isEnabled
enables link-layer reliability
size_t maxRetx
maximum number of retransmissions for an LpPacket
Scheduler & getScheduler()
Returns the global Scheduler instance for the calling thread.
Definition: global.cpp:45
size_t seqNumLossThreshold
a fragment is considered lost if this number of fragments with greater sequence numbers are acknowled...
Copyright (c) 2014-2015, Regents of the University of California, Arizona Board of Regents...
Definition: algorithm.hpp:32
LpReliability(const Options &options, GenericLinkService *linkService)
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 ...
time::nanoseconds idleAckTimerPeriod
period between sending pending Acks in an IDLE packet
void setOptions(const Options &options)
set options for reliability
signal::Signal< LpReliability, Interest > onDroppedInterest
signals on Interest dropped by reliability system for exceeding allowed number of retx ...