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>();
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;
84 unackedFragsIt->second.rtoTimer =
86 unackedFragsIt->second.netPkt = netPkt;
88 if (m_unackedFrags.size() == 1) {
89 m_firstUnackedFrag = m_unackedFrags.begin();
93 netPkt->unackedFrags.push_back(unackedFragsIt);
102 auto now = time::steady_clock::now();
105 for (lp::Sequence ackSeq : pkt.list<lp::AckField>()) {
106 auto fragIt = m_unackedFrags.find(ackSeq);
107 if (fragIt == m_unackedFrags.end()) {
111 auto& frag = fragIt->second;
114 frag.rtoTimer.cancel();
116 if (frag.retxCount == 0) {
118 m_rto.
addMeasurement(time::duration_cast<RttEstimator::Duration>(now - frag.sendTime));
124 auto lostLpPackets = findLostLpPackets(fragIt);
128 onLpPacketAcknowledged(fragIt);
132 this->onLpPacketLost(txSeqIt);
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();
149 BOOST_ASSERT(pkt.wireEncode().type() == lp::tlv::LpPacket);
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;
158 while (!m_ackQueue.empty()) {
159 lp::Sequence ackSeq = m_ackQueue.front();
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);
166 if (ackSize > remainingSpace) {
170 pkt.add<lp::AckField>(ackSeq);
172 remainingSpace -= ackSize;
177 LpReliability::assignTxSequence(lp::Packet& frag)
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"));
184 return m_lastTxSeqNo;
188 LpReliability::startIdleAckTimer()
190 BOOST_ASSERT(!m_isIdleAckTimerRunning);
191 m_isIdleAckTimerRunning =
true;
194 while (!m_ackQueue.empty()) {
195 m_linkService->requestIdlePacket();
198 m_isIdleAckTimerRunning =
false;
203 LpReliability::stopIdleAckTimer()
206 m_isIdleAckTimerRunning =
false;
209 std::vector<LpReliability::UnackedFrags::iterator>
212 std::vector<UnackedFrags::iterator> 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);
231 return lostLpPackets;
237 BOOST_ASSERT(m_unackedFrags.count(txSeqIt->first) > 0);
239 auto& txFrag = txSeqIt->second;
240 txFrag.rtoTimer.cancel();
241 auto netPkt = txFrag.netPkt;
244 if (txFrag.retxCount >= m_options.
maxRetx) {
246 for (
size_t i = 0; i < netPkt->unackedFrags.size(); i++) {
247 if (netPkt->unackedFrags[i] != txSeqIt) {
248 deleteUnackedFrag(netPkt->unackedFrags[i]);
253 deleteUnackedFrag(txSeqIt);
257 lp::Sequence newTxSeq = assignTxSequence(txFrag.pkt);
260 auto newTxFragIt = m_unackedFrags.emplace_hint(
261 m_firstUnackedFrag != m_unackedFrags.end() && m_firstUnackedFrag->first > newTxSeq
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;
272 auto fragInNetPkt = std::find(netPkt->unackedFrags.begin(), netPkt->unackedFrags.end(), txSeqIt);
273 BOOST_ASSERT(fragInNetPkt != netPkt->unackedFrags.end());
274 *fragInNetPkt = newTxFragIt;
276 deleteUnackedFrag(txSeqIt);
279 m_linkService->sendLpPacket(lp::Packet(newTxFrag.pkt));
283 bind(&LpReliability::onLpPacketLost,
this, newTxFragIt));
290 auto netPkt = fragIt->second.netPkt;
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();
299 if (netPkt->unackedFrags.empty()) {
300 if (netPkt->didRetx) {
308 deleteUnackedFrag(fragIt);
314 lp::Sequence firstUnackedTxSeq = m_firstUnackedFrag->first;
315 lp::Sequence currentTxSeq = fragIt->first;
316 auto nextFragIt = m_unackedFrags.erase(fragIt);
318 if (!m_unackedFrags.empty() && firstUnackedTxSeq == currentTxSeq) {
320 if (nextFragIt == m_unackedFrags.end()) {
321 m_firstUnackedFrag = m_unackedFrags.begin();
324 m_firstUnackedFrag = nextFragIt;
327 else if (m_unackedFrags.empty()) {
328 m_firstUnackedFrag = m_unackedFrags.end();
332 LpReliability::UnackedFrag::UnackedFrag(lp::Packet pkt)
333 : pkt(std::move(pkt))
334 , sendTime(time::steady_clock::now())
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
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...
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
void cancel()
cancels the event manually
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...