36 , m_linkService(linkService)
37 , m_firstUnackedFrag(m_unackedFrags.begin())
40 BOOST_ASSERT(m_linkService !=
nullptr);
50 m_idleAckTimer.cancel();
67 auto unackedFragsIt = m_unackedFrags.begin();
68 auto sendTime = time::steady_clock::now();
70 auto netPkt = make_shared<NetPkt>(std::move(pkt), isInterest);
71 netPkt->unackedFrags.reserve(frags.size());
73 for (lp::Packet& frag : frags) {
75 lp::Sequence txSeq = assignTxSequence(frag);
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;
87 if (m_unackedFrags.size() == 1) {
88 m_firstUnackedFrag = m_unackedFrags.begin();
92 netPkt->unackedFrags.push_back(unackedFragsIt);
101 auto now = time::steady_clock::now();
104 for (lp::Sequence ackSeq : pkt.list<lp::AckField>()) {
105 auto fragIt = m_unackedFrags.find(ackSeq);
106 if (fragIt == m_unackedFrags.end()) {
110 auto& frag = fragIt->second;
113 frag.rtoTimer.cancel();
115 if (frag.retxCount == 0) {
117 m_rttEst.addMeasurement(now - frag.sendTime);
123 auto lostLpPackets = findLostLpPackets(fragIt);
127 onLpPacketAcknowledged(fragIt);
133 std::set<lp::Sequence> removedLpPackets;
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);
147 if (pkt.has<lp::FragmentField>() && pkt.has<lp::TxSequenceField>()) {
148 m_ackQueue.push(pkt.get<lp::TxSequenceField>());
157 BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
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;
166 while (!m_ackQueue.empty()) {
167 lp::Sequence ackSeq = m_ackQueue.front();
169 const ssize_t ackSize = tlv::sizeOfVarNumber(lp::tlv::Ack) +
170 tlv::sizeOfVarNumber(
sizeof(lp::Sequence)) +
171 sizeof(lp::Sequence);
173 if (ackSize > remainingSpace) {
177 pkt.add<lp::AckField>(ackSeq);
179 remainingSpace -= ackSize;
184 LpReliability::assignTxSequence(lp::Packet& frag)
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"));
191 return m_lastTxSeqNo;
195 LpReliability::startIdleAckTimer()
197 if (m_idleAckTimer) {
203 while (!m_ackQueue.empty()) {
204 m_linkService->requestIdlePacket(0);
209 std::vector<lp::Sequence>
210 LpReliability::findLostLpPackets(LpReliability::UnackedFrags::iterator ackIt)
212 std::vector<lp::Sequence> lostLpPackets;
214 for (
auto it = m_firstUnackedFrag; ; ++it) {
215 if (it == m_unackedFrags.end()) {
216 it = m_unackedFrags.begin();
219 if (it->first == ackIt->first) {
223 auto& unackedFrag = it->second;
224 unackedFrag.nGreaterSeqAcks++;
227 lostLpPackets.push_back(it->first);
231 return lostLpPackets;
234 std::vector<lp::Sequence>
235 LpReliability::onLpPacketLost(lp::Sequence txSeq)
237 BOOST_ASSERT(m_unackedFrags.count(txSeq) > 0);
238 auto txSeqIt = m_unackedFrags.find(txSeq);
240 auto& txFrag = txSeqIt->second;
241 txFrag.rtoTimer.cancel();
242 auto netPkt = txFrag.netPkt;
243 std::vector<lp::Sequence> removedThisTxSeq;
246 if (txFrag.retxCount >= m_options.
maxRetx) {
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]);
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));
266 removedThisTxSeq.push_back(txSeqIt->first);
267 deleteUnackedFrag(txSeqIt);
271 lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
272 netPkt->didRetx =
true;
275 auto newTxFragIt = m_unackedFrags.emplace_hint(
276 m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
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;
287 auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
288 BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
289 *fragInNetPkt = newTxFragIt;
291 removedThisTxSeq.push_back(txSeqIt->first);
292 deleteUnackedFrag(txSeqIt);
295 m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt), 0);
298 newTxFrag.rtoTimer =
getScheduler().schedule(m_rttEst.getEstimatedRto(),
299 [=] { onLpPacketLost(newTxSeq); });
302 return removedThisTxSeq;
306 LpReliability::onLpPacketAcknowledged(UnackedFrags::iterator fragIt)
308 auto netPkt = fragIt->second.netPkt;
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();
317 if (netPkt->unackedFrags.empty()) {
318 if (netPkt->didRetx) {
326 deleteUnackedFrag(fragIt);
330 LpReliability::deleteUnackedFrag(UnackedFrags::iterator fragIt)
332 lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
333 lp::Sequence currentTxSeq = fragIt->first;
334 auto nextFragIt = m_unackedFrags.erase(fragIt);
336 if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
338 if (nextFragIt == m_unackedFrags.end()) {
339 m_firstUnackedFrag = m_unackedFrags.begin();
342 m_firstUnackedFrag = nextFragIt;
345 else if (m_unackedFrags.empty()) {
346 m_firstUnackedFrag = m_unackedFrags.end();
350 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
351 : pkt(std::move(pkt))
352 , sendTime(time::steady_clock::now())
358 LpReliability::NetPkt::NetPkt(lp::Packet&& pkt,
bool isInterest)
359 : pkt(std::move(pkt))
360 , isInterest(isInterest)
void processIncomingPacket(const lp::Packet &pkt)
extract and parse all Acks and add Ack for contained Fragment (if any) to AckQueue ...
const ssize_t MTU_UNLIMITED
indicates the transport has no limit on payload size
void piggyback(lp::Packet &pkt, ssize_t mtu)
called by GenericLinkService to attach Acks onto an outgoing LpPacket
bool isEnabled
enables link-layer reliability
PacketCounter nAcknowledged
count of network-layer packets that did not require retransmission of a fragment
GenericLinkService is a LinkService that implements the NDNLPv2 protocol.
size_t maxRetx
maximum number of retransmissions for an LpPacket
Scheduler & getScheduler()
Returns the global Scheduler instance for the calling thread.
const GenericLinkService * getLinkService() const
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...
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 ...
PacketCounter nRetxExhausted
count of network-layer packets dropped because a fragment reached the maximum number of retransmissio...
PacketCounter nRetransmitted
count of network-layer packets that had at least one fragment retransmitted, but were eventually rece...