From 08233d44a6cfde878d7e10bca38ae935ed1c8fd5 Mon Sep 17 00:00:00 2001 From: Mauro Date: Wed, 30 Jun 2021 07:57:22 +0000 Subject: [HICN-713] Transport Library Major Refactoring 2 Co-authored-by: Luca Muscariello Co-authored-by: Michele Papalini Co-authored-by: Olivier Roques Co-authored-by: Giulio Grassi Signed-off-by: Mauro Sardara Change-Id: I5b2c667bad66feb45abdb5effe22ed0f6c85d1c2 --- libtransport/src/protocols/raaqm.cc | 180 ++++++++++++++---------------------- 1 file changed, 69 insertions(+), 111 deletions(-) (limited to 'libtransport/src/protocols/raaqm.cc') diff --git a/libtransport/src/protocols/raaqm.cc b/libtransport/src/protocols/raaqm.cc index bc8500227..1247af400 100644 --- a/libtransport/src/protocols/raaqm.cc +++ b/libtransport/src/protocols/raaqm.cc @@ -17,7 +17,7 @@ #include #include #include -#include +#include #include #include @@ -31,7 +31,8 @@ using namespace interface; RaaqmTransportProtocol::RaaqmTransportProtocol( implementation::ConsumerSocket *icn_socket) - : TransportProtocol(icn_socket, new ByteStreamReassembly(icn_socket, this)), + : TransportProtocol(icn_socket, new IndexManager(icn_socket, this), + new ByteStreamReassembly(icn_socket, this)), current_window_size_(1), interests_in_flight_(0), cur_path_(nullptr), @@ -47,11 +48,34 @@ RaaqmTransportProtocol::~RaaqmTransportProtocol() { } } -int RaaqmTransportProtocol::start() { +void RaaqmTransportProtocol::reset() { + // Set first segment to retrieve + TransportProtocol::reset(); + core::Name *name; + socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name); + indexer_verifier_->setFirstSuffix(name->getSuffix()); + std::queue empty; + std::swap(interest_to_retransmit_, empty); + stats_->reset(); + + // Reset protocol variables + interests_in_flight_ = 0; + t0_ = utils::SteadyClock::now(); + + // Optionally reset congestion window + bool reset_window; + socket_->getSocketOption(RaaqmTransportOptions::PER_SESSION_CWINDOW_RESET, + reset_window); + if (reset_window) { + current_window_size_ = 1; + } + + // Reset rate estimator if (rate_estimator_) { rate_estimator_->onStart(); } + // If not cur_path exists, create one if (!cur_path_) { // RAAQM double drop_factor; @@ -94,37 +118,6 @@ int RaaqmTransportProtocol::start() { cur_path_ = cur_path.get(); path_table_[default_values::path_id] = std::move(cur_path); } - - portal_->setConsumerCallback(this); - return TransportProtocol::start(); -} - -void RaaqmTransportProtocol::resume() { return TransportProtocol::resume(); } - -void RaaqmTransportProtocol::reset() { - // Set first segment to retrieve - core::Name *name; - socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name); - index_manager_->reset(); - index_manager_->setFirstSuffix(name->getSuffix()); - std::queue empty; - std::swap(interest_to_retransmit_, empty); - stats_->reset(); - - // Reset reassembly component - reassembly_protocol_->reInitialize(); - - // Reset protocol variables - interests_in_flight_ = 0; - t0_ = utils::SteadyClock::now(); - - // Optionally reset congestion window - bool reset_window; - socket_->getSocketOption(RaaqmTransportOptions::PER_SESSION_CWINDOW_RESET, - reset_window); - if (reset_window) { - current_window_size_ = 1; - } } void RaaqmTransportProtocol::increaseWindow() { @@ -194,9 +187,8 @@ void RaaqmTransportProtocol::init() { lte_delay_ = 15000; if (!is) { - TRANSPORT_LOGW( - "WARNING: RAAQM parameters not found at %s, set default values", - RAAQM_CONFIG_PATH); + LOG(WARNING) << "RAAQM parameters not found at " << RAAQM_CONFIG_PATH + << ", set default values"; return; } @@ -322,12 +314,9 @@ void RaaqmTransportProtocol::init() { is.close(); } -void RaaqmTransportProtocol::onContentObject(Interest &interest, - ContentObject &content_object) { - // Check whether makes sense to continue - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { - return; - } +void RaaqmTransportProtocol::onContentObjectReceived( + Interest &interest, ContentObject &content_object, std::error_code &ec) { + uint32_t incremental_suffix = content_object.getName().getSuffix(); // Call application-defined callbacks if (*on_content_object_input_) { @@ -338,18 +327,12 @@ void RaaqmTransportProtocol::onContentObject(Interest &interest, (*on_interest_satisfied_)(*socket_->getInterface(), interest); } + ec = make_error_code(protocol_error::success); + if (content_object.getPayloadType() == PayloadType::DATA) { stats_->updateBytesRecv(content_object.payloadSize()); } - onContentSegment(interest, content_object); - scheduleNextInterests(); -} - -void RaaqmTransportProtocol::onContentSegment(Interest &interest, - ContentObject &content_object) { - uint32_t incremental_suffix = content_object.getName().getSuffix(); - // Decrease in-flight interests interests_in_flight_--; @@ -358,11 +341,13 @@ void RaaqmTransportProtocol::onContentSegment(Interest &interest, afterContentReception(interest, content_object); } - index_manager_->onContentObject(interest, content_object); + // Schedule next interests + scheduleNextInterests(); } void RaaqmTransportProtocol::onPacketDropped(Interest &interest, - ContentObject &content_object) { + ContentObject &content_object, + const std::error_code &reason) { uint32_t max_rtx = 0; socket_->getSocketOption(GeneralTransportOptions::MAX_INTEREST_RETX, max_rtx); @@ -380,16 +365,15 @@ void RaaqmTransportProtocol::onPacketDropped(Interest &interest, (*on_interest_output_)(*socket_->getInterface(), interest); } - if (!is_running_) { + if (!isRunning()) { return; } interest_retransmissions_[segment & mask]++; - interest_to_retransmit_.push(interest.shared_from_this()); + interest_to_retransmit_.push(segment); } else { - TRANSPORT_LOGE( - "Stop: received not trusted packet %llu times", - (unsigned long long)interest_retransmissions_[segment & mask]); + LOG(ERROR) << "Stop: received not trusted packet " + << interest_retransmissions_[segment & mask] << " times"; onContentReassembled( make_error_code(protocol_error::max_retransmissions_error)); } @@ -399,23 +383,25 @@ void RaaqmTransportProtocol::onReassemblyFailed(std::uint32_t missing_segment) { } -void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { - checkForStalePaths(); - - const Name &n = interest->getName(); - - TRANSPORT_LOGW("Timeout on content %s", n.toString().c_str()); +void RaaqmTransportProtocol::sendInterest( + const Name &interest_name, + std::array *additional_suffixes, + uint32_t len) { + interests_in_flight_++; + interest_retransmissions_[interest_name.getSuffix() & mask]++; + TransportProtocol::sendInterest(interest_name, additional_suffixes, len); +} - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { - return; - } +void RaaqmTransportProtocol::onInterestTimeout(Interest::Ptr &interest, + const Name &n) { + checkForStalePaths(); interests_in_flight_--; uint64_t segment = n.getSuffix(); // Do not retransmit interests asking contents that do not exist. - if (segment > index_manager_->getFinalSuffix()) { + if (segment > indexer_verifier_->getFinalSuffix()) { return; } @@ -436,32 +422,34 @@ void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { (*on_interest_retransmission_)(*socket_->getInterface(), *interest); } - if (!is_running_) { + if (!isRunning()) { return; } - interest_retransmissions_[segment & mask]++; - interest_to_retransmit_.push(std::move(interest)); - + interest_to_retransmit_.push(segment); scheduleNextInterests(); } else { - TRANSPORT_LOGE("Stop: reached max retx limit."); + LOG(ERROR) << "Stop: reached max retx limit."; onContentReassembled(std::make_error_code(std::errc(std::errc::io_error))); } } void RaaqmTransportProtocol::scheduleNextInterests() { - bool cancel = (!is_running_ && !is_first_) || !schedule_interests_; + bool cancel = (!isRunning() && !is_first_) || !schedule_interests_; if (TRANSPORT_EXPECT_FALSE(cancel)) { schedule_interests_ = true; return; } + core::Name *name; + socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name); + if (TRANSPORT_EXPECT_FALSE(interests_in_flight_ >= current_window_size_ && interest_to_retransmit_.size() > 0)) { // send at least one interest if there are retransmissions to perform and // there is no space left in the window - sendInterest(std::move(interest_to_retransmit_.front())); + auto suffix = interest_to_retransmit_.front(); + sendInterest(name->setSuffix(suffix)); interest_to_retransmit_.pop(); } @@ -470,55 +458,25 @@ void RaaqmTransportProtocol::scheduleNextInterests() { // Send the interest needed for filling the window while (interests_in_flight_ < current_window_size_) { if (interest_to_retransmit_.size() > 0) { - sendInterest(std::move(interest_to_retransmit_.front())); + auto suffix = interest_to_retransmit_.front(); + sendInterest(name->setSuffix(suffix)); interest_to_retransmit_.pop(); } else { - if (TRANSPORT_EXPECT_FALSE(!is_running_ && !is_first_)) { - TRANSPORT_LOGI("Adios"); + if (TRANSPORT_EXPECT_FALSE(!isRunning() && !is_first_)) { break; } - index = index_manager_->getNextSuffix(); + index = indexer_verifier_->getNextSuffix(); if (index == IndexManager::invalid_index) { break; } - sendInterest(index); + interest_retransmissions_[index & mask] = ~0; + sendInterest(name->setSuffix(index)); } } } -void RaaqmTransportProtocol::sendInterest(std::uint64_t next_suffix) { - auto interest = core::PacketManager<>::getInstance().getPacket(); - core::Name *name; - socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name); - name->setSuffix((uint32_t)next_suffix); - interest->setName(*name); - - uint32_t interest_lifetime; - socket_->getSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, - interest_lifetime); - interest->setLifetime(interest_lifetime); - - if (*on_interest_output_) { - on_interest_output_->operator()(*socket_->getInterface(), *interest); - } - // This is set to ~0 so that the next interest_retransmissions_ + 1, - // performed by sendInterest, will result in 0 - interest_retransmissions_[next_suffix & mask] = ~0; - interest_timepoints_[next_suffix & mask] = utils::SteadyClock::now(); - - sendInterest(std::move(interest)); -} - -void RaaqmTransportProtocol::sendInterest(Interest::Ptr &&interest) { - interests_in_flight_++; - interest_retransmissions_[interest->getName().getSuffix() & mask]++; - - TRANSPORT_LOGD("Send interest %s", interest->getName().toString().c_str()); - portal_->sendInterest(std::move(interest)); -} - void RaaqmTransportProtocol::onContentReassembled(std::error_code ec) { rate_estimator_->onDownloadFinished(); TransportProtocol::onContentReassembled(ec); -- cgit 1.2.3-korg