35 , m_linkService(linkService)
36 , m_firstUnackedFrag(m_unackedFrags.begin())
38 , m_isIdleAckTimerRunning(false)
40 BOOST_ASSERT(m_linkService !=
nullptr);
51 this->stopIdleAckTimer();
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 lp::Sequence txSeq = assignTxSequence(frag);
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;
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_rto.
addMeasurement(time::duration_cast<RttEstimator::Duration>(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 = this->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>());
149 if (!m_isIdleAckTimerRunning) {
150 this->startIdleAckTimer();
159 BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
162 ssize_t pktSize = pkt.wireEncode().size();
163 ssize_t reservedSpace = tlv::sizeOfVarNumber(ndn::MAX_NDN_PACKET_SIZE) -
164 tlv::sizeOfVarNumber(pktSize);
165 ssize_t remainingSpace = (mtu ==
MTU_UNLIMITED ? ndn::MAX_NDN_PACKET_SIZE : mtu) - reservedSpace;
166 remainingSpace -= pktSize;
168 while (!m_ackQueue.empty()) {
169 lp::Sequence ackSeq = m_ackQueue.front();
171 const ssize_t ackSize = tlv::sizeOfVarNumber(lp::tlv::Ack) +
172 tlv::sizeOfVarNumber(
sizeof(lp::Sequence)) +
173 sizeof(lp::Sequence);
175 if (ackSize > remainingSpace) {
179 pkt.add<lp::AckField>(ackSeq);
181 remainingSpace -= ackSize;
186 LpReliability::assignTxSequence(lp::Packet& frag)
188 lp::Sequence txSeq = ++m_lastTxSeqNo;
189 frag.set<lp::TxSequenceField>(txSeq);
190 if (m_unackedFrags.size() > 0 && m_lastTxSeqNo == m_firstUnackedFrag->first) {
191 BOOST_THROW_EXCEPTION(std::length_error(
"TxSequence range exceeded"));
193 return m_lastTxSeqNo;
197 LpReliability::startIdleAckTimer()
199 BOOST_ASSERT(!m_isIdleAckTimerRunning);
200 m_isIdleAckTimerRunning =
true;
203 while (!m_ackQueue.empty()) {
204 m_linkService->requestIdlePacket();
207 m_isIdleAckTimerRunning =
false;
212 LpReliability::stopIdleAckTimer()
214 m_idleAckTimer.cancel();
215 m_isIdleAckTimerRunning =
false;
218 std::vector<lp::Sequence>
221 std::vector<lp::Sequence> lostLpPackets;
223 for (
auto it = m_firstUnackedFrag; ; ++it) {
224 if (it == m_unackedFrags.end()) {
225 it = m_unackedFrags.begin();
228 if (it->first == ackIt->first) {
232 auto& unackedFrag = it->second;
233 unackedFrag.nGreaterSeqAcks++;
236 lostLpPackets.push_back(it->first);
240 return lostLpPackets;
243 std::vector<lp::Sequence>
244 LpReliability::onLpPacketLost(lp::Sequence txSeq)
246 BOOST_ASSERT(m_unackedFrags.count(txSeq) > 0);
247 auto txSeqIt = m_unackedFrags.find(txSeq);
249 auto& txFrag = txSeqIt->second;
250 txFrag.rtoTimer.cancel();
251 auto netPkt = txFrag.netPkt;
252 std::vector<lp::Sequence> removedThisTxSeq;
255 if (txFrag.retxCount >= m_options.
maxRetx) {
257 for (
size_t i = 0; i < netPkt->unackedFrags.size(); i++) {
258 if (netPkt->unackedFrags[i] != txSeqIt) {
259 removedThisTxSeq.push_back(netPkt->unackedFrags[i]->first);
260 deleteUnackedFrag(netPkt->unackedFrags[i]);
267 if (netPkt->isInterest) {
268 BOOST_ASSERT(netPkt->pkt.has<lp::FragmentField>());
269 ndn::Buffer::const_iterator fragBegin, fragEnd;
270 std::tie(fragBegin, fragEnd) = netPkt->pkt.get<lp::FragmentField>();
271 Block frag(&*fragBegin, std::distance(fragBegin, fragEnd));
275 removedThisTxSeq.push_back(txSeqIt->first);
276 deleteUnackedFrag(txSeqIt);
280 lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
281 netPkt->didRetx =
true;
284 auto newTxFragIt = m_unackedFrags.emplace_hint(
285 m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
287 : m_unackedFrags.end(),
288 std::piecewise_construct,
289 std::forward_as_tuple(newTxSeq),
290 std::forward_as_tuple(txFrag.pkt));
291 auto& newTxFrag = newTxFragIt->second;
292 newTxFrag.retxCount = txFrag.retxCount + 1;
293 newTxFrag.netPkt = netPkt;
296 auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
297 BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
298 *fragInNetPkt = newTxFragIt;
300 removedThisTxSeq.push_back(txSeqIt->first);
301 deleteUnackedFrag(txSeqIt);
304 m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt));
310 return removedThisTxSeq;
316 auto netPkt = fragIt->second.netPkt;
319 auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), fragIt);
320 BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
321 *fragInNetPkt = netPkt->unackedFrags.back();
322 netPkt->unackedFrags.pop_back();
325 if (netPkt->unackedFrags.empty()) {
326 if (netPkt->didRetx) {
334 deleteUnackedFrag(fragIt);
340 lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
341 lp::Sequence currentTxSeq = fragIt->first;
342 auto nextFragIt = m_unackedFrags.erase(fragIt);
344 if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
346 if (nextFragIt == m_unackedFrags.end()) {
347 m_firstUnackedFrag = m_unackedFrags.begin();
350 m_firstUnackedFrag = nextFragIt;
353 else if (m_unackedFrags.empty()) {
354 m_firstUnackedFrag = m_unackedFrags.end();
358 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
359 : pkt(std::move(pkt))
360 , sendTime(time::steady_clock::now())
366 LpReliability::NetPkt::NetPkt(lp::Packet&& pkt,
bool isInterest)
367 : pkt(std::move(pkt))
368 , isInterest(isInterest)
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
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
Table::const_iterator iterator
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 ...
EventId schedule(time::nanoseconds after, const EventCallback &event)
Schedule an event.
PacketCounter nRetxExhausted
count of network-layer packets dropped because a fragment reached the maximum number of retransmissio...
Duration computeRto() const
void addMeasurement(Duration measure)
PacketCounter nRetransmitted
count of network-layer packets that had at least one fragment retransmitted, but were eventually rece...