From 755c6833ae2d2eee87e80ed3b84c75e968f48c46 Mon Sep 17 00:00:00 2001 From: Alberto Compagno Date: Tue, 15 Oct 2019 18:08:41 +0200 Subject: [HICN-328] Reworking setSocketOption and getSocketOption to be thread-safe Change-Id: Ie22572822b9ac1e6c300fd7982035c799546bd76 Signed-off-by: Alberto Compagno --- .../src/hicn/transport/interfaces/CMakeLists.txt | 1 + .../src/hicn/transport/interfaces/callbacks.cc | 26 + .../src/hicn/transport/interfaces/callbacks.h | 4 + .../transport/interfaces/rtc_socket_producer.cc | 158 ++--- .../src/hicn/transport/interfaces/socket.h | 2 - .../hicn/transport/interfaces/socket_consumer.cc | 717 ++++++++++++++++++++- .../hicn/transport/interfaces/socket_consumer.h | 673 +++---------------- .../interfaces/socket_options_default_values.h | 2 - .../transport/interfaces/socket_options_keys.h | 31 +- .../hicn/transport/interfaces/socket_producer.cc | 659 +++++++++++++++++-- .../hicn/transport/interfaces/socket_producer.h | 508 ++------------- .../src/hicn/transport/protocols/protocol.cc | 15 +- .../src/hicn/transport/protocols/protocol.h | 6 +- libtransport/src/hicn/transport/protocols/raaqm.cc | 40 +- libtransport/src/hicn/transport/protocols/rtc.cc | 266 ++++---- .../src/hicn/transport/utils/content_store.cc | 7 +- .../src/hicn/transport/utils/content_store.h | 5 +- 17 files changed, 1755 insertions(+), 1365 deletions(-) create mode 100644 libtransport/src/hicn/transport/interfaces/callbacks.cc (limited to 'libtransport/src/hicn') diff --git a/libtransport/src/hicn/transport/interfaces/CMakeLists.txt b/libtransport/src/hicn/transport/interfaces/CMakeLists.txt index a5cca78a6..0c2c73623 100644 --- a/libtransport/src/hicn/transport/interfaces/CMakeLists.txt +++ b/libtransport/src/hicn/transport/interfaces/CMakeLists.txt @@ -28,6 +28,7 @@ list(APPEND SOURCE_FILES ${CMAKE_CURRENT_SOURCE_DIR}/rtc_socket_producer.cc ${CMAKE_CURRENT_SOURCE_DIR}/socket_producer.cc ${CMAKE_CURRENT_SOURCE_DIR}/socket_consumer.cc + ${CMAKE_CURRENT_SOURCE_DIR}/callbacks.cc ) set(SOURCE_FILES ${SOURCE_FILES} PARENT_SCOPE) diff --git a/libtransport/src/hicn/transport/interfaces/callbacks.cc b/libtransport/src/hicn/transport/interfaces/callbacks.cc new file mode 100644 index 000000000..af470898c --- /dev/null +++ b/libtransport/src/hicn/transport/interfaces/callbacks.cc @@ -0,0 +1,26 @@ +/* + * Copyright (c) 2017-2019 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "callbacks.h" + +namespace transport { + +namespace interface { + +nullptr_t VOID_HANDLER = nullptr; + +} // namespace interface + +} // namespace transport \ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/callbacks.h b/libtransport/src/hicn/transport/interfaces/callbacks.h index 24f47eb75..41b6213fe 100644 --- a/libtransport/src/hicn/transport/interfaces/callbacks.h +++ b/libtransport/src/hicn/transport/interfaces/callbacks.h @@ -18,6 +18,8 @@ #include #include +#include + namespace utils { class MemBuf; } @@ -105,6 +107,8 @@ using ProducerContentObjectCallback = using ProducerInterestCallback = std::function; +extern nullptr_t VOID_HANDLER; + } // namespace interface } // namespace transport \ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc index e7dc98868..c1a45ebb7 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc @@ -13,6 +13,7 @@ * limitations under the License. */ +#include #include #include #include @@ -31,9 +32,10 @@ #define HICN_MAX_DATA_SEQ 0xefffffff -//slow production rate param -#define MIN_PRODUCTION_RATE 10 // in pacekts per sec. this value is computed - // through experiments +// slow production rate param +#define MIN_PRODUCTION_RATE \ + 10 // in pacekts per sec. this value is computed + // through experiments #define LIFETIME_FRACTION 0.5 // NACK HEADER @@ -63,13 +65,12 @@ RTCProducerSocket::RTCProducerSocket(asio::io_service &io_service) bytesProductionRate_(INIT_PACKET_PRODUCTION_RATE * 1400), packetsProductionRate_(INIT_PACKET_PRODUCTION_RATE), perSecondFactor_(MILLI_IN_A_SEC / STATS_INTERVAL_DURATION), - timer_on_(false){ + timer_on_(false) { srand((unsigned int)time(NULL)); prodLabel_ = ((rand() % 255) << 24UL); - interests_cache_timer_ = std::make_unique( - this->getIoService()); - round_timer_ = std::make_unique( - this->getIoService()); + interests_cache_timer_ = + std::make_unique(this->getIoService()); + round_timer_ = std::make_unique(this->getIoService()); setSocketOption(GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 10000U); scheduleRoundTimer(); } @@ -82,13 +83,12 @@ RTCProducerSocket::RTCProducerSocket() bytesProductionRate_(INIT_PACKET_PRODUCTION_RATE * 1400), packetsProductionRate_(INIT_PACKET_PRODUCTION_RATE), perSecondFactor_(MILLI_IN_A_SEC / STATS_INTERVAL_DURATION), - timer_on_(false){ + timer_on_(false) { srand((unsigned int)time(NULL)); prodLabel_ = ((rand() % 255) << 24UL); - interests_cache_timer_ = std::make_unique( - this->getIoService()); - round_timer_ = std::make_unique( - this->getIoService()); + interests_cache_timer_ = + std::make_unique(this->getIoService()); + round_timer_ = std::make_unique(this->getIoService()); setSocketOption(GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 10000U); scheduleRoundTimer(); } @@ -113,13 +113,13 @@ void RTCProducerSocket::registerPrefix(const Prefix &producer_namespace) { } } -void RTCProducerSocket::scheduleRoundTimer(){ - round_timer_->expires_from_now( - std::chrono::milliseconds(STATS_INTERVAL_DURATION)); - round_timer_->async_wait([this](std::error_code ec) { - if (ec) return; - updateStats(); - }); +void RTCProducerSocket::scheduleRoundTimer() { + round_timer_->expires_from_now( + std::chrono::milliseconds(STATS_INTERVAL_DURATION)); + round_timer_->async_wait([this](std::error_code ec) { + if (ec) return; + updateStats(); + }); } void RTCProducerSocket::updateStats() { @@ -150,8 +150,8 @@ void RTCProducerSocket::produce(std::unique_ptr &&buffer) { producedBytes_ += (uint32_t)(buffer_size + headerSize_ + TIMESTAMP_LEN); producedPackets_++; - auto content_object = std::make_shared( - flowName_.setSuffix(currentSeg_.load())); + auto content_object = + std::make_shared(flowName_.setSuffix(currentSeg_.load())); auto payload = utils::MemBuf::create(TIMESTAMP_LEN); memcpy(payload->writableData(), &now, TIMESTAMP_LEN); @@ -166,27 +166,26 @@ void RTCProducerSocket::produce(std::unique_ptr &&buffer) { output_buffer_.insert(std::static_pointer_cast( content_object->shared_from_this())); - if (on_content_object_in_output_buffer_ != VOID_HANDLER) { + if (on_content_object_in_output_buffer_) { on_content_object_in_output_buffer_(*this, *content_object); } portal_->sendContentObject(*content_object); - if (on_content_object_output_ != VOID_HANDLER) { + if (on_content_object_output_) { on_content_object_output_(*this, *content_object); } uint32_t old_curr = currentSeg_.load(); currentSeg_ = (currentSeg_.load() + 1) % HICN_MAX_DATA_SEQ; - //remove interests from the interest cache if it exists - //this generates nacks that will tell to the consumer - //that a new data packet was produced - if(!seqs_map_.empty()){ + // remove interests from the interest cache if it exists + // this generates nacks that will tell to the consumer + // that a new data packet was produced + if (!seqs_map_.empty()) { utils::SpinLock::Acquire locked(interests_cache_lock_); - for(auto it = seqs_map_.begin(); it != seqs_map_.end(); it++){ - if(it->first != old_curr) - sendNack(it->first); + for (auto it = seqs_map_.begin(); it != seqs_map_.end(); it++) { + if (it->first != old_curr) sendNack(it->first); } seqs_map_.clear(); timers_map_.clear(); @@ -197,15 +196,15 @@ void RTCProducerSocket::onInterest(Interest::Ptr &&interest) { uint32_t interestSeg = interest->getName().getSuffix(); uint32_t lifetime = interest->getLifetime(); - if (on_interest_input_ != VOID_HANDLER) { + if (on_interest_input_) { on_interest_input_(*this, *interest); } uint64_t now = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); - if(interestSeg > HICN_MAX_DATA_SEQ){ + if (interestSeg > HICN_MAX_DATA_SEQ) { sendNack(interestSeg); return; } @@ -214,71 +213,73 @@ void RTCProducerSocket::onInterest(Interest::Ptr &&interest) { output_buffer_.find(*interest); if (content_object) { - if (on_interest_satisfied_output_buffer_ != VOID_HANDLER) { + if (on_interest_satisfied_output_buffer_) { on_interest_satisfied_output_buffer_(*this, *interest); } - if (on_content_object_output_ != VOID_HANDLER) { + if (on_content_object_output_) { on_content_object_output_(*this, *content_object); } portal_->sendContentObject(*content_object); return; } else { - if (on_interest_process_ != VOID_HANDLER) { + if (on_interest_process_) { on_interest_process_(*this, *interest); } } // if the production rate is less than MIN_PRODUCTION_RATE we put the // interest in a queue, otherwise we handle it in the usual way - if(packetsProductionRate_.load() < MIN_PRODUCTION_RATE && - interestSeg >= currentSeg_.load()){ - + if (packetsProductionRate_.load() < MIN_PRODUCTION_RATE && + interestSeg >= currentSeg_.load()) { utils::SpinLock::Acquire locked(interests_cache_lock_); uint64_t next_timer = ~0; - if(!timers_map_.empty()){ + if (!timers_map_.empty()) { next_timer = timers_map_.begin()->first; } uint64_t expiration = now + (lifetime * LIFETIME_FRACTION); - //check if the seq number exists already + // check if the seq number exists already auto it_seqs = seqs_map_.find(interestSeg); - if(it_seqs != seqs_map_.end()){ - //the seq already exists - if(expiration < it_seqs->second){ + if (it_seqs != seqs_map_.end()) { + // the seq already exists + if (expiration < it_seqs->second) { // we need to update the timer becasue we got a smaller one // 1) remove the entry from the multimap // 2) update this entry auto range = timers_map_.equal_range(it_seqs->second); - for(auto it_timers = range.first; it_timers != range.second; it_timers++){ - if(it_timers->second == it_seqs->first){ + for (auto it_timers = range.first; it_timers != range.second; + it_timers++) { + if (it_timers->second == it_seqs->first) { timers_map_.erase(it_timers); break; } } - timers_map_.insert(std::pair(expiration, interestSeg)); + timers_map_.insert( + std::pair(expiration, interestSeg)); it_seqs->second = expiration; - }else{ - //nothing to do here - return; + } else { + // nothing to do here + return; } - }else{ + } else { // add the new seq - timers_map_.insert(std::pair(expiration, interestSeg)); - seqs_map_.insert(std::pair(interestSeg, expiration)); + timers_map_.insert( + std::pair(expiration, interestSeg)); + seqs_map_.insert(std::pair(interestSeg, expiration)); } - //here we have at least one interest in the queue, we need to start or - //update the timer - if(!timer_on_){ - //set timeout + // here we have at least one interest in the queue, we need to start or + // update the timer + if (!timer_on_) { + // set timeout timer_on_ = true; scheduleCacheTimer(timers_map_.begin()->first - now); } else { - //re-schedule the timer because a new interest will expires sooner - if(next_timer > timers_map_.begin()->first){ + // re-schedule the timer because a new interest will expires sooner + if (next_timer > timers_map_.begin()->first) { interests_cache_timer_->cancel(); scheduleCacheTimer(timers_map_.begin()->first - now); } @@ -292,44 +293,43 @@ void RTCProducerSocket::onInterest(Interest::Ptr &&interest) { (double)packetsProductionRate_.load())); if (interestSeg < currentSeg_.load() || - interestSeg > (max_gap + currentSeg_.load())) { + interestSeg > (max_gap + currentSeg_.load())) { sendNack(interestSeg); } // else drop packet } -void RTCProducerSocket::scheduleCacheTimer(uint64_t wait){ - interests_cache_timer_->expires_from_now( - std::chrono::milliseconds(wait)); +void RTCProducerSocket::scheduleCacheTimer(uint64_t wait) { + interests_cache_timer_->expires_from_now(std::chrono::milliseconds(wait)); interests_cache_timer_->async_wait([this](std::error_code ec) { - if (ec) return; - interestCacheTimer(); - }); + if (ec) return; + interestCacheTimer(); + }); } -void RTCProducerSocket::interestCacheTimer(){ +void RTCProducerSocket::interestCacheTimer() { uint64_t now = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); utils::SpinLock::Acquire locked(interests_cache_lock_); - for(auto it_timers = timers_map_.begin(); it_timers != timers_map_.end();){ + for (auto it_timers = timers_map_.begin(); it_timers != timers_map_.end();) { uint64_t expire = it_timers->first; - if(expire <= now){ + if (expire <= now) { uint32_t seq = it_timers->second; sendNack(seq); - //remove the interest from the other map + // remove the interest from the other map seqs_map_.erase(seq); it_timers = timers_map_.erase(it_timers); - }else{ - //stop, we are done! + } else { + // stop, we are done! break; } } - if(timers_map_.empty()){ + if (timers_map_.empty()) { timer_on_ = false; - }else{ + } else { timer_on_ = true; scheduleCacheTimer(timers_map_.begin()->first - now); } @@ -351,7 +351,7 @@ void RTCProducerSocket::sendNack(uint32_t sequence) { nack.setLifetime(0); nack.setPathLabel(prodLabel_); - if (on_content_object_output_ != VOID_HANDLER) { + if (on_content_object_output_) { on_content_object_output_(*this, nack); } diff --git a/libtransport/src/hicn/transport/interfaces/socket.h b/libtransport/src/hicn/transport/interfaces/socket.h index 90f6a3ef6..f0194880a 100644 --- a/libtransport/src/hicn/transport/interfaces/socket.h +++ b/libtransport/src/hicn/transport/interfaces/socket.h @@ -27,8 +27,6 @@ #define SOCKET_OPTION_NOT_SET 3 #define SOCKET_OPTION_DEFAULT 12345 -#define VOID_HANDLER 0 - namespace transport { namespace interface { diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc index e1afd2161..6eae23c85 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc @@ -45,7 +45,6 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) rate_estimation_alpha_(default_values::rate_alpha), rate_estimation_observer_(nullptr), rate_estimation_choice_(0), - is_async_(false), verifier_(std::make_shared()), verify_signature_(false), on_interest_output_(VOID_HANDLER), @@ -58,8 +57,8 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) stats_summary_(VOID_HANDLER), read_callback_(nullptr), virtual_download_(false), - rtt_stats_(false), - timer_interval_milliseconds_(0) { + timer_interval_milliseconds_(0), + guard_raaqm_params_() { switch (protocol) { case TransportProtocolAlgorithms::CBR: transport_protocol_ = std::make_unique(this); @@ -88,7 +87,6 @@ int ConsumerSocket::consume(const Name &name) { network_name_ = name; network_name_.setSuffix(0); - is_async_ = false; transport_protocol_->start(); @@ -100,7 +98,6 @@ int ConsumerSocket::asyncConsume(const Name &name) { async_downloader_.add([this, name]() { network_name_ = std::move(name); network_name_.setSuffix(0); - is_async_ = true; transport_protocol_->start(); }); } @@ -108,20 +105,6 @@ int ConsumerSocket::asyncConsume(const Name &name) { return CONSUMER_RUNNING; } -void ConsumerSocket::asyncSendInterest(Interest::Ptr &&interest, - Portal::ConsumerCallback *callback) { - if (!async_downloader_.stopped()) { - // TODO Workaround, to be fixed! - auto i = interest.release(); - async_downloader_.add([this, i, callback]() mutable { - Interest::Ptr _interest(i); - portal_->setConsumerCallback(callback); - portal_->sendInterest(std::move(_interest)); - portal_->runEventsLoop(); - }); - } -} - void ConsumerSocket::stop() { if (transport_protocol_->isRunning()) { transport_protocol_->stop(); @@ -138,6 +121,702 @@ asio::io_service &ConsumerSocket::getIoService() { return portal_->getIoService(); } +// If the thread calling lambda_func is not the same of io_service, this +// function reschedule the function on it +template +int ConsumerSocket::rescheduleOnIOService(int socket_option_key, + arg2 socket_option_value, + Lambda lambda_func) { + // To enforce type check + std::function func = lambda_func; + int result = SOCKET_OPTION_SET; + if (transport_protocol_->isRunning()) { + std::mutex mtx; + /* Condition variable for the wait */ + std::condition_variable cv; + bool done = false; + io_service_.dispatch([this, &socket_option_key, &socket_option_value, &mtx, + &cv, &result, &done, &func]() { + std::unique_lock lck(mtx); + done = true; + result = func(socket_option_key, socket_option_value); + }); + std::unique_lock lck(mtx); + if (!done) { + cv.wait(lck); + } + } else { + result = func(socket_option_key, socket_option_value); + } + + return result; +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + ReadCallback *socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, ReadCallback *socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::READ_CALLBACK: + read_callback_ = socket_option_value; + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + ReadCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, ReadCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::READ_CALLBACK: + *socket_option_value = read_callback_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + double socket_option_value) { + utils::SpinLock::Acquire locked(guard_raaqm_params_); + switch (socket_option_key) { + case MIN_WINDOW_SIZE: + min_window_size_ = socket_option_value; + break; + + case MAX_WINDOW_SIZE: + max_window_size_ = socket_option_value; + break; + + case CURRENT_WINDOW_SIZE: + current_window_size_ = socket_option_value; + break; + + case GAMMA_VALUE: + gamma_ = socket_option_value; + break; + + case BETA_VALUE: + beta_ = socket_option_value; + break; + + case DROP_FACTOR: + drop_factor_ = socket_option_value; + break; + + case MINIMUM_DROP_PROBABILITY: + minimum_drop_probability_ = socket_option_value; + break; + + case RATE_ESTIMATION_ALPHA: + if (socket_option_value >= 0 && socket_option_value < 1) { + rate_estimation_alpha_ = socket_option_value; + } else { + rate_estimation_alpha_ = default_values::alpha; + } + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + uint32_t socket_option_value) { + utils::SpinLock::Acquire locked(guard_raaqm_params_); + switch (socket_option_key) { + case GeneralTransportOptions::MAX_INTEREST_RETX: + max_retransmissions_ = socket_option_value; + break; + + case GeneralTransportOptions::INTEREST_LIFETIME: + interest_lifetime_ = socket_option_value; + break; + + case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: + if (socket_option_value > 0) { + rate_estimation_batching_parameter_ = socket_option_value; + } else { + rate_estimation_batching_parameter_ = default_values::batch; + } + break; + + case RateEstimationOptions::RATE_ESTIMATION_CHOICE: + if (socket_option_value > 0) { + rate_estimation_choice_ = socket_option_value; + } else { + rate_estimation_choice_ = default_values::rate_choice; + } + break; + + case GeneralTransportOptions::STATS_INTERVAL: + timer_interval_milliseconds_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + std::nullptr_t socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, std::nullptr_t socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: + if (socket_option_value == VOID_HANDLER) { + on_interest_retransmission_ = VOID_HANDLER; + break; + } + + case ConsumerCallbacksOptions::INTEREST_EXPIRED: + if (socket_option_value == VOID_HANDLER) { + on_interest_timeout_ = VOID_HANDLER; + break; + } + + case ConsumerCallbacksOptions::INTEREST_SATISFIED: + if (socket_option_value == VOID_HANDLER) { + on_interest_satisfied_ = VOID_HANDLER; + break; + } + + case ConsumerCallbacksOptions::INTEREST_OUTPUT: + if (socket_option_value == VOID_HANDLER) { + on_interest_output_ = VOID_HANDLER; + break; + } + + case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: + if (socket_option_value == VOID_HANDLER) { + on_content_object_input_ = VOID_HANDLER; + break; + } + + case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: + if (socket_option_value == VOID_HANDLER) { + on_content_object_verification_ = VOID_HANDLER; + break; + } + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + bool socket_option_value) { + int result = SOCKET_OPTION_NOT_SET; + if (!transport_protocol_->isRunning()) { + switch (socket_option_key) { + case OtherOptions::VIRTUAL_DOWNLOAD: + virtual_download_ = socket_option_value; + result = SOCKET_OPTION_SET; + break; + + case GeneralTransportOptions::VERIFY_SIGNATURE: + verify_signature_ = socket_option_value; + result = SOCKET_OPTION_SET; + break; + + default: + return result; + } + } + return result; +} + +int ConsumerSocket::setSocketOption( + int socket_option_key, ConsumerContentObjectCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerContentObjectCallback socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: + on_content_object_input_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::setSocketOption( + int socket_option_key, + ConsumerContentObjectVerificationCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerContentObjectVerificationCallback socket_option_value) + -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: + on_content_object_verification_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::setSocketOption( + int socket_option_key, ConsumerInterestCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerInterestCallback socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: + on_interest_retransmission_ = socket_option_value; + break; + + case ConsumerCallbacksOptions::INTEREST_OUTPUT: + on_interest_output_ = socket_option_value; + break; + + case ConsumerCallbacksOptions::INTEREST_EXPIRED: + on_interest_timeout_ = socket_option_value; + break; + + case ConsumerCallbacksOptions::INTEREST_SATISFIED: + on_interest_satisfied_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::setSocketOption( + int socket_option_key, ConsumerManifestCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerManifestCallback socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::MANIFEST_INPUT: + on_manifest_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + IcnObserver *socket_option_value) { + utils::SpinLock::Acquire locked(guard_raaqm_params_); + switch (socket_option_key) { + case RateEstimationOptions::RATE_ESTIMATION_OBSERVER: + rate_estimation_observer_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ConsumerSocket::setSocketOption( + int socket_option_key, + const std::shared_ptr &socket_option_value) { + int result = SOCKET_OPTION_NOT_SET; + if (!transport_protocol_->isRunning()) { + switch (socket_option_key) { + case GeneralTransportOptions::VERIFIER: + verifier_ = socket_option_value; + result = SOCKET_OPTION_SET; + break; + default: + return result; + } + } + + return result; +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + const std::string &socket_option_value) { + int result = SOCKET_OPTION_NOT_SET; + if (!transport_protocol_->isRunning()) { + switch (socket_option_key) { + case GeneralTransportOptions::CERTIFICATE: + key_id_ = verifier_->addKeyFromCertificate(socket_option_value); + + if (key_id_ != nullptr) { + result = SOCKET_OPTION_SET; + } + break; + + case DataLinkOptions::OUTPUT_INTERFACE: + output_interface_ = socket_option_value; + portal_->setOutputInterface(output_interface_); + result = SOCKET_OPTION_SET; + break; + + default: + return result; + } + } + return result; +} + +int ConsumerSocket::setSocketOption(int socket_option_key, + ConsumerTimerCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerTimerCallback socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::STATS_SUMMARY: + stats_summary_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + double &socket_option_value) { + utils::SpinLock::Acquire locked(guard_raaqm_params_); + switch (socket_option_key) { + case GeneralTransportOptions::MIN_WINDOW_SIZE: + socket_option_value = min_window_size_; + break; + + case GeneralTransportOptions::MAX_WINDOW_SIZE: + socket_option_value = max_window_size_; + break; + + case GeneralTransportOptions::CURRENT_WINDOW_SIZE: + socket_option_value = current_window_size_; + break; + + // RAAQM parameters + + case RaaqmTransportOptions::GAMMA_VALUE: + socket_option_value = gamma_; + break; + + case RaaqmTransportOptions::BETA_VALUE: + socket_option_value = beta_; + break; + + case RaaqmTransportOptions::DROP_FACTOR: + socket_option_value = drop_factor_; + break; + + case RaaqmTransportOptions::MINIMUM_DROP_PROBABILITY: + socket_option_value = minimum_drop_probability_; + break; + + case RateEstimationOptions::RATE_ESTIMATION_ALPHA: + socket_option_value = rate_estimation_alpha_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + uint32_t &socket_option_value) { + utils::SpinLock::Acquire locked(guard_raaqm_params_); + switch (socket_option_key) { + case GeneralTransportOptions::MAX_INTEREST_RETX: + socket_option_value = max_retransmissions_; + break; + + case GeneralTransportOptions::INTEREST_LIFETIME: + socket_option_value = interest_lifetime_; + break; + + case RaaqmTransportOptions::SAMPLE_NUMBER: + socket_option_value = sample_number_; + break; + + case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: + socket_option_value = rate_estimation_batching_parameter_; + break; + + case RateEstimationOptions::RATE_ESTIMATION_CHOICE: + socket_option_value = rate_estimation_choice_; + break; + + case GeneralTransportOptions::STATS_INTERVAL: + socket_option_value = timer_interval_milliseconds_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + bool &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::RUNNING: + socket_option_value = transport_protocol_->isRunning(); + break; + + case OtherOptions::VIRTUAL_DOWNLOAD: + socket_option_value = virtual_download_; + break; + + case GeneralTransportOptions::VERIFY_SIGNATURE: + socket_option_value = verify_signature_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + Name **socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::NETWORK_NAME: + *socket_option_value = &network_name_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, + ConsumerContentObjectCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerContentObjectCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: + *socket_option_value = &on_content_object_input_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, + ConsumerContentObjectVerificationCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerContentObjectVerificationCallback **socket_option_value) + -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: + *socket_option_value = &on_content_object_verification_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, ConsumerInterestCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerInterestCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: + *socket_option_value = &on_interest_retransmission_; + break; + + case ConsumerCallbacksOptions::INTEREST_OUTPUT: + *socket_option_value = &on_interest_output_; + break; + + case ConsumerCallbacksOptions::INTEREST_EXPIRED: + *socket_option_value = &on_interest_timeout_; + break; + + case ConsumerCallbacksOptions::INTEREST_SATISFIED: + *socket_option_value = &on_interest_satisfied_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, ConsumerManifestCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerManifestCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::MANIFEST_INPUT: + *socket_option_value = &on_manifest_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, std::shared_ptr &socket_option_value) { + switch (socket_option_key) { + case PORTAL: + socket_option_value = portal_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + IcnObserver **socket_option_value) { + utils::SpinLock::Acquire locked(guard_raaqm_params_); + switch (socket_option_key) { + case RateEstimationOptions::RATE_ESTIMATION_OBSERVER: + *socket_option_value = (rate_estimation_observer_); + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, + std::shared_ptr &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::VERIFIER: + socket_option_value = verifier_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption(int socket_option_key, + std::string &socket_option_value) { + switch (socket_option_key) { + case DataLinkOptions::OUTPUT_INTERFACE: + socket_option_value = output_interface_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ConsumerSocket::getSocketOption( + int socket_option_key, ConsumerTimerCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ConsumerTimerCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ConsumerCallbacksOptions::STATS_SUMMARY: + *socket_option_value = &stats_summary_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + } // namespace interface } // end namespace transport \ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h index 8f7a9718c..e3620b269 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h @@ -178,20 +178,6 @@ class ConsumerSocket : public BaseSocket { int consume(const Name &name); int asyncConsume(const Name &name); - /** - * Send an interest asynchronously in another thread, which is the same used - * for asyncConsume. - * - * @param interest - An Interest::Ptr to the interest. Notice that the - * application looses the ownership of the interest, which is transferred to - * the library itself. - * @param callback - A ConsumerCallback containing the events to be trigger in - * case of timeout or content reception. - * - */ - void asyncSendInterest(Interest::Ptr &&interest, - Portal::ConsumerCallback *callback); - /** * Stops the consumer socket. If several downloads are queued (using * asyncConsume), this call stops just the current one. @@ -211,595 +197,94 @@ class ConsumerSocket : public BaseSocket { */ asio::io_service &getIoService() override; - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ReadCallback *socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::READ_CALLBACK: - read_callback_ = socket_option_value; - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ReadCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::READ_CALLBACK: - *socket_option_value = read_callback_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - double socket_option_value) { - switch (socket_option_key) { - case MIN_WINDOW_SIZE: - min_window_size_ = socket_option_value; - break; - - case MAX_WINDOW_SIZE: - max_window_size_ = socket_option_value; - break; - - case CURRENT_WINDOW_SIZE: - current_window_size_ = socket_option_value; - break; - - case GAMMA_VALUE: - gamma_ = socket_option_value; - break; - - case BETA_VALUE: - beta_ = socket_option_value; - break; - - case DROP_FACTOR: - drop_factor_ = socket_option_value; - break; - - case MINIMUM_DROP_PROBABILITY: - minimum_drop_probability_ = socket_option_value; - break; - - case RATE_ESTIMATION_ALPHA: - if (socket_option_value >= 0 && socket_option_value < 1) { - rate_estimation_alpha_ = socket_option_value; - } else { - rate_estimation_alpha_ = default_values::alpha; - } - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - uint32_t socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::INPUT_BUFFER_SIZE: - input_buffer_size_ = socket_option_value; - break; - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - output_buffer_size_ = socket_option_value; - break; - - case GeneralTransportOptions::MAX_INTEREST_RETX: - max_retransmissions_ = socket_option_value; - break; - - case GeneralTransportOptions::INTEREST_LIFETIME: - interest_lifetime_ = socket_option_value; - break; - - case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: - if (socket_option_value == VOID_HANDLER) { - on_interest_retransmission_ = VOID_HANDLER; - break; - } - - case ConsumerCallbacksOptions::INTEREST_EXPIRED: - if (socket_option_value == VOID_HANDLER) { - on_interest_timeout_ = VOID_HANDLER; - break; - } - - case ConsumerCallbacksOptions::INTEREST_SATISFIED: - if (socket_option_value == VOID_HANDLER) { - on_interest_satisfied_ = VOID_HANDLER; - break; - } - - case ConsumerCallbacksOptions::INTEREST_OUTPUT: - if (socket_option_value == VOID_HANDLER) { - on_interest_output_ = VOID_HANDLER; - break; - } - - case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: - if (socket_option_value == VOID_HANDLER) { - on_content_object_input_ = VOID_HANDLER; - break; - } - - case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: - if (socket_option_value == VOID_HANDLER) { - on_content_object_verification_ = VOID_HANDLER; - break; - } - - case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: - if (socket_option_value > 0) { - rate_estimation_batching_parameter_ = socket_option_value; - } else { - rate_estimation_batching_parameter_ = default_values::batch; - } - break; - - case RateEstimationOptions::RATE_ESTIMATION_CHOICE: - if (socket_option_value > 0) { - rate_estimation_choice_ = socket_option_value; - } else { - rate_estimation_choice_ = default_values::rate_choice; - } - break; - - case GeneralTransportOptions::STATS_INTERVAL: - timer_interval_milliseconds_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - bool socket_option_value) { - switch (socket_option_key) { - case OtherOptions::VIRTUAL_DOWNLOAD: - virtual_download_ = socket_option_value; - break; - - case RaaqmTransportOptions::RTT_STATS: - rtt_stats_ = socket_option_value; - break; - - case GeneralTransportOptions::VERIFY_SIGNATURE: - verify_signature_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - Name *socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::NETWORK_NAME: - network_name_ = *socket_option_value; - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, - ConsumerContentObjectCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: - on_content_object_input_ = socket_option_value; - break; + virtual int setSocketOption(int socket_option_key, + ReadCallback *socket_option_value); + + virtual int getSocketOption(int socket_option_key, + ReadCallback **socket_option_value); - default: - return SOCKET_OPTION_NOT_SET; - } + virtual int setSocketOption(int socket_option_key, + double socket_option_value); - return SOCKET_OPTION_SET; - } + virtual int setSocketOption(int socket_option_key, + uint32_t socket_option_value); - TRANSPORT_ALWAYS_INLINE int setSocketOption( + virtual int setSocketOption(int socket_option_key, + std::nullptr_t socket_option_value); + + virtual int setSocketOption(int socket_option_key, bool socket_option_value); + + virtual int setSocketOption( + int socket_option_key, ConsumerContentObjectCallback socket_option_value); + + virtual int setSocketOption( int socket_option_key, - ConsumerContentObjectVerificationCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: - on_content_object_verification_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ConsumerInterestCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: - on_interest_retransmission_ = socket_option_value; - break; - - case ConsumerCallbacksOptions::INTEREST_OUTPUT: - on_interest_output_ = socket_option_value; - break; - - case ConsumerCallbacksOptions::INTEREST_EXPIRED: - on_interest_timeout_ = socket_option_value; - break; - - case ConsumerCallbacksOptions::INTEREST_SATISFIED: - on_interest_satisfied_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ConsumerManifestCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::MANIFEST_INPUT: - on_manifest_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, IcnObserver *socket_option_value) { - switch (socket_option_key) { - case RateEstimationOptions::RATE_ESTIMATION_OBSERVER: - rate_estimation_observer_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( + ConsumerContentObjectVerificationCallback socket_option_value); + + virtual int setSocketOption(int socket_option_key, + ConsumerInterestCallback socket_option_value); + + virtual int setSocketOption(int socket_option_key, + ConsumerManifestCallback socket_option_value); + + virtual int setSocketOption(int socket_option_key, + IcnObserver *socket_option_value); + + virtual int setSocketOption( int socket_option_key, - const std::shared_ptr &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::VERIFIER: - verifier_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, const std::string &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::CERTIFICATE: - key_id_ = verifier_->addKeyFromCertificate(socket_option_value); - - if (key_id_ != nullptr) { - break; - } - - case DataLinkOptions::OUTPUT_INTERFACE: - output_interface_ = socket_option_value; - portal_->setOutputInterface(output_interface_); - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ConsumerTimerCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::STATS_SUMMARY: - stats_summary_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption(int socket_option_key, - double &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::MIN_WINDOW_SIZE: - socket_option_value = min_window_size_; - break; - - case GeneralTransportOptions::MAX_WINDOW_SIZE: - socket_option_value = max_window_size_; - break; - - case GeneralTransportOptions::CURRENT_WINDOW_SIZE: - socket_option_value = current_window_size_; - break; - - // RAAQM parameters - - case RaaqmTransportOptions::GAMMA_VALUE: - socket_option_value = gamma_; - break; - - case RaaqmTransportOptions::BETA_VALUE: - socket_option_value = beta_; - break; - - case RaaqmTransportOptions::DROP_FACTOR: - socket_option_value = drop_factor_; - break; - - case RaaqmTransportOptions::MINIMUM_DROP_PROBABILITY: - socket_option_value = minimum_drop_probability_; - break; - - case RateEstimationOptions::RATE_ESTIMATION_ALPHA: - socket_option_value = rate_estimation_alpha_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption(int socket_option_key, - uint32_t &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::INPUT_BUFFER_SIZE: - socket_option_value = (uint32_t)input_buffer_size_; - break; - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - socket_option_value = (uint32_t)output_buffer_size_; - break; - - case GeneralTransportOptions::MAX_INTEREST_RETX: - socket_option_value = max_retransmissions_; - break; - - case GeneralTransportOptions::INTEREST_LIFETIME: - socket_option_value = interest_lifetime_; - break; - - case RaaqmTransportOptions::SAMPLE_NUMBER: - socket_option_value = sample_number_; - break; - - case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: - socket_option_value = rate_estimation_batching_parameter_; - break; - - case RateEstimationOptions::RATE_ESTIMATION_CHOICE: - socket_option_value = rate_estimation_choice_; - break; - - case GeneralTransportOptions::STATS_INTERVAL: - socket_option_value = timer_interval_milliseconds_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption(int socket_option_key, - bool &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::ASYNC_MODE: - socket_option_value = is_async_; - break; + const std::shared_ptr &socket_option_value); - case GeneralTransportOptions::RUNNING: - socket_option_value = transport_protocol_->isRunning(); - break; + virtual int setSocketOption(int socket_option_key, + const std::string &socket_option_value); - case OtherOptions::VIRTUAL_DOWNLOAD: - socket_option_value = virtual_download_; - break; - - case RaaqmTransportOptions::RTT_STATS: - socket_option_value = rtt_stats_; - break; + virtual int setSocketOption(int socket_option_key, + ConsumerTimerCallback socket_option_value); - case GeneralTransportOptions::VERIFY_SIGNATURE: - socket_option_value = verify_signature_; - break; + virtual int getSocketOption(int socket_option_key, + double &socket_option_value); - default: - return SOCKET_OPTION_NOT_GET; - } + virtual int getSocketOption(int socket_option_key, + uint32_t &socket_option_value); - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption(int socket_option_key, - Name **socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::NETWORK_NAME: - *socket_option_value = &network_name_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, - ConsumerContentObjectCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: - *socket_option_value = &on_content_object_input_; - break; + virtual int getSocketOption(int socket_option_key, bool &socket_option_value); - default: - return SOCKET_OPTION_NOT_GET; - } + virtual int getSocketOption(int socket_option_key, + Name **socket_option_value); - return SOCKET_OPTION_GET; - } + virtual int getSocketOption( + int socket_option_key, + ConsumerContentObjectCallback **socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption( + virtual int getSocketOption( int socket_option_key, - ConsumerContentObjectVerificationCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: - *socket_option_value = &on_content_object_verification_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ConsumerInterestCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: - *socket_option_value = &on_interest_retransmission_; - break; - - case ConsumerCallbacksOptions::INTEREST_OUTPUT: - *socket_option_value = &on_interest_output_; - break; - - case ConsumerCallbacksOptions::INTEREST_EXPIRED: - *socket_option_value = &on_interest_timeout_; - break; - - case ConsumerCallbacksOptions::INTEREST_SATISFIED: - *socket_option_value = &on_interest_satisfied_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ConsumerManifestCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::MANIFEST_INPUT: - *socket_option_value = &on_manifest_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, std::shared_ptr &socket_option_value) { - switch (socket_option_key) { - case PORTAL: - socket_option_value = portal_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, IcnObserver **socket_option_value) { - switch (socket_option_key) { - case RateEstimationOptions::RATE_ESTIMATION_OBSERVER: - *socket_option_value = (rate_estimation_observer_); - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( + ConsumerContentObjectVerificationCallback **socket_option_value); + + virtual int getSocketOption(int socket_option_key, + ConsumerInterestCallback **socket_option_value); + + virtual int getSocketOption(int socket_option_key, + ConsumerManifestCallback **socket_option_value); + + virtual int getSocketOption(int socket_option_key, + std::shared_ptr &socket_option_value); + + virtual int getSocketOption(int socket_option_key, + IcnObserver **socket_option_value); + + virtual int getSocketOption( int socket_option_key, - std::shared_ptr &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::VERIFIER: - socket_option_value = verifier_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, std::string &socket_option_value) { - switch (socket_option_key) { - case DataLinkOptions::OUTPUT_INTERFACE: - socket_option_value = output_interface_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ConsumerTimerCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::STATS_SUMMARY: - *socket_option_value = &stats_summary_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } + std::shared_ptr &socket_option_value); + + virtual int getSocketOption(int socket_option_key, + std::string &socket_option_value); + + virtual int getSocketOption(int socket_option_key, + ConsumerTimerCallback **socket_option_value); + + protected: + template + int rescheduleOnIOService(int socket_option_key, arg2 socket_option_value, + Lambda lambda_func); private: asio::io_service internal_io_service_; @@ -808,6 +293,9 @@ class ConsumerSocket : public BaseSocket { std::shared_ptr portal_; utils::EventThread async_downloader_; + // No need to protect from multiple accesses in the async consumer + // The parameter is accessible only with a getSocketOption and + // set from the consume Name network_name_; int interest_lifetime_; @@ -816,8 +304,6 @@ class ConsumerSocket : public BaseSocket { double max_window_size_; double current_window_size_; uint32_t max_retransmissions_; - size_t output_buffer_size_; - size_t input_buffer_size_; // RAAQM Parameters double minimum_drop_probability_; @@ -832,12 +318,10 @@ class ConsumerSocket : public BaseSocket { int rate_estimation_batching_parameter_; int rate_estimation_choice_; - bool is_async_; - // Verification parameters std::shared_ptr verifier_; PARCKeyId *key_id_; - bool verify_signature_; + std::atomic_bool verify_signature_; ConsumerInterestCallback on_interest_retransmission_; ConsumerInterestCallback on_interest_output_; @@ -853,12 +337,13 @@ class ConsumerSocket : public BaseSocket { // Virtual download for traffic generator bool virtual_download_; - bool rtt_stats_; uint32_t timer_interval_milliseconds_; // Transport protocol std::unique_ptr transport_protocol_; + + utils::SpinLock guard_raaqm_params_; }; } // namespace interface diff --git a/libtransport/src/hicn/transport/interfaces/socket_options_default_values.h b/libtransport/src/hicn/transport/interfaces/socket_options_default_values.h index 13029e83a..bcf103b8c 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_options_default_values.h +++ b/libtransport/src/hicn/transport/interfaces/socket_options_default_values.h @@ -31,8 +31,6 @@ static constexpr uint32_t content_object_expiry_time = never_expire_time; // milliseconds -> 50 seconds static constexpr uint32_t content_object_packet_size = 1500; // The ethernet MTU -static constexpr uint32_t producer_socket_input_buffer_size = - 150000; // Interests static constexpr uint32_t producer_socket_output_buffer_size = 150000; // Content Object static constexpr uint32_t log_2_default_buffer_size = 12; diff --git a/libtransport/src/hicn/transport/interfaces/socket_options_keys.h b/libtransport/src/hicn/transport/interfaces/socket_options_keys.h index c21108186..e14f0f412 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_options_keys.h +++ b/libtransport/src/hicn/transport/interfaces/socket_options_keys.h @@ -35,22 +35,21 @@ typedef enum { INTEREST_LIFETIME = 107, CONTENT_OBJECT_EXPIRY_TIME = 108, KEY_LOCATOR = 110, - SIGNATURE_TYPE = 111, - MIN_WINDOW_SIZE = 112, - MAX_WINDOW_SIZE = 113, - CURRENT_WINDOW_SIZE = 114, - ASYNC_MODE = 115, - MAKE_MANIFEST = 116, - PORTAL = 117, - RUNNING = 118, - APPLICATION_BUFFER = 119, - HASH_ALGORITHM = 120, - CRYPTO_SUITE = 121, - IDENTITY = 122, - VERIFIER = 123, - CERTIFICATE = 124, - VERIFY_SIGNATURE = 125, - STATS_INTERVAL = 126 + MIN_WINDOW_SIZE = 111, + MAX_WINDOW_SIZE = 112, + CURRENT_WINDOW_SIZE = 113, + ASYNC_MODE = 114, + MAKE_MANIFEST = 115, + PORTAL = 116, + RUNNING = 117, + APPLICATION_BUFFER = 118, + HASH_ALGORITHM = 119, + CRYPTO_SUITE = 120, + IDENTITY = 121, + VERIFIER = 122, + CERTIFICATE = 123, + VERIFY_SIGNATURE = 124, + STATS_INTERVAL = 125 } GeneralTransportOptions; typedef enum { diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc index 9ca004c41..bc93e77c6 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc @@ -37,10 +37,7 @@ ProducerSocket::ProducerSocket(asio::io_service &io_service) output_buffer_(default_values::producer_socket_output_buffer_size), registration_status_(REGISTRATION_NOT_ATTEMPTED), making_manifest_(false), - signature_type_(SHA_256), hash_algorithm_(HashAlgorithm::SHA_256), - input_buffer_capacity_(default_values::producer_socket_input_buffer_size), - input_buffer_size_(0), on_interest_input_(VOID_HANDLER), on_interest_dropped_input_buffer_(VOID_HANDLER), on_interest_inserted_input_buffer_(VOID_HANDLER), @@ -98,22 +95,30 @@ void ProducerSocket::listen() { void ProducerSocket::passContentObjectToCallbacks( const std::shared_ptr &content_object) { if (content_object) { - if (on_new_segment_ != VOID_HANDLER) { - on_new_segment_(*this, *content_object); + if (on_new_segment_) { + io_service_.dispatch([this, content_object]() { + on_new_segment_(*this, *content_object); + }); } - if (on_content_object_to_sign_ != VOID_HANDLER) { - on_content_object_to_sign_(*this, *content_object); + if (on_content_object_to_sign_) { + io_service_.dispatch([this, content_object]() { + on_content_object_to_sign_(*this, *content_object); + }); } - if (on_content_object_in_output_buffer_ != VOID_HANDLER) { - on_content_object_in_output_buffer_(*this, *content_object); + if (on_content_object_in_output_buffer_) { + io_service_.dispatch([this, content_object]() { + on_content_object_in_output_buffer_(*this, *content_object); + }); } output_buffer_.insert(content_object); - if (on_content_object_output_ != VOID_HANDLER) { - on_content_object_output_(*this, *content_object); + if (on_content_object_output_) { + io_service_.dispatch([this, content_object]() { + on_content_object_output_(*this, *content_object); + }); } portal_->sendContentObject(*content_object); @@ -121,15 +126,19 @@ void ProducerSocket::passContentObjectToCallbacks( } void ProducerSocket::produce(ContentObject &content_object) { - if (on_content_object_in_output_buffer_ != VOID_HANDLER) { - on_content_object_in_output_buffer_(*this, content_object); + if (on_content_object_in_output_buffer_) { + io_service_.dispatch([this, &content_object]() { + on_content_object_in_output_buffer_(*this, content_object); + }); } output_buffer_.insert(std::static_pointer_cast( content_object.shared_from_this())); - if (on_content_object_output_ != VOID_HANDLER) { - on_content_object_output_(*this, content_object); + if (on_content_object_output_) { + io_service_.dispatch([this, &content_object]() { + on_content_object_output_(*this, content_object); + }); } portal_->sendContentObject(content_object); @@ -142,6 +151,15 @@ uint32_t ProducerSocket::produce(Name content_name, return 0; } + // Copy the atomic variable to ensure always the same value during the a + // production + std::size_t data_packet_size = data_packet_size_; + uint32_t content_object_expiry_time = content_object_expiry_time_; + HashAlgorithm algo = hash_algorithm_; + bool making_manifest = making_manifest_; + std::shared_ptr identity; + getSocketOption(GeneralTransportOptions::IDENTITY, identity); + auto buffer_size = buffer->length(); const std::size_t hash_size = 32; @@ -162,7 +180,7 @@ uint32_t ProducerSocket::produce(Name content_name, std::unique_ptr zero_hash; // TODO Manifest may still be used for indexing - if (making_manifest_ && !identity_) { + if (making_manifest && !identity) { throw errors::RuntimeException( "Making manifests without setting producer identity. Aborting."); } @@ -182,18 +200,18 @@ uint32_t ProducerSocket::produce(Name content_name, } format = hf_format; - if (making_manifest_) { + if (making_manifest) { format = hf_format; manifest_header_size = core::Packet::getHeaderSizeFromFormat( - hf_format_ah, identity_->getSignatureLength()); - } else if (identity_) { + hf_format_ah, identity->getSignatureLength()); + } else if (identity) { format = hf_format_ah; - signature_length = identity_->getSignatureLength(); + signature_length = identity->getSignatureLength(); } header_size = core::Packet::getHeaderSizeFromFormat(format, signature_length); - free_space_for_content = data_packet_size_ - header_size; + free_space_for_content = data_packet_size - header_size; uint32_t number_of_segments = uint32_t(std::ceil(double(buffer_size) / double(free_space_for_content))); @@ -204,9 +222,9 @@ uint32_t ProducerSocket::produce(Name content_name, // TODO allocate space for all the headers - if (making_manifest_) { + if (making_manifest) { auto segment_in_manifest = static_cast( - std::floor(double(data_packet_size_ - manifest_header_size - + std::floor(double(data_packet_size - manifest_header_size - ContentObjectManifest::getManifestHeaderSize()) / (4.0 + 32.0)) - 1.0); @@ -219,8 +237,8 @@ uint32_t ProducerSocket::produce(Name content_name, core::ManifestVersion::VERSION_1, core::ManifestType::INLINE_MANIFEST, hash_algorithm_, is_last_manifest, content_name, core::NextSegmentCalculationStrategy::INCREMENTAL, - identity_->getSignatureLength())); - manifest->setLifetime(content_object_expiry_time_); + identity->getSignatureLength())); + manifest->setLifetime(content_object_expiry_time); if (is_last) { manifest->setFinalBlockNumber(final_block_number); @@ -231,21 +249,21 @@ uint32_t ProducerSocket::produce(Name content_name, uint8_t hash[hash_size]; std::memset(hash, 0, hash_size); zero_hash = std::make_unique( - hash, hash_size, static_cast(hash_algorithm_)); + hash, hash_size, static_cast(algo)); } for (unsigned int packaged_segments = 0; packaged_segments < number_of_segments; packaged_segments++) { - if (making_manifest_) { + if (making_manifest) { if (manifest->estimateManifestSize(2) > - data_packet_size_ - manifest_header_size) { + data_packet_size - manifest_header_size) { // Add next manifest manifest->addSuffixHash(current_segment, *zero_hash); // Send the current manifest manifest->encode(); - identity_->getSigner().sign(*manifest); + identity->getSigner().sign(*manifest); passContentObjectToCallbacks(manifest); @@ -258,8 +276,8 @@ uint32_t ProducerSocket::produce(Name content_name, core::ManifestType::INLINE_MANIFEST, hash_algorithm_, is_last_manifest, content_name, core::NextSegmentCalculationStrategy::INCREMENTAL, - identity_->getSignatureLength())); - manifest->setLifetime(content_object_expiry_time_); + identity->getSignatureLength())); + manifest->setLifetime(content_object_expiry_time); if (is_last) { manifest->setFinalBlockNumber(final_block_number); } else { @@ -271,7 +289,7 @@ uint32_t ProducerSocket::produce(Name content_name, auto content_object = std::make_shared( content_name.setSuffix(current_segment), format); - content_object->setLifetime(content_object_expiry_time_); + content_object->setLifetime(content_object_expiry_time); auto b = buffer->cloneOne(); b->trimStart(free_space_for_content * packaged_segments); @@ -280,7 +298,7 @@ uint32_t ProducerSocket::produce(Name content_name, b->append(buffer_size - bytes_segmented); bytes_segmented += (int)(buffer_size - bytes_segmented); - if (is_last && making_manifest_) { + if (is_last && making_manifest) { is_last_manifest = true; } else if (is_last) { content_object->setRst(); @@ -293,19 +311,19 @@ uint32_t ProducerSocket::produce(Name content_name, content_object->appendPayload(std::move(b)); - if (making_manifest_) { + if (making_manifest) { using namespace std::chrono_literals; utils::CryptoHash hash = content_object->computeDigest(hash_algorithm_); manifest->addSuffixHash(current_segment, hash); - } else if (identity_) { - identity_->getSigner().sign(*content_object); + } else if (identity) { + identity->getSigner().sign(*content_object); } current_segment++; passContentObjectToCallbacks(content_object); } - if (making_manifest_) { + if (making_manifest) { if (is_last_manifest) { manifest->setFinalManifest(is_last_manifest); } @@ -315,13 +333,15 @@ uint32_t ProducerSocket::produce(Name content_name, } manifest->encode(); - identity_->getSigner().sign(*manifest); + identity->getSigner().sign(*manifest); passContentObjectToCallbacks(manifest); } - if (on_content_produced_ != VOID_HANDLER) { - on_content_produced_(*this, std::make_error_code(std::errc(0)), - buffer_size); + if (on_content_produced_) { + io_service_.dispatch([this, buffer_size]() { + on_content_produced_(*this, std::make_error_code(std::errc(0)), + buffer_size); + }); } return current_segment - start_offset; @@ -347,7 +367,7 @@ void ProducerSocket::asyncProduce(const Name &suffix, const uint8_t *buf, } void ProducerSocket::onInterest(Interest &interest) { - if (on_interest_input_ != VOID_HANDLER) { + if (on_interest_input_) { on_interest_input_(*this, interest); } @@ -355,22 +375,571 @@ void ProducerSocket::onInterest(Interest &interest) { output_buffer_.find(interest); if (content_object) { - if (on_interest_satisfied_output_buffer_ != VOID_HANDLER) { + if (on_interest_satisfied_output_buffer_) { on_interest_satisfied_output_buffer_(*this, interest); } - if (on_content_object_output_ != VOID_HANDLER) { + if (on_content_object_output_) { on_content_object_output_(*this, *content_object); } portal_->sendContentObject(*content_object); } else { - if (on_interest_process_ != VOID_HANDLER) { + if (on_interest_process_) { on_interest_process_(*this, interest); } } } +// If the thread calling lambda_func is not the same of io_service, this +// function reschedule the function on it +template +int ProducerSocket::rescheduleOnIOService(int socket_option_key, + arg2 socket_option_value, + Lambda lambda_func) { + // To enforce type check + std::function func = lambda_func; + int result = SOCKET_OPTION_SET; + if (listening_thread_.joinable() && + std::this_thread::get_id() != listening_thread_.get_id()) { + std::mutex mtx; + /* Condition variable for the wait */ + std::condition_variable cv; + bool done = false; + io_service_.dispatch([this, &socket_option_key, &socket_option_value, &mtx, + &cv, &result, &done, &func]() { + std::unique_lock lck(mtx); + done = true; + result = func(socket_option_key, socket_option_value); + }); + std::unique_lock lck(mtx); + if (!done) { + cv.wait(lck); + } + } else { + result = func(socket_option_key, socket_option_value); + } + + return result; +} + +// If the thread calling lambda_func is not the same of io_service, this +// function reschedule the function on it +template +int ProducerSocket::rescheduleOnIOServiceWithReference( + int socket_option_key, arg2 &socket_option_value, Lambda lambda_func) { + // To enforce type check + std::function func = lambda_func; + int result = SOCKET_OPTION_SET; + if (listening_thread_.joinable() && + std::this_thread::get_id() != this->listening_thread_.get_id()) { + std::mutex mtx; + /* Condition variable for the wait */ + std::condition_variable cv; + std::unique_lock lck(mtx); + bool done = false; + io_service_.dispatch([this, &socket_option_key, &socket_option_value, &mtx, + &cv, &result, &done, &func]() { + std::unique_lock lck(mtx); + done = true; + result = func(socket_option_key, socket_option_value); + + if (!done) { + cv.wait(lck); + } + }); + } else { + result = func(socket_option_key, socket_option_value); + } + + return result; +} + +int ProducerSocket::setSocketOption(int socket_option_key, + uint32_t socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::DATA_PACKET_SIZE: + if (socket_option_value < default_values::max_content_object_size && + socket_option_value > 0) { + data_packet_size_ = socket_option_value; + break; + } + + case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: + output_buffer_.setLimit(socket_option_value); + break; + + case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME: + content_object_expiry_time_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::setSocketOption(int socket_option_key, + std::nullptr_t socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerContentObjectCallback socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::INTEREST_INPUT: + if (socket_option_value == VOID_HANDLER) { + on_interest_input_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::INTEREST_DROP: + if (socket_option_value == VOID_HANDLER) { + on_interest_dropped_input_buffer_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::INTEREST_PASS: + if (socket_option_value == VOID_HANDLER) { + on_interest_inserted_input_buffer_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::CACHE_HIT: + if (socket_option_value == VOID_HANDLER) { + on_interest_satisfied_output_buffer_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::CACHE_MISS: + if (socket_option_value == VOID_HANDLER) { + on_interest_process_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: + if (socket_option_value == VOID_HANDLER) { + on_new_segment_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: + if (socket_option_value == VOID_HANDLER) { + on_content_object_to_sign_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::CONTENT_OBJECT_READY: + if (socket_option_value == VOID_HANDLER) { + on_content_object_in_output_buffer_ = VOID_HANDLER; + break; + } + + case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: + if (socket_option_value == VOID_HANDLER) { + on_content_object_output_ = VOID_HANDLER; + break; + } + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ProducerSocket::setSocketOption(int socket_option_key, + bool socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::MAKE_MANIFEST: + making_manifest_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::setSocketOption(int socket_option_key, + Name *socket_option_value) { + return SOCKET_OPTION_NOT_SET; +} + +int ProducerSocket::setSocketOption(int socket_option_key, + std::list socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::NETWORK_NAME: + served_namespaces_ = socket_option_value; + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::setSocketOption( + int socket_option_key, ProducerContentObjectCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerContentObjectCallback socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: + on_new_segment_ = socket_option_value; + break; + + case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: + on_content_object_to_sign_ = socket_option_value; + break; + + case ProducerCallbacksOptions::CONTENT_OBJECT_READY: + on_content_object_in_output_buffer_ = socket_option_value; + break; + + case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: + on_content_object_output_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ProducerSocket::setSocketOption( + int socket_option_key, ProducerInterestCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerInterestCallback socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::INTEREST_INPUT: + on_interest_input_ = socket_option_value; + break; + + case ProducerCallbacksOptions::INTEREST_DROP: + on_interest_dropped_input_buffer_ = socket_option_value; + break; + + case ProducerCallbacksOptions::INTEREST_PASS: + on_interest_inserted_input_buffer_ = socket_option_value; + break; + + case ProducerCallbacksOptions::CACHE_HIT: + on_interest_satisfied_output_buffer_ = socket_option_value; + break; + + case ProducerCallbacksOptions::CACHE_MISS: + on_interest_process_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ProducerSocket::setSocketOption( + int socket_option_key, ProducerContentCallback socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerContentCallback socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::CONTENT_PRODUCED: + on_content_produced_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + }); +} + +int ProducerSocket::setSocketOption(int socket_option_key, + HashAlgorithm socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::HASH_ALGORITHM: + hash_algorithm_ = socket_option_value; + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::setSocketOption(int socket_option_key, + utils::CryptoSuite socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::CRYPTO_SUITE: + crypto_suite_ = socket_option_value; + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::setSocketOption( + int socket_option_key, + const std::shared_ptr &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::IDENTITY: { + utils::SpinLock::Acquire locked(identity_lock_); + identity_.reset(); + identity_ = socket_option_value; + } break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::setSocketOption(int socket_option_key, + const std::string &socket_option_value) { + switch (socket_option_key) { + case DataLinkOptions::OUTPUT_INTERFACE: + output_interface_ = socket_option_value; + portal_->setOutputInterface(output_interface_); + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; +} + +int ProducerSocket::getSocketOption(int socket_option_key, + uint32_t &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: + socket_option_value = (uint32_t)output_buffer_.getLimit(); + break; + + case GeneralTransportOptions::DATA_PACKET_SIZE: + socket_option_value = (uint32_t)data_packet_size_; + break; + + case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME: + socket_option_value = content_object_expiry_time_; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption(int socket_option_key, + bool &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::MAKE_MANIFEST: + socket_option_value = making_manifest_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption(int socket_option_key, + std::list &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::NETWORK_NAME: + socket_option_value = served_namespaces_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption( + int socket_option_key, + ProducerContentObjectCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerContentObjectCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: + *socket_option_value = &on_new_segment_; + break; + + case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: + *socket_option_value = &on_content_object_to_sign_; + break; + + case ProducerCallbacksOptions::CONTENT_OBJECT_READY: + *socket_option_value = &on_content_object_in_output_buffer_; + break; + + case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: + *socket_option_value = &on_content_object_output_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ProducerSocket::getSocketOption( + int socket_option_key, ProducerContentCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerContentCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::CONTENT_PRODUCED: + *socket_option_value = &on_content_produced_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ProducerSocket::getSocketOption( + int socket_option_key, ProducerInterestCallback **socket_option_value) { + // Reschedule the function on the io_service to avoid race condition in case + // setSocketOption is called while the io_service is running. + return rescheduleOnIOService( + socket_option_key, socket_option_value, + [this](int socket_option_key, + ProducerInterestCallback **socket_option_value) -> int { + switch (socket_option_key) { + case ProducerCallbacksOptions::INTEREST_INPUT: + *socket_option_value = &on_interest_input_; + break; + + case ProducerCallbacksOptions::INTEREST_DROP: + *socket_option_value = &on_interest_dropped_input_buffer_; + break; + + case ProducerCallbacksOptions::INTEREST_PASS: + *socket_option_value = &on_interest_inserted_input_buffer_; + break; + + case CACHE_HIT: + *socket_option_value = &on_interest_satisfied_output_buffer_; + break; + + case CACHE_MISS: + *socket_option_value = &on_interest_process_; + break; + + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + }); +} + +int ProducerSocket::getSocketOption( + int socket_option_key, std::shared_ptr &socket_option_value) { + switch (socket_option_key) { + case PORTAL: + socket_option_value = portal_; + break; + default: + return SOCKET_OPTION_NOT_GET; + ; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption(int socket_option_key, + HashAlgorithm &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::HASH_ALGORITHM: + socket_option_value = hash_algorithm_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption(int socket_option_key, + utils::CryptoSuite &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::HASH_ALGORITHM: + socket_option_value = crypto_suite_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption( + int socket_option_key, + std::shared_ptr &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::IDENTITY: { + utils::SpinLock::Acquire locked(identity_lock_); + socket_option_value = identity_; + } break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + +int ProducerSocket::getSocketOption(int socket_option_key, + std::string &socket_option_value) { + switch (socket_option_key) { + case DataLinkOptions::OUTPUT_INTERFACE: + socket_option_value = output_interface_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; +} + asio::io_service &ProducerSocket::getIoService() { return io_service_; } } // namespace interface diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h index 18adbf4a7..5c617d761 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h @@ -91,500 +91,116 @@ class ProducerSocket : public Socket, onInterest(*interest); }; - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - uint32_t socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::DATA_PACKET_SIZE: - if (socket_option_value < default_values::max_content_object_size && - socket_option_value > 0) { - data_packet_size_ = socket_option_value; - break; - } - - case GeneralTransportOptions::INPUT_BUFFER_SIZE: - if (socket_option_value >= 1) { - input_buffer_capacity_ = socket_option_value; - break; - } - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - output_buffer_.setLimit(socket_option_value); - break; - - case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME: - content_object_expiry_time_ = socket_option_value; - break; - - case GeneralTransportOptions::SIGNATURE_TYPE: - if (socket_option_value == SOCKET_OPTION_DEFAULT) { - signature_type_ = SHA_256; - } else { - signature_type_ = socket_option_value; - } - - if (signature_type_ == SHA_256 || signature_type_ == RSA_256) { - signature_size_ = 32; - } - - break; - - case ProducerCallbacksOptions::INTEREST_INPUT: - if (socket_option_value == VOID_HANDLER) { - on_interest_input_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::INTEREST_DROP: - if (socket_option_value == VOID_HANDLER) { - on_interest_dropped_input_buffer_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::INTEREST_PASS: - if (socket_option_value == VOID_HANDLER) { - on_interest_inserted_input_buffer_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::CACHE_HIT: - if (socket_option_value == VOID_HANDLER) { - on_interest_satisfied_output_buffer_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::CACHE_MISS: - if (socket_option_value == VOID_HANDLER) { - on_interest_process_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: - if (socket_option_value == VOID_HANDLER) { - on_new_segment_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: - if (socket_option_value == VOID_HANDLER) { - on_content_object_to_sign_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::CONTENT_OBJECT_READY: - if (socket_option_value == VOID_HANDLER) { - on_content_object_in_output_buffer_ = VOID_HANDLER; - break; - } - - case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: - if (socket_option_value == VOID_HANDLER) { - on_content_object_output_ = VOID_HANDLER; - break; - } - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - bool socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::MAKE_MANIFEST: - making_manifest_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, - Name *socket_option_value) { - return SOCKET_OPTION_NOT_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, std::list socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::NETWORK_NAME: - served_namespaces_ = socket_option_value; - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } + virtual int setSocketOption(int socket_option_key, + uint32_t socket_option_value); - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, - ProducerContentObjectCallback socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: - on_new_segment_ = socket_option_value; - break; - - case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: - on_content_object_to_sign_ = socket_option_value; - break; - - case ProducerCallbacksOptions::CONTENT_OBJECT_READY: - on_content_object_in_output_buffer_ = socket_option_value; - break; + virtual int setSocketOption(int socket_option_key, + std::nullptr_t socket_option_value); - case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: - on_content_object_output_ = socket_option_value; - break; + virtual int setSocketOption(int socket_option_key, bool socket_option_value); - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } + virtual int setSocketOption(int socket_option_key, Name *socket_option_value); - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ProducerInterestCallback socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::INTEREST_INPUT: - on_interest_input_ = socket_option_value; - break; + virtual int setSocketOption(int socket_option_key, + std::list socket_option_value); - case ProducerCallbacksOptions::INTEREST_DROP: - on_interest_dropped_input_buffer_ = socket_option_value; - break; + virtual int setSocketOption( + int socket_option_key, ProducerContentObjectCallback socket_option_value); - case ProducerCallbacksOptions::INTEREST_PASS: - on_interest_inserted_input_buffer_ = socket_option_value; - break; + virtual int setSocketOption(int socket_option_key, + ProducerInterestCallback socket_option_value); - case ProducerCallbacksOptions::CACHE_HIT: - on_interest_satisfied_output_buffer_ = socket_option_value; - break; - - case ProducerCallbacksOptions::CACHE_MISS: - on_interest_process_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } + virtual int setSocketOption(int socket_option_key, + ProducerContentCallback socket_option_value); - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ProducerContentCallback socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::CONTENT_PRODUCED: - on_content_produced_ = socket_option_value; - break; + virtual int setSocketOption(int socket_option_key, + HashAlgorithm socket_option_value); - default: - return SOCKET_OPTION_NOT_SET; - } + virtual int setSocketOption(int socket_option_key, + utils::CryptoSuite socket_option_value); - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, HashAlgorithm socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::HASH_ALGORITHM: - hash_algorithm_ = socket_option_value; - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, utils::CryptoSuite socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::CRYPTO_SUITE: - crypto_suite_ = socket_option_value; - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( + virtual int setSocketOption( int socket_option_key, - const std::shared_ptr &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::IDENTITY: - identity_ = socket_option_value; - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, const std::string &socket_option_value) { - switch (socket_option_key) { - case DataLinkOptions::OUTPUT_INTERFACE: - output_interface_ = socket_option_value; - portal_->setOutputInterface(output_interface_); - break; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - ; - } + const std::shared_ptr &socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption(int socket_option_key, - uint32_t &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::INPUT_BUFFER_SIZE: - socket_option_value = (uint32_t)input_buffer_capacity_; - break; + virtual int setSocketOption(int socket_option_key, + const std::string &socket_option_value); - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - socket_option_value = (uint32_t)output_buffer_.getLimit(); - break; + virtual int getSocketOption(int socket_option_key, + uint32_t &socket_option_value); - case GeneralTransportOptions::DATA_PACKET_SIZE: - socket_option_value = (uint32_t)data_packet_size_; - break; + virtual int getSocketOption(int socket_option_key, bool &socket_option_value); - case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME: - socket_option_value = content_object_expiry_time_; - break; - - case GeneralTransportOptions::SIGNATURE_TYPE: - socket_option_value = signature_type_; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption(int socket_option_key, - bool &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::MAKE_MANIFEST: - socket_option_value = making_manifest_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } + virtual int getSocketOption(int socket_option_key, + std::list &socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, std::list &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::NETWORK_NAME: - socket_option_value = served_namespaces_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( + virtual int getSocketOption( int socket_option_key, - ProducerContentObjectCallback **socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: - *socket_option_value = &on_new_segment_; - break; - - case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: - *socket_option_value = &on_content_object_to_sign_; - break; - - case ProducerCallbacksOptions::CONTENT_OBJECT_READY: - *socket_option_value = &on_content_object_in_output_buffer_; - break; - - case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: - *socket_option_value = &on_content_object_output_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ProducerContentCallback **socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::CONTENT_PRODUCED: - *socket_option_value = &on_content_produced_; - break; - - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ProducerInterestCallback **socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::INTEREST_INPUT: - *socket_option_value = &on_interest_input_; - break; - - case ProducerCallbacksOptions::INTEREST_DROP: - *socket_option_value = &on_interest_dropped_input_buffer_; - break; - - case ProducerCallbacksOptions::INTEREST_PASS: - *socket_option_value = &on_interest_inserted_input_buffer_; - break; + ProducerContentObjectCallback **socket_option_value); - case CACHE_HIT: - *socket_option_value = &on_interest_satisfied_output_buffer_; - break; + virtual int getSocketOption(int socket_option_key, + ProducerContentCallback **socket_option_value); - case CACHE_MISS: - *socket_option_value = &on_interest_process_; - break; + virtual int getSocketOption(int socket_option_key, + ProducerInterestCallback **socket_option_value); - default: - return SOCKET_OPTION_NOT_GET; - } + virtual int getSocketOption(int socket_option_key, + std::shared_ptr &socket_option_value); - return SOCKET_OPTION_GET; - } + virtual int getSocketOption(int socket_option_key, + HashAlgorithm &socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, std::shared_ptr &socket_option_value) { - switch (socket_option_key) { - case PORTAL: - socket_option_value = portal_; - break; - default: - return SOCKET_OPTION_NOT_GET; - ; - } - - return SOCKET_OPTION_GET; - } + virtual int getSocketOption(int socket_option_key, + utils::CryptoSuite &socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, HashAlgorithm &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::HASH_ALGORITHM: - socket_option_value = hash_algorithm_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } + virtual int getSocketOption( + int socket_option_key, + std::shared_ptr &socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, utils::CryptoSuite &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::HASH_ALGORITHM: - socket_option_value = crypto_suite_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } + virtual int getSocketOption(int socket_option_key, + std::string &socket_option_value); - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, - std::shared_ptr &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::IDENTITY: - if (identity_) { - socket_option_value = identity_; - break; - } - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } + template + int rescheduleOnIOService(int socket_option_key, arg2 socket_option_value, + Lambda lambda_func); - TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, std::string &socket_option_value) { - switch (socket_option_key) { - case DataLinkOptions::OUTPUT_INTERFACE: - socket_option_value = output_interface_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } + template + int rescheduleOnIOServiceWithReference(int socket_option_key, + arg2 &socket_option_value, + Lambda lambda_func); - private: + protected: // Threads std::thread listening_thread_; - protected: asio::io_service internal_io_service_; asio::io_service &io_service_; std::shared_ptr portal_; - std::size_t data_packet_size_; - std::list served_namespaces_; - uint32_t content_object_expiry_time_; + std::atomic data_packet_size_; + std::list + served_namespaces_; // No need to be threadsafe, this is always modified + // by the application thread + std::atomic content_object_expiry_time_; // buffers + // ContentStore is thread-safe utils::ContentStore output_buffer_; - private: utils::EventThread async_thread_; int registration_status_; - bool making_manifest_; + std::atomic making_manifest_; // map for storing sequence numbers for several calls of the publish // function std::unordered_map> seq_number_map_; - int signature_type_; - int signature_size_; - - HashAlgorithm hash_algorithm_; - utils::CryptoSuite crypto_suite_; + std::atomic hash_algorithm_; + std::atomic crypto_suite_; + utils::SpinLock identity_lock_; std::shared_ptr identity_; - // buffers - - std::queue> input_buffer_; - std::atomic_size_t input_buffer_capacity_; - std::atomic_size_t input_buffer_size_; - // callbacks - protected: ProducerInterestCallback on_interest_input_; ProducerInterestCallback on_interest_dropped_input_buffer_; ProducerInterestCallback on_interest_inserted_input_buffer_; diff --git a/libtransport/src/hicn/transport/protocols/protocol.cc b/libtransport/src/hicn/transport/protocols/protocol.cc index 9caa2eca7..8da9529d6 100644 --- a/libtransport/src/hicn/transport/protocols/protocol.cc +++ b/libtransport/src/hicn/transport/protocols/protocol.cc @@ -23,23 +23,28 @@ namespace protocol { using namespace interface; TransportProtocol::TransportProtocol(interface::ConsumerSocket *icn_socket) - : socket_(icn_socket), is_running_(false) { + : socket_(icn_socket), is_running_(false), is_first_(false) { socket_->getSocketOption(GeneralTransportOptions::PORTAL, portal_); } int TransportProtocol::start() { - // If the protocol is already running, return + // If the protocol is already running, return otherwise set as running if (is_running_) return -1; - // Set the protocol as running - is_running_ = true; - // Reset the protocol state machine reset(); + // Set it is the first time we schedule an interest + is_first_ = true; + // Schedule next interests scheduleNextInterests(); + is_first_ = false; + + // Set the protocol as running + is_running_ = true; + // Start Event loop portal_->runEventsLoop(); diff --git a/libtransport/src/hicn/transport/protocols/protocol.h b/libtransport/src/hicn/transport/protocols/protocol.h index 88889bb8c..e4821b6a0 100644 --- a/libtransport/src/hicn/transport/protocols/protocol.h +++ b/libtransport/src/hicn/transport/protocols/protocol.h @@ -15,6 +15,8 @@ #pragma once +#include + #include #include #include @@ -60,7 +62,9 @@ class TransportProtocol : public interface::BasePortal::ConsumerCallback, protected: interface::ConsumerSocket *socket_; std::shared_ptr portal_; - volatile bool is_running_; + std::atomic is_running_; + // True if it si the first time we schedule an interest + std::atomic is_first_; TransportStatistics stats_; }; diff --git a/libtransport/src/hicn/transport/protocols/raaqm.cc b/libtransport/src/hicn/transport/protocols/raaqm.cc index 574693c51..c816158f9 100644 --- a/libtransport/src/hicn/transport/protocols/raaqm.cc +++ b/libtransport/src/hicn/transport/protocols/raaqm.cc @@ -318,17 +318,17 @@ void RaaqmTransportProtocol::onContentObject( } // Call application-defined callbacks - ConsumerContentObjectCallback *callback_content_object = nullptr; + ConsumerContentObjectCallback *callback_content_object = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT, &callback_content_object); - if (*callback_content_object != VOID_HANDLER) { + if (*callback_content_object) { (*callback_content_object)(*socket_, *content_object); } - ConsumerInterestCallback *callback_interest = nullptr; + ConsumerInterestCallback *callback_interest = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_SATISFIED, &callback_interest); - if (*callback_content_object != VOID_HANDLER) { + if (*callback_content_object) { (*callback_interest)(*socket_, *interest); } @@ -369,10 +369,10 @@ void RaaqmTransportProtocol::onContentSegment( reassemble(std::move(content_object)); } else if (TRANSPORT_EXPECT_FALSE(incremental_suffix == index_manager_->getFinalSuffix())) { - interface::ConsumerSocket::ReadCallback *on_payload = nullptr; + interface::ConsumerSocket::ReadCallback *on_payload = VOID_HANDLER; socket_->getSocketOption(READ_CALLBACK, &on_payload); - if (on_payload != nullptr) { + if (on_payload) { on_payload->readSuccess(stats_.getBytesRecv()); } } @@ -404,10 +404,10 @@ void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { return; } - ConsumerInterestCallback *callback = nullptr; + ConsumerInterestCallback *callback = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_EXPIRED, &callback); - if (*callback != VOID_HANDLER) { + if (*callback) { (*callback)(*socket_, *interest); } @@ -420,17 +420,17 @@ void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { max_rtx)) { stats_.updateRetxCount(1); - callback = nullptr; + callback = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_RETRANSMISSION, &callback); - if (*callback != VOID_HANDLER) { + if (*callback) { (*callback)(*socket_, *interest); } - callback = nullptr; + callback = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_OUTPUT, &callback); - if ((*callback) != VOID_HANDLER) { + if (*callback) { (*callback)(*socket_, *interest); } @@ -450,7 +450,7 @@ void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { } void RaaqmTransportProtocol::scheduleNextInterests() { - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + if (TRANSPORT_EXPECT_FALSE(!is_running_ && !is_first_)) { return; } @@ -490,14 +490,14 @@ void RaaqmTransportProtocol::sendInterest(std::uint64_t next_suffix) { interest_lifetime); interest->setLifetime(interest_lifetime); - ConsumerInterestCallback *callback = nullptr; + ConsumerInterestCallback *callback = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_OUTPUT, &callback); - if (*callback != VOID_HANDLER) { + if (*callback) { callback->operator()(*socket_, *interest); } - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + if (TRANSPORT_EXPECT_FALSE(!is_running_ && !is_first_)) { return; } @@ -516,10 +516,10 @@ void RaaqmTransportProtocol::sendInterest(Interest::Ptr &&interest) { } void RaaqmTransportProtocol::onContentReassembled(std::error_code ec) { - interface::ConsumerSocket::ReadCallback *on_payload = nullptr; + interface::ConsumerSocket::ReadCallback *on_payload = VOID_HANDLER; socket_->getSocketOption(READ_CALLBACK, &on_payload); - if (on_payload == nullptr) { + if (on_payload) { throw errors::RuntimeException( "The read callback must be installed in the transport before " "starting " @@ -581,10 +581,10 @@ void RaaqmTransportProtocol::updateStats(uint32_t suffix, uint64_t rtt, stats_.updateAverageWindowSize(current_window_size_); // Call statistics callback - ConsumerTimerCallback *stats_callback = nullptr; + ConsumerTimerCallback *stats_callback = VOID_HANDLER; socket_->getSocketOption(ConsumerCallbacksOptions::STATS_SUMMARY, &stats_callback); - if (*stats_callback != VOID_HANDLER) { + if (*stats_callback) { auto dt = std::chrono::duration_cast(now - t0_); uint32_t timer_interval_milliseconds = 0; diff --git a/libtransport/src/hicn/transport/protocols/rtc.cc b/libtransport/src/hicn/transport/protocols/rtc.cc index 620523cbe..e6134f767 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.cc +++ b/libtransport/src/hicn/transport/protocols/rtc.cc @@ -88,13 +88,13 @@ void RTCTransportProtocol::reset() { lastSegNacked_ = 0; lastReceived_ = 0; lastReceivedTime_ = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); highestReceived_ = 0; firstSequenceInRound_ = 0; rtx_timer_used_ = false; - for(int i = 0; i < (1 << default_values::log_2_default_buffer_size); i++){ + for (int i = 0; i < (1 << default_values::log_2_default_buffer_size); i++) { inflightInterests_[i] = {0}; } @@ -191,14 +191,13 @@ void RTCTransportProtocol::updateDelayStats( pathTable_[pathLabel]->insertOwdSample(OWD); pathTable_[pathLabel]->computeInterArrivalGap(segmentNumber); - }else{ + } else { pathTable_[pathLabel]->receivedNack(); } } void RTCTransportProtocol::updateStats(uint32_t round_duration) { - if(pathTable_.empty()) - return; + if (pathTable_.empty()) return; if (receivedBytes_ != 0) { double bytesPerSec = @@ -213,68 +212,70 @@ void RTCTransportProtocol::updateStats(uint32_t round_duration) { for (auto it = pathTable_.begin(); it != pathTable_.end(); it++) { it->second->roundEnd(); - if(it->second->isActive()){ - if(it->second->getMinRtt() < minRtt){ + if (it->second->isActive()) { + if (it->second->getMinRtt() < minRtt) { minRtt = it->second->getMinRtt(); producerPathLabels_[0] = it->first; } - if(it->second->getMinRtt() > maxRtt){ + if (it->second->getMinRtt() > maxRtt) { maxRtt = it->second->getMinRtt(); producerPathLabels_[1] = it->first; } } } - if(pathTable_.find(producerPathLabels_[0]) == pathTable_.end() || + if (pathTable_.find(producerPathLabels_[0]) == pathTable_.end() || pathTable_.find(producerPathLabels_[1]) == pathTable_.end()) - return; //this should not happen + return; // this should not happen - //as a queuing delay we keep the lowest one among the two paths - //if one path is congested the forwarder should decide to do not - //use it so it does not make sense to inform the application - //that maybe we have a problem - if(pathTable_[producerPathLabels_[0]]->getQueuingDealy() < + // as a queuing delay we keep the lowest one among the two paths + // if one path is congested the forwarder should decide to do not + // use it so it does not make sense to inform the application + // that maybe we have a problem + if (pathTable_[producerPathLabels_[0]]->getQueuingDealy() < pathTable_[producerPathLabels_[1]]->getQueuingDealy()) - queuingDelay_ = pathTable_[producerPathLabels_[0]]->getQueuingDealy(); + queuingDelay_ = pathTable_[producerPathLabels_[0]]->getQueuingDealy(); else - queuingDelay_ = pathTable_[producerPathLabels_[1]]->getQueuingDealy(); + queuingDelay_ = pathTable_[producerPathLabels_[1]]->getQueuingDealy(); if (sentInterest_ != 0 && currentState_ == HICN_RTC_NORMAL_STATE) { - uint32_t numberTheoricallyReceivedPackets_ = highestReceived_ - firstSequenceInRound_; + uint32_t numberTheoricallyReceivedPackets_ = + highestReceived_ - firstSequenceInRound_; double lossRate = 0; - if(numberTheoricallyReceivedPackets_ != 0) - lossRate = (double)((double)(packetLost_ - lossRecovered_) / (double)numberTheoricallyReceivedPackets_); + if (numberTheoricallyReceivedPackets_ != 0) + lossRate = (double)((double)(packetLost_ - lossRecovered_) / + (double)numberTheoricallyReceivedPackets_); - if(lossRate < 0) - lossRate = 0; + if (lossRate < 0) lossRate = 0; - if(initied){ + if (initied) { lossRate_ = lossRate_ * HICN_ESTIMATED_LOSSES_ALPHA + - (lossRate * (1 - HICN_ESTIMATED_LOSSES_ALPHA)); - }else { - lossRate_ =lossRate; + (lossRate * (1 - HICN_ESTIMATED_LOSSES_ALPHA)); + } else { + lossRate_ = lossRate; initied = true; } } if (avgPacketSize_ == 0) avgPacketSize_ = HICN_INIT_PACKET_SIZE; - //for the BDP we use the max rtt, so that we calibrate the window on the - //RTT of the slowest path. In this way we are sure that the window will - //never be too small + // for the BDP we use the max rtt, so that we calibrate the window on the + // RTT of the slowest path. In this way we are sure that the window will + // never be too small uint32_t BDP = (uint32_t)ceil( - (estimatedBw_ * (double)((double) pathTable_[producerPathLabels_[1]]->getMinRtt() / - (double)HICN_MILLI_IN_A_SEC) * - HICN_BANDWIDTH_SLACK_FACTOR) / - avgPacketSize_); + (estimatedBw_ * + (double)((double)pathTable_[producerPathLabels_[1]]->getMinRtt() / + (double)HICN_MILLI_IN_A_SEC) * + HICN_BANDWIDTH_SLACK_FACTOR) / + avgPacketSize_); uint32_t BW = (uint32_t)ceil(estimatedBw_); computeMaxWindow(BW, BDP); ConsumerTimerCallback *stats_callback = nullptr; socket_->getSocketOption(ConsumerCallbacksOptions::STATS_SUMMARY, &stats_callback); - if (*stats_callback != VOID_HANDLER) { - //Send the stats to the app + if (*stats_callback) { + // Send the stats to the app stats_.updateQueuingDelay(queuingDelay_); stats_.updateLossRatio(lossRate_); (*stats_callback)(*socket_, stats_); @@ -334,8 +335,8 @@ void RTCTransportProtocol::computeMaxWindow(uint32_t productionRate, // currentState = RTC_NORMAL_STATE if (BDPWin != 0) { - maxCWin_ = - (uint32_t)ceil((double)BDPWin + (((double)BDPWin * 30.0) / 100.0)); // BDP + 30% + maxCWin_ = (uint32_t)ceil((double)BDPWin + + (((double)BDPWin * 30.0) / 100.0)); // BDP + 30% } else { maxCWin_ = min(maxWaintingInterest, maxCWin_); } @@ -380,22 +381,22 @@ void RTCTransportProtocol::increaseWindow() { } } -void RTCTransportProtocol::probeRtt(){ +void RTCTransportProtocol::probeRtt() { time_sent_probe_ = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - &interest_name); - //get a random numbe in the probe seq range + &interest_name); + // get a random numbe in the probe seq range std::default_random_engine eng((std::random_device())()); - std::uniform_int_distribution idis( - HICN_MIN_PROBE_SEQ, HICN_MAX_PROBE_SEQ); + std::uniform_int_distribution idis(HICN_MIN_PROBE_SEQ, + HICN_MAX_PROBE_SEQ); probe_seq_number_ = idis(eng); interest_name->setSuffix(probe_seq_number_); - //we considere the probe as a rtx so that we do not incresea inFlightInt + // we considere the probe as a rtx so that we do not incresea inFlightInt received_probe_ = false; sendInterest(interest_name, true); @@ -406,7 +407,6 @@ void RTCTransportProtocol::probeRtt(){ }); } - void RTCTransportProtocol::sendInterest(Name *interest_name, bool rtx) { auto interest = getPacket(); interest->setName(*interest_name); @@ -421,11 +421,11 @@ void RTCTransportProtocol::sendInterest(Name *interest_name, bool rtx) { socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_OUTPUT, &on_interest_output); - if (*on_interest_output != VOID_HANDLER) { + if (*on_interest_output) { (*on_interest_output)(*socket_, *interest); } - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + if (TRANSPORT_EXPECT_FALSE(!is_running_ && !is_first_)) { return; } @@ -440,20 +440,20 @@ void RTCTransportProtocol::sendInterest(Name *interest_name, bool rtx) { void RTCTransportProtocol::scheduleNextInterests() { checkRound(); - if (!is_running_) return; + if (!is_running_ && !is_first_) return; while (inflightInterestsCount_ < currentCWin_) { Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &interest_name); - interest_name->setSuffix(actualSegment_); + interest_name->setSuffix(actualSegment_); // if the producer socket is not stated (does not reply even with nacks) // we keep asking for something without marking anything as lost (see // timeout). In this way when the producer socket will start the - //consumer socket will not miss any packet - if(TRANSPORT_EXPECT_FALSE(!firstPckReceived_)){ + // consumer socket will not miss any packet + if (TRANSPORT_EXPECT_FALSE(!firstPckReceived_)) { uint32_t pkt = actualSegment_ & modMask_; inflightInterests_[pkt].state = sent_; inflightInterests_[pkt].sequence = actualSegment_; @@ -477,9 +477,9 @@ void RTCTransportProtocol::scheduleNextInterests() { continue; } - //same if the packet is lost + // same if the packet is lost if (inflightInterests_[pkt].state == lost_ && - inflightInterests_[pkt].sequence == actualSegment_){ + inflightInterests_[pkt].sequence == actualSegment_) { actualSegment_ = (actualSegment_ + 1) % HICN_MIN_PROBE_SEQ; continue; } @@ -489,7 +489,7 @@ void RTCTransportProtocol::scheduleNextInterests() { std::chrono::steady_clock::now().time_since_epoch()) .count(); - //here the packet can be in any state except for lost or recevied + // here the packet can be in any state except for lost or recevied inflightInterests_[pkt].state = sent_; inflightInterests_[pkt].sequence = actualSegment_; actualSegment_ = (actualSegment_ + 1) % HICN_MIN_PROBE_SEQ; @@ -506,22 +506,21 @@ void RTCTransportProtocol::addRetransmissions(uint32_t val) { void RTCTransportProtocol::addRetransmissions(uint32_t start, uint32_t stop) { uint64_t now = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); bool new_rtx = false; for (uint32_t i = start; i < stop; i++) { auto it = interestRetransmissions_.find(i); if (it == interestRetransmissions_.end()) { uint32_t pkt = i & modMask_; - if (lastSegNacked_ <= i && - inflightInterests_[pkt].state != received_) { + if (lastSegNacked_ <= i && inflightInterests_[pkt].state != received_) { // it must be larger than the last past nack received packetLost_++; interestRetransmissions_[i] = 0; uint32_t pkt = i & modMask_; - //we reset the transmission time setting to now, so that rtx will - //happne in one RTT on waint one inter arrival gap + // we reset the transmission time setting to now, so that rtx will + // happne in one RTT on waint one inter arrival gap inflightInterests_[pkt].transmissionTime = now; new_rtx = true; } @@ -529,10 +528,10 @@ void RTCTransportProtocol::addRetransmissions(uint32_t start, uint32_t stop) { // take care of it } - //in case a new rtx is added to the map we need to run checkRtx() - if(new_rtx){ - if(rtx_timer_used_){ - //if a timer is pending we need to delete it + // in case a new rtx is added to the map we need to run checkRtx() + if (new_rtx) { + if (rtx_timer_used_) { + // if a timer is pending we need to delete it rtx_timer_->cancel(); rtx_timer_used_ = false; } @@ -553,8 +552,8 @@ uint64_t RTCTransportProtocol::retransmit() { it = interestRetransmissions_.begin(); uint64_t smallest_timeout = ULONG_MAX; uint64_t now = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); while (it != interestRetransmissions_.end()) { uint32_t pkt = it->first & modMask_; @@ -580,46 +579,46 @@ uint64_t RTCTransportProtocol::retransmit() { uint64_t rtx_time = now; - if(it->second == 0) { - //first rtx - if(producerPathLabels_[0] != producerPathLabels_[1]){ - //multipath - if(pathTable_.find(producerPathLabels_[0]) != pathTable_.end() && - pathTable_.find(producerPathLabels_[1]) != pathTable_.end() && - (pathTable_[producerPathLabels_[0]]->getInterArrivalGap() < - HICN_MIN_INTER_ARRIVAL_GAP)){ + if (it->second == 0) { + // first rtx + if (producerPathLabels_[0] != producerPathLabels_[1]) { + // multipath + if (pathTable_.find(producerPathLabels_[0]) != pathTable_.end() && + pathTable_.find(producerPathLabels_[1]) != pathTable_.end() && + (pathTable_[producerPathLabels_[0]]->getInterArrivalGap() < + HICN_MIN_INTER_ARRIVAL_GAP)) { rtx_time = lastReceivedTime_ + - (pathTable_[producerPathLabels_[1]]->getMinRtt() - - pathTable_[producerPathLabels_[0]]->getMinRtt()) + - pathTable_[producerPathLabels_[0]]->getInterArrivalGap(); - }//else low rate producer, send it immediatly - }else{ - //single path - if(pathTable_.find(producerPathLabels_[0]) != pathTable_.end() && + (pathTable_[producerPathLabels_[1]]->getMinRtt() - + pathTable_[producerPathLabels_[0]]->getMinRtt()) + + pathTable_[producerPathLabels_[0]]->getInterArrivalGap(); + } // else low rate producer, send it immediatly + } else { + // single path + if (pathTable_.find(producerPathLabels_[0]) != pathTable_.end() && (pathTable_[producerPathLabels_[0]]->getInterArrivalGap() < - HICN_MIN_INTER_ARRIVAL_GAP)){ + HICN_MIN_INTER_ARRIVAL_GAP)) { rtx_time = lastReceivedTime_ + - pathTable_[producerPathLabels_[0]]->getInterArrivalGap(); - }//else low rate producer send immediatly + pathTable_[producerPathLabels_[0]]->getInterArrivalGap(); + } // else low rate producer send immediatly } - }else{ - //second or plus rtx, wait for the min rtt - if(pathTable_.find(producerPathLabels_[0]) != pathTable_.end()){ + } else { + // second or plus rtx, wait for the min rtt + if (pathTable_.find(producerPathLabels_[0]) != pathTable_.end()) { uint64_t sent_time = inflightInterests_[pkt].transmissionTime; rtx_time = sent_time + pathTable_[producerPathLabels_[0]]->getMinRtt(); - }//if we don't have info we send it immediatly + } // if we don't have info we send it immediatly } - if(now >= rtx_time){ + if (now >= rtx_time) { inflightInterests_[pkt].transmissionTime = now; it->second++; Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - &interest_name); + &interest_name); interest_name->setSuffix(it->first); sendInterest(interest_name, true); - }else if(rtx_time < smallest_timeout){ + } else if (rtx_time < smallest_timeout) { smallest_timeout = rtx_time; } @@ -629,7 +628,7 @@ uint64_t RTCTransportProtocol::retransmit() { } void RTCTransportProtocol::checkRtx() { - if(interestRetransmissions_.empty()){ + if (interestRetransmissions_.empty()) { rtx_timer_used_ = false; return; } @@ -637,10 +636,10 @@ void RTCTransportProtocol::checkRtx() { uint64_t next_timeout = retransmit(); uint64_t wait = 1; uint64_t now = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); - if(next_timeout != ULONG_MAX && now < next_timeout){ - wait = next_timeout - now; + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + if (next_timeout != ULONG_MAX && now < next_timeout) { + wait = next_timeout - now; } rtx_timer_used_ = true; rtx_timer_->expires_from_now(std::chrono::milliseconds(wait)); @@ -656,14 +655,14 @@ void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { uint32_t segmentNumber = interest->getName().getSuffix(); - if(segmentNumber >= HICN_MIN_PROBE_SEQ){ + if (segmentNumber >= HICN_MIN_PROBE_SEQ) { // this is a timeout on a probe, do nothing return; } uint32_t pkt = segmentNumber & modMask_; - if(TRANSPORT_EXPECT_FALSE(!firstPckReceived_)){ + if (TRANSPORT_EXPECT_FALSE(!firstPckReceived_)) { inflightInterestsCount_--; // we do nothing, and we keep asking the same stuff over // and over until we get at least a packet @@ -698,7 +697,8 @@ void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { scheduleNextInterests(); } -bool RTCTransportProtocol::onNack(const ContentObject &content_object, bool rtx) { +bool RTCTransportProtocol::onNack(const ContentObject &content_object, + bool rtx) { uint32_t *payload = (uint32_t *)content_object.getPayload()->data(); uint32_t productionSeg = *payload; uint32_t productionRate = *(++payload); @@ -708,11 +708,11 @@ bool RTCTransportProtocol::onNack(const ContentObject &content_object, bool rtx) // if we did not received anything between lastReceived_ + 1 and productionSeg // most likelly some packets got lost - if(lastReceived_ != 0){ + if (lastReceived_ != 0) { addRetransmissions(lastReceived_ + 1, productionSeg); } - if(!rtx){ + if (!rtx) { gotNack_ = true; // we synch the estimated production rate with the actual one estimatedBw_ = (double)productionRate; @@ -722,7 +722,7 @@ bool RTCTransportProtocol::onNack(const ContentObject &content_object, bool rtx) // we are asking for stuff produced in the past actualSegment_ = max(productionSeg, actualSegment_) % HICN_MIN_PROBE_SEQ; - if(!rtx) { + if (!rtx) { if (currentState_ == HICN_RTC_NORMAL_STATE) { currentState_ = HICN_RTC_SYNC_STATE; } @@ -737,7 +737,7 @@ bool RTCTransportProtocol::onNack(const ContentObject &content_object, bool rtx) } else if (productionSeg < nackSegment) { actualSegment_ = productionSeg % HICN_MIN_PROBE_SEQ; - if(!rtx){ + if (!rtx) { // we are asking stuff in the future gotFutureNack_++; computeMaxWindow(productionRate, 0); @@ -748,8 +748,8 @@ bool RTCTransportProtocol::onNack(const ContentObject &content_object, bool rtx) } } } else { - //we are asking the right thing, but the producer is slow - //keep doing the same until the packet is produced + // we are asking the right thing, but the producer is slow + // keep doing the same until the packet is produced actualSegment_ = productionSeg % HICN_MIN_PROBE_SEQ; } @@ -758,7 +758,6 @@ bool RTCTransportProtocol::onNack(const ContentObject &content_object, bool rtx) void RTCTransportProtocol::onContentObject( Interest::Ptr &&interest, ContentObject::Ptr &&content_object) { - // as soon as we get a packet firstPckReceived_ will never be false firstPckReceived_ = true; @@ -770,24 +769,25 @@ void RTCTransportProtocol::onContentObject( ConsumerContentObjectCallback *callback_content_object = nullptr; socket_->getSocketOption(ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT, &callback_content_object); - if (*callback_content_object != VOID_HANDLER) { + if (*callback_content_object) { (*callback_content_object)(*socket_, *content_object); } - if(segmentNumber >= HICN_MIN_PROBE_SEQ){ - if(segmentNumber == probe_seq_number_ && !received_probe_){ + if (segmentNumber >= HICN_MIN_PROBE_SEQ) { + if (segmentNumber == probe_seq_number_ && !received_probe_) { received_probe_ = true; uint32_t pathLabel = content_object->getPathLabel(); - if (pathTable_.find(pathLabel) == pathTable_.end()){ - //if this path does not exists we cannot create a new one so drop + if (pathTable_.find(pathLabel) == pathTable_.end()) { + // if this path does not exists we cannot create a new one so drop return; } - //this is the expected probe, update the RTT and drop the packet + // this is the expected probe, update the RTT and drop the packet uint64_t RTT = std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count() - time_sent_probe_; + std::chrono::steady_clock::now().time_since_epoch()) + .count() - + time_sent_probe_; pathTable_[pathLabel]->insertRttSample(RTT); pathTable_[pathLabel]->receivedNack(); @@ -803,22 +803,23 @@ void RTCTransportProtocol::onContentObject( bool old_nack = false; if (interestRetransmissions_.find(segmentNumber) == - interestRetransmissions_.end()){ - //this is not a retransmitted packet + interestRetransmissions_.end()) { + // this is not a retransmitted packet old_nack = onNack(*content_object, false); updateDelayStats(*content_object); } else { old_nack = onNack(*content_object, true); } - //the nacked_ state is used only to avoid to decrease inflightInterestsCount_ - //multiple times. In fact, every time that we receive an event related to an - //interest (timeout, nacked, content) we cange the state. In this way we are - //sure that we do not decrease twice the counter - if(old_nack){ + // the nacked_ state is used only to avoid to decrease + // inflightInterestsCount_ multiple times. In fact, every time that we + // receive an event related to an interest (timeout, nacked, content) we + // cange the state. In this way we are sure that we do not decrease twice the + // counter + if (old_nack) { inflightInterests_[pkt].state = lost_; interestRetransmissions_.erase(segmentNumber); - }else{ + } else { inflightInterests_[pkt].state = nacked_; } @@ -827,7 +828,7 @@ void RTCTransportProtocol::onContentObject( ((1 - HICN_ESTIMATED_PACKET_SIZE) * payload->length()); receivedBytes_ += (uint32_t)(content_object->headerSize() + - content_object->payloadSize()); + content_object->payloadSize()); if (inflightInterests_[pkt].state == sent_) { inflightInterestsCount_--; // packet sent without timeouts @@ -841,22 +842,21 @@ void RTCTransportProtocol::onContentObject( } addRetransmissions(lastReceived_ + 1, segmentNumber); - if(segmentNumber > highestReceived_){ + if (segmentNumber > highestReceived_) { highestReceived_ = segmentNumber; } - if(segmentNumber > lastReceived_){ + if (segmentNumber > lastReceived_) { lastReceived_ = segmentNumber; - lastReceivedTime_ = std::chrono::duration_cast< - std::chrono::milliseconds>( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + lastReceivedTime_ = + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); } receivedData_++; inflightInterests_[pkt].state = received_; auto it = interestRetransmissions_.find(segmentNumber); - if(it != interestRetransmissions_.end()) - lossRecovered_ ++; + if (it != interestRetransmissions_.end()) lossRecovered_++; interestRetransmissions_.erase(segmentNumber); diff --git a/libtransport/src/hicn/transport/utils/content_store.cc b/libtransport/src/hicn/transport/utils/content_store.cc index 1e6b9fcea..8e3435507 100644 --- a/libtransport/src/hicn/transport/utils/content_store.cc +++ b/libtransport/src/hicn/transport/utils/content_store.cc @@ -85,12 +85,17 @@ void ContentStore::erase(const Name &exact_name) { } void ContentStore::setLimit(size_t max_packets) { + utils::SpinLock::Acquire locked(cs_mutex_); max_content_store_size_ = max_packets; } -std::size_t ContentStore::getLimit() const { return max_content_store_size_; } +std::size_t ContentStore::getLimit() const { + utils::SpinLock::Acquire locked(cs_mutex_); + return max_content_store_size_; +} std::size_t ContentStore::size() const { + utils::SpinLock::Acquire locked(cs_mutex_); return content_store_hash_table_.size(); } diff --git a/libtransport/src/hicn/transport/utils/content_store.h b/libtransport/src/hicn/transport/utils/content_store.h index a89403a01..f7dc41835 100644 --- a/libtransport/src/hicn/transport/utils/content_store.h +++ b/libtransport/src/hicn/transport/utils/content_store.h @@ -68,8 +68,9 @@ class ContentStore { ContentStoreHashTable content_store_hash_table_; FIFOList fifo_list_; std::shared_ptr empty_reference_; - std::size_t max_content_store_size_; - utils::SpinLock cs_mutex_; + // Must be atomic + std::atomic_size_t max_content_store_size_; + mutable utils::SpinLock cs_mutex_; }; } // end namespace utils \ No newline at end of file -- cgit 1.2.3-korg