From d8d8beda0052a65800c38886060e6b75afff1c69 Mon Sep 17 00:00:00 2001 From: michele papalini Date: Thu, 11 Apr 2019 11:25:11 +0200 Subject: [HICN-168] RTC retransmissions Change-Id: Ie7c24674715e593e1479b027226dbe8eda310421 Signed-off-by: michele papalini --- libtransport/src/hicn/transport/protocols/rtc.cc | 329 ++++++++++++++--------- libtransport/src/hicn/transport/protocols/rtc.h | 31 ++- 2 files changed, 228 insertions(+), 132 deletions(-) (limited to 'libtransport/src/hicn') diff --git a/libtransport/src/hicn/transport/protocols/rtc.cc b/libtransport/src/hicn/transport/protocols/rtc.cc index 882508b41..9402d3b02 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.cc +++ b/libtransport/src/hicn/transport/protocols/rtc.cc @@ -18,14 +18,6 @@ #include #include -/* - * TODO - * 2) start/constructor/rest variable implementation - * 3) interest retransmission: now I always recover, we should recover only if - * we have enough time 4) returnContentToApplication: rememeber to remove the - * first 32bits from the payload - */ - namespace transport { namespace protocol { @@ -39,6 +31,7 @@ RTCTransportProtocol::RTCTransportProtocol( modMask_((1 << default_values::log_2_default_buffer_size) - 1) { icnet_socket->getSocketOption(PORTAL, portal_); nack_timer_ = std::make_unique(portal_->getIoService()); + rtx_timer_ = std::make_unique(portal_->getIoService()); nack_timer_used_ = false; reset(); } @@ -49,7 +42,10 @@ RTCTransportProtocol::~RTCTransportProtocol() { } } -int RTCTransportProtocol::start() { return TransportProtocol::start(); } +int RTCTransportProtocol::start() { + checkRtx(); + return TransportProtocol::start(); +} void RTCTransportProtocol::stop() { if (!is_running_) return; @@ -67,6 +63,7 @@ void RTCTransportProtocol::resume() { inflightInterestsCount_ = 0; scheduleNextInterests(); + checkRtx(); portal_->runEventsLoop(); @@ -104,7 +101,7 @@ void RTCTransportProtocol::reset() { // names/packets var actualSegment_ = 0; inflightInterestsCount_ = 0; - while (interestRetransmissions_.size() != 0) interestRetransmissions_.pop(); + interestRetransmissions_.clear(); lastSegNacked_ = 0; lastReceived_ = 0; nackedByProducer_.clear(); @@ -120,10 +117,6 @@ void RTCTransportProtocol::reset() { gotFutureNack_ = 0; roundsWithoutNacks_ = 0; pathTable_.clear(); - // roundCounter_ = 0; - // minRTTwin_.clear(); - // for (int i = 0; i < MIN_RTT_WIN; i++) - // minRTTwin_.push_back(UINT_MAX); minRtt_ = UINT_MAX; // CC var @@ -135,7 +128,7 @@ void RTCTransportProtocol::reset() { producerPathLabel_ = 0; socket_->setSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, (uint32_t)HICN_RTC_INTEREST_LIFETIME); - // XXX this should bedone by the application + // XXX this should be done by the application } uint32_t max(uint32_t a, uint32_t b) { @@ -168,9 +161,12 @@ void RTCTransportProtocol::updateDelayStats( uint32_t segmentNumber = content_object.getName().getSuffix(); uint32_t pkt = segmentNumber & modMask_; - if (inflightInterests_[pkt].transmissionTime == - 0) // this is always the case if we have a retransmitted packet (timeout - // or RTCP) + if (inflightInterests_[pkt].state != sent_) + return; + + if(interestRetransmissions_.find(segmentNumber) != + interestRetransmissions_.end()) + //this packet was rtx at least once return; uint32_t pathLabel = content_object.getPathLabel(); @@ -283,8 +279,8 @@ void RTCTransportProtocol::computeMaxWindow(uint32_t productionRate, (double)HICN_MILLI_IN_A_SEC)); if (currentState_ == HICN_RTC_SYNC_STATE) { - // in this case we do not limit the window with the BDP, beacuse most likly - // it is wrong + // in this case we do not limit the window with the BDP, beacuse most + // likely it is wrong maxCWin_ = maxWaintingInterest; return; } @@ -333,68 +329,12 @@ void RTCTransportProtocol::increaseWindow() { } else { currentCWin_ = min( maxCWin_, - (uint32_t)ceil(currentCWin_ + (1.0 / (double)currentCWin_))); // linear + (uint32_t)ceil(currentCWin_ + + (1.0 / (double)currentCWin_))); // linear } } -void RTCTransportProtocol::sendInterest() { - Name *interest_name = nullptr; - socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - &interest_name); - bool isRTX = false; - // uint32_t sentInt = 0; - - if (interestRetransmissions_.size() > 0) { - // handle retransmission - // here we have two possibile retransmissions: retransmissions due to - // timeouts and retransmissions due to RTCP NACKs. we will send the interest - // anyway, even if it is pending (this is possible only in the second case) - uint32_t rtxSeg = interestRetransmissions_.front(); - interestRetransmissions_.pop(); - - // a packet recovery means that there was a loss - packetLost_++; - - uint32_t pkt = rtxSeg & modMask_; - interest_name->setSuffix(rtxSeg); - - // if the interest is not pending anymore we encrease the retrasnmission - // counter in order to avoid to handle a recovered packt as a normal one - if (!portal_->interestIsPending(*interest_name)) { - inflightInterests_[pkt].retransmissions++; - } - - inflightInterests_[pkt].transmissionTime = 0; - inflightInterests_[pkt].received = 0; - inflightInterests_[pkt].sequence = rtxSeg; - isRTX = true; - } else { - // in this case we send the packet only if it is not pending yet - interest_name->setSuffix(actualSegment_); - if (portal_->interestIsPending(*interest_name)) { - actualSegment_++; - return; - } - uint32_t pkt = actualSegment_ & modMask_; - - //if we already reacevied the content we don't ask it again - if(inflightInterests_[pkt].received == 1 && - inflightInterests_[pkt].sequence == actualSegment_) { - actualSegment_++; - return; - } - - // sentInt = actualSegment_; - inflightInterests_[pkt].transmissionTime = - std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); - inflightInterests_[pkt].retransmissions = 0; - inflightInterests_[pkt].received = 0; - inflightInterests_[pkt].sequence = actualSegment_; - actualSegment_++; - } - +void RTCTransportProtocol::sendInterest(Name *interest_name, bool rtx) { auto interest = getPacket(); interest->setName(*interest_name); @@ -420,27 +360,51 @@ void RTCTransportProtocol::sendInterest() { sentInterest_++; - if (!isRTX) { + if (!rtx) { inflightInterestsCount_++; } + } void RTCTransportProtocol::scheduleNextInterests() { checkRound(); if (!is_running_) return; - while (interestRetransmissions_.size() > 0) { - sendInterest(); - checkRound(); - } - while (inflightInterestsCount_ < currentCWin_) { - sendInterest(); + Name *interest_name = nullptr; + socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, + &interest_name); + + //we send the packet only if it is not pending yet + interest_name->setSuffix(actualSegment_); + if (portal_->interestIsPending(*interest_name)) { + actualSegment_++; + continue; + } + + uint32_t pkt = actualSegment_ & modMask_; + //if we already reacevied the content we don't ask it again + if(inflightInterests_[pkt].state == received_ && + inflightInterests_[pkt].sequence == actualSegment_) { + actualSegment_++; + continue; + } + + inflightInterests_[pkt].transmissionTime = + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + inflightInterests_[pkt].state = sent_; + inflightInterests_[pkt].sequence = actualSegment_; + actualSegment_++; + + sendInterest(interest_name, false); checkRound(); } } void RTCTransportProtocol::scheduleAppNackRtx(std::vector &nacks) { +#if 0 for (uint32_t i = 0; i < nacks.size(); i++) { if (nackedByProducer_.find(nacks[i]) != nackedByProducer_.end()) { continue; @@ -452,22 +416,142 @@ void RTCTransportProtocol::scheduleAppNackRtx(std::vector &nacks) { } scheduleNextInterests(); +#endif +} + +void RTCTransportProtocol::addRetransmissions(uint32_t val){ + //add only val in the rtx list + addRetransmissions(val, val + 1); } + +void RTCTransportProtocol::addRetransmissions(uint32_t start, uint32_t stop){ + for(uint32_t i = start; i < stop; i++){ + auto it = interestRetransmissions_.find(i); + if(it == interestRetransmissions_.end()){ + if (lastSegNacked_ <= i) { + //i must be larger than the last past nack received + interestRetransmissions_[i] = 0; + } + }//if the retransmission is already there the rtx timer will + //take care of it + } + retransmit(true); +} + +void RTCTransportProtocol::retransmit(bool first_rtx){ + auto it = interestRetransmissions_.begin(); + + //cut len to max HICN_MAX_RTX_SIZE + //since we use a map, the smaller (and so the older) sequence number are at + //the beginnin of the map + while(interestRetransmissions_.size() > HICN_MAX_RTX_SIZE){ + it = interestRetransmissions_.erase(it); + } + + it = interestRetransmissions_.begin(); + + while (it != interestRetransmissions_.end()){ + uint32_t pkt = it->first & modMask_; + + if(inflightInterests_[pkt].sequence != it->first){ + //this packet is not anymore in the inflight buffer, erase it + it = interestRetransmissions_.erase(it); + continue; + } + + //we retransmitted the packet too many times + if(it->second >= HICN_MAX_RTX){ + it = interestRetransmissions_.erase(it); + continue; + } + + //this packet is too old + if((lastReceived_ > it->first) && + (lastReceived_ - it->first) > HICN_MAX_RTX_MAX_AGE){ + it = interestRetransmissions_.erase(it); + continue; + } + + if(first_rtx){ + //TODO (optimization) + //the rtx that we never sent (it->second == 0) are all at the + //end, so we can go directly there + if(it->second == 0){ + inflightInterests_[pkt].transmissionTime = + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + it->second++; + + Name *interest_name = nullptr; + socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, + &interest_name); + interest_name->setSuffix(it->first); + sendInterest(interest_name, true); + } + ++it; + }else{ + //base on time + uint64_t now = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + if((now - inflightInterests_[pkt].transmissionTime) > 20){ + //XXX replace 20 with rtt + inflightInterests_[pkt].transmissionTime = + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + it->second++; + + Name *interest_name = nullptr; + socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, + &interest_name); + interest_name->setSuffix(it->first); + sendInterest(interest_name, true); + } + ++it; + } + } +} + +void RTCTransportProtocol::checkRtx(){ + retransmit(false); + rtx_timer_->expires_from_now(std::chrono::milliseconds(20)); + rtx_timer_->async_wait([this](std::error_code ec) { + if (ec) return; + checkRtx(); + }); +} + void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { // packetLost_++; uint32_t segmentNumber = interest->getName().getSuffix(); uint32_t pkt = segmentNumber & modMask_; - if (inflightInterests_[pkt].retransmissions == 0) { + if (inflightInterests_[pkt].state == sent_) { inflightInterestsCount_--; } - if (inflightInterests_[pkt].retransmissions < HICN_MAX_RTX && - lastSegNacked_ <= segmentNumber && - actualSegment_ < segmentNumber) { - interestRetransmissions_.push(segmentNumber); - } + //check how many times we sent this packet + auto it = interestRetransmissions_.find(segmentNumber); + if(it != interestRetransmissions_.end() && it->second >= HICN_MAX_RTX){ + inflightInterests_[pkt].state = lost_; + } + + if(inflightInterests_[pkt].state == sent_) { + inflightInterests_[pkt].state = timeout1_; + } else if (inflightInterests_[pkt].state == timeout1_) { + inflightInterests_[pkt].state = timeout2_; + } else if (inflightInterests_[pkt].state == timeout2_) { + inflightInterests_[pkt].state = lost_; + } + + if(inflightInterests_[pkt].state == lost_) { + interestRetransmissions_.erase(segmentNumber); + }else{ + addRetransmissions(segmentNumber); + } scheduleNextInterests(); } @@ -478,6 +562,7 @@ bool RTCTransportProtocol::checkIfProducerIsActive( uint32_t productionSeg = *payload; uint32_t productionRate = *(++payload); + if (productionRate == 0) { // the producer socket is not active // in this case we consider only the first nack @@ -486,7 +571,7 @@ bool RTCTransportProtocol::checkIfProducerIsActive( } nack_timer_used_ = true; - // actualSegment_ should be the one in the nack, which will the next in + // actualSegment_ should be the one in the nack, which will be the next in // production actualSegment_ = productionSeg; // all the rest (win size should not change) @@ -522,7 +607,7 @@ void RTCTransportProtocol::onNack(const ContentObject &content_object) { computeMaxWindow(productionRate, 0); increaseWindow(); - while (interestRetransmissions_.size() != 0) interestRetransmissions_.pop(); + interestRetransmissions_.clear(); lastSegNacked_ = productionSeg; if (nackedByProducer_.size() >= nackedByProducerMaxSize_) @@ -533,30 +618,13 @@ void RTCTransportProtocol::onNack(const ContentObject &content_object) { // we are asking stuff in the future gotFutureNack_++; - if(lastReceived_ > productionSeg){ - //if this happens the producer socket was restarted - //we erase all the inflight + timeout state (only for the first NACK) - if(gotFutureNack_ == 1){ - while (interestRetransmissions_.size() != 0) - interestRetransmissions_.pop(); - for (uint32_t i = 0; i < inflightInterests_.size(); i++){ - inflightInterests_[i].transmissionTime = 0; - inflightInterests_[i].sequence = 0; - inflightInterests_[i].retransmissions = 0; - inflightInterests_[i].received = 0; - } - } - actualSegment_ = productionSeg + 1; - //we can't say much abouit the window, so keep it as it is - }else{ - actualSegment_ = productionSeg + 1; + actualSegment_ = productionSeg + 1; - computeMaxWindow(productionRate, 0); - decreaseWindow(); + computeMaxWindow(productionRate, 0); + decreaseWindow(); - if (currentState_ == HICN_RTC_SYNC_STATE) { - currentState_ = HICN_RTC_NORMAL_STATE; - } + if (currentState_ == HICN_RTC_SYNC_STATE) { + currentState_ = HICN_RTC_NORMAL_STATE; } } // equal should not happen } @@ -570,11 +638,10 @@ void RTCTransportProtocol::onNackForRtx(const ContentObject &content_object) { // we are asking for stuff produced in the past actualSegment_ = max(productionSeg + 1, actualSegment_); - while (interestRetransmissions_.size() != 0) interestRetransmissions_.pop(); + interestRetransmissions_.clear(); lastSegNacked_ = productionSeg; } else if (productionSeg < nackSegment) { - actualSegment_ = productionSeg + 1; } // equal should not happen } @@ -598,8 +665,7 @@ void RTCTransportProtocol::onContentObject( // Nacks always come form the producer, so we set the producerPathLabel_; producerPathLabel_ = content_object->getPathLabel(); schedule_next_interest = checkIfProducerIsActive(*content_object); - if (inflightInterests_[pkt].retransmissions == 0) { - // discard nacks for rtx packets + if (inflightInterests_[pkt].state == sent_) { inflightInterestsCount_--; // if checkIfProducerIsActive returns false, we did all we need to do // inside that function, no need to call onNack @@ -610,26 +676,37 @@ void RTCTransportProtocol::onContentObject( } } else { - receivedData_++; - inflightInterests_[pkt].received = 1; - lastReceived_ = segmentNumber; - avgPacketSize_ = (HICN_ESTIMATED_PACKET_SIZE * avgPacketSize_) + ((1 - HICN_ESTIMATED_PACKET_SIZE) * payload->length()); - if (inflightInterests_[pkt].retransmissions == 0) { - inflightInterestsCount_--; + if (inflightInterests_[pkt].state == sent_) { + inflightInterestsCount_--; //packet sent without timeouts + } + + if (inflightInterests_[pkt].state == sent_ && + interestRetransmissions_.find(segmentNumber) == + interestRetransmissions_.end()){ // we count only non retransmitted data in order to take into accunt only // the transmition rate of the producer receivedBytes_ += (uint32_t)(content_object->headerSize() + content_object->payloadSize()); updateDelayStats(*content_object); + + addRetransmissions(lastReceived_ + 1, segmentNumber); + //lastReceived_ is updated only for data packets received without RTX + lastReceived_ = segmentNumber; } + receivedData_++; + inflightInterests_[pkt].state = received_; + reassemble(std::move(content_object)); increaseWindow(); } + //in any case we remove the packet from the rtx list + interestRetransmissions_.erase(segmentNumber); + if (schedule_next_interest) { scheduleNextInterests(); } diff --git a/libtransport/src/hicn/transport/protocols/rtc.h b/libtransport/src/hicn/transport/protocols/rtc.h index 227f1d737..55deead6e 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.h +++ b/libtransport/src/hicn/transport/protocols/rtc.h @@ -16,7 +16,7 @@ #pragma once #include -#include +#include #include #include @@ -38,7 +38,9 @@ // controller constant #define HICN_ROUND_LEN \ 200 // ms interval of time on which we take decisions / measurements -#define HICN_MAX_RTX 3 +#define HICN_MAX_RTX 10 +#define HICN_MAX_RTX_SIZE 1024 +#define HICN_MAX_RTX_MAX_AGE 10000 #define HICN_MIN_RTT_WIN 30 // rounds // cwin @@ -82,11 +84,21 @@ namespace transport { namespace protocol { +enum packetState { + sent_, + received_, + timeout1_, + timeout2_, + lost_ +}; + +typedef enum packetState packetState_t; + struct sentInterest { uint64_t transmissionTime; uint32_t sequence; //sequence number of the interest sent - uint8_t retransmissions; - uint8_t received; //1 = received, 0 = not received + //to handle seq % buffer_size + packetState_t state; //see packet state }; class RTCTransportProtocol : public TransportProtocol, public Reassembly { @@ -119,9 +131,13 @@ class RTCTransportProtocol : public TransportProtocol, public Reassembly { void resetPreviousWindow(); // packet functions - void sendInterest(); + void sendInterest(Name *interest_name, bool rtx); void scheduleNextInterests() override; void scheduleAppNackRtx(std::vector &nacks); + void addRetransmissions(uint32_t val); + void addRetransmissions(uint32_t start, uint32_t stop); + void retransmit(bool first_rtx); + void checkRtx(); void onTimeout(Interest::Ptr &&interest) override; // checkIfProducerIsActive: return true if we need to schedule an interest // immediatly after, false otherwise (this happens when the producer socket @@ -160,7 +176,10 @@ class RTCTransportProtocol : public TransportProtocol, public Reassembly { uint32_t actualSegment_; int32_t RTPhICN_offset_; uint32_t inflightInterestsCount_; - std::queue interestRetransmissions_; + //map seq to rtx + std::map interestRetransmissions_; + std::unique_ptr rtx_timer_; + //std::queue interestRetransmissions_; std::vector inflightInterests_; uint32_t lastSegNacked_; //indicates the segment id in the last received // past Nack. we do not ask for retransmissions -- cgit 1.2.3-korg