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)
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>();
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
146 LpReliability::piggyback(lp::Packet& pkt, ssize_t mtu)
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) +
162  tlv::sizeOfVarNumber(
163  tlv::sizeOfNonNegativeInteger(std::numeric_limits<lp::Sequence>::max())) +
164  tlv::sizeOfNonNegativeInteger(ackSeq);
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 
176 lp::Sequence
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  deleteUnackedFrag(txSeqIt);
254  }
255  else {
256  // Assign new TxSequence
257  lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
258 
259  // Move fragment to new TxSequence mapping
260  auto newTxFragIt = m_unackedFrags.emplace_hint(
261  m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
262  ? m_firstUnackedFrag
263  : m_unackedFrags.end(),
264  std::piecewise_construct,
265  std::forward_as_tuple(newTxSeq),
266  std::forward_as_tuple(txFrag.pkt));
267  auto& newTxFrag = newTxFragIt->second;
268  newTxFrag.retxCount = txFrag.retxCount + 1;
269  newTxFrag.netPkt = netPkt;
270 
271  // Update associated NetPkt
272  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
273  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
274  *fragInNetPkt = newTxFragIt;
275 
276  deleteUnackedFrag(txSeqIt);
277 
278  // Retransmit fragment
279  m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt));
280 
281  // Start RTO timer for this sequence
282  newTxFrag.rtoTimer = scheduler::schedule(m_rto.computeRto(),
283  bind(&LpReliability::onLpPacketLost, this, newTxFragIt));
284  }
285 }
286 
287 void
288 LpReliability::onLpPacketAcknowledged(UnackedFrags::iterator fragIt)
289 {
290  auto netPkt = fragIt->second.netPkt;
291 
292  // Remove from NetPkt unacked fragment list
293  auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), fragIt);
294  BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
295  *fragInNetPkt = netPkt->unackedFrags.back();
296  netPkt->unackedFrags.pop_back();
297 
298  // Check if network-layer packet completely received. If so, increment counters
299  if (netPkt->unackedFrags.empty()) {
300  if (netPkt->didRetx) {
301  ++m_linkService->nRetransmitted;
302  }
303  else {
304  ++m_linkService->nAcknowledged;
305  }
306  }
307 
308  deleteUnackedFrag(fragIt);
309 }
310 
311 void
312 LpReliability::deleteUnackedFrag(UnackedFrags::iterator fragIt)
313 {
314  lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
315  lp::Sequence currentTxSeq = fragIt->first;
316  auto nextFragIt = m_unackedFrags.erase(fragIt);
317 
318  if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
319  // If "first" fragment in send window (allowing for wraparound), increment window begin
320  if (nextFragIt == m_unackedFrags.end()) {
321  m_firstUnackedFrag = m_unackedFrags.begin();
322  }
323  else {
324  m_firstUnackedFrag = nextFragIt;
325  }
326  }
327  else if (m_unackedFrags.empty()) {
328  m_firstUnackedFrag = m_unackedFrags.end();
329  }
330 }
331 
332 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
333  : pkt(std::move(pkt))
334  , sendTime(time::steady_clock::now())
335  , retxCount(0)
336  , nGreaterSeqAcks(0)
337 {
338 }
339 
340 } // namespace face
341 } // 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:95
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
Table::const_iterator iterator
Definition: cs-internal.hpp:41
void handleOutgoing(std::vector< lp::Packet > &frags)
observe outgoing fragment(s) of a network packet and store for potential retransmission ...
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)
time::nanoseconds idleAckTimerPeriod
period between sending pending Acks in an IDLE packet
void setOptions(const Options &options)
set options for reliability
EventId schedule(time::nanoseconds after, const EventCallback &event)
schedule an event
Definition: scheduler.cpp:47
void cancel()
cancels the event manually
Definition: scheduler.cpp:95
Duration computeRto() const
void addMeasurement(Duration measure)