aboutsummaryrefslogtreecommitdiffstats
path: root/libtransport
diff options
context:
space:
mode:
authorMichele Papalini <micpapal@cisco.com>2019-10-22 10:00:39 +0000
committerGerrit Code Review <gerrit@fd.io>2019-10-22 10:00:39 +0000
commitc37f17534d4f7eaab692a5ffb252858d48358679 (patch)
treefb128fe2d59ef551016a4d063fee13762856000b /libtransport
parent49b9891dc32ea1fbd0af4711a200b425c7482da0 (diff)
parent755c6833ae2d2eee87e80ed3b84c75e968f48c46 (diff)
Merge "[HICN-328] Reworking setSocketOption and getSocketOption to be thread-safe"
Diffstat (limited to 'libtransport')
-rw-r--r--libtransport/src/hicn/transport/interfaces/CMakeLists.txt1
-rw-r--r--libtransport/src/hicn/transport/interfaces/callbacks.cc26
-rw-r--r--libtransport/src/hicn/transport/interfaces/callbacks.h4
-rw-r--r--libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc158
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket.h2
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_consumer.cc717
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_consumer.h673
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_options_default_values.h2
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_options_keys.h31
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_producer.cc659
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_producer.h508
-rw-r--r--libtransport/src/hicn/transport/protocols/protocol.cc15
-rw-r--r--libtransport/src/hicn/transport/protocols/protocol.h6
-rw-r--r--libtransport/src/hicn/transport/protocols/raaqm.cc40
-rw-r--r--libtransport/src/hicn/transport/protocols/rtc.cc266
-rw-r--r--libtransport/src/hicn/transport/utils/content_store.cc7
-rw-r--r--libtransport/src/hicn/transport/utils/content_store.h5
17 files changed, 1755 insertions, 1365 deletions
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 <functional>
#include <system_error>
+#include <hicn/transport/core/facade.h>
+
namespace utils {
class MemBuf;
}
@@ -105,6 +107,8 @@ using ProducerContentObjectCallback =
using ProducerInterestCallback =
std::function<void(ProducerSocket &, core::Interest &)>;
+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 <hicn/transport/interfaces/callbacks.h>
#include <hicn/transport/interfaces/rtc_socket_producer.h>
#include <stdlib.h>
#include <time.h>
@@ -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<asio::steady_timer>(
- this->getIoService());
- round_timer_ = std::make_unique<asio::steady_timer>(
- this->getIoService());
+ interests_cache_timer_ =
+ std::make_unique<asio::steady_timer>(this->getIoService());
+ round_timer_ = std::make_unique<asio::steady_timer>(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<asio::steady_timer>(
- this->getIoService());
- round_timer_ = std::make_unique<asio::steady_timer>(
- this->getIoService());
+ interests_cache_timer_ =
+ std::make_unique<asio::steady_timer>(this->getIoService());
+ round_timer_ = std::make_unique<asio::steady_timer>(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<utils::MemBuf> &&buffer) {
producedBytes_ += (uint32_t)(buffer_size + headerSize_ + TIMESTAMP_LEN);
producedPackets_++;
- auto content_object = std::make_shared<ContentObject>(
- flowName_.setSuffix(currentSeg_.load()));
+ auto content_object =
+ std::make_shared<ContentObject>(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<utils::MemBuf> &&buffer) {
output_buffer_.insert(std::static_pointer_cast<ContentObject>(
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::milliseconds>(
- 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<uint64_t,uint32_t>(expiration, interestSeg));
+ timers_map_.insert(
+ std::pair<uint64_t, uint32_t>(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<uint64_t,uint32_t>(expiration, interestSeg));
- seqs_map_.insert(std::pair<uint32_t,uint64_t>(interestSeg, expiration));
+ timers_map_.insert(
+ std::pair<uint64_t, uint32_t>(expiration, interestSeg));
+ seqs_map_.insert(std::pair<uint32_t, uint64_t>(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::milliseconds>(
- 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<utils::Verifier>()),
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<CbrTransportProtocol>(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 <typename Lambda, typename arg2>
+int ConsumerSocket::rescheduleOnIOService(int socket_option_key,
+ arg2 socket_option_value,
+ Lambda lambda_func) {
+ // To enforce type check
+ std::function<int(int, arg2)> 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<std::mutex> lck(mtx);
+ done = true;
+ result = func(socket_option_key, socket_option_value);
+ });
+ std::unique_lock<std::mutex> 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<utils::Verifier> &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<Portal> &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<utils::Verifier> &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
@@ -179,20 +179,6 @@ class ConsumerSocket : public BaseSocket {
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<utils::Verifier> &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<utils::Verifier> &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<Portal> &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<Portal> &socket_option_value);
+
+ virtual int getSocketOption(int socket_option_key,
+ IcnObserver **socket_option_value);
+
+ virtual int getSocketOption(
int socket_option_key,
- std::shared_ptr<utils::Verifier> &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<utils::Verifier> &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 <typename Lambda, typename arg2>
+ 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> 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<utils::Verifier> 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<TransportProtocol> 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<ContentObject> &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<ContentObject>(
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<utils::Identity> 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<utils::CryptoHash> 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<float>(
- 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<utils::CryptoHash>(
- hash, hash_size, static_cast<utils::CryptoHashType>(hash_algorithm_));
+ hash, hash_size, static_cast<utils::CryptoHashType>(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<ContentObject>(
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 <typename Lambda, typename arg2>
+int ProducerSocket::rescheduleOnIOService(int socket_option_key,
+ arg2 socket_option_value,
+ Lambda lambda_func) {
+ // To enforce type check
+ std::function<int(int, arg2)> 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<std::mutex> lck(mtx);
+ done = true;
+ result = func(socket_option_key, socket_option_value);
+ });
+ std::unique_lock<std::mutex> 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 <typename Lambda, typename arg2>
+int ProducerSocket::rescheduleOnIOServiceWithReference(
+ int socket_option_key, arg2 &socket_option_value, Lambda lambda_func) {
+ // To enforce type check
+ std::function<int(int, arg2 &)> 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<std::mutex> lck(mtx);
+ bool done = false;
+ io_service_.dispatch([this, &socket_option_key, &socket_option_value, &mtx,
+ &cv, &result, &done, &func]() {
+ std::unique_lock<std::mutex> 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<Prefix> 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<utils::Identity> &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<Prefix> &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<Portal> &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<utils::Identity> &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<BasePortal>,
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<Prefix> 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<Prefix> 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<utils::Identity> &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<utils::Identity> &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<Prefix> &socket_option_value);
- TRANSPORT_ALWAYS_INLINE int getSocketOption(
- int socket_option_key, std::list<Prefix> &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<Portal> &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<Portal> &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<utils::Identity> &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<utils::Identity> &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 <typename Lambda, typename arg2>
+ 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 <typename Lambda, typename arg2>
+ 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> portal_;
- std::size_t data_packet_size_;
- std::list<Prefix> served_namespaces_;
- uint32_t content_object_expiry_time_;
+ std::atomic<size_t> data_packet_size_;
+ std::list<Prefix>
+ served_namespaces_; // No need to be threadsafe, this is always modified
+ // by the application thread
+ std::atomic<uint32_t> 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<bool> making_manifest_;
// map for storing sequence numbers for several calls of the publish
// function
std::unordered_map<Name, std::unordered_map<int, uint32_t>> seq_number_map_;
- int signature_type_;
- int signature_size_;
-
- HashAlgorithm hash_algorithm_;
- utils::CryptoSuite crypto_suite_;
+ std::atomic<HashAlgorithm> hash_algorithm_;
+ std::atomic<utils::CryptoSuite> crypto_suite_;
+ utils::SpinLock identity_lock_;
std::shared_ptr<utils::Identity> identity_;
- // buffers
-
- std::queue<std::shared_ptr<const Interest>> 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 <atomic>
+
#include <hicn/transport/interfaces/socket.h>
#include <hicn/transport/protocols/packet_manager.h>
#include <hicn/transport/protocols/statistics.h>
@@ -60,7 +62,9 @@ class TransportProtocol : public interface::BasePortal::ConsumerCallback,
protected:
interface::ConsumerSocket *socket_;
std::shared_ptr<interface::BasePortal> portal_;
- volatile bool is_running_;
+ std::atomic<bool> is_running_;
+ // True if it si the first time we schedule an interest
+ std::atomic<bool> 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<utils::Milliseconds>(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::milliseconds>(
- 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::milliseconds>(
- 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<uint32_t> idis(
- HICN_MIN_PROBE_SEQ, HICN_MAX_PROBE_SEQ);
+ std::uniform_int_distribution<uint32_t> 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::milliseconds>(
- 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::milliseconds>(
- 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::milliseconds>(
- 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::milliseconds>(
- 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::milliseconds>(
+ 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<ContentObject> 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