37 , m_linkService(linkService)
38 , m_firstUnackedFrag(m_unackedFrags.begin())
41 BOOST_ASSERT(m_linkService !=
nullptr);
51 m_idleAckTimer.cancel();
68 auto unackedFragsIt = m_unackedFrags.begin();
69 auto sendTime = time::steady_clock::now();
71 auto netPkt = make_shared<NetPkt>(std::move(pkt), isInterest);
72 netPkt->unackedFrags.reserve(frags.size());
74 for (lp::Packet& frag : frags) {
76 BOOST_ASSERT(frag.has<lp::SequenceField>());
79 lp::Sequence txSeq = assignTxSequence(frag);
82 unackedFragsIt = m_unackedFrags.try_emplace(unackedFragsIt, txSeq, frag);
83 unackedFragsIt->second.sendTime = sendTime;
84 auto rto = m_rttEst.getEstimatedRto();
85 lp::Sequence seq = frag.get<lp::SequenceField>();
87 time::duration_cast<time::milliseconds>(rto).count() <<
"ms");
88 unackedFragsIt->second.rtoTimer =
getScheduler().schedule(rto, [=] {
89 onLpPacketLost(txSeq,
true);
91 unackedFragsIt->second.netPkt = netPkt;
93 if (m_unackedFrags.size() == 1) {
94 m_firstUnackedFrag = m_unackedFrags.begin();
98 netPkt->unackedFrags.push_back(unackedFragsIt);
107 bool isDuplicate =
false;
108 auto now = time::steady_clock::now();
111 for (lp::Sequence ackTxSeq : pkt.list<lp::AckField>()) {
112 auto fragIt = m_unackedFrags.find(ackTxSeq);
113 if (fragIt == m_unackedFrags.end()) {
118 auto& frag = fragIt->second;
121 frag.rtoTimer.cancel();
123 if (frag.retxCount == 0) {
124 NFD_LOG_FACE_TRACE(
"received ack for seq=" << frag.pkt.get<lp::SequenceField>() <<
", txseq=" <<
125 ackTxSeq <<
", retx=0, rtt=" <<
126 time::duration_cast<time::milliseconds>(now - frag.sendTime).count() <<
"ms");
128 m_rttEst.addMeasurement(now - frag.sendTime);
131 NFD_LOG_FACE_TRACE(
"received ack for seq=" << frag.pkt.get<lp::SequenceField>() <<
", txseq=" <<
132 ackTxSeq <<
", retx=" << frag.retxCount);
138 auto lostLpPackets = findLostLpPackets(fragIt);
142 onLpPacketAcknowledged(fragIt);
148 std::set<lp::Sequence> removedLpPackets;
151 for (lp::Sequence txSeq : lostLpPackets) {
152 if (removedLpPackets.find(txSeq) == removedLpPackets.end()) {
153 auto removedTxSeqs = onLpPacketLost(txSeq,
false);
154 for (
auto removedTxSeq : removedTxSeqs) {
155 removedLpPackets.insert(removedTxSeq);
162 if (pkt.has<lp::FragmentField>() && pkt.has<lp::TxSequenceField>()) {
163 NFD_LOG_FACE_TRACE(
"queueing ack for remote txseq=" << pkt.get<lp::TxSequenceField>());
164 m_ackQueue.push(pkt.get<lp::TxSequenceField>());
167 if (pkt.has<lp::SequenceField>()) {
168 lp::Sequence pktSequence = pkt.get<lp::SequenceField>();
169 isDuplicate = m_recentRecvSeqs.count(pktSequence) > 0;
171 auto now = time::steady_clock::now();
172 auto rto = m_rttEst.getEstimatedRto();
173 while (!m_recentRecvSeqsQueue.empty() &&
174 now > m_recentRecvSeqs[m_recentRecvSeqsQueue.front()] + rto) {
175 m_recentRecvSeqs.erase(m_recentRecvSeqsQueue.front());
176 m_recentRecvSeqsQueue.pop();
178 m_recentRecvSeqs.try_emplace(pktSequence, now);
179 m_recentRecvSeqsQueue.push(pktSequence);
192 BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
195 ssize_t pktSize = pkt.wireEncode().size();
196 ssize_t reservedSpace = tlv::sizeOfVarNumber(ndn::MAX_NDN_PACKET_SIZE) -
197 tlv::sizeOfVarNumber(pktSize);
198 ssize_t remainingSpace = (mtu ==
MTU_UNLIMITED ? ndn::MAX_NDN_PACKET_SIZE : mtu) - reservedSpace;
199 remainingSpace -= pktSize;
201 while (!m_ackQueue.empty()) {
202 lp::Sequence ackTxSeq = m_ackQueue.front();
204 const ssize_t ackSize = tlv::sizeOfVarNumber(lp::tlv::Ack) +
205 tlv::sizeOfVarNumber(
sizeof(lp::Sequence)) +
206 sizeof(lp::Sequence);
208 if (ackSize > remainingSpace) {
214 pkt.add<lp::AckField>(ackTxSeq);
216 remainingSpace -= ackSize;
221 LpReliability::assignTxSequence(lp::Packet& frag)
223 lp::Sequence txSeq = ++m_lastTxSeqNo;
224 frag.set<lp::TxSequenceField>(txSeq);
225 if (!m_unackedFrags.empty() && m_lastTxSeqNo == m_firstUnackedFrag->first) {
226 NDN_THROW(std::length_error(
"TxSequence range exceeded"));
228 return m_lastTxSeqNo;
232 LpReliability::startIdleAckTimer()
234 if (m_idleAckTimer) {
240 while (!m_ackQueue.empty()) {
241 m_linkService->requestIdlePacket();
246 std::vector<lp::Sequence>
247 LpReliability::findLostLpPackets(LpReliability::UnackedFrags::iterator ackIt)
249 std::vector<lp::Sequence> lostLpPackets;
251 for (
auto it = m_firstUnackedFrag; ; ++it) {
252 if (it == m_unackedFrags.end()) {
253 it = m_unackedFrags.begin();
256 if (it->first == ackIt->first) {
260 auto& unackedFrag = it->second;
261 unackedFrag.nGreaterSeqAcks++;
263 ", before count=" << unackedFrag.nGreaterSeqAcks);
265 if (unackedFrag.nGreaterSeqAcks >= m_options.seqNumLossThreshold) {
266 lostLpPackets.push_back(it->first);
270 return lostLpPackets;
273 std::vector<lp::Sequence>
274 LpReliability::onLpPacketLost(lp::Sequence txSeq,
bool isTimeout)
276 BOOST_ASSERT(m_unackedFrags.count(txSeq) > 0);
277 auto txSeqIt = m_unackedFrags.find(txSeq);
279 auto& txFrag = txSeqIt->second;
280 txFrag.rtoTimer.cancel();
281 auto netPkt = txFrag.netPkt;
282 std::vector<lp::Sequence> removedThisTxSeq;
283 lp::Sequence seq = txFrag.pkt.get<lp::SequenceField>();
290 " considered lost from acks for more recent txseqs");
294 if (txFrag.retxCount >= m_options.maxRetx) {
297 for (
size_t i = 0; i < netPkt->unackedFrags.size(); i++) {
298 if (netPkt->unackedFrags[i] != txSeqIt) {
299 removedThisTxSeq.push_back(netPkt->unackedFrags[i]->first);
300 deleteUnackedFrag(netPkt->unackedFrags[i]);
304 ++m_linkService->nRetxExhausted;
307 if (netPkt->isInterest) {
308 BOOST_ASSERT(netPkt->pkt.has<lp::FragmentField>());
309 auto frag = netPkt->pkt.get<lp::FragmentField>();
310 onDroppedInterest(Interest(Block({frag.first, frag.second})));
314 removedThisTxSeq.push_back(txSeqIt->first);
315 deleteUnackedFrag(txSeqIt);
319 lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
320 netPkt->didRetx =
true;
323 auto hint = m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
325 : m_unackedFrags.end();
326 auto newTxFragIt = m_unackedFrags.try_emplace(hint, newTxSeq, txFrag.pkt);
327 auto& newTxFrag = newTxFragIt->second;
328 newTxFrag.retxCount = txFrag.retxCount + 1;
329 newTxFrag.netPkt = netPkt;
332 auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
333 BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
334 *fragInNetPkt = newTxFragIt;
336 removedThisTxSeq.push_back(txSeqIt->first);
337 deleteUnackedFrag(txSeqIt);
340 m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt));
342 auto rto = m_rttEst.getEstimatedRto();
343 NFD_LOG_FACE_TRACE(
"retransmitting seq=" << seq <<
", txseq=" << newTxSeq <<
", retx=" <<
344 txFrag.retxCount <<
", rto=" <<
345 time::duration_cast<time::milliseconds>(rto).count() <<
"ms");
349 onLpPacketLost(newTxSeq,
true);
353 return removedThisTxSeq;
357 LpReliability::onLpPacketAcknowledged(UnackedFrags::iterator fragIt)
359 auto netPkt = fragIt->second.netPkt;
362 auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), fragIt);
363 BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
364 *fragInNetPkt = netPkt->unackedFrags.back();
365 netPkt->unackedFrags.pop_back();
368 if (netPkt->unackedFrags.empty()) {
369 if (netPkt->didRetx) {
370 ++m_linkService->nRetransmitted;
373 ++m_linkService->nAcknowledged;
377 deleteUnackedFrag(fragIt);
381 LpReliability::deleteUnackedFrag(UnackedFrags::iterator fragIt)
383 lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
384 lp::Sequence currentTxSeq = fragIt->first;
385 auto nextFragIt = m_unackedFrags.erase(fragIt);
387 if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
389 if (nextFragIt == m_unackedFrags.end()) {
390 m_firstUnackedFrag = m_unackedFrags.begin();
393 m_firstUnackedFrag = nextFragIt;
396 else if (m_unackedFrags.empty()) {
397 m_firstUnackedFrag = m_unackedFrags.end();
401 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
402 : pkt(std::move(pkt))
403 , sendTime(time::steady_clock::now())
409 LpReliability::NetPkt::NetPkt(lp::Packet&& pkt,
bool isInterest)
410 : pkt(std::move(pkt))
411 , isInterest(isInterest)
419 if (flh.
obj.getLinkService() ==
nullptr) {
420 os <<
"[id=0,local=unknown,remote=unknown] ";
423 os << FaceLogHelper<LinkService>(*flh.
obj.getLinkService());
For internal use by FaceLogging macros.
GenericLinkService is a LinkService that implements the NDNLPv2 protocol.
LpReliability(const Options &options, GenericLinkService *linkService)
void piggyback(lp::Packet &pkt, ssize_t mtu)
Called by GenericLinkService to attach Acks onto an outgoing LpPacket.
const GenericLinkService * getLinkService() const
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.
bool processIncomingPacket(const lp::Packet &pkt)
Extract and parse all Acks and add Ack for contained Fragment (if any) to AckQueue.
void setOptions(const Options &options)
Set options for reliability.
#define NFD_LOG_FACE_DEBUG(msg)
Log a message at DEBUG level.
#define NFD_LOG_FACE_TRACE(msg)
Log a message at TRACE level.
#define NFD_LOG_INIT(name)
std::ostream & operator<<(std::ostream &os, const FaceLogHelper< Face > &flh)
constexpr ssize_t MTU_UNLIMITED
Indicates that the transport has no limit on payload size.
Scheduler & getScheduler()
Returns the global Scheduler instance for the calling thread.
time::nanoseconds idleAckTimerPeriod
Period between sending pending Acks in an IDLE packet.
bool isEnabled
Enables link-layer reliability.