diff options
87 files changed, 2915 insertions, 4264 deletions
diff --git a/hicn-light/CMakeLists.txt b/hicn-light/CMakeLists.txt index 0d9e4d261..eb6a50121 100644 --- a/hicn-light/CMakeLists.txt +++ b/hicn-light/CMakeLists.txt @@ -1,3 +1,16 @@ +# 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. + cmake_minimum_required(VERSION 3.5 FATAL_ERROR) project(hicn-light) diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt index a18777cd1..0c10759fa 100644 --- a/lib/CMakeLists.txt +++ b/lib/CMakeLists.txt @@ -12,8 +12,8 @@ # limitations under the License. cmake_minimum_required (VERSION 3.5 FATAL_ERROR) -project(Hicn C) -include(CTest) +project(libhicn C) +#include(CTest) set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} diff --git a/lib/src/CMakeLists.txt b/lib/src/CMakeLists.txt index 453119d13..d7c39831a 100644 --- a/lib/src/CMakeLists.txt +++ b/lib/src/CMakeLists.txt @@ -57,7 +57,7 @@ include(WindowsMacros) build_library(${LIBHICN} SHARED STATIC - SOURCES ${LIBHICN_SOURCE_FILES} + SOURCES ${LIBHICN_SOURCE_FILES} ${LIBHICN_HEADER_FILES} ${LIBHICN_HEADER_FILES_PROTOCOL} COMPONENT lib${LIBHICN} INCLUDE_DIRS ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_SOURCE_DIR}/.. DEFINITIONS ${COMPILER_DEFINITIONS} diff --git a/libtransport/CMakeLists.txt b/libtransport/CMakeLists.txt index d2567ed77..131067888 100644 --- a/libtransport/CMakeLists.txt +++ b/libtransport/CMakeLists.txt @@ -13,7 +13,7 @@ cmake_minimum_required(VERSION 3.5 FATAL_ERROR) -project(Transport VERSION 1.0) +project(libtransport) set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} @@ -39,8 +39,6 @@ set(TRANSPORT_HTTP ${TRANSPORT_ROOT_PATH}/http) set(TRANSPORT_PORTABILITY ${TRANSPORT_ROOT_PATH}/portability) set(TRANSPORT_INTERFACES ${TRANSPORT_ROOT_PATH}/interfaces) -set(raaqm_config_path ${CMAKE_INSTALL_PREFIX}/etc/hicn-consumer.conf) - # Install includes set(INSTALL_INCLUDE_DIR include/hicn/transport) diff --git a/libtransport/src/hicn/transport/CMakeLists.txt b/libtransport/src/hicn/transport/CMakeLists.txt index bc827ad7f..82d478b65 100644 --- a/libtransport/src/hicn/transport/CMakeLists.txt +++ b/libtransport/src/hicn/transport/CMakeLists.txt @@ -52,7 +52,7 @@ endif() build_library(${LIBTRANSPORT} STATIC SHARED - SOURCES ${SOURCE_FILES} + SOURCES ${SOURCE_FILES} ${HEADER_FILES} INSTALL_HEADERS ${HEADER_FILES} LINK_LIBRARIES ${LIBRARIES} DEPENDS ${DEPENDENCIES} diff --git a/libtransport/src/hicn/transport/core/content_object.cc b/libtransport/src/hicn/transport/core/content_object.cc index 4cd5623c4..c00535512 100644 --- a/libtransport/src/hicn/transport/core/content_object.cc +++ b/libtransport/src/hicn/transport/core/content_object.cc @@ -92,6 +92,8 @@ const Name &ContentObject::getName() const { return name_; } +Name &ContentObject::getWritableName() { return const_cast<Name &>(getName()); } + ContentObject &ContentObject::setName(const Name &name) { if (hicn_data_set_name(format_, (hicn_header_t *)packet_start_, name.getStructReference()) < 0) { diff --git a/libtransport/src/hicn/transport/core/content_object.h b/libtransport/src/hicn/transport/core/content_object.h index c85259f20..8770e8cb4 100644 --- a/libtransport/src/hicn/transport/core/content_object.h +++ b/libtransport/src/hicn/transport/core/content_object.h @@ -46,7 +46,9 @@ class ContentObject : public Packet { ~ContentObject() override; - const Name &getName() const; + const Name &getName() const override; + + Name &getWritableName() override; ContentObject &setName(const Name &name); diff --git a/libtransport/src/hicn/transport/core/forwarder_interface.h b/libtransport/src/hicn/transport/core/forwarder_interface.h index de9f3b568..b3403cf5a 100644 --- a/libtransport/src/hicn/transport/core/forwarder_interface.h +++ b/libtransport/src/hicn/transport/core/forwarder_interface.h @@ -18,6 +18,8 @@ #include <hicn/transport/core/prefix.h> #include <hicn/transport/core/udp_socket_connector.h> #include <hicn/transport/portability/portability.h> +#include <hicn/transport/utils/chrono_typedefs.h> +#include <hicn/transport/utils/log.h> #include <deque> @@ -97,6 +99,9 @@ class ForwarderInterface { packet.setLocator(inet6_address_); } + // TRANSPORT_LOGI("Sending packet %s at %lu", + // packet.getName().toString().c_str(), + // utils::SteadyClock::now().time_since_epoch().count()); packet.setChecksum(); connector_.send(packet.data()); } diff --git a/libtransport/src/hicn/transport/core/interest.cc b/libtransport/src/hicn/transport/core/interest.cc index 4ea0c4419..49d452d19 100644 --- a/libtransport/src/hicn/transport/core/interest.cc +++ b/libtransport/src/hicn/transport/core/interest.cc @@ -79,16 +79,7 @@ const Name &Interest::getName() const { return name_; } -Name &Interest::getWritableName() { - if (!name_) { - if (hicn_interest_get_name(format_, (hicn_header_t *)packet_start_, - (hicn_name_t *)name_.getStructReference()) < 0) { - throw errors::MalformedPacketException(); - } - } - - return name_; -} +Name &Interest::getWritableName() { return const_cast<Name &>(getName()); } Interest &Interest::setName(const Name &name) { if (hicn_interest_set_name(format_, (hicn_header_t *)packet_start_, @@ -148,4 +139,4 @@ void Interest::resetForHash() { } // end namespace core -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/core/interest.h b/libtransport/src/hicn/transport/core/interest.h index 75fcba8eb..95b45ac9a 100644 --- a/libtransport/src/hicn/transport/core/interest.h +++ b/libtransport/src/hicn/transport/core/interest.h @@ -45,9 +45,9 @@ class Interest ~Interest() override; - const Name &getName() const; + const Name &getName() const override; - Name &getWritableName(); + Name &getWritableName() override; Interest &setName(const Name &name); diff --git a/libtransport/src/hicn/transport/core/manifest.h b/libtransport/src/hicn/transport/core/manifest.h index 558a96804..9f7dc5984 100644 --- a/libtransport/src/hicn/transport/core/manifest.h +++ b/libtransport/src/hicn/transport/core/manifest.h @@ -38,39 +38,28 @@ class Manifest : public Base { using Encoder = typename FormatTraits::Encoder; using Decoder = typename FormatTraits::Decoder; - Manifest() - : packet_(new Base(HF_INET6_TCP_AH), nullptr), - encoder_(*packet_), - decoder_(*packet_) { + Manifest(std::size_t signature_size = 0) + : Base(HF_INET6_TCP_AH), + encoder_(*this, signature_size), + decoder_(*this) { Base::setPayloadType(PayloadType::MANIFEST); } - Manifest(const core::Name &name) - : packet_(new Base(name, HF_INET6_TCP_AH), nullptr), - encoder_(*packet_), - decoder_(*packet_) { - Base::setPayloadType(PayloadType::MANIFEST); - } - - Manifest(typename Base::Ptr &&base) - : packet_(std::move(base)), encoder_(*packet_), decoder_(*packet_) { + Manifest(const core::Name &name, std::size_t signature_size = 0) + : Base(name, HF_INET6_TCP_AH), + encoder_(*this, signature_size), + decoder_(*this) { Base::setPayloadType(PayloadType::MANIFEST); } template <typename T> Manifest(T &&base) - : packet_(new Base(std::move<T &&>(base)), nullptr), - encoder_(*packet_), - decoder_(*packet_) { + : Base(std::forward<T &&>(base)), encoder_(*this), decoder_(*this) { Base::setPayloadType(PayloadType::MANIFEST); } virtual ~Manifest() = default; - bool operator==(const Manifest &other) { - return this->packet_ == other.packet_; - } - std::size_t estimateManifestSize(std::size_t additional_entries = 0) { return static_cast<ManifestImpl &>(*this).estimateManifestSizeImpl( additional_entries); @@ -142,15 +131,7 @@ class Manifest : public Base { return *this; } - void setSignatureSize(std::size_t size_bits) { - Packet::setSignatureSize(size_bits); - encoder_.update(); - } - - typename Base::Ptr &&getPacket() { return std::move(packet_); } - protected: - typename Base::Ptr packet_; ManifestType manifest_type_; HashAlgorithm hash_algorithm_; bool is_last_; diff --git a/libtransport/src/hicn/transport/core/manifest_format_fixed.cc b/libtransport/src/hicn/transport/core/manifest_format_fixed.cc index 6141ae311..73b33268c 100644 --- a/libtransport/src/hicn/transport/core/manifest_format_fixed.cc +++ b/libtransport/src/hicn/transport/core/manifest_format_fixed.cc @@ -15,7 +15,6 @@ #include <hicn/transport/core/manifest_format_fixed.h> #include <hicn/transport/core/packet.h> -#include <hicn/transport/utils/endianess.h> #include <hicn/transport/utils/literals.h> namespace transport { @@ -23,26 +22,34 @@ namespace transport { namespace core { // TODO use preallocated pool of membufs -FixedManifestEncoder::FixedManifestEncoder(Packet &packet) +FixedManifestEncoder::FixedManifestEncoder(Packet &packet, + std::size_t signature_size) : packet_(packet), - max_size_(Packet::default_mtu - packet_.headerSize()), + max_size_(Packet::default_mtu - packet_.headerSize() - signature_size), manifest_( utils::MemBuf::create(Packet::default_mtu - packet_.headerSize())), manifest_header_( reinterpret_cast<ManifestHeader *>(manifest_->writableData())), manifest_entries_(reinterpret_cast<ManifestEntry *>( manifest_->writableData() + sizeof(ManifestHeader))), - current_entry_(0) {} + current_entry_(0), + signature_size_(signature_size) { + *manifest_header_ = {0}; +} FixedManifestEncoder::~FixedManifestEncoder() {} FixedManifestEncoder &FixedManifestEncoder::encodeImpl() { + manifest_->append(sizeof(ManifestHeader) + + manifest_header_->number_of_entries * + sizeof(ManifestEntry)); packet_.appendPayload(std::move(manifest_)); return *this; } FixedManifestEncoder &FixedManifestEncoder::clearImpl() { - manifest_ = utils::MemBuf::create(Packet::default_mtu - packet_.headerSize()); + manifest_ = utils::MemBuf::create(Packet::default_mtu - packet_.headerSize() - + signature_size_); return *this; } @@ -84,7 +91,7 @@ FixedManifestEncoder &FixedManifestEncoder::addSuffixAndHashImpl( void FixedManifestEncoder::addSuffixHashBytes(uint32_t suffix, const uint8_t *hash, std::size_t length) { - manifest_entries_[current_entry_].suffix = utils::hton(suffix); + manifest_entries_[current_entry_].suffix = htonl(suffix); // std::copy(hash, hash + length, // manifest_entries_[current_entry_].hash); std::memcpy( @@ -119,18 +126,13 @@ std::size_t FixedManifestEncoder::estimateSerializedLengthImpl( } FixedManifestEncoder &FixedManifestEncoder::updateImpl() { - max_size_ = Packet::default_mtu - packet_.headerSize(); - manifest_header_ = reinterpret_cast<ManifestHeader *>( - const_cast<uint8_t *>(packet_.getPayload().data())); - manifest_entries_ = reinterpret_cast<ManifestEntry *>( - const_cast<uint8_t *>(packet_.getPayload().data()) + - sizeof(ManifestHeader)); + max_size_ = Packet::default_mtu - packet_.headerSize() - signature_size_; return *this; } FixedManifestEncoder &FixedManifestEncoder::setFinalBlockNumberImpl( std::uint32_t final_block_number) { - manifest_header_->final_block_number = utils::hton(final_block_number); + manifest_header_->final_block_number = htonl(final_block_number); return *this; } @@ -179,7 +181,7 @@ typename Fixed::SuffixList FixedManifestDecoder::getSuffixHashListImpl() { for (int i = 0; i < manifest_header_->number_of_entries; i++) { hash_list.insert(hash_list.end(), - std::make_pair(utils::ntoh(manifest_entries_[i].suffix), + std::make_pair(ntohl(manifest_entries_[i].suffix), reinterpret_cast<uint8_t *>( &manifest_entries_[i].hash[0]))); } @@ -213,7 +215,7 @@ std::size_t FixedManifestDecoder::estimateSerializedLengthImpl( } uint32_t FixedManifestDecoder::getFinalBlockNumberImpl() const { - return utils::ntoh(manifest_header_->final_block_number); + return ntohl(manifest_header_->final_block_number); } } // end namespace core diff --git a/libtransport/src/hicn/transport/core/manifest_format_fixed.h b/libtransport/src/hicn/transport/core/manifest_format_fixed.h index 82817ddfb..c9bc3f8e5 100644 --- a/libtransport/src/hicn/transport/core/manifest_format_fixed.h +++ b/libtransport/src/hicn/transport/core/manifest_format_fixed.h @@ -83,7 +83,7 @@ static const constexpr std::uint8_t manifest_version = 1; class FixedManifestEncoder : public ManifestEncoder<FixedManifestEncoder> { public: - FixedManifestEncoder(Packet &packet); + FixedManifestEncoder(Packet &packet, std::size_t signature_size = 0); ~FixedManifestEncoder(); @@ -126,6 +126,7 @@ class FixedManifestEncoder : public ManifestEncoder<FixedManifestEncoder> { ManifestHeader *manifest_header_; ManifestEntry *manifest_entries_; std::size_t current_entry_; + std::size_t signature_size_; }; class FixedManifestDecoder : public ManifestDecoder<FixedManifestDecoder> { diff --git a/libtransport/src/hicn/transport/core/manifest_format_json_jsoncpp.cc b/libtransport/src/hicn/transport/core/manifest_format_json_jsoncpp.cc deleted file mode 100644 index 0ffca39f6..000000000 --- a/libtransport/src/hicn/transport/core/manifest_format_json_jsoncpp.cc +++ /dev/null @@ -1,244 +0,0 @@ -/* - * 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 <hicn/transport/core/packet.h> -#include <hicn/transport/portability/transport_portability.h> - -#include <array> - -namespace transport { - -namespace core { - -namespace { - -template <typename T> -TRANSPORT_ALWAYS_INLINE void checkPointer(T *pointer) { - if (pointer == nullptr) { - throw errors::NullPointerException(); - } -} - -template <typename EnumType> -TRANSPORT_ALWAYS_INLINE void setValueToJson(Json::Value &root, EnumType value) { - root[JSONKey<EnumType>::key] = static_cast<uint8_t>(value); -} - -template <typename EnumType> -TRANSPORT_ALWAYS_INLINE EnumType getValueFromJson(const Json::Value &root) { - return static_cast<EnumType>(root[JSONKey<EnumType>::key].asUInt()); -}; - -} // namespace - -JSONManifestEncoder::JSONManifestEncoder(Packet &packet) : packet_(packet) {} - -JSONManifestEncoder::~JSONManifestEncoder() {} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder &JSONManifestEncoder::encodeImpl() { - Json::StreamWriterBuilder writer_builder; - Json::StreamWriter *fast_writer = writer_builder.newStreamWriter(); - - asio::streambuf strbuf; - strbuf.prepare(1500); - std::ostream stream(&strbuf); - fast_writer->write(root_, &stream); - - const uint8_t *buffer = asio::buffer_cast<const uint8_t *>(strbuf.data()); - - packet_.setPayload(buffer, strbuf.size()); - - delete fast_writer; - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder &JSONManifestEncoder::clearImpl() { - root_.clear(); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setHashAlgorithmImpl(HashAlgorithm algorithm) { - setValueToJson(root_, algorithm); - return *this; -} - -JSONManifestEncoder &JSONManifestEncoder::setManifestTypeImpl( - ManifestType manifest_type) { - setValueToJson(root_, manifest_type); - return *this; -} - -JSONManifestEncoder &JSONManifestEncoder::setNextSegmentCalculationStrategyImpl( - NextSegmentCalculationStrategy strategy) { - setValueToJson(root_, strategy); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setBaseNameImpl(const core::Name &base_name) { - root_[JSONKey<core::Name>::key] = base_name.toString().c_str(); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::addSuffixAndHashImpl(uint32_t suffix, - const utils::CryptoHash &hash) { - throw errors::NotImplementedException(); - // Json::Value value(Json::arrayValue); - // value.append(Json::Value(suffix)); - // value.append(Json::Value(Json::Value::UInt64 (hash))); - // root_[JSONKey<SuffixHashList>::key].append(value); - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setIsFinalManifestImpl(bool is_last) { - root_[JSONKey<bool>::final_manifest] = is_last; - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setVersionImpl(ManifestVersion version) { - setValueToJson(root_, version); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setSuffixHashListImpl( - const typename JSON::SuffixList &name_hash_list) { - throw errors::NotImplementedException(); - // for (auto &suffix : name_hash_list) { - // addSuffixAndHashImpl(suffix.first, suffix.second); - // } - // - // return *this; -} - -TRANSPORT_ALWAYS_INLINE std::size_t -JSONManifestEncoder::estimateSerializedLengthImpl( - std::size_t number_of_entries) { - Json::StreamWriterBuilder writer_builder; - Json::StreamWriter *fast_writer = writer_builder.newStreamWriter(); - - asio::streambuf strbuf; - strbuf.prepare(1500); - std::ostream stream(&strbuf); - fast_writer->write(root_, &stream); - - return strbuf.size(); -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder &JSONManifestEncoder::updateImpl() { - throw errors::NotImplementedException(); -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setFinalBlockNumberImpl(std::uint32_t final_block_number) { - throw errors::NotImplementedException(); -} - -TRANSPORT_ALWAYS_INLINE std::size_t -JSONManifestEncoder::getManifestHeaderSizeImpl() { - return 0; -} - -JSONManifestDecoder::JSONManifestDecoder(Packet &packet) : packet_(packet) {} - -JSONManifestDecoder::~JSONManifestDecoder() {} - -TRANSPORT_ALWAYS_INLINE void JSONManifestDecoder::decodeImpl() { - auto array = packet_.getPayload(); - auto payload = array.data(); - auto payload_size = array.length(); - - Json::CharReaderBuilder reader_builder; - Json::CharReader *reader = reader_builder.newCharReader(); - std::string errors; - - if (!reader->parse((char *)payload, (char *)payload + payload_size, &root_, - &errors)) { - TRANSPORT_LOGE("Error parsing manifest!"); - TRANSPORT_LOGE("%s", errors.c_str()); - - delete reader; - - throw errors::MalformedPacketException(); - } - - delete reader; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestDecoder &JSONManifestDecoder::clearImpl() { - root_.clear(); - return *this; -} - -TRANSPORT_ALWAYS_INLINE ManifestType -JSONManifestDecoder::getManifestTypeImpl() const { - return getValueFromJson<ManifestType>(root_); -} - -TRANSPORT_ALWAYS_INLINE HashAlgorithm -JSONManifestDecoder::getHashAlgorithmImpl() const { - return getValueFromJson<HashAlgorithm>(root_); -} - -TRANSPORT_ALWAYS_INLINE NextSegmentCalculationStrategy -JSONManifestDecoder::getNextSegmentCalculationStrategyImpl() const { - return getValueFromJson<NextSegmentCalculationStrategy>(root_); -} - -TRANSPORT_ALWAYS_INLINE typename JSON::SuffixList -JSONManifestDecoder::getSuffixHashListImpl() { - throw errors::NotImplementedException(); - // SuffixHashList hash_list; - // - // Json::Value &array = root_[JSONKey<SuffixHashList>::key]; - // - // for (Json::Value::ArrayIndex i = 0; - // i != array.size(); - // i++) { - // hash_list[array[i][0].asUInt()] = array[i][1].asUInt64(); - // } - // - // return hash_list; -} - -TRANSPORT_ALWAYS_INLINE core::Name JSONManifestDecoder::getBaseNameImpl() - const { - return core::Name(root_[JSONKey<core::Name>::key].asCString()); -} - -TRANSPORT_ALWAYS_INLINE bool JSONManifestDecoder::getIsFinalManifestImpl() - const { - return root_[JSONKey<bool>::final_manifest].asBool(); -} - -TRANSPORT_ALWAYS_INLINE ManifestVersion -JSONManifestDecoder::getVersionImpl() const { - return getValueFromJson<ManifestVersion>(root_); -} - -TRANSPORT_ALWAYS_INLINE uint32_t -JSONManifestDecoder::getFinalBlockNumberImpl() const { - return 0; -} - -} // end namespace core - -} // end namespace transport diff --git a/libtransport/src/hicn/transport/core/manifest_format_json_jsoncpp.h b/libtransport/src/hicn/transport/core/manifest_format_json_jsoncpp.h deleted file mode 100644 index 84d2ba29a..000000000 --- a/libtransport/src/hicn/transport/core/manifest_format_json_jsoncpp.h +++ /dev/null @@ -1,162 +0,0 @@ -/* - * 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. - */ - -#pragma once - -#include <hicn/transport/core/manifest_format.h> -#include <hicn/transport/core/name.h> - -#if defined(__APPLE__) || defined(__ANDROID__) -#include <json/json.h> -#else -#include <jsoncpp/json/json.h> -#endif /* __APPLE__ || __ANDROID__*/ - -#include <string> - -namespace transport { - -namespace core { - -class JSONManifestEncoder; -class JSONManifestDecoder; -class Packet; - -struct JSON { - using Encoder = JSONManifestEncoder; - using Decoder = JSONManifestDecoder; - using HashType = utils::CryptoHash; - using SuffixList = std::unordered_map<std::uint32_t, std::uint8_t *>; -}; - -template <typename T> -struct JSONKey; - -template <> -struct JSONKey<ManifestVersion> { - static const constexpr char *key = "manifest_version"; -}; - -template <> -struct JSONKey<HashAlgorithm> { - static const constexpr char *key = "hash_algorithm"; -}; - -template <> -struct JSONKey<ManifestType> { - static const constexpr char *key = "manifest_type"; -}; - -template <> -struct JSONKey<NextSegmentCalculationStrategy> { - static const constexpr char *key = "next_segment_strategy"; -}; - -template <> -struct JSONKey<typename JSON::SuffixList> { - static const constexpr char *key = "suffix_hash_list"; -}; - -template <> -struct JSONKey<core::Name> { - static const constexpr char *key = "base_name"; -}; - -template <> -struct JSONKey<bool> { - static const constexpr char *final_manifest = "final_manifest"; -}; - -class JSONManifestEncoder : public ManifestEncoder<JSONManifestEncoder> { - public: - JSONManifestEncoder(Packet &packet); - - ~JSONManifestEncoder() override; - - JSONManifestEncoder &encodeImpl(); - - JSONManifestEncoder &clearImpl(); - - JSONManifestEncoder &setManifestTypeImpl(ManifestType manifest_type); - - JSONManifestEncoder &setHashAlgorithmImpl(HashAlgorithm algorithm); - - JSONManifestEncoder &setNextSegmentCalculationStrategyImpl( - NextSegmentCalculationStrategy strategy); - - JSONManifestEncoder &setSuffixHashListImpl( - const typename JSON::SuffixList &name_hash_list); - - JSONManifestEncoder &setBaseNameImpl(const core::Name &base_name); - - JSONManifestEncoder &addSuffixAndHashImpl(uint32_t suffix, - const utils::CryptoHash &hash); - - JSONManifestEncoder &setIsFinalManifestImpl(bool is_last); - - JSONManifestEncoder &setVersionImpl(ManifestVersion version); - - std::size_t estimateSerializedLengthImpl(std::size_t number_of_entries); - - JSONManifestEncoder &updateImpl(); - - JSONManifestEncoder &setFinalBlockNumberImpl( - std::uint32_t final_block_number); - - static std::size_t getManifestHeaderSizeImpl(); - - private: - Packet &packet_; - Json::Value root_; -}; - -class JSONManifestDecoder : public ManifestDecoder<JSONManifestDecoder> { - public: - JSONManifestDecoder(Packet &packet); - - ~JSONManifestDecoder() override; - - void decodeImpl(); - - JSONManifestDecoder &clearImpl(); - - ManifestType getManifestTypeImpl() const; - - HashAlgorithm getHashAlgorithmImpl() const; - - uint32_t getFinalChunkImpl() const; - - NextSegmentCalculationStrategy getNextSegmentCalculationStrategyImpl() const; - - typename JSON::SuffixList getSuffixHashListImpl(); - - core::Name getBaseNameImpl() const; - - bool getIsFinalManifestImpl() const; - - std::size_t estimateSerializedLengthImpl(std::size_t number_of_entries) const; - - ManifestVersion getVersionImpl() const; - - uint32_t getFinalBlockNumberImpl() const; - - private: - Packet &packet_; - Json::Value root_; -}; - -} // namespace core - -} // namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/core/manifest_format_json_libparc_deprecated.cc b/libtransport/src/hicn/transport/core/manifest_format_json_libparc_deprecated.cc deleted file mode 100644 index a3a47e62b..000000000 --- a/libtransport/src/hicn/transport/core/manifest_format_json_libparc_deprecated.cc +++ /dev/null @@ -1,298 +0,0 @@ -/* - * 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 <hicn/transport/core/manifest_format_json_libparc_deprecated.h> -#include <hicn/transport/core/packet.h> -#include <hicn/transport/errors/errors.h> -#include <hicn/transport/portability/transport_portability.h> - -extern "C" { -#include <parc/algol/parc_Memory.h> -} - -namespace transport { - -namespace core { - -namespace { - -template <typename T> -TRANSPORT_ALWAYS_INLINE void checkPointer(T *pointer) { - if (pointer == nullptr) { - throw errors::NullPointerException(); - } -} - -template <typename EnumType> -TRANSPORT_ALWAYS_INLINE void setValueToJson(PARCJSON *root, EnumType value) { - parcJSON_AddInteger(root, JSONKey<EnumType>::key, - static_cast<int64_t>(value)); -} - -template <typename EnumType> -TRANSPORT_ALWAYS_INLINE EnumType getValueFromJson(PARCJSON *root) { - checkPointer(root); - - PARCJSONValue *value = parcJSON_GetValueByName(root, JSONKey<EnumType>::key); - - EnumType ret = static_cast<EnumType>(parcJSONValue_GetInteger(value)); - // parcJSONValue_Release(&value); - - return ret; -}; - -} // namespace - -JSONManifestEncoder::JSONManifestEncoder() : root_(parcJSON_Create()) { - parcJSON_Acquire(root_); -} - -JSONManifestEncoder::~JSONManifestEncoder() { - if (root_) { - parcJSON_Release(&root_); - } -} - -TRANSPORT_ALWAYS_INLINE SONManifestEncoder &JSONManifestEncoder::encodeImpl( - Packet &packet) { - char *json_string = parcJSON_ToString(root_); - packet.setPayload(reinterpret_cast<uint8_t *>(json_string), - std::strlen(json_string)); - parcMemory_Deallocate(&json_string); - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder &JSONManifestEncoder::clearImpl() { - if (root_) { - parcJSON_Release(&root_); - } - - root_ = parcJSON_Create(); - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setHashAlgorithmImpl(HashAlgorithm algorithm) { - setValueToJson(root_, algorithm); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setManifestTypeImpl(ManifestType manifest_type) { - setValueToJson(root_, manifest_type); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setNextSegmentCalculationStrategyImpl( - NextSegmentCalculationStrategy strategy) { - setValueToJson(root_, strategy); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setBaseNameImpl(const core::Name &base_name) { - parcJSON_AddString(root_, JSONKey<core::Name>::key, - base_name.toString().c_str()); - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::addSuffixAndHashImpl(uint32_t suffix, - utils::CryptoHash &hash) { - throw errors::NotImplementedException(); - // PARCJSONValue *value = parcJSON_GetValueByName(root_, - // JSONKey<SuffixHashList>::key); - // - // // Create the pair to store in the array. - // // It will be segment number + Hash of the segment - // PARCJSONArray * pair = parcJSONArray_Create(); - // - // PARCJSONValue *v = parcJSONValue_CreateFromInteger(suffix); - // parcJSONArray_AddValue(pair, v); - // parcJSONValue_Release(&v); - // - // v = parcJSONValue_CreateFromInteger(hash); - // parcJSONArray_AddValue(pair, v); - // parcJSONValue_Release(&v); - // - // if (value == nullptr /* || !parcJSONValue_IsArray(value) */) { - // // Create the array - // PARCJSONArray *array = parcJSONArray_Create(); - // parcJSON_AddArray(root_, - // JSONKey<SuffixHashList>::key, - // array); - // parcJSONArray_Release(&array); - // - // value = parcJSON_GetValueByName(root_, JSONKey<SuffixHashList>::key); - // } - // - // v = parcJSONValue_CreateFromJSONArray(pair); - // parcJSONArray_AddValue(parcJSONValue_GetArray(value), v); - // parcJSONValue_Release(&v); - // - // parcJSONArray_Release(&pair); - // // parcJSONValue_Release(&value); - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setIsFinalManifestImpl(bool is_last) { - parcJSON_AddBoolean(root_, JSONKey<bool>::final_manifest, is_last); - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestEncoder & -JSONManifestEncoder::setSuffixHashListImpl( - const SuffixHashList &name_hash_list) { - for (auto &suffix : name_hash_list) { - addSuffixAndHashImpl(suffix.first, suffix.second); - } - - return *this; -} - -TRANSPORT_ALWAYS_INLINE JSONManifestDecoder::JSONManifestDecoder() - : root_(nullptr) {} - -TRANSPORT_ALWAYS_INLINE JSONManifestDecoder::~JSONManifestDecoder() { - if (root_) { - parcJSON_Release(&root_); - } -} - -TRANSPORT_ALWAYS_INLINE void JSONManifestDecoder::decodeImpl( - const uint8_t *payload, std::size_t payload_size) { - PARCBuffer *b = parcBuffer_Wrap(const_cast<uint8_t *>(payload), payload_size, - 0, payload_size); - clearImpl(); - - root_ = parcJSON_ParseBuffer(b); - parcBuffer_Release(&b); - - char *str = parcJSON_ToString(root_); -} - -TRANSPORT_ALWAYS_INLINE JSONManifestDecoder &JSONManifestDecoder::clearImpl() { - if (root_) { - parcJSON_Release(&root_); - } - - return *this; -} - -TRANSPORT_ALWAYS_INLINE ManifestType -JSONManifestDecoder::getManifestTypeImpl() const { - return getValueFromJson<ManifestType>(root_); -} - -TRANSPORT_ALWAYS_INLINE HashAlgorithm -JSONManifestDecoder::getHashAlgorithmImpl() const { - return getValueFromJson<HashAlgorithm>(root_); -} - -TRANSPORT_ALWAYS_INLINE NextSegmentCalculationStrategy -JSONManifestDecoder::getNextSegmentCalculationStrategyImpl() const { - return getValueFromJson<NextSegmentCalculationStrategy>(root_); -} - -TRANSPORT_ALWAYS_INLINE SuffixHashList -JSONManifestDecoder::getSuffixHashListImpl() { - throw errors::NotImplementedException(); - // SuffixHashList hash_list; - // - // char * str = parcJSON_ToString(root_); - // - // PARCJSONValue *value = parcJSON_GetValueByName(root_, - // JSONKey<SuffixHashList>::key); - // - // if (value == nullptr || !parcJSONValue_IsArray(value)) { - // throw errors::RuntimeException("Manifest does not contain suffix-hash - // list"); - // } - // - // PARCJSONArray *array = parcJSONValue_GetArray(value); - // std::size_t array_size = parcJSONArray_GetLength(array); - // - // for (std::size_t i = 0; i < array_size; i++) { - // PARCJSONValue *v = parcJSONArray_GetValue(array, i); - // checkPointer(v); - // PARCJSONArray *a = parcJSONValue_GetArray(v); - // PARCJSONValue *_suffix = parcJSONArray_GetValue(a, 0); - // PARCJSONValue *_hash = parcJSONArray_GetValue(a, 1); - // - // uint32_t value1 = - // static_cast<uint32_t>(parcJSONValue_GetInteger(_suffix)); uint64_t - // value2 = static_cast<uint64_t>(parcJSONValue_GetInteger(_hash)); - // - // hash_list[static_cast<uint32_t>(parcJSONValue_GetInteger(_suffix))] = - // static_cast<uint64_t>(parcJSONValue_GetInteger(_hash)); - // - //// parcJSONValue_Release(&_hash); - //// parcJSONValue_Release(&_suffix); - //// parcJSONArray_Release(&a); - //// parcJSONValue_Release(&v); - // } - // - //// parcJSONArray_Release(&array); - //// parcJSONValue_Release(&value); - // - // char * str2 = parcJSON_ToString(root_); - // - // return hash_list; -} - -TRANSPORT_ALWAYS_INLINE core::Name JSONManifestDecoder::getBaseNameImpl() - const { - checkPointer(root_); - PARCJSONValue *value = - parcJSON_GetValueByName(root_, JSONKey<core::Name>::key); - - PARCBuffer *b = parcJSONValue_GetString(value); - char *string = parcBuffer_ToString(b); - - core::Name ret(string); - - // parcJSONValue_Release(&value); - parcMemory_Deallocate(&string); - - return ret; -} - -TRANSPORT_ALWAYS_INLINE bool JSONManifestDecoder::getIsFinalManifestImpl() { - checkPointer(root_); - PARCJSONValue *value = - parcJSON_GetValueByName(root_, JSONKey<bool>::final_manifest); - - bool ret = parcJSONValue_GetBoolean(value); - - // parcJSONValue_Release(&value); - - return ret; -} - -TRANSPORT_ALWAYS_INLINE std::size_t -JSONManifestDecoder::estimateSerializedLengthImpl( - std::size_t number_of_entries) { - return 0; -} - -} // end namespace core - -} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/core/manifest_format_json_libparc_deprecated.h b/libtransport/src/hicn/transport/core/manifest_format_json_libparc_deprecated.h deleted file mode 100644 index c89f4af7b..000000000 --- a/libtransport/src/hicn/transport/core/manifest_format_json_libparc_deprecated.h +++ /dev/null @@ -1,152 +0,0 @@ -/* - * 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. - */ - -#pragma once - -#include <hicn/transport/core/manifest_format.h> -#include <hicn/transport/core/name.h> - -extern "C" { -#include <parc/algol/parc_JSON.h> -} - -#include <string> - -namespace transport { - -namespace core { - -class JSONManifestEncoder; -class JSONManifestDecoder; -class Packet; - -struct JSON { - using Encoder = JSONManifestEncoder; - using Decoder = JSONManifestDecoder; -}; - -template <typename T> -struct JSONKey; - -template <> -struct JSONKey<HashAlgorithm> { - static const constexpr char *key = "hash_algorithm"; -}; - -template <> -struct JSONKey<ManifestType> { - static const constexpr char *key = "manifest_type"; -}; - -template <> -struct JSONKey<NextSegmentCalculationStrategy> { - static const constexpr char *key = "next_segment_strategy"; -}; - -template <> -struct JSONKey<NameHashList> { - static const constexpr char *key = "name_hash_list"; -}; - -template <> -struct JSONKey<SuffixHashList> { - static const constexpr char *key = "suffix_hash_list"; -}; - -template <> -struct JSONKey<core::Name> { - static const constexpr char *key = "base_name"; -}; - -template <> -struct JSONKey<bool> { - static const constexpr char *final_manifest = "final_manifest"; -}; - -// template <> -// struct JSONKey<base_name> { -// static const std::string key = "name_hash_list"; -//}; - -// namespace JSONManifestEncoding { -// static const std::string base_name = "base_name"; -// static const std::string final_chunk_number = "final_chunk_number"; -// static const std::string hash_algorithm = "hash_algorithm"; -// static const std::string manifest_type = "manifest_type"; -// static const std::string name_hash_list = "name_hash_list"; -// static const std::string next_segment_strategy = "next_segment_strategy"; -//} - -class JSONManifestEncoder : public ManifestEncoder<JSONManifestEncoder> { - public: - JSONManifestEncoder(); - - ~JSONManifestEncoder(); - - JSONManifestEncoder &encodeImpl(Packet &packet); - - JSONManifestEncoder &clearImpl(); - - JSONManifestEncoder &setManifestTypeImpl(ManifestType manifest_type); - - JSONManifestEncoder &setHashAlgorithmImpl(HashAlgorithm algorithm); - - JSONManifestEncoder &setNextSegmentCalculationStrategyImpl( - NextSegmentCalculationStrategy strategy); - - JSONManifestEncoder &setSuffixHashListImpl( - const SuffixHashList &name_hash_list); - - JSONManifestEncoder &setBaseNameImpl(const core::Name &base_name); - - JSONManifestEncoder &addSuffixAndHashImpl(uint32_t suffix, uint64_t hash); - - JSONManifestEncoder &setIsFinalManifestImpl(bool is_last); - - private: - PARCJSON *root_; -}; - -class JSONManifestDecoder : public ManifestDecoder<JSONManifestDecoder> { - public: - JSONManifestDecoder(); - - ~JSONManifestDecoder(); - - void decodeImpl(const uint8_t *payload, std::size_t payload_size); - - JSONManifestDecoder &clearImpl(); - - ManifestType getManifestTypeImpl() const; - - HashAlgorithm getHashAlgorithmImpl() const; - - uint32_t getFinalChunkImpl() const; - - NextSegmentCalculationStrategy getNextSegmentCalculationStrategyImpl() const; - - SuffixHashList getSuffixHashListImpl(); - - core::Name getBaseNameImpl() const; - - bool getIsFinalManifestImpl(); - - private: - PARCJSON *root_; -}; - -} // namespace core - -} // namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/core/manifest_inline.h b/libtransport/src/hicn/transport/core/manifest_inline.h index 1f2c4edb4..60eb08433 100644 --- a/libtransport/src/hicn/transport/core/manifest_inline.h +++ b/libtransport/src/hicn/transport/core/manifest_inline.h @@ -37,7 +37,8 @@ class ManifestInline public: ManifestInline() : ManifestBase() {} - ManifestInline(const core::Name &name) : ManifestBase(name) {} + ManifestInline(const core::Name &name, std::size_t signature_size = 0) + : ManifestBase(name, signature_size) {} template <typename T> ManifestInline(T &&base) : ManifestBase(std::forward<T &&>(base)) {} @@ -47,8 +48,7 @@ class ManifestInline ManifestType type, HashAlgorithm algorithm, bool is_last, const Name &base_name, NextSegmentCalculationStrategy strategy, std::size_t signature_size) { - auto manifest = new ManifestInline(manifest_name); - manifest->setSignatureSize(signature_size); + auto manifest = new ManifestInline(manifest_name, signature_size); manifest->setVersion(version); manifest->setManifestType(type); manifest->setHashAlgorithm(algorithm); diff --git a/libtransport/src/hicn/transport/core/packet.cc b/libtransport/src/hicn/transport/core/packet.cc index b3e5526ab..ec99bb3f7 100644 --- a/libtransport/src/hicn/transport/core/packet.cc +++ b/libtransport/src/hicn/transport/core/packet.cc @@ -80,7 +80,7 @@ Packet::Packet(const uint8_t *buffer, std::size_t size) Packet::Packet(Packet &&other) : packet_(std::move(other.packet_)), - packet_start_(packet_->writableData()), + packet_start_(other.packet_start_), header_head_(other.header_head_), payload_head_(other.payload_head_), format_(other.format_) { @@ -269,7 +269,7 @@ void Packet::dump() const { std::cout << std::endl << "PAYLOAD -- Length: " << payloadSize() << std::endl; for (utils::MemBuf *current = payload_head_; current && current != header_head_; current = current->next()) { - std::cout << "First MemBuf Length: " << current->length() << std::endl; + std::cout << "MemBuf Length: " << current->length() << std::endl; hicn_packet_dump((uint8_t *)current->data(), current->length()); } } @@ -310,17 +310,6 @@ std::size_t Packet::getSignatureSize() const { return size_bytes; } -void Packet::setSignature(std::unique_ptr<utils::MemBuf> &&signature) { - // Check if packet already contains a signature - auto header = header_head_->next(); - while (header != payload_head_) { - header->unlink(); - header = header->next(); - } - - appendHeader(std::move(signature)); -} - void Packet::setSignatureTimestamp(const uint64_t ×tamp) { int ret = hicn_packet_set_signature_timestamp( format_, (hicn_header_t *)packet_start_, timestamp); @@ -397,12 +386,11 @@ utils::CryptoHash Packet::computeDigest(HashAlgorithm algorithm) const { const_cast<Packet *>(this)->resetForHash(); - std::size_t payload_len = getPayloadSizeFromBuffer(format_, packet_start_); - std::size_t header_length = getHeaderSizeFromFormat(format_); - std::size_t signature_size = _is_ah(format_) ? getSignatureSize() : 0; - - hasher.updateBytes(packet_start_, - payload_len + header_length + signature_size); + auto current = header_head_; + do { + hasher.updateBytes(current->data(), current->length()); + current = current->next(); + } while (current != header_head_); hicn_packet_copy_header(format_, &header_copy, (hicn_header_t *)packet_start_, false); diff --git a/libtransport/src/hicn/transport/core/packet.h b/libtransport/src/hicn/transport/core/packet.h index 49a558333..add830c1e 100644 --- a/libtransport/src/hicn/transport/core/packet.h +++ b/libtransport/src/hicn/transport/core/packet.h @@ -103,6 +103,10 @@ class Packet : public std::enable_shared_from_this<Packet> { virtual void setLifetime(uint32_t lifetime); + virtual const Name &getName() const = 0; + + virtual Name &getWritableName() = 0; + virtual uint32_t getLifetime() const; Packet &appendPayload(const uint8_t *buffer, std::size_t length); @@ -129,12 +133,6 @@ class Packet : public std::enable_shared_from_this<Packet> { virtual ip_address_t getLocator() const = 0; - void setSignatureSize(std::size_t size_bytes); - - std::size_t getSignatureSize() const; - - uint8_t *getSignature() const; - void setSignatureTimestamp(const uint64_t ×tamp); uint64_t getSignatureTimestamp() const; @@ -147,8 +145,6 @@ class Packet : public std::enable_shared_from_this<Packet> { utils::KeyId getKeyId() const; - void setSignature(std::unique_ptr<utils::MemBuf> &&signature); - virtual utils::CryptoHash computeDigest(HashAlgorithm algorithm) const; void setChecksum(); @@ -180,6 +176,9 @@ class Packet : public std::enable_shared_from_this<Packet> { private: virtual void resetForHash() = 0; + void setSignatureSize(std::size_t size_bytes); + std::size_t getSignatureSize() const; + uint8_t *getSignature() const; protected: Name name_; diff --git a/libtransport/src/hicn/transport/core/raw_socket_connector.cc b/libtransport/src/hicn/transport/core/raw_socket_connector.cc index fe16d2132..78241b2ea 100644 --- a/libtransport/src/hicn/transport/core/raw_socket_connector.cc +++ b/libtransport/src/hicn/transport/core/raw_socket_connector.cc @@ -181,7 +181,7 @@ void RawSocketConnector::doSendPacket() { } void RawSocketConnector::doRecvPacket() { - read_msg_ = std::move(getPacket()); + read_msg_ = getPacket(); socket_.async_receive( asio::buffer(read_msg_->writableData(), packet_size), [this](std::error_code ec, std::size_t bytes_transferred) mutable { diff --git a/libtransport/src/hicn/transport/core/raw_socket_connector.h b/libtransport/src/hicn/transport/core/raw_socket_connector.h index a54b55e7e..57a2bc067 100644 --- a/libtransport/src/hicn/transport/core/raw_socket_connector.h +++ b/libtransport/src/hicn/transport/core/raw_socket_connector.h @@ -19,11 +19,11 @@ #include <hicn/transport/core/connector.h> #include <hicn/transport/core/name.h> -#include <asio/steady_timer.hpp> #include <linux/if_packet.h> #include <net/ethernet.h> #include <sys/socket.h> #include <asio.hpp> +#include <asio/steady_timer.hpp> #include <deque> namespace transport { diff --git a/libtransport/src/hicn/transport/core/tcp_socket_connector.cc b/libtransport/src/hicn/transport/core/tcp_socket_connector.cc index ade0f2611..4c5f90db3 100644 --- a/libtransport/src/hicn/transport/core/tcp_socket_connector.cc +++ b/libtransport/src/hicn/transport/core/tcp_socket_connector.cc @@ -111,8 +111,25 @@ void TcpSocketConnector::close() { } void TcpSocketConnector::doWrite() { - // TODO improve this piece of code for sending many buffers togethers - // if list contains more than one packet +#if 1 + auto array = std::vector<asio::const_buffer>(); + std::vector<Packet::MemBufPtr> packet_store(packet_store_size); + uint8_t i = 0; + + utils::MemBuf *packet = nullptr; + const utils::MemBuf *current = nullptr; + // Send vectors of 32 packets + while (!output_buffer_.empty() && i++ < packet_store_size) { + packet_store[i] = output_buffer_.front(); + output_buffer_.pop_front(); + packet = packet_store[i].get(); + current = packet; + do { + array.push_back(asio::const_buffer(current->data(), current->length())); + current = current->next(); + } while (current != packet); + } +#else auto packet = output_buffer_.front().get(); auto array = std::vector<asio::const_buffer>(); @@ -121,12 +138,13 @@ void TcpSocketConnector::doWrite() { array.push_back(asio::const_buffer(current->data(), current->length())); current = current->next(); } while (current != packet); +#endif asio::async_write( socket_, std::move(array), - [this /*, packet*/](std::error_code ec, std::size_t length) { + [this, packet_store = std::move(packet_store)](std::error_code ec, + std::size_t length) { if (TRANSPORT_EXPECT_TRUE(!ec)) { - output_buffer_.pop_front(); if (!output_buffer_.empty()) { doWrite(); } diff --git a/libtransport/src/hicn/transport/core/tcp_socket_connector.h b/libtransport/src/hicn/transport/core/tcp_socket_connector.h index ca9d2b663..465eeb912 100644 --- a/libtransport/src/hicn/transport/core/tcp_socket_connector.h +++ b/libtransport/src/hicn/transport/core/tcp_socket_connector.h @@ -30,6 +30,8 @@ namespace core { using asio::ip::tcp; class TcpSocketConnector : public Connector { + static constexpr uint16_t packet_store_size = 32; + public: TcpSocketConnector(PacketReceivedCallback &&receive_callback, OnReconnect &&reconnect_callback, diff --git a/libtransport/src/hicn/transport/errors/CMakeLists.txt b/libtransport/src/hicn/transport/errors/CMakeLists.txt index 1c19a9070..7b0d1332d 100644 --- a/libtransport/src/hicn/transport/errors/CMakeLists.txt +++ b/libtransport/src/hicn/transport/errors/CMakeLists.txt @@ -23,6 +23,8 @@ list(APPEND HEADER_FILES ${CMAKE_CURRENT_SOURCE_DIR}/tokenizer_exception.h ${CMAKE_CURRENT_SOURCE_DIR}/null_pointer_exception.h ${CMAKE_CURRENT_SOURCE_DIR}/malformed_ahpacket_exception.h + ${CMAKE_CURRENT_SOURCE_DIR}/unexpected_manifest_exception.h + ${CMAKE_CURRENT_SOURCE_DIR}/indexing_exception.h ) set(SOURCE_FILES ${SOURCE_FILES} PARENT_SCOPE) diff --git a/libtransport/src/hicn/transport/errors/indexing_exception.h b/libtransport/src/hicn/transport/errors/indexing_exception.h new file mode 100644 index 000000000..731314f0e --- /dev/null +++ b/libtransport/src/hicn/transport/errors/indexing_exception.h @@ -0,0 +1,31 @@ +/* + * 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. + */ + +#pragma once + +#include <stdexcept> + +namespace errors { + +class IndexingException : public std::logic_error { + public: + IndexingException() : std::logic_error("") {} + + virtual char const *what() const noexcept override { + return "Impossible to retrieve next index to download."; + } +}; + +} // end namespace errors diff --git a/libtransport/src/hicn/transport/errors/unexpected_manifest_exception.h b/libtransport/src/hicn/transport/errors/unexpected_manifest_exception.h new file mode 100644 index 000000000..6f71471e4 --- /dev/null +++ b/libtransport/src/hicn/transport/errors/unexpected_manifest_exception.h @@ -0,0 +1,31 @@ +/* + * 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. + */ + +#pragma once + +#include <stdexcept> + +namespace errors { + +class UnexpectedManifestException : public std::logic_error { + public: + UnexpectedManifestException() : std::logic_error("") {} + + virtual char const *what() const noexcept override { + return "Received unexpected manifest."; + } +}; + +} // end namespace errors diff --git a/libtransport/src/hicn/transport/http/callbacks.h b/libtransport/src/hicn/transport/http/callbacks.h index 5ca5fcbe2..a0f3d5999 100644 --- a/libtransport/src/hicn/transport/http/callbacks.h +++ b/libtransport/src/hicn/transport/http/callbacks.h @@ -35,9 +35,9 @@ using OnHttpRequest = std::size_t, int request_id)>; using DeadlineTimerCallback = std::function<void(const std::error_code& e)>; using ReceiveCallback = std::function<void(const std::vector<uint8_t>&)>; -using OnPayloadCallback = std::function<RC( - const std::error_code& ec, const core::Name& name, - const std::shared_ptr<utils::SharableVector<uint8_t>>& payload)>; +using OnPayloadCallback = + std::function<RC(const std::error_code& ec, const core::Name& name, + const ContentBuffer& payload)>; using ContentSentCallback = std::function<void(const std::error_code&, const core::Name&)>; diff --git a/libtransport/src/hicn/transport/http/client_connection.cc b/libtransport/src/hicn/transport/http/client_connection.cc index d4207bb81..e2ee6478d 100644 --- a/libtransport/src/hicn/transport/http/client_connection.cc +++ b/libtransport/src/hicn/transport/http/client_connection.cc @@ -118,14 +118,17 @@ std::string HTTPClientConnection::sendRequestGetReply( stream << "|0"; - consumer_.consume(Name(stream.str()), *response); + ContentBuffer response_ptr = + std::static_pointer_cast<std::vector<uint8_t>>(response); + + consumer_.consume(Name(stream.str()), response_ptr); consumer_.stop(); return stream.str(); } -HTTPResponse &&HTTPClientConnection::response() { +HTTPResponse HTTPClientConnection::response() { // response_->parse(); return std::move(*response_); } diff --git a/libtransport/src/hicn/transport/http/client_connection.h b/libtransport/src/hicn/transport/http/client_connection.h index f6e1fa03e..faad1864c 100644 --- a/libtransport/src/hicn/transport/http/client_connection.h +++ b/libtransport/src/hicn/transport/http/client_connection.h @@ -44,7 +44,7 @@ class HTTPClientConnection { HTTPPayload payload = {}, std::shared_ptr<HTTPResponse> response = nullptr); - HTTPResponse &&response(); + HTTPResponse response(); HTTPClientConnection &stop(); diff --git a/libtransport/src/hicn/transport/http/message.h b/libtransport/src/hicn/transport/http/message.h index b2f271bcb..270dd3f0e 100644 --- a/libtransport/src/hicn/transport/http/message.h +++ b/libtransport/src/hicn/transport/http/message.h @@ -19,8 +19,6 @@ #include <hicn/transport/portability/win_portability.h> #endif -#include <hicn/transport/utils/sharable_vector.h> - #include <map> #include <sstream> #include <vector> diff --git a/libtransport/src/hicn/transport/http/request.h b/libtransport/src/hicn/transport/http/request.h index 88d67d4ad..1202144c0 100644 --- a/libtransport/src/hicn/transport/http/request.h +++ b/libtransport/src/hicn/transport/http/request.h @@ -16,7 +16,6 @@ #pragma once #include <hicn/transport/http/message.h> -#include <hicn/transport/utils/sharable_vector.h> #include <map> #include <sstream> diff --git a/libtransport/src/hicn/transport/http/response.h b/libtransport/src/hicn/transport/http/response.h index e7dec8c40..b261a128a 100644 --- a/libtransport/src/hicn/transport/http/response.h +++ b/libtransport/src/hicn/transport/http/response.h @@ -17,7 +17,6 @@ #include <hicn/transport/http/message.h> #include <hicn/transport/utils/array.h> -#include <hicn/transport/utils/sharable_vector.h> #include <map> #include <sstream> @@ -27,7 +26,7 @@ namespace transport { namespace http { -class HTTPResponse : public HTTPMessage, public utils::SharableVector<uint8_t> { +class HTTPResponse : public HTTPMessage, public std::vector<uint8_t> { public: HTTPResponse(const HTTPHeaders &headers, const HTTPPayload &payload); diff --git a/libtransport/src/hicn/transport/http/server_publisher.h b/libtransport/src/hicn/transport/http/server_publisher.h index 91f7e43e9..f3c39c3fe 100644 --- a/libtransport/src/hicn/transport/http/server_publisher.h +++ b/libtransport/src/hicn/transport/http/server_publisher.h @@ -64,7 +64,7 @@ class HTTPServerPublisher { ProducerInterestCallback interest_enter_callback_; utils::UserCallback wait_callback_; - utils::SharableVector<uint8_t> receive_buffer_; + ContentBuffer receive_buffer_; }; } // end namespace http diff --git a/libtransport/src/hicn/transport/interfaces/async_transport.h b/libtransport/src/hicn/transport/interfaces/async_transport.h index 2911377a4..692dd318c 100644 --- a/libtransport/src/hicn/transport/interfaces/async_transport.h +++ b/libtransport/src/hicn/transport/interfaces/async_transport.h @@ -17,8 +17,8 @@ #pragma once #include <hicn/transport/interfaces/publication_options.h> +#include <hicn/transport/interfaces/socket.h> #include <hicn/transport/portability/portability.h> -#include <hicn/transport/utils/sharable_vector.h> #ifndef _WIN32 #include <sys/uio.h> @@ -528,8 +528,7 @@ class AsyncReader { // virtual void readBufferAvailable(uint8_t** buffer, std::size_t // *buf_length) noexcept {} - virtual void readBufferAvailable( - utils::SharableVector<uint8_t> &&buffer) noexcept {} + virtual void readBufferAvailable(ContentBuffer &&buffer) noexcept {} // virtual void readBufferAvailable(utils::SharableBuffer<uint8_t>&& buffer) // noexcept {} @@ -612,8 +611,7 @@ class AsyncWriter { * shutdownWriteNow()). When closing the socket this way, writeErr() will * still be invoked once for each outstanding write operation. */ - virtual void write(WriteCallback *callback, - utils::SharableVector<uint8_t> &&output_buffer, + virtual void write(WriteCallback *callback, ContentBuffer &&output_buffer, const PublicationOptions &options, WriteFlags flags = WriteFlags::NONE) = 0; diff --git a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc index 0a091d94e..3bb51e72e 100644 --- a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc +++ b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc @@ -15,7 +15,6 @@ #include <hicn/transport/interfaces/full_duplex_socket.h> #include <hicn/transport/interfaces/socket_options_default_values.h> -#include <hicn/transport/utils/sharable_vector.h> #include <memory> @@ -45,7 +44,7 @@ AsyncFullDuplexSocket::AsyncFullDuplexSocket(const Prefix &locator, internal_signal_callback_(new OnSignalCallback(*this)), send_timeout_milliseconds_(~0), counters_({0}), - receive_buffer_(std::make_shared<utils::SharableVector<uint8_t>>()) { + receive_buffer_(ContentBuffer()) { using namespace transport; using namespace std::placeholders; producer_->registerPrefix(locator); @@ -64,11 +63,10 @@ AsyncFullDuplexSocket::AsyncFullDuplexSocket(const Prefix &locator, producer_->connect(); - consumer_->setSocketOption( - ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY, - (ConsumerContentObjectVerificationCallback)[](ConsumerSocket & s, - const ContentObject &c) - ->bool { return true; }); + consumer_->setSocketOption(ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY, + (ConsumerContentObjectVerificationCallback)[]( + ConsumerSocket & s, const ContentObject &c) + ->bool { return true; }); ConsumerContentCallback consumer_callback = std::bind(&AsyncFullDuplexSocket::onContentRetrieved, this, _1, _2, _3); @@ -200,17 +198,18 @@ void AsyncFullDuplexSocket::write(WriteCallback *callback, const void *buf, } } -void AsyncFullDuplexSocket::write( - WriteCallback *callback, utils::SharableVector<uint8_t> &&output_buffer, - const PublicationOptions &options, WriteFlags flags) { +void AsyncFullDuplexSocket::write(WriteCallback *callback, + ContentBuffer &&output_buffer, + const PublicationOptions &options, + WriteFlags flags) { using namespace transport; // 1 asynchronously write the content. I assume here the // buffer contains the whole application frame. FIXME: check // if this is true and fix it accordingly - std::cout << "Size of the PAYLOAD: " << output_buffer.size() << std::endl; + std::cout << "Size of the PAYLOAD: " << output_buffer->size() << std::endl; - if (output_buffer.size() > + if (output_buffer->size() > core::Packet::default_mtu - sizeof(PayloadMessage)) { TRANSPORT_LOGI("Producing content with name %s", options.name.toString().c_str()); @@ -218,8 +217,8 @@ void AsyncFullDuplexSocket::write( signalProductionToSubscribers(options.name); } else { TRANSPORT_LOGI("Sending payload through interest"); - piggybackPayloadToSubscribers(options.name, &output_buffer[0], - output_buffer.size()); + piggybackPayloadToSubscribers(options.name, &(*output_buffer)[0], + output_buffer->size()); } } @@ -350,10 +349,10 @@ AsyncFullDuplexSocket::decodeSynchronizationMessage( // The interest contains the payload directly. // We saved one round trip :) - auto buffer = std::make_shared<utils::SharableVector<uint8_t>>(); + auto buffer = ContentBuffer(); const uint8_t *data = mesg.data() + sizeof(PayloadMessage); buffer->assign(data, data + mesg.length() - sizeof(PayloadMessage)); - read_callback_->readBufferAvailable(std::move(*buffer)); + read_callback_->readBufferAvailable(std::move(buffer)); return createAck(); } default: { @@ -401,7 +400,7 @@ void AsyncFullDuplexSocket::onContentRetrieved(ConsumerSocket &s, TRANSPORT_LOGI("Received content with size %zu", size); if (!ec) { - read_callback_->readBufferAvailable(std::move(*receive_buffer_)); + read_callback_->readBufferAvailable(std::move(receive_buffer_)); } else { TRANSPORT_LOGE("Error retrieving content."); } diff --git a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h index f881bea54..b47432460 100644 --- a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h +++ b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h @@ -24,7 +24,6 @@ #include <hicn/transport/interfaces/socket_consumer.h> #include <hicn/transport/interfaces/socket_producer.h> #include <hicn/transport/portability/portability.h> -#include <hicn/transport/utils/sharable_vector.h> #include <unordered_set> #include <vector> @@ -136,8 +135,7 @@ class AsyncFullDuplexSocket : public AsyncSocket, const PublicationOptions &options, WriteFlags flags = WriteFlags::NONE) override; - virtual void write(WriteCallback *callback, - utils::SharableVector<uint8_t> &&output_buffer, + virtual void write(WriteCallback *callback, ContentBuffer &&output_buffer, const PublicationOptions &options, WriteFlags flags = WriteFlags::NONE) override; @@ -247,7 +245,7 @@ class AsyncFullDuplexSocket : public AsyncSocket, uint32_t send_timeout_milliseconds_; struct Counters counters_; - std::shared_ptr<utils::SharableVector<uint8_t>> receive_buffer_; + ContentBuffer receive_buffer_; }; } // namespace interface diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/rtc_socket_consumer.cc index ef952a891..a24e3883c 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_consumer.cc @@ -14,6 +14,7 @@ */ #include <hicn/transport/interfaces/rtc_socket_consumer.h> +#include <hicn/transport/protocols/rtc.h> namespace transport { diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc index ede1ff247..c07ca7989 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc @@ -159,7 +159,6 @@ void RTCProducerSocket::onInterest(Interest::Ptr &&interest) { if (interestSeg < currentSeg_ || interestSeg > (max_gap + currentSeg_)) { sendNack(*interest); } - // else drop packet } void RTCProducerSocket::sendNack(const Interest &interest) { diff --git a/libtransport/src/hicn/transport/interfaces/socket.h b/libtransport/src/hicn/transport/interfaces/socket.h index 14ef86422..7d50d0fbd 100644 --- a/libtransport/src/hicn/transport/interfaces/socket.h +++ b/libtransport/src/hicn/transport/interfaces/socket.h @@ -24,6 +24,7 @@ #include <hicn/transport/core/name.h> #include <hicn/transport/interfaces/socket_options_default_values.h> #include <hicn/transport/interfaces/socket_options_keys.h> +#include <hicn/transport/protocols/statistics.h> #include <hicn/transport/utils/crypto_suite.h> #include <hicn/transport/utils/identity.h> #include <hicn/transport/utils/verifier.h> @@ -40,7 +41,8 @@ namespace transport { namespace protocol { class IcnObserver; -} +class TransportStatistics; +} // namespace protocol namespace interface { @@ -77,6 +79,7 @@ using BasePortal = HicnForwarderPortal; using PayloadType = core::PayloadType; using Prefix = core::Prefix; using Array = utils::Array<uint8_t>; +using ContentBuffer = std::shared_ptr<std::vector<uint8_t>>; using ConsumerInterestCallback = std::function<void(ConsumerSocket &, const core::Interest &)>; @@ -84,9 +87,8 @@ using ConsumerInterestCallback = using ConsumerContentCallback = std::function<void(ConsumerSocket &, std::size_t, const std::error_code &)>; -using ConsumerTimerCallback = - std::function<void(ConsumerSocket &, std::size_t, - std::chrono::milliseconds &, float, uint32_t, uint32_t)>; +using ConsumerTimerCallback = std::function<void( + ConsumerSocket &, const protocol::TransportStatistics &stats)>; using ProducerContentCallback = std::function<void( ProducerSocket &, const std::error_code &, uint64_t bytes_written)>; @@ -132,129 +134,6 @@ class Socket { virtual void connect() = 0; - virtual int setSocketOption(int socket_option_key, - uint32_t socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - double socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - bool socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - core::Name socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - std::list<Prefix> socket_option_value) = 0; - - virtual int setSocketOption( - int socket_option_key, - ProducerContentObjectCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ProducerInterestCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ProducerContentCallback socket_option_value) = 0; - - virtual int setSocketOption( - int socket_option_key, - ConsumerContentObjectVerificationCallback socket_option_value) = 0; - - virtual int setSocketOption( - int socket_option_key, - ConsumerContentObjectCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerInterestCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerContentCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerManifestCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - IcnObserver *socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - core::HashAlgorithm socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - utils::CryptoSuite socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - const utils::Identity &socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerTimerCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - const std::string &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - uint32_t &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - double &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - bool &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - core::Name &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - std::list<Prefix> &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, - ProducerContentObjectCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, ProducerInterestCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, - ConsumerContentObjectVerificationCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, - ConsumerContentObjectCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, ConsumerInterestCallback &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - ConsumerContentCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, ConsumerManifestCallback &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - ProducerContentCallback &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - std::shared_ptr<Portal> &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - IcnObserver **socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - core::HashAlgorithm &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - utils::CryptoSuite &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - utils::Identity &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - std::string &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - ConsumerTimerCallback &socket_option_value) = 0; - protected: virtual ~Socket(){}; diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc index eacaf932b..febe66853 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc @@ -14,6 +14,9 @@ */ #include <hicn/transport/interfaces/socket_consumer.h> +#include <hicn/transport/protocols/cbr.h> +#include <hicn/transport/protocols/raaqm.h> +#include <hicn/transport/protocols/rtc.h> namespace transport { @@ -43,6 +46,7 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) rate_estimation_observer_(nullptr), rate_estimation_choice_(0), is_async_(false), + verifier_(std::make_shared<utils::Verifier>()), verify_signature_(false), content_buffer_(nullptr), on_interest_output_(VOID_HANDLER), @@ -55,12 +59,8 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) on_payload_retrieved_(VOID_HANDLER), virtual_download_(false), rtt_stats_(false), - timer_(portal_->getIoService()), timer_interval_milliseconds_(0) { switch (protocol) { - case TransportProtocolAlgorithms::VEGAS: - transport_protocol_ = std::make_shared<VegasTransportProtocol>(this); - break; case TransportProtocolAlgorithms::CBR: transport_protocol_ = std::make_shared<CbrTransportProtocol>(this); break; @@ -85,33 +85,31 @@ ConsumerSocket::~ConsumerSocket() { void ConsumerSocket::connect() { portal_->connect(); } -int ConsumerSocket::consume(const Name &name, - utils::SharableVector<uint8_t> &receive_buffer) { +int ConsumerSocket::consume(const Name &name, ContentBuffer &receive_buffer) { if (transport_protocol_->isRunning()) { return CONSUMER_BUSY; } - content_buffer_ = receive_buffer.shared_from_this(); + content_buffer_ = receive_buffer; network_name_ = name; network_name_.setSuffix(0); is_async_ = false; - transport_protocol_->start(receive_buffer); + transport_protocol_->start(); return CONSUMER_FINISHED; } -int ConsumerSocket::asyncConsume( - const Name &name, - std::shared_ptr<utils::SharableVector<uint8_t>> receive_buffer) { - // XXX Try to move the name here, instead of copying it!! +int ConsumerSocket::asyncConsume(const Name &name, + ContentBuffer &receive_buffer) { if (!async_downloader_.stopped()) { async_downloader_.add([this, receive_buffer, name]() { network_name_ = std::move(name); network_name_.setSuffix(0); is_async_ = true; - transport_protocol_->start(*receive_buffer); + content_buffer_ = receive_buffer; + transport_protocol_->start(); }); } @@ -150,586 +148,6 @@ asio::io_service &ConsumerSocket::getIoService() { return portal_->getIoService(); } -int ConsumerSocket::setSocketOption(int socket_option_key, - double socket_option_value) { - switch (socket_option_key) { - case MIN_WINDOW_SIZE: - min_window_size_ = socket_option_value; - return SOCKET_OPTION_SET; - - case MAX_WINDOW_SIZE: - max_window_size_ = socket_option_value; - return SOCKET_OPTION_SET; - - case CURRENT_WINDOW_SIZE: - current_window_size_ = socket_option_value; - return SOCKET_OPTION_SET; - - case GAMMA_VALUE: - gamma_ = socket_option_value; - return SOCKET_OPTION_SET; - - case BETA_VALUE: - beta_ = socket_option_value; - return SOCKET_OPTION_SET; - - case DROP_FACTOR: - drop_factor_ = socket_option_value; - return SOCKET_OPTION_SET; - - case MINIMUM_DROP_PROBABILITY: - minimum_drop_probability_ = socket_option_value; - return SOCKET_OPTION_SET; - - case RATE_ESTIMATION_ALPHA: - if (socket_option_value >= 0 && socket_option_value < 1) { - rate_estimation_alpha_ = socket_option_value; - } else { - rate_estimation_alpha_ = ALPHA; - } - return SOCKET_OPTION_SET; - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::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; - return SOCKET_OPTION_SET; - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - output_buffer_size_ = socket_option_value; - return SOCKET_OPTION_SET; - - case GeneralTransportOptions::MAX_INTEREST_RETX: - max_retransmissions_ = socket_option_value; - return SOCKET_OPTION_SET; - - case GeneralTransportOptions::INTEREST_LIFETIME: - interest_lifetime_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: - if (socket_option_value == VOID_HANDLER) { - on_interest_retransmission_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ConsumerCallbacksOptions::INTEREST_EXPIRED: - if (socket_option_value == VOID_HANDLER) { - on_interest_timeout_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ConsumerCallbacksOptions::INTEREST_SATISFIED: - if (socket_option_value == VOID_HANDLER) { - on_interest_satisfied_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ConsumerCallbacksOptions::INTEREST_OUTPUT: - if (socket_option_value == VOID_HANDLER) { - on_interest_output_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT: - if (socket_option_value == VOID_HANDLER) { - on_content_object_input_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY: - if (socket_option_value == VOID_HANDLER) { - on_content_object_verification_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ConsumerCallbacksOptions::CONTENT_RETRIEVED: - if (socket_option_value == VOID_HANDLER) { - on_payload_retrieved_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: - if (socket_option_value > 0) { - rate_estimation_batching_parameter_ = socket_option_value; - } else { - rate_estimation_batching_parameter_ = BATCH; - } - return SOCKET_OPTION_SET; - - case RateEstimationOptions::RATE_ESTIMATION_CHOICE: - if (socket_option_value > 0) { - rate_estimation_choice_ = socket_option_value; - } else { - rate_estimation_choice_ = RATE_CHOICE; - } - return SOCKET_OPTION_SET; - - case GeneralTransportOptions::TIMER_INTERVAL: - timer_interval_milliseconds_ = socket_option_value; - TRANSPORT_LOGD("Ok set %d", timer_interval_milliseconds_); - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - bool socket_option_value) { - switch (socket_option_key) { - case OtherOptions::VIRTUAL_DOWNLOAD: - virtual_download_ = socket_option_value; - return SOCKET_OPTION_SET; - - case RaaqmTransportOptions::RTT_STATS: - rtt_stats_ = socket_option_value; - return SOCKET_OPTION_SET; - - case GeneralTransportOptions::VERIFY_SIGNATURE: - verify_signature_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - Name socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::NETWORK_NAME: - network_name_ = socket_option_value; - return SOCKET_OPTION_SET; - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - std::list<Prefix> socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::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; - ; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, ProducerContentObjectCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::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; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, ConsumerInterestCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: - on_interest_retransmission_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ConsumerCallbacksOptions::INTEREST_OUTPUT: - on_interest_output_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ConsumerCallbacksOptions::INTEREST_EXPIRED: - on_interest_timeout_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ConsumerCallbacksOptions::INTEREST_SATISFIED: - on_interest_satisfied_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, ProducerInterestCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, ConsumerContentCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_RETRIEVED: - on_payload_retrieved_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, ConsumerManifestCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::MANIFEST_INPUT: - on_manifest_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, ProducerContentCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - IcnObserver *socket_option_value) { - if (socket_option_key == RateEstimationOptions::RATE_ESTIMATION_OBSERVER) { - rate_estimation_observer_ = socket_option_value; - return SOCKET_OPTION_SET; - } - - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - HashAlgorithm socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - utils::CryptoSuite socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::setSocketOption( - int socket_option_key, const utils::Identity &socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::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) { - return SOCKET_OPTION_SET; - } - - break; - - case DataLinkOptions::OUTPUT_INTERFACE: - output_interface_ = socket_option_value; - portal_->setOutputInterface(output_interface_); - return SOCKET_OPTION_SET; - } - - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::setSocketOption(int socket_option_key, - ConsumerTimerCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::TIMER_EXPIRES: - on_timer_expires_ = socket_option_value; - return SOCKET_OPTION_SET; - } - - return SOCKET_OPTION_NOT_SET; -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - double &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::MIN_WINDOW_SIZE: - socket_option_value = min_window_size_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::MAX_WINDOW_SIZE: - socket_option_value = max_window_size_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::CURRENT_WINDOW_SIZE: - socket_option_value = current_window_size_; - return SOCKET_OPTION_GET; - - // RAAQM parameters - - case RaaqmTransportOptions::GAMMA_VALUE: - socket_option_value = gamma_; - return SOCKET_OPTION_GET; - - case RaaqmTransportOptions::BETA_VALUE: - socket_option_value = beta_; - return SOCKET_OPTION_GET; - - case RaaqmTransportOptions::DROP_FACTOR: - socket_option_value = drop_factor_; - return SOCKET_OPTION_GET; - - case RaaqmTransportOptions::MINIMUM_DROP_PROBABILITY: - socket_option_value = minimum_drop_probability_; - return SOCKET_OPTION_GET; - - case RateEstimationOptions::RATE_ESTIMATION_ALPHA: - socket_option_value = rate_estimation_alpha_; - return SOCKET_OPTION_GET; - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::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_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - socket_option_value = (uint32_t)output_buffer_size_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::MAX_INTEREST_RETX: - socket_option_value = max_retransmissions_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::INTEREST_LIFETIME: - socket_option_value = interest_lifetime_; - return SOCKET_OPTION_GET; - - case RaaqmTransportOptions::SAMPLE_NUMBER: - socket_option_value = sample_number_; - return SOCKET_OPTION_GET; - - case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: - socket_option_value = rate_estimation_batching_parameter_; - return SOCKET_OPTION_GET; - - case RateEstimationOptions::RATE_ESTIMATION_CHOICE: - socket_option_value = rate_estimation_choice_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - bool &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::ASYNC_MODE: - socket_option_value = is_async_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::RUNNING: - socket_option_value = transport_protocol_->isRunning(); - return SOCKET_OPTION_GET; - - case OtherOptions::VIRTUAL_DOWNLOAD: - socket_option_value = virtual_download_; - return SOCKET_OPTION_GET; - - case RaaqmTransportOptions::RTT_STATS: - socket_option_value = rtt_stats_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::VERIFY_SIGNATURE: - socket_option_value = verify_signature_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - std::list<Prefix> &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::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_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ProducerContentObjectCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::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_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ConsumerInterestCallback &socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::INTEREST_RETRANSMISSION: - socket_option_value = on_interest_retransmission_; - return SOCKET_OPTION_GET; - - case ConsumerCallbacksOptions::INTEREST_OUTPUT: - socket_option_value = on_interest_output_; - return SOCKET_OPTION_GET; - - case ConsumerCallbacksOptions::INTEREST_EXPIRED: - socket_option_value = on_interest_timeout_; - return SOCKET_OPTION_GET; - - case ConsumerCallbacksOptions::INTEREST_SATISFIED: - socket_option_value = on_interest_satisfied_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ProducerInterestCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ConsumerContentCallback &socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_RETRIEVED: - socket_option_value = on_payload_retrieved_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ConsumerManifestCallback &socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::MANIFEST_INPUT: - socket_option_value = on_manifest_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - IcnObserver **socket_option_value) { - if (socket_option_key == RATE_ESTIMATION_OBSERVER) { - *socket_option_value = (rate_estimation_observer_); - return SOCKET_OPTION_GET; - } - - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - HashAlgorithm &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - utils::CryptoSuite &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::getSocketOption(int socket_option_key, - utils::Identity &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ProducerContentCallback &socket_option_value) { - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - } - - return SOCKET_OPTION_NOT_GET; -} - -int ConsumerSocket::getSocketOption( - int socket_option_key, ConsumerTimerCallback &socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::TIMER_EXPIRES: - socket_option_value = on_timer_expires_; - return SOCKET_OPTION_GET; - } - - return SOCKET_OPTION_NOT_GET; -} - } // namespace interface } // end namespace transport diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h index 536d2fde3..0f1ad537c 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h @@ -16,15 +16,9 @@ #pragma once #include <hicn/transport/interfaces/socket.h> - -#include <hicn/transport/protocols/cbr.h> +#include <hicn/transport/interfaces/socket_options_default_values.h> #include <hicn/transport/protocols/protocol.h> -#include <hicn/transport/protocols/raaqm.h> -#include <hicn/transport/protocols/rtc.h> -#include <hicn/transport/protocols/vegas.h> - #include <hicn/transport/utils/event_thread.h> -#include <hicn/transport/utils/sharable_vector.h> #define CONSUMER_FINISHED 0 #define CONSUMER_BUSY 1 @@ -35,11 +29,6 @@ namespace transport { namespace interface { class ConsumerSocket : public BaseSocket { - friend class protocol::TransportProtocol; - friend class protocol::VegasTransportProtocol; - friend class protocol::RaaqmTransportProtocol; - friend class protocol::CbrTransportProtocol; - public: explicit ConsumerSocket(int protocol); explicit ConsumerSocket(int protocol, asio::io_service &io_service); @@ -48,11 +37,9 @@ class ConsumerSocket : public BaseSocket { void connect() override; - int consume(const Name &name, utils::SharableVector<uint8_t> &receive_buffer); + int consume(const Name &name, ContentBuffer &receive_buffer); - int asyncConsume( - const Name &name, - std::shared_ptr<utils::SharableVector<uint8_t>> receive_buffer); + int asyncConsume(const Name &name, ContentBuffer &receive_buffer); void asyncSendInterest(Interest::Ptr &&interest, Portal::ConsumerCallback *callback); @@ -63,126 +50,630 @@ class ConsumerSocket : public BaseSocket { asio::io_service &getIoService() override; - int setSocketOption(int socket_option_key, - uint32_t socket_option_value) override; - - int setSocketOption(int socket_option_key, - double socket_option_value) override; - - int setSocketOption(int socket_option_key, bool socket_option_value) override; - - int setSocketOption(int socket_option_key, Name socket_option_value) override; + 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 ConsumerCallbacksOptions::CONTENT_RETRIEVED: + if (socket_option_value == VOID_HANDLER) { + on_payload_retrieved_ = 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; - int setSocketOption(int socket_option_key, - std::list<Prefix> socket_option_value) override; + default: + return SOCKET_OPTION_NOT_SET; + } - int setSocketOption( - int socket_option_key, - ProducerContentObjectCallback socket_option_value) override; + return SOCKET_OPTION_SET; + } - int setSocketOption( + TRANSPORT_ALWAYS_INLINE int setSocketOption( int socket_option_key, - ConsumerContentObjectVerificationCallback socket_option_value) override; - - int setSocketOption( + 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, ConsumerContentCallback socket_option_value) { + switch (socket_option_key) { + case ConsumerCallbacksOptions::CONTENT_RETRIEVED: + on_payload_retrieved_ = 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( int socket_option_key, - ConsumerContentObjectCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ConsumerInterestCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ProducerInterestCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ConsumerContentCallback socket_option_value) override; + const std::shared_ptr<utils::Verifier> &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::VERIFIER: + verifier_ = socket_option_value; + break; - int setSocketOption(int socket_option_key, - ConsumerManifestCallback socket_option_value) override; + default: + return SOCKET_OPTION_NOT_SET; + } - int setSocketOption(int socket_option_key, - IcnObserver *socket_option_value) override; + return SOCKET_OPTION_SET; + } - int setSocketOption(int socket_option_key, - HashAlgorithm socket_option_value) override; - - int setSocketOption(int socket_option_key, - utils::CryptoSuite crypto_suite) override; - - int setSocketOption(int socket_option_key, - const utils::Identity &crypto_suite) override; - - int setSocketOption(int socket_option_key, - const std::string &socket_option_value) override; + TRANSPORT_ALWAYS_INLINE int setSocketOption( + int socket_option_key, + const std::shared_ptr<std::vector<uint8_t>> &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::APPLICATION_BUFFER: + content_buffer_ = 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 = input_buffer_size_; + break; + + case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: + socket_option_value = 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; - int setSocketOption(int socket_option_key, - ConsumerTimerCallback socket_option_value) override; + case GeneralTransportOptions::RUNNING: + socket_option_value = transport_protocol_->isRunning(); + break; - int setSocketOption(int socket_option_key, - ProducerContentCallback socket_option_value) override; + case OtherOptions::VIRTUAL_DOWNLOAD: + socket_option_value = virtual_download_; + break; + + case RaaqmTransportOptions::RTT_STATS: + socket_option_value = rtt_stats_; + break; - int getSocketOption(int socket_option_key, - uint32_t &socket_option_value) override; + case GeneralTransportOptions::VERIFY_SIGNATURE: + socket_option_value = verify_signature_; + break; - int getSocketOption(int socket_option_key, - double &socket_option_value) override; + default: + return SOCKET_OPTION_NOT_GET; + } - int getSocketOption(int socket_option_key, - bool &socket_option_value) override; + 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; - int getSocketOption(int socket_option_key, - Name &socket_option_value) override; + default: + return SOCKET_OPTION_NOT_GET; + } - int getSocketOption(int socket_option_key, - std::list<Prefix> &socket_option_value) override; + return SOCKET_OPTION_GET; + } - int getSocketOption( + TRANSPORT_ALWAYS_INLINE int getSocketOption( int socket_option_key, - ProducerContentObjectCallback &socket_option_value) override; - - int getSocketOption( + 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, ConsumerContentCallback **socket_option_value) { + switch (socket_option_key) { + case ConsumerCallbacksOptions::CONTENT_RETRIEVED: + *socket_option_value = &on_payload_retrieved_; + return SOCKET_OPTION_GET; + + default: + return SOCKET_OPTION_NOT_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( int socket_option_key, - ConsumerContentObjectVerificationCallback &socket_option_value) override; - - int getSocketOption( + 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, - ConsumerContentObjectCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerInterestCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ProducerInterestCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerContentCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerManifestCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - std::shared_ptr<Portal> &socket_option_value) override; - - int getSocketOption(int socket_option_key, - IcnObserver **socket_option_value) override; - - int getSocketOption(int socket_option_key, - HashAlgorithm &socket_option_value) override; - - int getSocketOption(int socket_option_key, - utils::CryptoSuite &crypto_suite) override; - - int getSocketOption(int socket_option_key, - utils::Identity &crypto_suite) override; - - int getSocketOption(int socket_option_key, - std::string &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerTimerCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ProducerContentCallback &socket_option_value) override; + std::shared_ptr<std::vector<uint8_t>> &socket_option_value) { + switch (socket_option_key) { + case GeneralTransportOptions::APPLICATION_BUFFER: + socket_option_value = content_buffer_; + 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; + } protected: std::shared_ptr<TransportProtocol> transport_protocol_; @@ -208,7 +699,6 @@ class ConsumerSocket : public BaseSocket { size_t input_buffer_size_; // RAAQM Parameters - double minimum_drop_probability_; unsigned int sample_number_; double gamma_; @@ -223,11 +713,12 @@ class ConsumerSocket : public BaseSocket { bool is_async_; - utils::Verifier verifier_; + // Verification parameters + std::shared_ptr<utils::Verifier> verifier_; PARCKeyId *key_id_; bool verify_signature_; - std::shared_ptr<utils::SharableVector<uint8_t>> content_buffer_; + ContentBuffer content_buffer_; ConsumerInterestCallback on_interest_retransmission_; ConsumerInterestCallback on_interest_output_; @@ -242,16 +733,13 @@ class ConsumerSocket : public BaseSocket { ConsumerContentCallback on_payload_retrieved_; - ConsumerTimerCallback on_timer_expires_; + ConsumerTimerCallback stats_summary_; // Virtual download for traffic generator bool virtual_download_; bool rtt_stats_; - Time t0_; - Time t1_; - asio::steady_timer timer_; uint32_t timer_interval_milliseconds_; }; 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 f9cbb5a44..142228aeb 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_options_default_values.h +++ b/libtransport/src/hicn/transport/interfaces/socket_options_default_values.h @@ -25,37 +25,39 @@ namespace interface { namespace default_values { -const uint32_t interest_lifetime = 1001; // milliseconds -const uint32_t never_expire_time = HICN_MAX_LIFETIME; -const uint32_t content_object_expiry_time = - never_expire_time; // milliseconds -> 50 seconds -const uint32_t content_object_packet_size = 1500; // The ethernet MTU -const uint32_t producer_socket_input_buffer_size = 150000; // Interests -const uint32_t producer_socket_output_buffer_size = 150000; // Content Object -const uint32_t log_2_default_buffer_size = 12; -const uint32_t signature_size = 260; // bytes -const uint32_t key_locator_size = 60; // bytes -const uint32_t limit_guard = 80; // bytes -const uint32_t min_window_size = 1; // Interests -const uint32_t max_window_size = 128000; // Interests -const uint32_t digest_size = 34; // bytes -const uint32_t max_out_of_order_segments = 3; // content object +static constexpr uint32_t interest_lifetime = 1001; // milliseconds +static constexpr uint32_t never_expire_time = HICN_MAX_LIFETIME; +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; +static constexpr uint32_t signature_size = 260; // bytes +static constexpr uint32_t key_locator_size = 60; // bytes +static constexpr uint32_t limit_guard = 80; // bytes +static constexpr uint32_t min_window_size = 1; // Interests +static constexpr uint32_t max_window_size = 256; // Interests +static constexpr uint32_t digest_size = 34; // bytes +static constexpr uint32_t max_out_of_order_segments = 3; // content object // RAAQM -const int sample_number = 30; -const double gamma_value = 1; -const double beta_value = 0.8; -const double drop_factor = 0.2; -const double minimum_drop_probability = 0.00001; -const int path_id = 0; -const double rate_alpha = 0.8; +static constexpr int sample_number = 30; +static constexpr double gamma_value = 1; +static constexpr double beta_value = 0.8; +static constexpr double drop_factor = 0.2; +static constexpr double minimum_drop_probability = 0.00001; +static constexpr int path_id = 0; +static constexpr double rate_alpha = 0.8; -// Vegas -const double alpha = 1 / 8; -const double beta = 1 / 4; -const uint16_t k = 4; -const std::chrono::milliseconds clock_granularity = - std::chrono::milliseconds(100); +// Rate estimation +static constexpr uint32_t batch = 50; +static constexpr uint32_t kv = 20; +static constexpr double alpha = 0.8; +static constexpr uint32_t rate_choice = 0; // maximum allowed values const uint32_t transport_protocol_min_retransmissions = 0; diff --git a/libtransport/src/hicn/transport/interfaces/socket_options_keys.h b/libtransport/src/hicn/transport/interfaces/socket_options_keys.h index 1afad2b48..bcf049310 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_options_keys.h +++ b/libtransport/src/hicn/transport/interfaces/socket_options_keys.h @@ -21,9 +21,8 @@ namespace interface { typedef enum { RAAQM = 0, - VEGAS = 1, - CBR = 3, - RTC = 4, + CBR = 1, + RTC = 2, } TransportProtocolAlgorithms; typedef enum { @@ -44,12 +43,14 @@ typedef enum { MAKE_MANIFEST = 116, PORTAL = 117, RUNNING = 118, - HASH_ALGORITHM = 119, - CRYPTO_SUITE = 120, - IDENTITY = 121, - CERTIFICATE = 122, - VERIFY_SIGNATURE = 123, - TIMER_INTERVAL = 124 + APPLICATION_BUFFER = 119, + HASH_ALGORITHM = 120, + CRYPTO_SUITE = 121, + IDENTITY = 122, + VERIFIER = 123, + CERTIFICATE = 124, + VERIFY_SIGNATURE = 125, + STATS_INTERVAL = 126 } GeneralTransportOptions; typedef enum { @@ -78,7 +79,7 @@ typedef enum { MANIFEST_INPUT = 412, CONTENT_OBJECT_TO_VERIFY = 413, CONTENT_RETRIEVED = 414, - TIMER_EXPIRES = 415 + STATS_SUMMARY = 415 } ConsumerCallbacksOptions; typedef enum { @@ -96,11 +97,11 @@ typedef enum { typedef enum { OUTPUT_INTERFACE = 601 } DataLinkOptions; -typedef enum { VIRTUAL_DOWNLOAD = 601, USE_CFG_FILE = 603 } OtherOptions; +typedef enum { VIRTUAL_DOWNLOAD = 701, USE_CFG_FILE = 702 } OtherOptions; typedef enum { - SHA_256 = 701, - RSA_256 = 702, + SHA_256 = 801, + RSA_256 = 802, } SignatureType; } // namespace interface diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc index 66c656924..497c40c99 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc @@ -122,23 +122,7 @@ void ProducerSocket::passContentObjectToCallbacks( on_content_object_output_(*this, *content_object); } -#ifndef PUSH_API - std::unordered_map<Name, std::shared_ptr<const Interest>>::iterator it; - - { - std::lock_guard<std::mutex> lock(pending_interests_mtx_); - it = pending_interests_.find(content_object->getName()); - } - - if (it != pending_interests_.end()) { - content_object->setLocator(it->second->getLocator()); - portal_->sendContentObject(*content_object); - std::lock_guard<std::mutex> lock(pending_interests_mtx_); - pending_interests_.erase(it); - } -#else portal_->sendContentObject(*content_object); -#endif } } @@ -154,23 +138,7 @@ void ProducerSocket::produce(ContentObject &content_object) { on_content_object_output_(*this, content_object); } -#ifndef PUSH_API - std::unordered_map<Name, std::shared_ptr<const Interest>>::iterator it; - - { - std::lock_guard<std::mutex> lock(pending_interests_mtx_); - it = pending_interests_.find(content_object.getName()); - } - - if (it != pending_interests_.end()) { - content_object.setLocator(it->second->getLocator()); - portal_->sendContentObject(content_object); - std::lock_guard<std::mutex> lock(pending_interests_mtx_); - pending_interests_.erase(it); - } -#else portal_->sendContentObject(content_object); -#endif } uint32_t ProducerSocket::produce(Name content_name, const uint8_t *buf, @@ -307,10 +275,6 @@ uint32_t ProducerSocket::produce(Name content_name, const uint8_t *buf, content_name.setSuffix(current_segment), format); content_object->setLifetime(content_object_expiry_time_); - if (!making_manifest_ && identity_) { - content_object->setSignatureSize(signature_length); - } - if (packaged_segments == number_of_segments - 1) { content_object->appendPayload(&buf[bytes_segmented], buffer_size - bytes_segmented); @@ -389,14 +353,12 @@ void ProducerSocket::asyncProduce(const Name &suffix, const uint8_t *buf, } } -void ProducerSocket::asyncProduce( - const Name &suffix, utils::SharableVector<uint8_t> &&output_buffer) { +void ProducerSocket::asyncProduce(const Name &suffix, + ContentBuffer &&output_buffer) { if (!async_thread_.stopped()) { async_thread_.add( [this, suff = suffix, buffer = std::move(output_buffer)]() { - TRANSPORT_LOGI("FOR REAL!!!!!! --> Producing content with name %s", - suff.toString().c_str()); - produce(suff, &buffer[0], buffer.size(), true); + produce(suff, &(*buffer)[0], buffer->size(), true); }); } } @@ -420,524 +382,14 @@ void ProducerSocket::onInterest(Interest &interest) { portal_->sendContentObject(*content_object); } else { -#ifndef PUSH_API - { - std::lock_guard<std::mutex> lock(pending_interests_mtx_); - pending_interests_[interest.getName()] = - std::static_pointer_cast<const Interest>(interest.shared_from_this()); - } -#endif - if (on_interest_process_ != VOID_HANDLER) { - // external_io_service_.post([this, &interest] () { on_interest_process_(*this, interest); - // }); } } } asio::io_service &ProducerSocket::getIoService() { return io_service_; } -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; - return SOCKET_OPTION_SET; - } else { - return SOCKET_OPTION_NOT_SET; - } - - case GeneralTransportOptions::INPUT_BUFFER_SIZE: - if (socket_option_value >= 1) { - input_buffer_capacity_ = socket_option_value; - return SOCKET_OPTION_SET; - } else { - return SOCKET_OPTION_NOT_SET; - } - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - output_buffer_.setLimit(socket_option_value); - return SOCKET_OPTION_SET; - - case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME: - content_object_expiry_time_ = socket_option_value; - return SOCKET_OPTION_SET; - - 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; - } - - case ProducerCallbacksOptions::INTEREST_INPUT: - if (socket_option_value == VOID_HANDLER) { - on_interest_input_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::INTEREST_DROP: - if (socket_option_value == VOID_HANDLER) { - on_interest_dropped_input_buffer_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::INTEREST_PASS: - if (socket_option_value == VOID_HANDLER) { - on_interest_inserted_input_buffer_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::CACHE_HIT: - if (socket_option_value == VOID_HANDLER) { - on_interest_satisfied_output_buffer_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::CACHE_MISS: - if (socket_option_value == VOID_HANDLER) { - on_interest_process_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: - if (socket_option_value == VOID_HANDLER) { - on_new_segment_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: - if (socket_option_value == VOID_HANDLER) { - on_content_object_to_sign_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::CONTENT_OBJECT_READY: - if (socket_option_value == VOID_HANDLER) { - on_content_object_in_output_buffer_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: - if (socket_option_value == VOID_HANDLER) { - on_content_object_output_ = VOID_HANDLER; - return SOCKET_OPTION_SET; - } - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ProducerSocket::setSocketOption(int socket_option_key, - double socket_option_value) { - return SOCKET_OPTION_NOT_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; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_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; - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ProducerContentObjectCallback socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: - on_new_segment_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: - on_content_object_to_sign_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::CONTENT_OBJECT_READY: - on_content_object_in_output_buffer_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: - on_content_object_output_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ProducerInterestCallback socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::INTEREST_INPUT: - on_interest_input_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::INTEREST_DROP: - on_interest_dropped_input_buffer_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::INTEREST_PASS: - on_interest_inserted_input_buffer_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::CACHE_HIT: - on_interest_satisfied_output_buffer_ = socket_option_value; - return SOCKET_OPTION_SET; - - case ProducerCallbacksOptions::CACHE_MISS: - on_interest_process_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ProducerContentCallback socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::CONTENT_PRODUCED: - on_content_produced_ = socket_option_value; - return SOCKET_OPTION_SET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ConsumerContentObjectCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::setSocketOption( - int socket_option_key, - ConsumerContentObjectVerificationCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ConsumerInterestCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ConsumerContentCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::setSocketOption( - int socket_option_key, ConsumerManifestCallback socket_option_value) { - return SOCKET_OPTION_NOT_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; - return SOCKET_OPTION_SET; - default: - return SOCKET_OPTION_NOT_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; - return SOCKET_OPTION_SET; - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ProducerSocket::setSocketOption( - int socket_option_key, const utils::Identity &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::IDENTITY: - identity_.reset(); - identity_ = std::make_unique<utils::Identity>(socket_option_value); - return SOCKET_OPTION_SET; - default: - return SOCKET_OPTION_NOT_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_); - return SOCKET_OPTION_SET; - } - - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::setSocketOption( - int socket_option_key, - interface::ConsumerTimerCallback socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::getSocketOption(int socket_option_key, - uint32_t &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::INPUT_BUFFER_SIZE: - socket_option_value = (int)input_buffer_capacity_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: - socket_option_value = (uint32_t)output_buffer_.getLimit(); - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::DATA_PACKET_SIZE: - socket_option_value = (uint32_t)data_packet_size_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME: - socket_option_value = content_object_expiry_time_; - return SOCKET_OPTION_GET; - - case GeneralTransportOptions::SIGNATURE_TYPE: - socket_option_value = signature_type_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_SET; - } -} - -int ProducerSocket::getSocketOption(int socket_option_key, - double &socket_option_value) { - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ProducerSocket::getSocketOption(int socket_option_key, - Name &socket_option_value) { - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ProducerContentObjectCallback &socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::NEW_CONTENT_OBJECT: - socket_option_value = on_new_segment_; - return SOCKET_OPTION_GET; - - case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN: - socket_option_value = on_content_object_to_sign_; - return SOCKET_OPTION_GET; - - case ProducerCallbacksOptions::CONTENT_OBJECT_READY: - socket_option_value = on_content_object_in_output_buffer_; - return SOCKET_OPTION_GET; - - case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: - socket_option_value = on_content_object_output_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ProducerContentCallback &socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::CONTENT_PRODUCED: - socket_option_value = on_content_produced_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ProducerInterestCallback &socket_option_value) { - switch (socket_option_key) { - case ProducerCallbacksOptions::INTEREST_INPUT: - socket_option_value = on_interest_input_; - return SOCKET_OPTION_GET; - - case ProducerCallbacksOptions::INTEREST_DROP: - socket_option_value = on_interest_dropped_input_buffer_; - return SOCKET_OPTION_GET; - - case ProducerCallbacksOptions::INTEREST_PASS: - socket_option_value = on_interest_inserted_input_buffer_; - return SOCKET_OPTION_GET; - - case CACHE_HIT: - socket_option_value = on_interest_satisfied_output_buffer_; - return SOCKET_OPTION_GET; - - case CACHE_MISS: - socket_option_value = on_interest_process_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ConsumerContentObjectCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::getSocketOption( - int socket_option_key, - ConsumerContentObjectVerificationCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ConsumerInterestCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ConsumerContentCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::getSocketOption( - int socket_option_key, ConsumerManifestCallback &socket_option_value) { - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - } - - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::getSocketOption(int socket_option_key, - IcnObserver **socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::setSocketOption(int socket_option_key, - IcnObserver *socket_option_value) { - return SOCKET_OPTION_NOT_SET; -} - -int ProducerSocket::getSocketOption(int socket_option_key, - HashAlgorithm &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::HASH_ALGORITHM: - socket_option_value = hash_algorithm_; - return SOCKET_OPTION_GET; - default: - return SOCKET_OPTION_NOT_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_; - return SOCKET_OPTION_GET; - default: - return SOCKET_OPTION_NOT_GET; - } -} - -int ProducerSocket::getSocketOption(int socket_option_key, - utils::Identity &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::IDENTITY: - if (identity_) { - socket_option_value = *identity_; - return SOCKET_OPTION_SET; - } - default: - return SOCKET_OPTION_NOT_SET; - } -} - -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_; - return SOCKET_OPTION_GET; - } - - return SOCKET_OPTION_NOT_GET; -} - -int ProducerSocket::getSocketOption( - int socket_option_key, - interface::ConsumerTimerCallback &socket_option_value) { - return SOCKET_OPTION_NOT_GET; -} - } // namespace interface } // end namespace transport diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h index d51464b77..4f38fb30e 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h @@ -18,7 +18,6 @@ #include <hicn/transport/interfaces/socket.h> #include <hicn/transport/utils/content_store.h> #include <hicn/transport/utils/event_thread.h> -#include <hicn/transport/utils/sharable_vector.h> #include <atomic> #include <cmath> @@ -26,8 +25,6 @@ #include <queue> #include <thread> -#define PUSH_API 1 - #define REGISTRATION_NOT_ATTEMPTED 0 #define REGISTRATION_SUCCESS 1 #define REGISTRATION_FAILURE 2 @@ -56,8 +53,7 @@ class ProducerSocket : public Socket<BasePortal>, void asyncProduce(const Name &suffix, const uint8_t *buf, size_t buffer_size); - void asyncProduce(const Name &suffix, - utils::SharableVector<uint8_t> &&output_buffer); + void asyncProduce(const Name &suffix, ContentBuffer &&output_buffer); void asyncProduce(ContentObject &content_object); @@ -75,125 +71,459 @@ class ProducerSocket : public Socket<BasePortal>, onInterest(*interest); }; - int setSocketOption(int socket_option_key, - uint32_t socket_option_value) override; - - int setSocketOption(int socket_option_key, - double socket_option_value) override; - - int setSocketOption(int socket_option_key, bool socket_option_value) override; - - int setSocketOption(int socket_option_key, Name socket_option_value) override; - - int setSocketOption(int socket_option_key, - std::list<Prefix> socket_option_value) override; - - int setSocketOption( + 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; + } + + TRANSPORT_ALWAYS_INLINE int setSocketOption( int socket_option_key, - ProducerContentObjectCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ProducerInterestCallback socket_option_value) override; - - int setSocketOption( + 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; + + case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT: + on_content_object_output_ = socket_option_value; + break; + + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + } + + 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; + + 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; + } + + 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; + + default: + return SOCKET_OPTION_NOT_SET; + } + + 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( int socket_option_key, - ConsumerContentObjectVerificationCallback socket_option_value) override; - - int setSocketOption( + 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; + ; + } + + 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 = input_buffer_capacity_; + break; + + case GeneralTransportOptions::OUTPUT_BUFFER_SIZE: + socket_option_value = output_buffer_.getLimit(); + break; + + case GeneralTransportOptions::DATA_PACKET_SIZE: + socket_option_value = data_packet_size_; + break; + + 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; + } + + 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( int socket_option_key, - ConsumerContentObjectCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ConsumerInterestCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ConsumerContentCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ConsumerManifestCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, IcnObserver *obs) override; - - int setSocketOption(int socket_option_key, - HashAlgorithm socket_option_value) override; - - int setSocketOption(int socket_option_key, - utils::CryptoSuite socket_option_value) override; - - int setSocketOption(int socket_option_key, - const utils::Identity &socket_option_value) override; - - int setSocketOption(int socket_option_key, - const std::string &socket_option_value) override; - - int setSocketOption(int socket_option_key, - ConsumerTimerCallback socket_option_value) override; - - int setSocketOption(int socket_option_key, - ProducerContentCallback socket_option_value) override; - - int getSocketOption(int socket_option_key, - uint32_t &socket_option_value) override; - - int getSocketOption(int socket_option_key, - double &socket_option_value) override; - - int getSocketOption(int socket_option_key, - bool &socket_option_value) override; - - int getSocketOption(int socket_option_key, - Name &socket_option_value) override; - - int getSocketOption(int socket_option_key, - std::list<Prefix> &socket_option_value) override; - - int getSocketOption( - int socket_option_key, - ProducerContentObjectCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ProducerInterestCallback &socket_option_value) override; - - int getSocketOption( + 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; + + 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; + } + + 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, 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; + } + + 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; + } + + TRANSPORT_ALWAYS_INLINE int getSocketOption( int socket_option_key, - ConsumerContentObjectVerificationCallback &socket_option_value) override; - - int getSocketOption( - int socket_option_key, - ConsumerContentObjectCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerInterestCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerContentCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerManifestCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - std::shared_ptr<Portal> &socket_option_value) override; - - int getSocketOption(int socket_option_key, - IcnObserver **socket_option_value) override; - - int getSocketOption(int socket_option_key, - HashAlgorithm &socket_option_value) override; - - int getSocketOption(int socket_option_key, - utils::CryptoSuite &socket_option_value) override; - - int getSocketOption(int socket_option_key, - utils::Identity &socket_option_value) override; - - int getSocketOption(int socket_option_key, - std::string &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ProducerContentCallback &socket_option_value) override; - - int getSocketOption(int socket_option_key, - ConsumerTimerCallback &socket_option_value) override; + 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; + } + + 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; + } protected: asio::io_service internal_io_service_; @@ -206,8 +536,6 @@ class ProducerSocket : public Socket<BasePortal>, // buffers utils::ContentStore output_buffer_; - std::unique_ptr<utils::Identity> identity_; - private: utils::EventThread async_thread_; @@ -215,7 +543,8 @@ class ProducerSocket : public Socket<BasePortal>, bool making_manifest_; - // map for storing sequence numbers for several calls of the publish function + // 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_; @@ -223,8 +552,7 @@ class ProducerSocket : public Socket<BasePortal>, HashAlgorithm hash_algorithm_; utils::CryptoSuite crypto_suite_; - // std::unique_ptr<utils::Identity> identity_; - // utils::Signer& signer_; + std::shared_ptr<utils::Identity> identity_; // buffers @@ -232,11 +560,6 @@ class ProducerSocket : public Socket<BasePortal>, std::atomic_size_t input_buffer_capacity_; std::atomic_size_t input_buffer_size_; -#ifndef PUSH_API - std::mutex pending_interests_mtx_; - std::unordered_map<Name, std::shared_ptr<const Interest>> pending_interests_; -#endif - // threads std::thread listening_thread_; std::thread processing_thread_; @@ -268,4 +591,4 @@ class ProducerSocket : public Socket<BasePortal>, } // namespace interface -} // end namespace transport +} // namespace transport diff --git a/libtransport/src/hicn/transport/portability/win_portability.h b/libtransport/src/hicn/transport/portability/win_portability.h index ca262144b..65d949291 100644 --- a/libtransport/src/hicn/transport/portability/win_portability.h +++ b/libtransport/src/hicn/transport/portability/win_portability.h @@ -18,19 +18,19 @@ #pragma once
#define WIN32_LEAN_AND_MEAN
#define NOMINMAX
-#include <parc/windows/parc_Utils.h>
-#include <windows.h>
-#include <winsock2.h>
-#include <ws2ipdef.h>
-#include <ws2tcpip.h>
#include <fcntl.h>
#include <io.h>
+#include <parc/windows/parc_Utils.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
+#include <windows.h>
+#include <winsock2.h>
+#include <ws2ipdef.h>
+#include <ws2tcpip.h>
#include <algorithm>
#define __ORDER_LITTLE_ENDIAN__ 0x41424344UL
diff --git a/libtransport/src/hicn/transport/protocols/CMakeLists.txt b/libtransport/src/hicn/transport/protocols/CMakeLists.txt index 1c3b76c24..84a4d5279 100644 --- a/libtransport/src/hicn/transport/protocols/CMakeLists.txt +++ b/libtransport/src/hicn/transport/protocols/CMakeLists.txt @@ -14,33 +14,52 @@ cmake_minimum_required(VERSION 3.5 FATAL_ERROR) list(APPEND HEADER_FILES + ${CMAKE_CURRENT_SOURCE_DIR}/indexing_manager.h + ${CMAKE_CURRENT_SOURCE_DIR}/reassembly.h + ${CMAKE_CURRENT_SOURCE_DIR}/congestion_window_protocol.h + ${CMAKE_CURRENT_SOURCE_DIR}/packet_manager.h + ${CMAKE_CURRENT_SOURCE_DIR}/statistics.h ${CMAKE_CURRENT_SOURCE_DIR}/rate_estimation.h ${CMAKE_CURRENT_SOURCE_DIR}/download_observer.h - ${CMAKE_CURRENT_SOURCE_DIR}/vegas.h ${CMAKE_CURRENT_SOURCE_DIR}/protocol.h ${CMAKE_CURRENT_SOURCE_DIR}/raaqm.h - ${CMAKE_CURRENT_SOURCE_DIR}/vegas_rto_estimator.h ${CMAKE_CURRENT_SOURCE_DIR}/raaqm_data_path.h ${CMAKE_CURRENT_SOURCE_DIR}/cbr.h ${CMAKE_CURRENT_SOURCE_DIR}/rtc.h ${CMAKE_CURRENT_SOURCE_DIR}/rtc_data_path.h + ${CMAKE_CURRENT_SOURCE_DIR}/manifest_indexing_manager.h ) list(APPEND SOURCE_FILES - ${CMAKE_CURRENT_SOURCE_DIR}/vegas.cc + ${CMAKE_CURRENT_SOURCE_DIR}/reassembly.cc ${CMAKE_CURRENT_SOURCE_DIR}/protocol.cc ${CMAKE_CURRENT_SOURCE_DIR}/raaqm.cc - ${CMAKE_CURRENT_SOURCE_DIR}/vegas_rto_estimator.cc ${CMAKE_CURRENT_SOURCE_DIR}/rate_estimation.cc ${CMAKE_CURRENT_SOURCE_DIR}/raaqm_data_path.cc ${CMAKE_CURRENT_SOURCE_DIR}/cbr.cc ${CMAKE_CURRENT_SOURCE_DIR}/rtc.cc ${CMAKE_CURRENT_SOURCE_DIR}/rtc_data_path.cc + ${CMAKE_CURRENT_SOURCE_DIR}/manifest_indexing_manager.cc +) + +set(RAAQM_CONFIG_INSTALL_PREFIX +${CMAKE_INSTALL_PREFIX}/etc/hicn +) + +set(raaqm_config_path + ${RAAQM_CONFIG_INSTALL_PREFIX}/consumer.conf + PARENT_SCOPE ) set(TRANSPORT_CONFIG ${CMAKE_CURRENT_SOURCE_DIR}/consumer.conf ) +install( + FILES ${TRANSPORT_CONFIG} + DESTINATION etc/hicn + COMPONENT lib${LIBTRANSPORT} +) + set(SOURCE_FILES ${SOURCE_FILES} PARENT_SCOPE) set(HEADER_FILES ${HEADER_FILES} PARENT_SCOPE)
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/protocols/cbr.cc b/libtransport/src/hicn/transport/protocols/cbr.cc index 3da4819c3..efd2149ad 100644 --- a/libtransport/src/hicn/transport/protocols/cbr.cc +++ b/libtransport/src/hicn/transport/protocols/cbr.cc @@ -22,25 +22,27 @@ namespace protocol { using namespace interface; -CbrTransportProtocol::CbrTransportProtocol(BaseSocket *icnet_socket) - : VegasTransportProtocol(icnet_socket) {} - -void CbrTransportProtocol::start( - utils::SharableVector<uint8_t> &receive_buffer) { - current_window_size_ = socket_->current_window_size_; - VegasTransportProtocol::start(receive_buffer); +CbrTransportProtocol::CbrTransportProtocol( + interface::ConsumerSocket *icnet_socket) + : RaaqmTransportProtocol(icnet_socket) {} + +int CbrTransportProtocol::start() { + socket_->getSocketOption(GeneralTransportOptions::CURRENT_WINDOW_SIZE, + current_window_size_); + return RaaqmTransportProtocol::start(); } -void CbrTransportProtocol::changeInterestLifetime(uint64_t segment) { return; } - -void CbrTransportProtocol::increaseWindow() {} - -void CbrTransportProtocol::decreaseWindow() {} - void CbrTransportProtocol::afterDataUnsatisfied(uint64_t segment) {} void CbrTransportProtocol::afterContentReception( - const Interest &interest, const ContentObject &content_object) {} + const Interest &interest, const ContentObject &content_object) { + auto segment = content_object.getName().getSuffix(); + auto now = utils::SteadyClock::now(); + auto rtt = std::chrono::duration_cast<utils::Microseconds>( + now - interest_timepoints_[segment & mask]); + // Update stats + updateStats(segment, rtt.count(), now); +} } // end namespace protocol diff --git a/libtransport/src/hicn/transport/protocols/cbr.h b/libtransport/src/hicn/transport/protocols/cbr.h index 0a572292a..a8eff2182 100644 --- a/libtransport/src/hicn/transport/protocols/cbr.h +++ b/libtransport/src/hicn/transport/protocols/cbr.h @@ -15,32 +15,22 @@ #pragma once -#include <hicn/transport/protocols/raaqm_data_path.h> -#include <hicn/transport/protocols/rate_estimation.h> -#include <hicn/transport/protocols/vegas.h> -#include <hicn/transport/protocols/vegas_rto_estimator.h> +#include <hicn/transport/protocols/raaqm.h> namespace transport { namespace protocol { -class CbrTransportProtocol : public VegasTransportProtocol { +class CbrTransportProtocol : public RaaqmTransportProtocol { public: - CbrTransportProtocol(interface::BaseSocket *icnet_socket); + CbrTransportProtocol(interface::ConsumerSocket *icnet_socket); - void start(utils::SharableVector<uint8_t> &receive_buffer) override; + int start() override; private: void afterContentReception(const Interest &interest, const ContentObject &content_object) override; - void afterDataUnsatisfied(uint64_t segment) override; - - void increaseWindow() override; - - void decreaseWindow() override; - - void changeInterestLifetime(uint64_t segment) override; }; } // end namespace protocol diff --git a/libtransport/src/hicn/transport/protocols/vegas_rto_estimator.h b/libtransport/src/hicn/transport/protocols/congestion_window_protocol.h index e84afc49c..36ac6eb17 100644 --- a/libtransport/src/hicn/transport/protocols/vegas_rto_estimator.h +++ b/libtransport/src/hicn/transport/protocols/congestion_window_protocol.h @@ -15,34 +15,16 @@ #pragma once -#include <chrono> - -// Implementation inspired from RFC6298 -// (https://tools.ietf.org/search/rfc6298#ref-JK88) - namespace transport { namespace protocol { -class RtoEstimator { - public: - typedef std::chrono::microseconds Duration; - - static Duration getInitialRtt() { return std::chrono::seconds(1); } - - RtoEstimator(Duration min_rto = std::chrono::seconds(1)); - - void addMeasurement(Duration measure); - - Duration computeRto() const; - - private: - double smoothed_rtt_; - double rtt_variation_; - bool first_measurement_; - double last_rto_; +class CWindowProtocol { + protected: + virtual void increaseWindow() = 0; + virtual void decreaseWindow() = 0; }; } // end namespace protocol -} // end namespace transport
\ No newline at end of file +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/indexing_manager.h b/libtransport/src/hicn/transport/protocols/indexing_manager.h new file mode 100644 index 000000000..888b17d9d --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/indexing_manager.h @@ -0,0 +1,193 @@ +/* + * 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. + */ + +#pragma once + +#include <hicn/transport/errors/runtime_exception.h> +#include <hicn/transport/errors/unexpected_manifest_exception.h> +#include <hicn/transport/interfaces/socket_consumer.h> +#include <hicn/transport/protocols/verification_manager.h> +#include <hicn/transport/utils/literals.h> + +#include <deque> + +namespace transport { + +namespace protocol { + +class IndexManager { + public: + static constexpr uint32_t invalid_index = ~0; + + /** + * + */ + virtual ~IndexManager() = default; + /** + * Retrieve from the manifest the next suffix to retrieve. + */ + virtual uint32_t getNextSuffix() = 0; + + /** + * Retrive the next segment to be reassembled. + */ + virtual uint32_t getNextReassemblySegment() = 0; + + virtual bool isFinalSuffixDiscovered() = 0; + + virtual uint32_t getFinalSuffix() = 0; + + virtual void reset() = 0; +}; + +class IndexVerificationManager : public IndexManager { + public: + /** + * + */ + virtual ~IndexVerificationManager() = default; + + /** + * The ownership of the ContentObjectManifest is moved + * from the caller to the VerificationManager + */ + virtual bool onManifest(core::ContentObject::Ptr &&content_object) = 0; + + /** + * The content object must just be verified; the ownership is still of the + * caller. + */ + virtual bool onContentObject(const core::ContentObject &content_object) = 0; +}; + +class ZeroIndexManager : public IndexVerificationManager { + public: + ZeroIndexManager() : reset_(true) {} + + TRANSPORT_ALWAYS_INLINE virtual void reset() override { reset_ = true; } + + /** + * Retrieve from the manifest the next suffix to retrieve. + */ + TRANSPORT_ALWAYS_INLINE virtual uint32_t getNextSuffix() override { + uint32_t ret = reset_ ? 0 : IndexManager::invalid_index; + reset_ = false; + return ret; + } + + /** + * Retrive the next segment to be reassembled. + */ + TRANSPORT_ALWAYS_INLINE virtual uint32_t getNextReassemblySegment() override { + return IndexManager::invalid_index; + } + + TRANSPORT_ALWAYS_INLINE virtual bool isFinalSuffixDiscovered() override { + return false; + } + + TRANSPORT_ALWAYS_INLINE virtual uint32_t getFinalSuffix() override { + return IndexManager::invalid_index; + } + + TRANSPORT_ALWAYS_INLINE bool onManifest( + core::ContentObject::Ptr &&content_object) override { + throw errors::UnexpectedManifestException(); + } + + TRANSPORT_ALWAYS_INLINE bool onContentObject( + const core::ContentObject &content_object) override { + throw errors::RuntimeException( + "Called onContentObject on a ZeroIndexManager, which is not able to " + "process packets."); + } + + private: + bool reset_; +}; + +class IncrementalIndexManager : public IndexVerificationManager { + public: + IncrementalIndexManager(interface::ConsumerSocket *icn_socket) + : socket_(icn_socket), + final_suffix_(std::numeric_limits<uint64_t>::max()), + next_download_suffix_(0), + next_reassembly_suffix_(0), + verification_manager_( + std::make_unique<SignatureVerificationManager>(icn_socket)) {} + + /** + * + */ + virtual ~IncrementalIndexManager() {} + + TRANSPORT_ALWAYS_INLINE virtual void reset() override { + final_suffix_ = std::numeric_limits<uint64_t>::max(); + next_download_suffix_ = 0; + next_reassembly_suffix_ = 0; + } + + /** + * Retrieve from the manifest the next suffix to retrieve. + */ + TRANSPORT_ALWAYS_INLINE virtual uint32_t getNextSuffix() override { + return next_download_suffix_ <= final_suffix_ ? next_download_suffix_++ + : IndexManager::invalid_index; + } + + /** + * Retrive the next segment to be reassembled. + */ + TRANSPORT_ALWAYS_INLINE virtual uint32_t getNextReassemblySegment() override { + return next_reassembly_suffix_ <= final_suffix_ + ? next_reassembly_suffix_++ + : IndexManager::invalid_index; + } + + TRANSPORT_ALWAYS_INLINE virtual bool isFinalSuffixDiscovered() override { + return final_suffix_ != std::numeric_limits<uint64_t>::max(); + } + + TRANSPORT_ALWAYS_INLINE virtual uint32_t getFinalSuffix() override { + return final_suffix_; + } + + TRANSPORT_ALWAYS_INLINE bool onManifest( + core::ContentObject::Ptr &&content_object) override { + throw errors::UnexpectedManifestException(); + } + + TRANSPORT_ALWAYS_INLINE bool onContentObject( + const core::ContentObject &content_object) override { + auto ret = verification_manager_->onPacketToVerify(content_object); + + if (TRANSPORT_EXPECT_FALSE(content_object.testRst())) { + final_suffix_ = content_object.getName().getSuffix(); + } + + return ret; + } + + protected: + interface::ConsumerSocket *socket_; + uint64_t final_suffix_; + uint64_t next_download_suffix_; + uint64_t next_reassembly_suffix_; + std::unique_ptr<VerificationManager> verification_manager_; +}; + +} // end namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/manifest_indexing_manager.cc b/libtransport/src/hicn/transport/protocols/manifest_indexing_manager.cc new file mode 100644 index 000000000..1afb4eaac --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/manifest_indexing_manager.cc @@ -0,0 +1,217 @@ +/* + * 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 <hicn/transport/interfaces/socket_consumer.h> +#include <hicn/transport/protocols/manifest_indexing_manager.h> + +#include <cmath> +#include <deque> + +namespace transport { + +namespace protocol { + +using namespace interface; + +ManifestIndexManager::ManifestIndexManager( + interface::ConsumerSocket *icn_socket) + : IncrementalIndexManager(icn_socket), + PacketManager<Interest>(1024), + next_reassembly_segment_(suffix_queue_.end()), + next_to_retrieve_segment_(suffix_queue_.end()), + next_manifest_(0) {} + +bool ManifestIndexManager::onManifest( + core::ContentObject::Ptr &&content_object) { + auto manifest = + std::make_unique<ContentObjectManifest>(std::move(*content_object)); + + bool manifest_verified = verification_manager_->onPacketToVerify(*manifest); + + if (manifest_verified) { + manifest->decode(); + + if (TRANSPORT_EXPECT_FALSE(manifest->getVersion() != + core::ManifestVersion::VERSION_1)) { + throw errors::RuntimeException("Received manifest with unknown version."); + } + + switch (manifest->getManifestType()) { + case core::ManifestType::INLINE_MANIFEST: { + auto _it = manifest->getSuffixList().begin(); + auto _end = --manifest->getSuffixList().end(); + + if (TRANSPORT_EXPECT_FALSE(manifest->isFinalManifest())) { + _end++; + } + + // Get final block number + final_suffix_ = manifest->getFinalBlockNumber(); + + suffix_hash_map_[_it->first] = + std::make_pair(std::vector<uint8_t>(_it->second, _it->second + 32), + manifest->getHashAlgorithm()); + suffix_queue_.push_back(_it->first); + + // If the transport protocol finished the list of segments to retrieve, + // reset the next_to_retrieve_segment_ iterator to the next segment + // provided by this manifest. + if (TRANSPORT_EXPECT_FALSE(next_to_retrieve_segment_ == + suffix_queue_.end())) { + next_to_retrieve_segment_ = --suffix_queue_.end(); + } + + std::advance(_it, 1); + for (; _it != _end; _it++) { + suffix_hash_map_[_it->first] = std::make_pair( + std::vector<uint8_t>(_it->second, _it->second + 32), + manifest->getHashAlgorithm()); + suffix_queue_.push_back(_it->first); + } + + if (TRANSPORT_EXPECT_FALSE(manifest->getName().getSuffix()) == 0) { + // Set the iterators to the beginning of the suffix queue + next_reassembly_segment_ = suffix_queue_.begin(); + } + + if (TRANSPORT_EXPECT_FALSE(manifest->isFinalManifest() || + next_manifest_ > final_suffix_)) { + break; + } + + // Get current window size + double current_window_size = 0.; + socket_->getSocketOption(GeneralTransportOptions::CURRENT_WINDOW_SIZE, + current_window_size); + + // Get portal + std::shared_ptr<interface::BasePortal> portal; + socket_->getSocketOption(GeneralTransportOptions::PORTAL, portal); + + // Number of segments in manifest + std::size_t segments_in_manifest = std::distance( + manifest->getSuffixList().begin(), manifest->getSuffixList().end()); + std::size_t segment_count = 0; + + // Manifest namespace + Name &name = manifest->getWritableName(); + + // Send as many manifest as required for filling window. + do { + segment_count += segments_in_manifest; + next_manifest_ += segments_in_manifest; + + Interest::Ptr interest = getPacket(); + name.setSuffix(next_manifest_); + interest->setName(name); + + uint32_t interest_lifetime; + socket_->getSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, + interest_lifetime); + interest->setLifetime(interest_lifetime); + + // Send requests for manifest out of the congestion window (no + // in_flight_interest++) + portal->sendInterest(std::move(interest)); + } while (segment_count < current_window_size && + next_manifest_ < final_suffix_); + + break; + } + case core::ManifestType::FLIC_MANIFEST: { + throw errors::NotImplementedException(); + } + case core::ManifestType::FINAL_CHUNK_NUMBER: { + throw errors::NotImplementedException(); + } + } + } + + return manifest_verified; +} + +bool ManifestIndexManager::onContentObject( + const core::ContentObject &content_object) { + bool verify_signature; + socket_->getSocketOption(GeneralTransportOptions::VERIFY_SIGNATURE, + verify_signature); + + if (!verify_signature) { + return true; + } + + uint64_t segment = content_object.getName().getSuffix(); + + bool ret = false; + + auto it = suffix_hash_map_.find(segment); + if (it != suffix_hash_map_.end()) { + auto hash_type = static_cast<utils::CryptoHashType>(it->second.second); + auto data_packet_digest = content_object.computeDigest(it->second.second); + auto data_packet_digest_bytes = + data_packet_digest.getDigest<uint8_t>().data(); + std::vector<uint8_t> &manifest_digest_bytes = it->second.first; + + if (utils::CryptoHash::compareBinaryDigest(data_packet_digest_bytes, + manifest_digest_bytes.data(), + hash_type)) { + suffix_hash_map_.erase(it); + ret = true; + } else { + throw errors::RuntimeException( + "Verification failure policy has to be implemented."); + } + } + + return ret; +} + +uint32_t ManifestIndexManager::getNextSuffix() { + if (TRANSPORT_EXPECT_FALSE(next_to_retrieve_segment_ == + suffix_queue_.end())) { + return invalid_index; + } + + return *next_to_retrieve_segment_++; +} + +uint32_t ManifestIndexManager::getFinalSuffix() { return final_suffix_; } + +bool ManifestIndexManager::isFinalSuffixDiscovered() { + return IncrementalIndexManager::isFinalSuffixDiscovered(); +} + +uint32_t ManifestIndexManager::getNextReassemblySegment() { + if (TRANSPORT_EXPECT_FALSE(next_reassembly_segment_ == suffix_queue_.end())) { + return invalid_index; + } + + if (TRANSPORT_EXPECT_TRUE(next_reassembly_segment_ != + suffix_queue_.begin())) { + suffix_queue_.erase(std::prev(next_reassembly_segment_)); + } + + return *next_reassembly_segment_++; +} + +void ManifestIndexManager::reset() { + IncrementalIndexManager::reset(); + suffix_queue_.clear(); + suffix_hash_map_.clear(); +} + +} // end namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/manifest_indexing_manager.h b/libtransport/src/hicn/transport/protocols/manifest_indexing_manager.h new file mode 100644 index 000000000..ee2c531c7 --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/manifest_indexing_manager.h @@ -0,0 +1,69 @@ +/* + * 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. + */ + +#pragma once + +#include <hicn/transport/interfaces/socket.h> +#include <hicn/transport/protocols/indexing_manager.h> + +#include <list> + +namespace transport { + +namespace protocol { + +class ManifestIndexManager : public IncrementalIndexManager, + public PacketManager<Interest> { + static constexpr double alpha = 0.3; + + public: + using SuffixQueue = std::list<uint32_t>; + using HashEntry = std::pair<std::vector<uint8_t>, core::HashAlgorithm>; + + ManifestIndexManager(interface::ConsumerSocket *icn_socket); + + virtual ~ManifestIndexManager() = default; + + void reset() override; + + bool onManifest(core::ContentObject::Ptr &&content_object) override; + + bool onContentObject(const core::ContentObject &content_object) override; + + uint32_t getNextSuffix() override; + + uint32_t getNextReassemblySegment() override; + + bool isFinalSuffixDiscovered() override; + + uint32_t getFinalSuffix() override; + + protected: + SuffixQueue suffix_queue_; + SuffixQueue::iterator next_reassembly_segment_; + SuffixQueue::iterator next_to_retrieve_segment_; + + // Hash verification + std::unordered_map<uint32_t, + std::pair<std::vector<uint8_t>, core::HashAlgorithm>> + suffix_hash_map_; + + // Next Manifest + std::uint32_t next_manifest_; +}; + +} // end namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/packet_manager.h b/libtransport/src/hicn/transport/protocols/packet_manager.h new file mode 100644 index 000000000..53486edde --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/packet_manager.h @@ -0,0 +1,66 @@ +/* + * 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. + */ + +#pragma once + +#include <hicn/transport/utils/object_pool.h> + +namespace transport { + +namespace protocol { + +using namespace core; + +template <typename PacketType> +class PacketManager { + static_assert(std::is_base_of<Packet, PacketType>::value, + "The packet manager support just Interest and Data."); + + static constexpr std::size_t packet_pool_size = 4096; + + public: + PacketManager(std::size_t size = packet_pool_size) : size_(0) { + // Create pool of interests + increasePoolSize(size); + } + + TRANSPORT_ALWAYS_INLINE void increasePoolSize(std::size_t size) { + for (std::size_t i = 0; i < size; i++) { + interest_pool_.add(new PacketType()); + } + + size_ += size; + } + + TRANSPORT_ALWAYS_INLINE typename PacketType::Ptr getPacket() { + auto result = interest_pool_.get(); + + while (TRANSPORT_EXPECT_FALSE(!result.first)) { + // Add packets to the pool + increasePoolSize(size_); + result = interest_pool_.get(); + } + + return std::move(result.second); + } + + private: + utils::ObjectPool<PacketType> interest_pool_; + std::size_t size_; +}; + +} // end namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/protocol.cc b/libtransport/src/hicn/transport/protocols/protocol.cc index ea4fd6dbf..9caa2eca7 100644 --- a/libtransport/src/hicn/transport/protocols/protocol.cc +++ b/libtransport/src/hicn/transport/protocols/protocol.cc @@ -20,24 +20,50 @@ namespace transport { namespace protocol { -TransportProtocol::TransportProtocol(interface::BaseSocket *icn_socket) - : socket_(dynamic_cast<interface::ConsumerSocket *>(icn_socket)), - is_running_(false), - interest_pool_() { - // Create pool of interests - increasePoolSize(); +using namespace interface; + +TransportProtocol::TransportProtocol(interface::ConsumerSocket *icn_socket) + : socket_(icn_socket), is_running_(false) { + socket_->getSocketOption(GeneralTransportOptions::PORTAL, portal_); +} + +int TransportProtocol::start() { + // If the protocol is already running, return + if (is_running_) return -1; + + // Set the protocol as running + is_running_ = true; + + // Reset the protocol state machine + reset(); + + // Schedule next interests + scheduleNextInterests(); + + // Start Event loop + portal_->runEventsLoop(); + + // Not running anymore + is_running_ = false; + + return 0; } -TransportProtocol::~TransportProtocol() {} +void TransportProtocol::stop() { + is_running_ = false; + portal_->stopEventsLoop(); +} + +void TransportProtocol::resume() { + if (is_running_) return; + + is_running_ = true; -void TransportProtocol::updatePortal() { portal_ = socket_->portal_; } + scheduleNextInterests(); -bool TransportProtocol::isRunning() { return is_running_; } + portal_->runEventsLoop(); -void TransportProtocol::increasePoolSize(std::size_t size) { - for (std::size_t i = 0; i < size; i++) { - interest_pool_.add(new Interest()); - } + is_running_ = false; } } // end namespace protocol diff --git a/libtransport/src/hicn/transport/protocols/protocol.h b/libtransport/src/hicn/transport/protocols/protocol.h index 56c57e025..6a3b23753 100644 --- a/libtransport/src/hicn/transport/protocols/protocol.h +++ b/libtransport/src/hicn/transport/protocols/protocol.h @@ -16,8 +16,9 @@ #pragma once #include <hicn/transport/interfaces/socket.h> +#include <hicn/transport/protocols/packet_manager.h> +#include <hicn/transport/protocols/statistics.h> #include <hicn/transport/utils/object_pool.h> -#include <hicn/transport/utils/sharable_vector.h> namespace transport { @@ -31,39 +32,28 @@ class TransportProtocolCallback { virtual void onTimeout(const core::Interest &interest) = 0; }; -class TransportProtocol : public interface::BasePortal::ConsumerCallback { +class TransportProtocol : public interface::BasePortal::ConsumerCallback, + public PacketManager<Interest> { static constexpr std::size_t interest_pool_size = 4096; public: - TransportProtocol(interface::BaseSocket *icn_socket); + TransportProtocol(interface::ConsumerSocket *icn_socket); - virtual ~TransportProtocol(); + virtual ~TransportProtocol() { stop(); }; - void updatePortal(); + TRANSPORT_ALWAYS_INLINE bool isRunning() { return is_running_; } - bool isRunning(); + virtual int start(); - virtual void start(utils::SharableVector<uint8_t> &content_buffer) = 0; + virtual void stop(); - virtual void stop() = 0; + virtual void resume(); - virtual void resume() = 0; + virtual void scheduleNextInterests() = 0; protected: - virtual void increasePoolSize(std::size_t size = interest_pool_size); - - TRANSPORT_ALWAYS_INLINE Interest::Ptr getInterest() { - auto result = interest_pool_.get(); - - while (TRANSPORT_EXPECT_FALSE(!result.first)) { - // Add packets to the pool - increasePoolSize(); - result = interest_pool_.get(); - } - - return std::move(result.second); - } // Consumer Callback + virtual void reset() = 0; virtual void onContentObject(Interest::Ptr &&i, ContentObject::Ptr &&c) = 0; virtual void onTimeout(Interest::Ptr &&i) = 0; @@ -71,7 +61,8 @@ class TransportProtocol : public interface::BasePortal::ConsumerCallback { interface::ConsumerSocket *socket_; std::shared_ptr<interface::BasePortal> portal_; volatile bool is_running_; - utils::ObjectPool<Interest> interest_pool_; + TransportStatistics stats_; + std::shared_ptr<std::vector<uint8_t>> content_buffer_; }; } // end namespace protocol diff --git a/libtransport/src/hicn/transport/protocols/raaqm.cc b/libtransport/src/hicn/transport/protocols/raaqm.cc index 3e04689e9..f5eb48bd8 100644 --- a/libtransport/src/hicn/transport/protocols/raaqm.cc +++ b/libtransport/src/hicn/transport/protocols/raaqm.cc @@ -14,8 +14,10 @@ */ #include <hicn/transport/interfaces/socket_consumer.h> +#include <hicn/transport/protocols/manifest_indexing_manager.h> #include <hicn/transport/protocols/raaqm.h> +#include <cstdlib> #include <fstream> namespace transport { @@ -24,8 +26,14 @@ namespace protocol { using namespace interface; -RaaqmTransportProtocol::RaaqmTransportProtocol(BaseSocket *icnet_socket) - : VegasTransportProtocol(icnet_socket), rate_estimator_(NULL) { +RaaqmTransportProtocol::RaaqmTransportProtocol(ConsumerSocket *icnet_socket) + : TransportProtocol(icnet_socket), + BaseReassembly(icnet_socket, this), + current_window_size_(1), + interests_in_flight_(0), + cur_path_(nullptr), + t0_(utils::SteadyClock::now()), + rate_estimator_(nullptr) { init(); } @@ -35,16 +43,123 @@ RaaqmTransportProtocol::~RaaqmTransportProtocol() { } } +int RaaqmTransportProtocol::start() { + if (this->rate_estimator_) { + this->rate_estimator_->onStart(); + } + + if (!cur_path_) { + // RAAQM + double drop_factor; + double minimum_drop_probability; + uint32_t sample_number; + uint32_t interest_lifetime; + + socket_->getSocketOption(RaaqmTransportOptions::DROP_FACTOR, drop_factor); + socket_->getSocketOption(RaaqmTransportOptions::MINIMUM_DROP_PROBABILITY, + minimum_drop_probability); + socket_->getSocketOption(RaaqmTransportOptions::SAMPLE_NUMBER, + sample_number); + socket_->getSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, + interest_lifetime); + + // Rate Estimation + double alpha = 0.0; + uint32_t batching_param = 0; + uint32_t choice_param = 0; + socket_->getSocketOption(RateEstimationOptions::RATE_ESTIMATION_ALPHA, + alpha); + socket_->getSocketOption( + RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER, batching_param); + socket_->getSocketOption(RateEstimationOptions::RATE_ESTIMATION_CHOICE, + choice_param); + + if (choice_param == 1) { + this->rate_estimator_ = new ALaTcpEstimator(); + } else { + this->rate_estimator_ = new SimpleEstimator(alpha, batching_param); + } + + socket_->getSocketOption(RateEstimationOptions::RATE_ESTIMATION_OBSERVER, + &this->rate_estimator_->observer_); + + // Current path + auto cur_path = std::make_unique<RaaqmDataPath>( + drop_factor, minimum_drop_probability, interest_lifetime * 1000, + sample_number); + cur_path_ = cur_path.get(); + path_table_[default_values::path_id] = std::move(cur_path); + } + + portal_->setConsumerCallback(this); + return TransportProtocol::start(); +} + +void RaaqmTransportProtocol::resume() { return TransportProtocol::resume(); } + +void RaaqmTransportProtocol::reset() { + // Reset reassembly component + BaseReassembly::reset(); + + // Reset protocol variables + interests_in_flight_ = 0; + t0_ = utils::SteadyClock::now(); +} + +void RaaqmTransportProtocol::increaseWindow() { + double max_window_size = 0.; + socket_->getSocketOption(GeneralTransportOptions::MAX_WINDOW_SIZE, + max_window_size); + if (current_window_size_ < max_window_size) { + double gamma = 0.; + socket_->getSocketOption(RaaqmTransportOptions::GAMMA_VALUE, gamma); + + current_window_size_ += gamma / current_window_size_; + socket_->setSocketOption(GeneralTransportOptions::CURRENT_WINDOW_SIZE, + current_window_size_); + } + this->rate_estimator_->onWindowIncrease(current_window_size_); +} + +void RaaqmTransportProtocol::decreaseWindow() { + double min_window_size = 0.; + socket_->getSocketOption(GeneralTransportOptions::MIN_WINDOW_SIZE, + min_window_size); + if (current_window_size_ > min_window_size) { + double beta = 0.; + socket_->getSocketOption(RaaqmTransportOptions::BETA_VALUE, beta); + + current_window_size_ = current_window_size_ * beta; + if (current_window_size_ < min_window_size) { + current_window_size_ = min_window_size; + } + + socket_->setSocketOption(GeneralTransportOptions::CURRENT_WINDOW_SIZE, + current_window_size_); + } + this->rate_estimator_->onWindowDecrease(current_window_size_); +} + +void RaaqmTransportProtocol::afterDataUnsatisfied(uint64_t segment) { + // Decrease the window because the timeout happened + decreaseWindow(); +} + +void RaaqmTransportProtocol::afterContentReception( + const Interest &interest, const ContentObject &content_object) { + updatePathTable(content_object); + increaseWindow(); + updateRtt(interest.getName().getSuffix()); + this->rate_estimator_->onDataReceived((int)content_object.payloadSize() + + content_object.headerSize()); + // Set drop probablility and window size accordingly + RAAQM(); +} + void RaaqmTransportProtocol::init() { std::ifstream is(RAAQM_CONFIG_PATH); std::string line; - - socket_->beta_ = default_values::beta_value; - socket_->drop_factor_ = default_values::drop_factor; - socket_->interest_lifetime_ = default_values::interest_lifetime; - socket_->max_retransmissions_ = - default_values::transport_protocol_max_retransmissions; raaqm_autotune_ = false; default_beta_ = default_values::beta_value; default_drop_ = default_values::drop_factor; @@ -56,7 +171,9 @@ void RaaqmTransportProtocol::init() { lte_delay_ = 15000; if (!is) { - TRANSPORT_LOGW("WARNING: RAAQM parameters not found, set default values"); + TRANSPORT_LOGW( + "WARNING: RAAQM parameters not found at %s, set default values", + RAAQM_CONFIG_PATH); return; } @@ -86,7 +203,8 @@ void RaaqmTransportProtocol::init() { std::string tmp; uint32_t lifetime; line_s >> tmp >> lifetime; - socket_->interest_lifetime_ = lifetime; + socket_->setSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, + lifetime); continue; } @@ -94,21 +212,23 @@ void RaaqmTransportProtocol::init() { std::string tmp; uint32_t rtx; line_s >> tmp >> rtx; - socket_->max_retransmissions_ = rtx; + socket_->setSocketOption(GeneralTransportOptions::MAX_INTEREST_RETX, rtx); continue; } if (command == "beta") { std::string tmp; line_s >> tmp >> default_beta_; - socket_->beta_ = default_beta_; + socket_->setSocketOption(RaaqmTransportOptions::BETA_VALUE, + default_beta_); continue; } if (command == "drop") { std::string tmp; line_s >> tmp >> default_drop_; - socket_->drop_factor_ = default_drop_; + socket_->setSocketOption(RaaqmTransportOptions::DROP_FACTOR, + default_drop_); continue; } @@ -151,7 +271,8 @@ void RaaqmTransportProtocol::init() { std::string tmp; double rate_alpha = 0.0; line_s >> tmp >> rate_alpha; - socket_->rate_estimation_alpha_ = rate_alpha; + socket_->setSocketOption(RateEstimationOptions::RATE_ESTIMATION_ALPHA, + rate_alpha); continue; } @@ -159,7 +280,9 @@ void RaaqmTransportProtocol::init() { std::string tmp; uint32_t batching_param = 0; line_s >> tmp >> batching_param; - socket_->rate_estimation_batching_parameter_ = batching_param; + socket_->setSocketOption( + RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER, + batching_param); continue; } @@ -167,133 +290,332 @@ void RaaqmTransportProtocol::init() { std::string tmp; uint32_t choice_param = 0; line_s >> tmp >> choice_param; - socket_->rate_estimation_choice_ = choice_param; + socket_->setSocketOption(RateEstimationOptions::RATE_ESTIMATION_CHOICE, + choice_param); continue; } } + is.close(); } -void RaaqmTransportProtocol::start( - utils::SharableVector<uint8_t> &content_buffer) { - if (this->rate_estimator_) { - this->rate_estimator_->onStart(); +void RaaqmTransportProtocol::onContentObject( + Interest::Ptr &&interest, ContentObject::Ptr &&content_object) { + uint32_t incremental_suffix = content_object->getName().getSuffix(); + + // Check whether makes sense to continue + if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + return; } - if (!cur_path_) { - double drop_factor; - double minimum_drop_probability; - uint32_t sample_number; - uint32_t interest_lifetime; - // double beta; + // Call application-defined callbacks + ConsumerContentObjectCallback *callback_content_object = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT, + &callback_content_object); + if (*callback_content_object != VOID_HANDLER) { + (*callback_content_object)(*socket_, *content_object); + } - drop_factor = socket_->drop_factor_; - minimum_drop_probability = socket_->minimum_drop_probability_; - sample_number = socket_->sample_number_; - interest_lifetime = socket_->interest_lifetime_; - // beta = socket_->beta_; + ConsumerInterestCallback *callback_interest = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_SATISFIED, + &callback_interest); + if (*callback_content_object != VOID_HANDLER) { + (*callback_interest)(*socket_, *interest); + } - double alpha = 0.0; - uint32_t batching_param = 0; - uint32_t choice_param = 0; - alpha = socket_->rate_estimation_alpha_; - batching_param = socket_->rate_estimation_batching_parameter_; - choice_param = socket_->rate_estimation_choice_; + if (TRANSPORT_EXPECT_FALSE(content_object->getPayloadType() == + PayloadType::MANIFEST)) { + if (TRANSPORT_EXPECT_FALSE(incremental_suffix == 0)) { + index_manager_ = manifest_index_manager_.get(); + interests_in_flight_--; + } - if (choice_param == 1) { - this->rate_estimator_ = new ALaTcpEstimator(); - } else { - this->rate_estimator_ = new SimpleEstimator(alpha, batching_param); + index_manager_->onManifest(std::move(content_object)); + + } else if (content_object->getPayloadType() == PayloadType::CONTENT_OBJECT) { + if (TRANSPORT_EXPECT_FALSE(incremental_suffix == 0)) { + index_manager_ = incremental_index_manager_.get(); } - this->rate_estimator_->observer_ = socket_->rate_estimation_observer_; + onContentSegment(std::move(interest), std::move(content_object)); + } - cur_path_ = std::make_shared<RaaqmDataPath>( - drop_factor, minimum_drop_probability, interest_lifetime * 1000, - sample_number); - path_table_[default_values::path_id] = cur_path_; + if (TRANSPORT_EXPECT_FALSE(incremental_suffix == 0)) { + BaseReassembly::index_ = index_manager_->getNextReassemblySegment(); } - VegasTransportProtocol::start(content_buffer); + scheduleNextInterests(); } -void RaaqmTransportProtocol::copyContent(const ContentObject &content_object) { - if (TRANSPORT_EXPECT_FALSE( - (content_object.getName().getSuffix() == final_block_number_) || - !(is_running_))) { - this->rate_estimator_->onDownloadFinished(); +void RaaqmTransportProtocol::onContentSegment( + Interest::Ptr &&interest, ContentObject::Ptr &&content_object) { + uint32_t incremental_suffix = content_object->getName().getSuffix(); + bool virtual_download = false; + socket_->getSocketOption(OtherOptions::VIRTUAL_DOWNLOAD, virtual_download); + + // Decrease in-flight interests + interests_in_flight_--; + + // Update stats + if (!interest_retransmissions_[incremental_suffix & mask]) { + afterContentReception(*interest, *content_object); + } + + if (index_manager_->onContentObject(*content_object)) { + stats_.updateBytesRecv(content_object->payloadSize()); + + if (!virtual_download) { + reassemble(std::move(content_object)); + } else if (TRANSPORT_EXPECT_FALSE(incremental_suffix == + index_manager_->getFinalSuffix())) { + onContentReassembled(std::make_error_code(std::errc(0))); + } + } else { + // TODO Application policy check + // unverified_segments_.emplace( + // std::make_pair(incremental_suffix, std::move(content_object))); + TRANSPORT_LOGE("Received not trusted segment."); } - VegasTransportProtocol::copyContent(content_object); } -void RaaqmTransportProtocol::updatePathTable( - const ContentObject &content_object) { - uint32_t path_id = content_object.getPathLabel(); +void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { + checkForStalePaths(); - if (path_table_.find(path_id) == path_table_.end()) { - if (cur_path_) { - // Create a new path with some default param - if (path_table_.empty()) { - throw errors::RuntimeException( - "No path initialized for path table, error could be in default " - "path initialization."); - } else { - // Initiate the new path default param - std::shared_ptr<RaaqmDataPath> new_path = - std::make_shared<RaaqmDataPath>( - *(path_table_.at(default_values::path_id))); - // Insert the new path into hash table - path_table_[path_id] = new_path; - } + const Name &n = interest->getName(); + + TRANSPORT_LOGW("Timeout on %s", n.toString().c_str()); + + if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + return; + } + + interests_in_flight_--; + + uint64_t segment = n.getSuffix(); + + // Do not retransmit interests asking contents that do not exist. + if (segment >= index_manager_->getFinalSuffix()) { + return; + } + + ConsumerInterestCallback *callback = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_EXPIRED, + &callback); + if (*callback != VOID_HANDLER) { + (*callback)(*socket_, *interest); + } + + afterDataUnsatisfied(segment); + + uint32_t max_rtx = 0; + socket_->getSocketOption(GeneralTransportOptions::MAX_INTEREST_RETX, max_rtx); + + if (TRANSPORT_EXPECT_TRUE(interest_retransmissions_[segment & mask] < + max_rtx)) { + stats_.updateRetxCount(1); + + callback = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_RETRANSMISSION, + &callback); + if (*callback != VOID_HANDLER) { + (*callback)(*socket_, *interest); + } + + callback = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_OUTPUT, + &callback); + if ((*callback) != VOID_HANDLER) { + (*callback)(*socket_, *interest); + } + + if (!is_running_) { + return; + } + + interest_retransmissions_[segment & mask]++; + + interest_to_retransmit_.push(std::move(interest)); + + scheduleNextInterests(); + } else { + TRANSPORT_LOGE("Stop: reached max retx limit."); + onContentReassembled(std::make_error_code(std::errc(std::errc::io_error))); + } +} + +void RaaqmTransportProtocol::scheduleNextInterests() { + if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + return; + } + + uint32_t index = IndexManager::invalid_index; + // Send the interest needed for filling the window + while (interests_in_flight_ < current_window_size_) { + if (interest_to_retransmit_.size() > 0) { + sendInterest(std::move(interest_to_retransmit_.front())); + interest_to_retransmit_.pop(); } else { - throw errors::RuntimeException( - "UNEXPECTED ERROR: when running,current path not found."); + index = index_manager_->getNextSuffix(); + if (index == IndexManager::invalid_index) { + break; + } + sendInterest(index); } } +} - cur_path_ = path_table_[path_id]; +void RaaqmTransportProtocol::sendInterest(std::uint64_t next_suffix) { + auto interest = getPacket(); + core::Name *name; + socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name); + name->setSuffix(next_suffix); + interest->setName(*name); + + uint32_t interest_lifetime; + socket_->getSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, + interest_lifetime); + interest->setLifetime(interest_lifetime); + + ConsumerInterestCallback *callback = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_OUTPUT, + &callback); + if (*callback != VOID_HANDLER) { + callback->operator()(*socket_, *interest); + } - size_t header_size = content_object.headerSize(); - size_t data_size = content_object.payloadSize(); + if (TRANSPORT_EXPECT_FALSE(!is_running_)) { + return; + } - // Update measurements for path - cur_path_->updateReceivedStats(header_size + data_size, data_size); + interest_retransmissions_[next_suffix & mask] = ~0; + interest_timepoints_[next_suffix & mask] = utils::SteadyClock::now(); + sendInterest(std::move(interest)); +} + +void RaaqmTransportProtocol::sendInterest(Interest::Ptr &&interest) { + interests_in_flight_++; + interest_retransmissions_[interest->getName().getSuffix() & mask]++; + + portal_->sendInterest(std::move(interest)); +} + +void RaaqmTransportProtocol::onContentReassembled(std::error_code ec) { + interface::ConsumerContentCallback *on_payload = nullptr; + socket_->getSocketOption(CONTENT_RETRIEVED, &on_payload); + if (*on_payload != VOID_HANDLER) { + std::shared_ptr<std::vector<uint8_t>> content_buffer; + socket_->getSocketOption( + interface::GeneralTransportOptions::APPLICATION_BUFFER, content_buffer); + (*on_payload)(*socket_, content_buffer->size(), ec); + } + + this->rate_estimator_->onDownloadFinished(); + stop(); } void RaaqmTransportProtocol::updateRtt(uint64_t segment) { if (TRANSPORT_EXPECT_FALSE(!cur_path_)) { - throw std::runtime_error("ERROR: no current path found, exit"); + throw std::runtime_error("RAAQM ERROR: no current path found, exit"); } else { - std::chrono::microseconds rtt; + auto now = utils::SteadyClock::now(); + utils::Microseconds rtt = std::chrono::duration_cast<utils::Microseconds>( + now - interest_timepoints_[segment & mask]); - std::chrono::steady_clock::duration duration = - std::chrono::steady_clock::now() - - interest_timepoints_[segment & mask_]; - rtt = std::chrono::duration_cast<std::chrono::microseconds>(duration); + // Update stats + updateStats(segment, rtt.count(), now); if (this->rate_estimator_) { this->rate_estimator_->onRttUpdate((double)rtt.count()); } + cur_path_->insertNewRtt(rtt.count()); cur_path_->smoothTimer(); if (cur_path_->newPropagationDelayAvailable()) { - check_drop_probability(); + checkDropProbability(); } } } -void RaaqmTransportProtocol::changeInterestLifetime(uint64_t segment) { - return; +void RaaqmTransportProtocol::RAAQM() { + if (!cur_path_) { + throw errors::RuntimeException("ERROR: no current path found, exit"); + exit(EXIT_FAILURE); + } else { + // Change drop probability according to RTT statistics + cur_path_->updateDropProb(); + + if (std::rand() % 10000 <= cur_path_->getDropProb() * 10000) { + decreaseWindow(); + } + } } -void RaaqmTransportProtocol::check_drop_probability() { +void RaaqmTransportProtocol::updateStats(uint32_t suffix, uint64_t rtt, + utils::TimePoint &now) { + // Update RTT statistics + stats_.updateAverageRtt(rtt); + stats_.updateAverageWindowSize(current_window_size_); + + // Call statistics callback + ConsumerTimerCallback *stats_callback = nullptr; + socket_->getSocketOption(ConsumerCallbacksOptions::STATS_SUMMARY, + &stats_callback); + if (*stats_callback != VOID_HANDLER) { + auto dt = std::chrono::duration_cast<utils::Milliseconds>(now - t0_); + + uint32_t timer_interval_milliseconds = 0; + socket_->getSocketOption(GeneralTransportOptions::STATS_INTERVAL, + timer_interval_milliseconds); + if (dt.count() > timer_interval_milliseconds) { + (*stats_callback)(*socket_, stats_); + t0_ = utils::SteadyClock::now(); + } + } +} + +void RaaqmTransportProtocol::updatePathTable( + const ContentObject &content_object) { + uint32_t path_id = content_object.getPathLabel(); + + if (path_table_.find(path_id) == path_table_.end()) { + if (TRANSPORT_EXPECT_TRUE(cur_path_ != nullptr)) { + // Create a new path with some default param + + if (TRANSPORT_EXPECT_FALSE(path_table_.empty())) { + throw errors::RuntimeException( + "[RAAQM] No path initialized for path table, error could be in " + "default path initialization."); + } + + // Initiate the new path default param + auto new_path = std::make_unique<RaaqmDataPath>( + *(path_table_.at(default_values::path_id))); + + // Insert the new path into hash table + path_table_[path_id] = std::move(new_path); + } else { + throw errors::RuntimeException( + "UNEXPECTED ERROR: when running,current path not found."); + } + } + + cur_path_ = path_table_[path_id].get(); + + size_t header_size = content_object.headerSize(); + size_t data_size = content_object.payloadSize(); + + // Update measurements for path + cur_path_->updateReceivedStats(header_size + data_size, data_size); +} + +void RaaqmTransportProtocol::checkDropProbability() { if (!raaqm_autotune_) { return; } unsigned int max_pd = 0; - std::unordered_map<uint32_t, std::shared_ptr<RaaqmDataPath>>::iterator it; + PathTable::iterator it; for (auto it = path_table_.begin(); it != path_table_.end(); ++it) { if (it->second->getPropagationDelay() > max_pd && it->second->getPropagationDelay() != UINT_MAX && @@ -317,28 +639,28 @@ void RaaqmTransportProtocol::check_drop_probability() { double old_drop_prob = 0; double old_beta = 0; - old_beta = socket_->beta_; - old_drop_prob = socket_->drop_factor_; + socket_->getSocketOption(RaaqmTransportOptions::BETA_VALUE, old_beta); + socket_->getSocketOption(RaaqmTransportOptions::DROP_FACTOR, old_drop_prob); if (drop_prob == old_drop_prob && beta == old_beta) { return; } - socket_->beta_ = beta; - socket_->drop_factor_ = drop_prob; + socket_->setSocketOption(RaaqmTransportOptions::BETA_VALUE, beta); + socket_->setSocketOption(RaaqmTransportOptions::DROP_FACTOR, drop_prob); for (it = path_table_.begin(); it != path_table_.end(); it++) { it->second->setDropProb(drop_prob); } } -void RaaqmTransportProtocol::check_for_stale_paths() { +void RaaqmTransportProtocol::checkForStalePaths() { if (!raaqm_autotune_) { return; } bool stale = false; - std::unordered_map<uint32_t, std::shared_ptr<RaaqmDataPath>>::iterator it; + PathTable::iterator it; for (it = path_table_.begin(); it != path_table_.end(); ++it) { if (it->second->isStale()) { stale = true; @@ -346,71 +668,10 @@ void RaaqmTransportProtocol::check_for_stale_paths() { } } if (stale) { - check_drop_probability(); + checkDropProbability(); } } -void RaaqmTransportProtocol::onTimeout(Interest::Ptr &&interest) { - check_for_stale_paths(); - VegasTransportProtocol::onTimeout(std::move(interest)); -} - -void RaaqmTransportProtocol::increaseWindow() { - double max_window_size = socket_->max_window_size_; - if (current_window_size_ < max_window_size) { - double gamma = socket_->gamma_; - - current_window_size_ += gamma / current_window_size_; - socket_->current_window_size_ = current_window_size_; - } - this->rate_estimator_->onWindowIncrease(current_window_size_); -} - -void RaaqmTransportProtocol::decreaseWindow() { - double min_window_size = socket_->min_window_size_; - if (current_window_size_ > min_window_size) { - double beta = socket_->beta_; - - current_window_size_ = current_window_size_ * beta; - if (current_window_size_ < min_window_size) { - current_window_size_ = min_window_size; - } - - socket_->current_window_size_ = current_window_size_; - } - this->rate_estimator_->onWindowDecrease(current_window_size_); -} - -void RaaqmTransportProtocol::RAAQM() { - if (!cur_path_) { - throw errors::RuntimeException("ERROR: no current path found, exit"); - exit(EXIT_FAILURE); - } else { - // Change drop probability according to RTT statistics - cur_path_->updateDropProb(); - - if (rand() % 10000 <= cur_path_->getDropProb() * 10000) { - decreaseWindow(); - } - } -} - -void RaaqmTransportProtocol::afterDataUnsatisfied(uint64_t segment) { - // Decrease the window because the timeout happened - decreaseWindow(); -} - -void RaaqmTransportProtocol::afterContentReception( - const Interest &interest, const ContentObject &content_object) { - updatePathTable(content_object); - increaseWindow(); - updateRtt(interest.getName().getSuffix()); - this->rate_estimator_->onDataReceived( - (int)(content_object.payloadSize() + content_object.headerSize())); - // Set drop probablility and window size accordingly - RAAQM(); -} - } // end namespace protocol } // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/raaqm.h b/libtransport/src/hicn/transport/protocols/raaqm.h index 6ca410251..09d22cd4f 100644 --- a/libtransport/src/hicn/transport/protocols/raaqm.h +++ b/libtransport/src/hicn/transport/protocols/raaqm.h @@ -15,65 +15,108 @@ #pragma once +#include <hicn/transport/protocols/congestion_window_protocol.h> +#include <hicn/transport/protocols/protocol.h> #include <hicn/transport/protocols/raaqm_data_path.h> #include <hicn/transport/protocols/rate_estimation.h> -#include <hicn/transport/protocols/vegas.h> -#include <hicn/transport/protocols/vegas_rto_estimator.h> +#include <hicn/transport/protocols/reassembly.h> +#include <hicn/transport/utils/chrono_typedefs.h> + +#include <queue> +#include <vector> namespace transport { namespace protocol { -class RaaqmTransportProtocol : public VegasTransportProtocol { +class RaaqmTransportProtocol + : public TransportProtocol, + public BaseReassembly, + public CWindowProtocol, + public BaseReassembly::ContentReassembledCallback { public: - RaaqmTransportProtocol(interface::BaseSocket *icnet_socket); + RaaqmTransportProtocol(interface::ConsumerSocket *icnet_socket); ~RaaqmTransportProtocol(); - void start(utils::SharableVector<uint8_t> &content_buffer) override; + int start() override; + + void resume() override; + + void reset() override; protected: - void copyContent(const ContentObject &content_object) override; + static constexpr uint32_t buffer_size = + 1 << interface::default_values::log_2_default_buffer_size; + static constexpr uint16_t mask = buffer_size - 1; + using PathTable = + std::unordered_map<uint32_t, std::unique_ptr<RaaqmDataPath>>; + + void increaseWindow() override; + void decreaseWindow() override; + + virtual void afterContentReception(const Interest &interest, + const ContentObject &content_object); + virtual void afterDataUnsatisfied(uint64_t segment); + + virtual void updateStats(uint32_t suffix, uint64_t rtt, + utils::TimePoint &now); private: void init(); - void afterContentReception(const Interest &interest, - const ContentObject &content_object) override; + void onContentObject(Interest::Ptr &&i, ContentObject::Ptr &&c) override; - void afterDataUnsatisfied(uint64_t segment) override; + void onContentSegment(Interest::Ptr &&interest, + ContentObject::Ptr &&content_object); - void increaseWindow() override; + void onTimeout(Interest::Ptr &&i) override; - void updateRtt(uint64_t segment); + virtual void scheduleNextInterests() override; - void decreaseWindow() override; + void sendInterest(std::uint64_t next_suffix); - void changeInterestLifetime(uint64_t segment) override; + void sendInterest(Interest::Ptr &&interest); - void onTimeout(Interest::Ptr &&interest) override; + void onContentReassembled(std::error_code ec) override; + + void updateRtt(uint64_t segment); void RAAQM(); void updatePathTable(const ContentObject &content_object); - void check_drop_probability(); + void checkDropProbability(); - void check_for_stale_paths(); + void checkForStalePaths(); void printRtt(); + protected: + // Congestion window management + double current_window_size_; + // Protocol management + uint64_t interests_in_flight_; + std::array<std::uint32_t, buffer_size> interest_retransmissions_; + std::array<utils::TimePoint, buffer_size> interest_timepoints_; + std::queue<Interest::Ptr> interest_to_retransmit_; + + private: /** * Current download path */ - std::shared_ptr<RaaqmDataPath> cur_path_; + RaaqmDataPath *cur_path_; /** * Hash table for path: each entry is a pair path ID(key) - path object */ - std::unordered_map<uint32_t, std::shared_ptr<RaaqmDataPath>> path_table_; + PathTable path_table_; + + // TimePoints for statistic + utils::TimePoint t0_; bool set_interest_filter_; + // for rate-estimation at packet level IcnRateEstimator *rate_estimator_; diff --git a/libtransport/src/hicn/transport/protocols/raaqm_data_path.cc b/libtransport/src/hicn/transport/protocols/raaqm_data_path.cc index ef26eabb5..e25646205 100644 --- a/libtransport/src/hicn/transport/protocols/raaqm_data_path.cc +++ b/libtransport/src/hicn/transport/protocols/raaqm_data_path.cc @@ -14,6 +14,7 @@ */ #include <hicn/transport/protocols/raaqm_data_path.h> +#include <hicn/transport/utils/chrono_typedefs.h> namespace transport { @@ -42,7 +43,7 @@ RaaqmDataPath::RaaqmDataPath(double drop_factor, raw_data_bytes_received_(0), last_raw_data_bytes_received_(0), rtt_samples_(samples_), - last_received_pkt_(std::chrono::steady_clock::now()), + last_received_pkt_(utils::SteadyClock::now()), average_rtt_(0), alpha_(ALPHA) {} @@ -58,7 +59,7 @@ RaaqmDataPath &RaaqmDataPath::insertNewRtt(uint64_t new_rtt) { prop_delay_ = rtt_min_; } - last_received_pkt_ = std::chrono::steady_clock::now(); + last_received_pkt_ = utils::SteadyClock::now(); return *this; } @@ -124,10 +125,6 @@ RaaqmDataPath &RaaqmDataPath::updateDropProb() { return *this; } -double RaaqmDataPath::getMicroSeconds(struct timeval &time) { - return (double)(time.tv_sec) * 1000000 + (double)(time.tv_usec); -} - void RaaqmDataPath::setAlpha(double alpha) { if (alpha >= 0 && alpha <= 1) { alpha_ = alpha; @@ -145,9 +142,10 @@ unsigned int RaaqmDataPath::getPropagationDelay() { } bool RaaqmDataPath::isStale() { - TimePoint now = std::chrono::steady_clock::now(); - auto time = std::chrono::duration_cast<Microseconds>(now - last_received_pkt_) - .count(); + utils::TimePoint now = utils::SteadyClock::now(); + auto time = + std::chrono::duration_cast<utils::Microseconds>(now - last_received_pkt_) + .count(); if (time > 2000000) { return true; } diff --git a/libtransport/src/hicn/transport/protocols/raaqm_data_path.h b/libtransport/src/hicn/transport/protocols/raaqm_data_path.h index a0b9ec9ca..9e4accfa5 100644 --- a/libtransport/src/hicn/transport/protocols/raaqm_data_path.h +++ b/libtransport/src/hicn/transport/protocols/raaqm_data_path.h @@ -15,6 +15,7 @@ #pragma once +#include <hicn/transport/utils/chrono_typedefs.h> #include <hicn/transport/utils/min_filter.h> #include <chrono> @@ -30,9 +31,6 @@ namespace transport { namespace protocol { class RaaqmDataPath { - using TimePoint = std::chrono::steady_clock::time_point; - using Microseconds = std::chrono::microseconds; - public: RaaqmDataPath(double drop_factor, double minimum_drop_probability, unsigned new_timer, unsigned int samples, @@ -125,12 +123,6 @@ class RaaqmDataPath { */ RaaqmDataPath &updateDropProb(); - /** - * @brief This function convert the time from struct timeval to its value in - * microseconds - */ - static double getMicroSeconds(struct timeval &time); - void setAlpha(double alpha); /** @@ -222,7 +214,7 @@ class RaaqmDataPath { /** * Time of the last call to the path reporter method */ - TimePoint last_received_pkt_; + utils::TimePoint last_received_pkt_; double average_rtt_; double alpha_; diff --git a/libtransport/src/hicn/transport/protocols/rate_estimation.cc b/libtransport/src/hicn/transport/protocols/rate_estimation.cc index a0bb69153..99fc78c3e 100644 --- a/libtransport/src/hicn/transport/protocols/rate_estimation.cc +++ b/libtransport/src/hicn/transport/protocols/rate_estimation.cc @@ -13,6 +13,7 @@ * limitations under the License. */ +#include <hicn/transport/interfaces/socket_options_default_values.h> #include <hicn/transport/protocols/rate_estimation.h> #include <hicn/transport/utils/log.h> @@ -33,8 +34,8 @@ void *Timer(void *data) { pthread_mutex_unlock(&(estimator->mutex_)); while (estimator->is_running_) { - std::this_thread::sleep_for( - std::chrono::microseconds((uint64_t)(KV * dat_rtt))); + std::this_thread::sleep_for(std::chrono::microseconds( + (uint64_t)(interface::default_values::kv * dat_rtt))); pthread_mutex_lock(&(estimator->mutex_)); diff --git a/libtransport/src/hicn/transport/protocols/rate_estimation.h b/libtransport/src/hicn/transport/protocols/rate_estimation.h index 91964ec1d..616501b24 100644 --- a/libtransport/src/hicn/transport/protocols/rate_estimation.h +++ b/libtransport/src/hicn/transport/protocols/rate_estimation.h @@ -20,11 +20,6 @@ #include <chrono> -#define BATCH 50 -#define KV 20 -#define ALPHA 0.8 -#define RATE_CHOICE 0 - namespace transport { namespace protocol { diff --git a/libtransport/src/hicn/transport/protocols/reassembly.cc b/libtransport/src/hicn/transport/protocols/reassembly.cc new file mode 100644 index 000000000..36cfb89a7 --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/reassembly.cc @@ -0,0 +1,83 @@ +/* + * 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 <hicn/transport/interfaces/socket_consumer.h> +#include <hicn/transport/protocols/indexing_manager.h> +#include <hicn/transport/protocols/reassembly.h> +#include <hicn/transport/utils/array.h> + +namespace transport { + +namespace protocol { + +BaseReassembly::BaseReassembly(interface::ConsumerSocket *icn_socket, + ContentReassembledCallback *content_callback) + : reassembly_consumer_socket_(icn_socket), + zero_index_manager_(std::make_unique<ZeroIndexManager>()), + incremental_index_manager_( + std::make_unique<IncrementalIndexManager>(icn_socket)), + manifest_index_manager_( + std::make_unique<ManifestIndexManager>(icn_socket)), + index_manager_(zero_index_manager_.get()), + index_(0) { + setContentCallback(content_callback); +} + +void BaseReassembly::reassemble(ContentObject::Ptr &&content_object) { + if (TRANSPORT_EXPECT_TRUE(content_object != nullptr)) { + received_packets_.emplace(std::make_pair( + content_object->getName().getSuffix(), std::move(content_object))); + } + + auto it = received_packets_.find(index_); + while (it != received_packets_.end()) { + if (it->second->getPayloadType() == PayloadType::CONTENT_OBJECT) { + copyContent(*it->second); + received_packets_.erase(it); + } + + index_ = index_manager_->getNextReassemblySegment(); + it = received_packets_.find(index_); + } +} + +void BaseReassembly::copyContent(const ContentObject &content_object) { + utils::Array<> a = content_object.getPayload(); + + std::shared_ptr<std::vector<uint8_t>> content_buffer; + reassembly_consumer_socket_->getSocketOption( + interface::GeneralTransportOptions::APPLICATION_BUFFER, content_buffer); + + content_buffer->insert(content_buffer->end(), (uint8_t *)a.data(), + (uint8_t *)a.data() + a.length()); + + bool download_completed = + index_manager_->getFinalSuffix() == content_object.getName().getSuffix(); + + if (TRANSPORT_EXPECT_FALSE(download_completed)) { + content_callback_->onContentReassembled(std::make_error_code(std::errc(0))); + } +} + +void BaseReassembly::reset() { + manifest_index_manager_->reset(); + incremental_index_manager_->reset(); + + received_packets_.clear(); +} + +} // namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/reassembly.h b/libtransport/src/hicn/transport/protocols/reassembly.h new file mode 100644 index 000000000..ef3e99fc5 --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/reassembly.h @@ -0,0 +1,71 @@ +/* + * 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. + */ + +#pragma once + +#include <hicn/transport/core/content_object.h> +#include <hicn/transport/protocols/manifest_indexing_manager.h> + +namespace transport { + +namespace protocol { + +// Forward Declaration +class ManifestManager; + +class Reassembly { + public: + class ContentReassembledCallback { + public: + virtual void onContentReassembled(std::error_code ec) = 0; + }; + + virtual void reassemble(ContentObject::Ptr &&content_object) = 0; + virtual void reset() = 0; + virtual void setContentCallback(ContentReassembledCallback *callback) { + content_callback_ = callback; + } + + protected: + ContentReassembledCallback *content_callback_; +}; + +class BaseReassembly : public Reassembly { + public: + BaseReassembly(interface::ConsumerSocket *icn_socket, + ContentReassembledCallback *content_callback); + + protected: + virtual void reassemble(ContentObject::Ptr &&content_object) override; + + virtual void copyContent(const ContentObject &content_object); + + virtual void reset() override; + + protected: + // The consumer socket + interface::ConsumerSocket *reassembly_consumer_socket_; + std::unique_ptr<ZeroIndexManager> zero_index_manager_; + std::unique_ptr<IncrementalIndexManager> incremental_index_manager_; + std::unique_ptr<ManifestIndexManager> manifest_index_manager_; + IndexVerificationManager *index_manager_; + std::unordered_map<std::uint32_t, ContentObject::Ptr> received_packets_; + + uint64_t index_; +}; + +} // namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/rtc.cc b/libtransport/src/hicn/transport/protocols/rtc.cc index 1356ad566..c2323345f 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.cc +++ b/libtransport/src/hicn/transport/protocols/rtc.cc @@ -22,8 +22,8 @@ * TODO * 2) start/constructor/rest variable implementation * 3) interest retransmission: now I always recover, we should recover only if - * we have enough time 4) returnContentToUser: rememeber to remove the first - * 32bits from the payload + * we have enough time 4) returnContentToApplication: rememeber to remove the + * first 32bits from the payload */ namespace transport { @@ -32,7 +32,8 @@ namespace protocol { using namespace interface; -RTCTransportProtocol::RTCTransportProtocol(BaseSocket *icnet_socket) +RTCTransportProtocol::RTCTransportProtocol( + interface::ConsumerSocket *icnet_socket) : TransportProtocol(icnet_socket), inflightInterests_(1 << default_values::log_2_default_buffer_size), modMask_((1 << default_values::log_2_default_buffer_size) - 1) { @@ -46,19 +47,7 @@ RTCTransportProtocol::~RTCTransportProtocol() { } } -void RTCTransportProtocol::start( - utils::SharableVector<uint8_t> &content_buffer) { - if (is_running_) return; - - is_running_ = true; - content_buffer_ = content_buffer.shared_from_this(); - - reset(); - scheduleNextInterest(); - - portal_->runEventsLoop(); - is_running_ = false; -} +int RTCTransportProtocol::start() { return TransportProtocol::start(); } void RTCTransportProtocol::stop() { if (!is_running_) return; @@ -74,9 +63,8 @@ void RTCTransportProtocol::resume() { lastRoundBegin_ = std::chrono::steady_clock::now(); inflightInterestsCount_ = 0; - if (content_buffer_) content_buffer_->clear(); - scheduleNextInterest(); + scheduleNextInterests(); portal_->runEventsLoop(); @@ -117,7 +105,6 @@ void RTCTransportProtocol::reset() { while (interestRetransmissions_.size() != 0) interestRetransmissions_.pop(); nackedByProducer_.clear(); nackedByProducerMaxSize_ = 512; - if (content_buffer_) content_buffer_->clear(); // stats receivedBytes_ = 0; @@ -364,9 +351,9 @@ void RTCTransportProtocol::increaseWindow() { } void RTCTransportProtocol::sendInterest() { - Name interest_name; + Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - interest_name); + &interest_name); bool isRTX = false; // uint32_t sentInt = 0; @@ -382,11 +369,11 @@ void RTCTransportProtocol::sendInterest() { packetLost_++; uint32_t pkt = rtxSeg & modMask_; - interest_name.setSuffix(rtxSeg); + interest_name->setSuffix(rtxSeg); // if the interest is not pending anymore we encrease the retrasnmission // counter in order to avoid to handle a recovered packt as a normal one - if (!portal_->interestIsPending(interest_name)) { + if (!portal_->interestIsPending(*interest_name)) { inflightInterests_[pkt].retransmissions++; } @@ -394,8 +381,8 @@ void RTCTransportProtocol::sendInterest() { isRTX = true; } else { // in this case we send the packet only if it is not pending yet - interest_name.setSuffix(actualSegment_); - if (portal_->interestIsPending(interest_name)) { + interest_name->setSuffix(actualSegment_); + if (portal_->interestIsPending(*interest_name)) { actualSegment_++; return; } @@ -410,21 +397,21 @@ void RTCTransportProtocol::sendInterest() { actualSegment_++; } - auto interest = getInterest(); - interest->setName(interest_name); + auto interest = getPacket(); + interest->setName(*interest_name); uint32_t interestLifetime = default_values::interest_lifetime; socket_->getSocketOption(GeneralTransportOptions::INTEREST_LIFETIME, interestLifetime); interest->setLifetime(uint32_t(interestLifetime)); - ConsumerInterestCallback on_interest_output = VOID_HANDLER; + ConsumerInterestCallback *on_interest_output = nullptr; socket_->getSocketOption(ConsumerCallbacksOptions::INTEREST_OUTPUT, - on_interest_output); + &on_interest_output); - if (on_interest_output != VOID_HANDLER) { - on_interest_output(*dynamic_cast<ConsumerSocket *>(socket_), *interest); + if (*on_interest_output != VOID_HANDLER) { + (*on_interest_output)(*socket_, *interest); } if (TRANSPORT_EXPECT_FALSE(!is_running_)) { @@ -441,7 +428,7 @@ void RTCTransportProtocol::sendInterest() { } } -void RTCTransportProtocol::scheduleNextInterest() { +void RTCTransportProtocol::scheduleNextInterests() { checkRound(); if (!is_running_) return; @@ -467,7 +454,7 @@ void RTCTransportProtocol::scheduleAppNackRtx(std::vector<uint32_t> &nacks) { interestRetransmissions_.push(nacks[i]); } - scheduleNextInterest(); + scheduleNextInterests(); } void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { // packetLost_++; @@ -483,7 +470,7 @@ void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { interestRetransmissions_.push(segmentNumber); } - scheduleNextInterest(); + scheduleNextInterests(); } void RTCTransportProtocol::onNack(const ContentObject &content_object) { @@ -573,14 +560,14 @@ void RTCTransportProtocol::onContentObject( updateDelayStats(*content_object); } - returnContentToUser(*content_object); + reassemble(std::move(content_object)); increaseWindow(); } - scheduleNextInterest(); + scheduleNextInterests(); } -void RTCTransportProtocol::returnContentToUser( +void RTCTransportProtocol::returnContentToApplication( const ContentObject &content_object) { // return content to the user Array a = content_object.getPayload(); @@ -592,13 +579,14 @@ void RTCTransportProtocol::returnContentToUser( uint16_t rtp_seq = ntohs(*(((uint16_t *)start) + 1)); RTPhICN_offset_ = content_object.getName().getSuffix() - rtp_seq; - content_buffer_->insert(content_buffer_->end(), start, start + size); + std::shared_ptr<std::vector<uint8_t>> content_buffer; + socket_->getSocketOption(APPLICATION_BUFFER, content_buffer); + content_buffer->insert(content_buffer_->end(), start, start + size); - ConsumerContentCallback on_payload = VOID_HANDLER; - socket_->getSocketOption(CONTENT_RETRIEVED, on_payload); - if (on_payload != VOID_HANDLER) { - on_payload(*dynamic_cast<ConsumerSocket *>(socket_), size, - std::make_error_code(std::errc(0))); + ConsumerContentCallback *on_payload = nullptr; + socket_->getSocketOption(CONTENT_RETRIEVED, &on_payload); + if ((*on_payload) != VOID_HANDLER) { + (*on_payload)(*socket_, size, std::make_error_code(std::errc(0))); } } diff --git a/libtransport/src/hicn/transport/protocols/rtc.h b/libtransport/src/hicn/transport/protocols/rtc.h index 61590fc8e..58c143988 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.h +++ b/libtransport/src/hicn/transport/protocols/rtc.h @@ -20,6 +20,7 @@ #include <unordered_map> #include <hicn/transport/protocols/protocol.h> +#include <hicn/transport/protocols/reassembly.h> #include <hicn/transport/protocols/rtc_data_path.h> // algorithm state @@ -86,23 +87,23 @@ struct sentInterest { uint8_t retransmissions; }; -class RTCTransportProtocol : public TransportProtocol { +class RTCTransportProtocol : public TransportProtocol, public Reassembly { public: - RTCTransportProtocol(interface::BaseSocket *icnet_socket); + RTCTransportProtocol(interface::ConsumerSocket *icnet_socket); ~RTCTransportProtocol(); - void start(utils::SharableVector<uint8_t> &content_buffer); + int start() override; - void stop(); + void stop() override; - void resume(); + void resume() override; void onRTCPPacket(uint8_t *packet, size_t len); private: // algo functions - void reset(); + void reset() override; void checkRound(); // CC functions @@ -117,13 +118,18 @@ class RTCTransportProtocol : public TransportProtocol { // packet functions void sendInterest(); - void scheduleNextInterest(); + void scheduleNextInterests() override; void scheduleAppNackRtx(std::vector<uint32_t> &nacks); - void onTimeout(Interest::Ptr &&interest); + void onTimeout(Interest::Ptr &&interest) override; void onNack(const ContentObject &content_object); void onContentObject(Interest::Ptr &&interest, - ContentObject::Ptr &&content_object); - void returnContentToUser(const ContentObject &content_object); + ContentObject::Ptr &&content_object) override; + void returnContentToApplication(const ContentObject &content_object); + + TRANSPORT_ALWAYS_INLINE virtual void reassemble( + ContentObject::Ptr &&content_object) override { + returnContentToApplication(*content_object); + } // RTCP functions uint32_t hICN2RTP(uint32_t hicn_seq); @@ -161,7 +167,7 @@ class RTCTransportProtocol : public TransportProtocol { // application for pakets for which we already got a // past NACK by the producer these packet are too old, // they will never be retrived - std::shared_ptr<utils::SharableVector<uint8_t>> content_buffer_; + uint32_t modMask_; // stats diff --git a/libtransport/src/hicn/transport/protocols/statistics.h b/libtransport/src/hicn/transport/protocols/statistics.h new file mode 100644 index 000000000..47d164158 --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/statistics.h @@ -0,0 +1,94 @@ +/* + * 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. + */ + +#pragma once + +#include <hicn/transport/portability/c_portability.h> + +#include <cstdint> + +namespace transport { + +namespace protocol { + +class TransportStatistics { + static constexpr double default_alpha = 0.7; + + public: + TransportStatistics(double alpha = default_alpha) + : retx_count_(0), + bytes_received_(0), + average_rtt_(0), + avg_window_size_(0), + interest_tx_(0), + alpha_(alpha) {} + + TRANSPORT_ALWAYS_INLINE void updateRetxCount(uint64_t retx) { + retx_count_ += retx; + } + + TRANSPORT_ALWAYS_INLINE void updateBytesRecv(uint64_t bytes) { + bytes_received_ += bytes; + } + + TRANSPORT_ALWAYS_INLINE void updateAverageRtt(uint64_t rtt) { + average_rtt_ = (alpha_ * average_rtt_) + ((1. - alpha_) * double(rtt)); + } + + TRANSPORT_ALWAYS_INLINE void updateAverageWindowSize(double current_window) { + avg_window_size_ = + (alpha_ * avg_window_size_) + ((1. - alpha_) * current_window); + } + + TRANSPORT_ALWAYS_INLINE void updateInterestTx(uint64_t int_tx) { + interest_tx_ += int_tx; + } + + TRANSPORT_ALWAYS_INLINE uint64_t getRetxCount() const { return retx_count_; } + + TRANSPORT_ALWAYS_INLINE uint64_t getBytesRecv() const { + return bytes_received_; + } + + TRANSPORT_ALWAYS_INLINE double getAverageRtt() const { return average_rtt_; } + + TRANSPORT_ALWAYS_INLINE double getAverageWindowSize() const { + return avg_window_size_; + } + + TRANSPORT_ALWAYS_INLINE uint64_t getInterestTx() const { + return interest_tx_; + } + + TRANSPORT_ALWAYS_INLINE void reset() { + retx_count_ = 0; + bytes_received_ = 0; + average_rtt_ = 0; + avg_window_size_ = 0; + interest_tx_ = 0; + } + + private: + uint64_t retx_count_; + uint64_t bytes_received_; + double average_rtt_; + double avg_window_size_; + uint64_t interest_tx_; + double alpha_; +}; + +} // end namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/vegas.cc b/libtransport/src/hicn/transport/protocols/vegas.cc deleted file mode 100644 index f99eb83e0..000000000 --- a/libtransport/src/hicn/transport/protocols/vegas.cc +++ /dev/null @@ -1,627 +0,0 @@ -/* - * 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 <hicn/transport/errors/not_implemented_exception.h> -#include <hicn/transport/interfaces/socket_consumer.h> -#include <hicn/transport/protocols/vegas.h> -#include <hicn/transport/utils/literals.h> - -#include <cmath> - -namespace transport { - -namespace protocol { - -using namespace interface; - -VegasTransportProtocol::VegasTransportProtocol(BaseSocket *icnet_socket) - : TransportProtocol(icnet_socket), - is_final_block_number_discovered_(false), - final_block_number_(std::numeric_limits<uint32_t>::max()), - last_reassembled_segment_(0), - content_buffer_size_(0), - current_window_size_(default_values::min_window_size), - interests_in_flight_(0), - next_suffix_(0), - interest_retransmissions_(1 << default_values::log_2_default_buffer_size), - interest_timepoints_(1 << default_values::log_2_default_buffer_size), - retx_count_(0), - receive_buffer_(1 << default_values::log_2_default_buffer_size), - unverified_segments_(1 << default_values::log_2_default_buffer_size), - verified_manifests_(1 << default_values::log_2_default_buffer_size), - mask_((1 << default_values::log_2_default_buffer_size) - 1), - incremental_suffix_index_(0), - suffix_queue_completed_(false), - download_with_manifest_(false), - next_manifest_interval_(0_U16), - interest_tx_(0), - interest_count_(0), - byte_count_(0), - average_rtt_(0.0) { - portal_ = socket_->portal_; - incremental_suffix_index_++; -} - -VegasTransportProtocol::~VegasTransportProtocol() { stop(); } - -void VegasTransportProtocol::reset() { - portal_->setConsumerCallback(this); - - is_final_block_number_discovered_ = false; - interest_pool_index_ = 0; - final_block_number_ = std::numeric_limits<uint32_t>::max(); - next_suffix_ = 0; - interests_in_flight_ = 0; - last_reassembled_segment_ = 0; - content_buffer_size_ = 0; - content_buffer_->clear(); - interest_retransmissions_.clear(); - interest_retransmissions_.resize( - 1 << default_values::log_2_default_buffer_size, 0); - interest_timepoints_.clear(); - interest_timepoints_.resize(1 << default_values::log_2_default_buffer_size, - std::chrono::steady_clock::time_point()); - receive_buffer_.clear(); - unverified_segments_.clear(); - verified_manifests_.clear(); - next_manifest_interval_ = 0; - next_manifest_ = 0; - download_with_manifest_ = false; - incremental_suffix_index_ = 0; - - interest_tx_ = 0; - interest_count_ = 0; - byte_count_ = 0; - average_rtt_ = 0; - - // asio::io_service &io_service = portal_->getIoService(); - - // if (io_service.stopped()) { - // io_service.reset(); - // } -} - -void VegasTransportProtocol::start( - utils::SharableVector<uint8_t> &content_buffer) { - if (is_running_) return; - - socket_->t0_ = std::chrono::steady_clock::now(); - - is_running_ = true; - content_buffer_ = content_buffer.shared_from_this(); - - reset(); - - sendInterest(next_suffix_++); - portal_->runEventsLoop(); - removeAllPendingInterests(); - is_running_ = false; -} - -void VegasTransportProtocol::resume() { - if (is_running_) return; - - is_running_ = true; - sendInterest(next_suffix_++); - portal_->runEventsLoop(); - removeAllPendingInterests(); - is_running_ = false; -} - -void VegasTransportProtocol::sendInterest(std::uint64_t next_suffix) { - auto interest = getInterest(); - socket_->network_name_.setSuffix((uint32_t)next_suffix); - interest->setName(socket_->network_name_); - - interest->setLifetime(uint32_t(socket_->interest_lifetime_)); - - if (socket_->on_interest_output_ != VOID_HANDLER) { - socket_->on_interest_output_(*socket_, *interest); - } - - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { - return; - } - - interests_in_flight_++; - interest_retransmissions_[next_suffix & mask_] = 0; - interest_timepoints_[next_suffix & mask_] = std::chrono::steady_clock::now(); - - using namespace std::placeholders; - portal_->sendInterest(std::move(interest)); -} - -void VegasTransportProtocol::stop() { - is_running_ = false; - portal_->stopEventsLoop(); -} - -void VegasTransportProtocol::onContentSegment( - Interest::Ptr &&interest, ContentObject::Ptr &&content_object) { - uint32_t incremental_suffix = content_object->getName().getSuffix(); - bool virtual_download = socket_->virtual_download_; - - if (verifyContentObject(*content_object)) { - byte_count_ += content_object->getPayload().length(); - - if (TRANSPORT_EXPECT_FALSE(content_object->testRst())) { - is_final_block_number_discovered_ = true; - final_block_number_ = incremental_suffix; - } - - if (!virtual_download) { - receive_buffer_.emplace( - std::make_pair(incremental_suffix, std::move(content_object))); - reassemble(); - } else if (TRANSPORT_EXPECT_FALSE(is_final_block_number_discovered_ && - incremental_suffix == - final_block_number_)) { - returnContentToUser(); - } - } else { - unverified_segments_.emplace( - std::make_pair(incremental_suffix, std::move(content_object))); - } -} - -void VegasTransportProtocol::afterContentReception( - const Interest &interest, const ContentObject &content_object) { - increaseWindow(); -} - -void VegasTransportProtocol::afterDataUnsatisfied(uint64_t segment) { - decreaseWindow(); -} - -void VegasTransportProtocol::scheduleNextInterests() { - if (is_running_) { - uint32_t next_suffix; - while (interests_in_flight_ < current_window_size_) { - if (download_with_manifest_) { - if (suffix_queue_.size() * 2 < current_window_size_ && - next_manifest_ < final_block_number_ && next_manifest_interval_) { - next_manifest_ += next_manifest_interval_; - sendInterest(next_manifest_); - continue; - } - - if (suffix_queue_.pop(next_suffix)) { - // next_suffix = suffix_queue_.front(); - sendInterest(next_suffix); - // suffix_queue_.pop_front(); - } else { - if (!suffix_queue_completed_) { - TRANSPORT_LOGE("Empty queue!!!!!!"); - } - break; - } - } else { - if (is_final_block_number_discovered_) { - if (next_suffix_ > final_block_number_) { - return; - } - } - - sendInterest(next_suffix_++); - } - } - } -} - -void VegasTransportProtocol::decreaseWindow() { - if (current_window_size_ > socket_->min_window_size_) { - current_window_size_ = std::ceil(current_window_size_ / 2); - socket_->current_window_size_ = current_window_size_; - } -} - -void VegasTransportProtocol::increaseWindow() { - if (current_window_size_ < socket_->max_window_size_) { - current_window_size_++; - socket_->max_window_size_ = current_window_size_; - } -}; - -void VegasTransportProtocol::changeInterestLifetime(uint64_t segment) { - std::chrono::steady_clock::duration duration = - std::chrono::steady_clock::now() - interest_timepoints_[segment]; - rtt_estimator_.addMeasurement( - std::chrono::duration_cast<std::chrono::microseconds>(duration)); - - RtoEstimator::Duration rto = rtt_estimator_.computeRto(); - std::chrono::milliseconds lifetime = - std::chrono::duration_cast<std::chrono::milliseconds>(rto); - - socket_->interest_lifetime_ = (int)lifetime.count(); -} - -void VegasTransportProtocol::returnContentToUser() { - if (socket_->on_payload_retrieved_ != VOID_HANDLER) { - socket_->on_payload_retrieved_(*socket_, byte_count_, - std::make_error_code(std::errc(0))); - } - - stop(); -} - -void VegasTransportProtocol::onManifest( - std::unique_ptr<ContentObjectManifest> &&manifest) { - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { - return; - } - - download_with_manifest_ = true; - - uint32_t segment = manifest->getName().getSuffix(); - - if (verifyManifest(*manifest)) { - manifest->decode(); - - if (TRANSPORT_EXPECT_TRUE(manifest->getVersion() == - core::ManifestVersion::VERSION_1)) { - switch (manifest->getManifestType()) { - case core::ManifestType::INLINE_MANIFEST: { - auto _it = manifest->getSuffixList().begin(); - auto _end = --manifest->getSuffixList().end(); - - if (TRANSPORT_EXPECT_FALSE(manifest->isFinalManifest())) { - _end++; - } - - // Get final block number - is_final_block_number_discovered_ = true; - final_block_number_ = manifest->getFinalBlockNumber(); - - for (; _it != _end; _it++) { - suffix_hash_map_[_it->first] = std::make_pair( - std::vector<uint8_t>(_it->second, _it->second + 32), - manifest->getHashAlgorithm()); - suffix_queue_.push(_it->first); - } - - next_manifest_interval_ = - (unsigned short)manifest->getSuffixList().size(); - - if (manifest->isFinalManifest()) { - suffix_queue_completed_ = true; - // Give it a try - if (verifier_thread_) { - asio::io_service &io_service = portal_->getIoService(); - io_service.post([this]() { scheduleNextInterests(); }); - } - } - - break; - } - case core::ManifestType::FLIC_MANIFEST: { - throw errors::NotImplementedException(); - } - case core::ManifestType::FINAL_CHUNK_NUMBER: { - throw errors::NotImplementedException(); - } - } - } - - if (!socket_->virtual_download_) { - receive_buffer_.emplace( - std::make_pair(segment, std::move(manifest->getPacket()))); - reassemble(); - } else { - if (segment >= final_block_number_) { - stop(); - } - } - } -} - -bool VegasTransportProtocol::verifyManifest( - const ContentObjectManifest &manifest) { - if (!socket_->verify_signature_) { - return true; - } - - bool is_data_secure = false; - - if (socket_->on_content_object_verification_ == VOID_HANDLER) { - is_data_secure = static_cast<bool>(socket_->verifier_.verify(manifest)); - } else if (socket_->on_content_object_verification_(*socket_, manifest)) { - is_data_secure = true; - } - - if (TRANSPORT_EXPECT_FALSE(!is_data_secure)) { - TRANSPORT_LOGE("Verification failed for %s\n", - manifest.getName().toString().c_str()); - } - - return is_data_secure; -} - -// TODO Add the name in the digest computation! -void VegasTransportProtocol::onContentObject( - Interest::Ptr &&interest, ContentObject::Ptr &&content_object) { - uint32_t incremental_suffix = content_object->getName().getSuffix(); - - std::chrono::microseconds rtt; - Time now = std::chrono::steady_clock::now(); - std::chrono::steady_clock::duration duration = - now - interest_timepoints_[incremental_suffix & mask_]; - rtt = std::chrono::duration_cast<std::chrono::microseconds>(duration); - - average_rtt_ = (0.7 * average_rtt_) + (0.3 * (double)rtt.count()); - - if (socket_->on_timer_expires_ != VOID_HANDLER) { - auto dt = std::chrono::duration_cast<TimeDuration>(now - socket_->t0_); - if (dt.count() > socket_->timer_interval_milliseconds_) { - socket_->on_timer_expires_(*socket_, byte_count_, dt, - (float)current_window_size_, retx_count_, - (uint32_t)std::round(average_rtt_)); - socket_->t0_ = std::chrono::steady_clock::now(); - } - } - - interests_in_flight_--; - - if (TRANSPORT_EXPECT_FALSE(!is_running_ || incremental_suffix == ~0_U64 || - receive_buffer_.find(incremental_suffix) != - receive_buffer_.end())) { - return; - } - - changeInterestLifetime(incremental_suffix); - - if (socket_->on_content_object_input_ != VOID_HANDLER) { - socket_->on_content_object_input_(*socket_, *content_object); - } - - if (socket_->on_interest_satisfied_ != VOID_HANDLER) { - socket_->on_interest_satisfied_(*socket_, *interest); - } - - if (!interest_retransmissions_[incremental_suffix & mask_]) { - afterContentReception(*interest, *content_object); - } - - if (TRANSPORT_EXPECT_FALSE(content_object->getPayloadType() == - PayloadType::MANIFEST)) { - // TODO Fix manifest!! - auto manifest = - std::make_unique<ContentObjectManifest>(std::move(content_object)); - - if (verifier_thread_ && incremental_suffix != 0) { - // verifier_thread_->add(std::bind(&VegasTransportProtocol::onManifest, - // this, std::move(manifest))); - } else { - onManifest(std::move(manifest)); - } - } else if (content_object->getPayloadType() == PayloadType::CONTENT_OBJECT) { - if (verifier_thread_) { - // verifier_thread_->add(std::bind(&VegasTransportProtocol::onContentSegment, - // this, std::move(content_object))); - } else { - onContentSegment(std::move(interest), std::move(content_object)); - } - } - - scheduleNextInterests(); -} - -bool VegasTransportProtocol::verifyContentObject( - const ContentObject &content_object) { - if (!dynamic_cast<ConsumerSocket *>(socket_)->verify_signature_) { - return true; - } - - uint64_t segment = content_object.getName().getSuffix(); - - bool ret = false; - - if (download_with_manifest_) { - auto it = suffix_hash_map_.find((const unsigned int)segment); - if (it != suffix_hash_map_.end()) { - auto hash_type = static_cast<utils::CryptoHashType>(it->second.second); - auto data_packet_digest = content_object.computeDigest(it->second.second); - auto data_packet_digest_bytes = - data_packet_digest.getDigest<uint8_t>().data(); - std::vector<uint8_t> &manifest_digest_bytes = it->second.first; - - if (utils::CryptoHash::compareBinaryDigest(data_packet_digest_bytes, - manifest_digest_bytes.data(), - hash_type)) { - suffix_hash_map_.erase(it); - ret = true; - } else { - throw errors::RuntimeException( - "Verification failure policy has to be implemented."); - } - } - } else { - ret = static_cast<bool>( - dynamic_cast<ConsumerSocket *>(socket_)->verifier_.verify( - content_object)); - - if (!ret) { - throw errors::RuntimeException( - "Verification failure policy has to be implemented."); - } - } - - return ret; - ; -} - -void VegasTransportProtocol::onTimeout(Interest::Ptr &&interest) { - TRANSPORT_LOGW("Timeout on %s", interest->getName().toString().c_str()); - - if (TRANSPORT_EXPECT_FALSE(!is_running_)) { - return; - } - - interests_in_flight_--; - - uint64_t segment = interest->getName().getSuffix(); - - // Do not retransmit interests asking contents that do not exist. - if (is_final_block_number_discovered_) { - if (segment > final_block_number_) { - return; - } - } - - if (socket_->on_interest_timeout_ != VOID_HANDLER) { - socket_->on_interest_timeout_(*socket_, *interest); - } - - afterDataUnsatisfied(segment); - - if (TRANSPORT_EXPECT_TRUE(interest_retransmissions_[segment & mask_] < - socket_->max_retransmissions_)) { - retx_count_++; - - if (socket_->on_interest_retransmission_ != VOID_HANDLER) { - socket_->on_interest_retransmission_(*socket_, *interest); - } - - if (socket_->on_interest_output_ != VOID_HANDLER) { - socket_->on_interest_output_(*socket_, *interest); - } - - if (!is_running_) { - return; - } - - // retransmit - interests_in_flight_++; - interest_retransmissions_[segment & mask_]++; - - using namespace std::placeholders; - portal_->sendInterest(std::move(interest)); - } else { - TRANSPORT_LOGE("Stop: reached max retx limit."); - partialDownload(); - stop(); - } -} - -void VegasTransportProtocol::copyContent(const ContentObject &content_object) { - Array a = content_object.getPayload(); - - content_buffer_->insert(content_buffer_->end(), (uint8_t *)a.data(), - (uint8_t *)a.data() + a.length()); - - bool download_completed = - is_final_block_number_discovered_ && - content_object.getName().getSuffix() == final_block_number_; - - if (TRANSPORT_EXPECT_FALSE(download_completed || !is_running_)) { - // asio::io_service& io_service = portal_->getIoService(); - // io_service.post([this] () { - returnContentToUser(); - // }); - } -} - -void VegasTransportProtocol::reassemble() { - uint64_t index = last_reassembled_segment_; - auto it = receive_buffer_.find((const unsigned int)index); - - while (it != receive_buffer_.end()) { - if (it->second->getPayloadType() == PayloadType::CONTENT_OBJECT) { - copyContent(*it->second); - receive_buffer_.erase(it); - } - - index = ++last_reassembled_segment_; - it = receive_buffer_.find((const unsigned int)index); - } -} - -void VegasTransportProtocol::partialDownload() { - if (!socket_->virtual_download_) { - reassemble(); - } - - if (socket_->on_payload_retrieved_ != VOID_HANDLER) { - socket_->on_payload_retrieved_( - *socket_, byte_count_, - std::make_error_code(std::errc(std::errc::io_error))); - } -} - -// TODO Check vegas protocol -// void VegasTransportProtocol::checkForFastRetransmission(const Interest -// &interest) { -// uint64_t segNumber = interest.getName().getSuffix(); -// received_segments_[segNumber] = true; -// fast_retransmitted_segments.erase(segNumber); - -// uint64_t possibly_lost_segment = 0; -// uint64_t highest_received_segment = received_segments_.rbegin()->first; - -// for (uint64_t i = 0; i <= highest_received_segment; i++) { -// if (received_segments_.find(i) == received_segments_.end()) { -// if (fast_retransmitted_segments.find(i) == -// fast_retransmitted_segments.end()) { -// possibly_lost_segment = i; -// uint8_t out_of_order_segments = 0; -// for (uint64_t j = i; j <= highest_received_segment; j++) { -// if (received_segments_.find(j) != received_segments_.end()) { -// out_of_order_segments++; -// if (out_of_order_segments >= -// default_values::max_out_of_order_segments) { -// fast_retransmitted_segments[possibly_lost_segment] = true; -// fastRetransmit(interest, possibly_lost_segment); -// } -// } -// } -// } -// } -// } -// } - -// void VegasTransportProtocol::fastRetransmit(const Interest &interest, -// uint32_t chunk_number) { -// if (interest_retransmissions_[chunk_number & mask_] < -// socket_->max_retransmissions_) { -// Name name = interest.getName(); -// name.setSuffix(chunk_number); - -// std::shared_ptr<Interest> retx_interest = -// std::make_shared<Interest>(name); - -// if (socket_->on_interest_retransmission_ != VOID_HANDLER) { -// socket_->on_interest_retransmission_(*socket_, *retx_interest); -// } - -// if (socket_->on_interest_output_ != VOID_HANDLER) { -// socket_->on_interest_output_(*socket_, *retx_interest); -// } - -// if (!is_running_) { -// return; -// } - -// interests_in_flight_++; -// interest_retransmissions_[chunk_number & mask_]++; - -// using namespace std::placeholders; -// portal_->sendInterest(std::move(retx_interest)); -// } -// } - -void VegasTransportProtocol::removeAllPendingInterests() { portal_->clear(); } - -} // end namespace protocol - -} // namespace transport diff --git a/libtransport/src/hicn/transport/protocols/vegas.h b/libtransport/src/hicn/transport/protocols/vegas.h deleted file mode 100644 index 7791ffc94..000000000 --- a/libtransport/src/hicn/transport/protocols/vegas.h +++ /dev/null @@ -1,161 +0,0 @@ -/* - * 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. - */ - -#pragma once - -#include <hicn/transport/protocols/protocol.h> -#include <hicn/transport/protocols/vegas_rto_estimator.h> -#include <hicn/transport/utils/event_thread.h> -#include <hicn/transport/utils/ring_buffer.h> -#include <hicn/transport/utils/sharable_vector.h> - -#include <map> - -namespace transport { - -namespace protocol { - -typedef utils::CircularFifo<uint32_t, 1024 * 128> SuffixQueue; -typedef std::chrono::time_point<std::chrono::steady_clock> Time; -typedef std::chrono::milliseconds TimeDuration; - -class VegasTransportProtocol : public TransportProtocol { - public: - VegasTransportProtocol(interface::BaseSocket *icnet_socket); - - virtual ~VegasTransportProtocol(); - - virtual void start(utils::SharableVector<uint8_t> &content_buffer) override; - - void stop() override; - - void resume() override; - - protected: - void reset(); - - void sendInterest(std::uint64_t next_suffix); - - void onContentSegment(Interest::Ptr &&interest, - ContentObject::Ptr &&content_object); - - bool verifyContentObject(const ContentObject &content_object); - - bool verifyManifest(const interface::ContentObjectManifest &manifest); - - virtual void onTimeout(Interest::Ptr &&interest) override; - - void onManifest(std::unique_ptr<interface::ContentObjectManifest> &&manifest); - - void onContentObject(Interest::Ptr &&interest, - ContentObject::Ptr &&content_object) override; - - virtual void changeInterestLifetime(uint64_t segment); - - void scheduleNextInterests(); - - virtual void decreaseWindow(); - - virtual void increaseWindow(); - - virtual void afterContentReception(const Interest &interest, - const ContentObject &content_object); - - virtual void afterDataUnsatisfied(uint64_t segment); - - void reassemble(); - - void returnContentToUser(); - - void partialDownload(); - - virtual void copyContent(const ContentObject &content_object); - - // virtual void checkForFastRetransmission(const Interest &interest); - - // void fastRetransmit(const Interest &interest, uint32_t chunk_number); - - void removeAllPendingInterests(); - - protected: - void handleTimeout(const std::error_code &ec); - - // reassembly variables - volatile bool is_final_block_number_discovered_; - std::atomic<uint64_t> final_block_number_; - uint64_t last_reassembled_segment_; - std::shared_ptr<utils::SharableVector<uint8_t>> content_buffer_; - size_t content_buffer_size_; - - // transmission variablesis_final_block_number_discovered_ - double current_window_size_; - double pending_window_size_; - uint64_t interests_in_flight_; - uint64_t next_suffix_; - std::vector<std::uint32_t> interest_retransmissions_; - std::vector<std::chrono::steady_clock::time_point> interest_timepoints_; - RtoEstimator rtt_estimator_; - - uint32_t retx_count_; - - // buffers - std::unordered_map<std::uint32_t, ContentObject::Ptr> - receive_buffer_; // verified segments by segment number - std::unordered_map<std::uint32_t, ContentObject::Ptr> - unverified_segments_; // used with embedded manifests - std::unordered_map<std::uint32_t, ContentObject::Ptr> - verified_manifests_; // by segment number - - std::uint16_t interest_pool_index_; - std::uint16_t mask_; - - // suffix randomization: since the suffixes in the manifests could not be in a - // sequential order, we need to map those suffixes into an ordered sequence. - std::unordered_map<std::uint64_t, std::uint64_t> - incremental_suffix_to_real_suffix_map_; - std::unordered_map<std::uint64_t, std::uint64_t> - real_suffix_to_incremental_suffix_map_; - std::uint32_t incremental_suffix_index_; - - // verification - std::unordered_map<uint32_t, std::pair<std::vector<uint8_t>, HashAlgorithm>> - suffix_hash_map_; - - // Fast Retransmission - std::map<uint64_t, bool> received_segments_; - std::unordered_map<uint64_t, bool> fast_retransmitted_segments; - - // Suffix queue - volatile bool suffix_queue_completed_; - SuffixQueue suffix_queue_; - - volatile bool download_with_manifest_; - uint32_t next_manifest_; - std::atomic<uint16_t> next_manifest_interval_; - - std::unique_ptr<utils::EventThread> verifier_thread_; - - uint32_t interest_tx_; - uint32_t interest_count_; - - uint64_t byte_count_; - double average_rtt_; - - std::unordered_map<uint32_t, uint64_t> sign_time_; -}; - -} // namespace protocol - -} // end namespace transport diff --git a/libtransport/src/hicn/transport/protocols/vegas_rto_estimator.cc b/libtransport/src/hicn/transport/protocols/vegas_rto_estimator.cc deleted file mode 100644 index a61fd05f0..000000000 --- a/libtransport/src/hicn/transport/protocols/vegas_rto_estimator.cc +++ /dev/null @@ -1,57 +0,0 @@ -/* - * 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 <hicn/transport/interfaces/socket_options_default_values.h> -#include <hicn/transport/protocols/vegas_rto_estimator.h> - -#include <algorithm> -#include <cmath> - -namespace transport { - -namespace protocol { - -using namespace interface; - -RtoEstimator::RtoEstimator(Duration min_rto) - : smoothed_rtt_((double)RtoEstimator::getInitialRtt().count()), - rtt_variation_(0), - first_measurement_(true), - last_rto_((double)min_rto.count()) {} - -void RtoEstimator::addMeasurement(Duration rtt) { - double duration = static_cast<double>(rtt.count()); - if (first_measurement_) { - smoothed_rtt_ = duration; - rtt_variation_ = duration / 2; - first_measurement_ = false; - } else { - rtt_variation_ = (1 - default_values::beta) * rtt_variation_ + - default_values::beta * std::abs(smoothed_rtt_ - duration); - smoothed_rtt_ = (1 - default_values::alpha) * smoothed_rtt_ + - default_values::alpha * duration; - } -} - -RtoEstimator::Duration RtoEstimator::computeRto() const { - double rto = smoothed_rtt_ + - std::max(double(default_values::clock_granularity.count()), - default_values::k* rtt_variation_); - return Duration(static_cast<Duration::rep>(rto)); -} - -} // end namespace protocol - -} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/protocols/verification_manager.h b/libtransport/src/hicn/transport/protocols/verification_manager.h new file mode 100644 index 000000000..da67e86f8 --- /dev/null +++ b/libtransport/src/hicn/transport/protocols/verification_manager.h @@ -0,0 +1,72 @@ +/* + * 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. + */ + +#pragma once + +#include <hicn/transport/interfaces/socket_consumer.h> + +#include <deque> + +namespace transport { + +namespace protocol { + +class VerificationManager { + public: + virtual ~VerificationManager() = default; + virtual bool onPacketToVerify(const Packet& packet) = 0; +}; + +class SignatureVerificationManager : public VerificationManager { + public: + SignatureVerificationManager(interface::ConsumerSocket* icn_socket) + : icn_socket_(icn_socket) {} + + TRANSPORT_ALWAYS_INLINE bool onPacketToVerify(const Packet& packet) override { + using namespace interface; + + bool verify_signature, ret = false; + icn_socket_->getSocketOption(GeneralTransportOptions::VERIFY_SIGNATURE, + verify_signature); + + if (!verify_signature) { + return true; + } + + std::shared_ptr<utils::Verifier> verifier; + icn_socket_->getSocketOption(GeneralTransportOptions::VERIFIER, verifier); + + if (TRANSPORT_EXPECT_FALSE(!verifier)) { + throw errors::RuntimeException( + "No certificate provided by the application."); + } + + ret = verifier->verify(packet); + + if (!ret) { + throw errors::RuntimeException( + "Verification failure policy has to be implemented."); + } + + return ret; + } + + private: + interface::ConsumerSocket* icn_socket_; +}; + +} // end namespace protocol + +} // end namespace transport diff --git a/libtransport/src/hicn/transport/utils/CMakeLists.txt b/libtransport/src/hicn/transport/utils/CMakeLists.txt index a5daf785e..c6b09fc5f 100644 --- a/libtransport/src/hicn/transport/utils/CMakeLists.txt +++ b/libtransport/src/hicn/transport/utils/CMakeLists.txt @@ -30,7 +30,6 @@ list(APPEND HEADER_FILES ${CMAKE_CURRENT_SOURCE_DIR}/string_tokenizer.h ${CMAKE_CURRENT_SOURCE_DIR}/hash.h ${CMAKE_CURRENT_SOURCE_DIR}/uri.h - ${CMAKE_CURRENT_SOURCE_DIR}/sharable_vector.h ${CMAKE_CURRENT_SOURCE_DIR}/branch_prediction.h ${CMAKE_CURRENT_SOURCE_DIR}/event_reactor.h ${CMAKE_CURRENT_SOURCE_DIR}/deadline_timer.h @@ -38,7 +37,6 @@ list(APPEND HEADER_FILES ${CMAKE_CURRENT_SOURCE_DIR}/event_reactor.h ${CMAKE_CURRENT_SOURCE_DIR}/min_filter.h ${CMAKE_CURRENT_SOURCE_DIR}/stream_buffer.h - ${CMAKE_CURRENT_SOURCE_DIR}/endianess.h ${CMAKE_CURRENT_SOURCE_DIR}/literals.h ${CMAKE_CURRENT_SOURCE_DIR}/signer.h ${CMAKE_CURRENT_SOURCE_DIR}/verifier.h diff --git a/libtransport/src/hicn/transport/utils/array.h b/libtransport/src/hicn/transport/utils/array.h index a3a66e498..7c0ed65d8 100644 --- a/libtransport/src/hicn/transport/utils/array.h +++ b/libtransport/src/hicn/transport/utils/array.h @@ -21,7 +21,7 @@ namespace utils { -template <typename T> +template <typename T = uint8_t> class Array { public: explicit Array(const T *array, size_t size) : array_(array), size_(size) { diff --git a/libtransport/src/hicn/transport/utils/sharable_vector.h b/libtransport/src/hicn/transport/utils/chrono_typedefs.h index 31adff1ad..8f28e763c 100644 --- a/libtransport/src/hicn/transport/utils/sharable_vector.h +++ b/libtransport/src/hicn/transport/utils/chrono_typedefs.h @@ -15,16 +15,13 @@ #pragma once -#include <memory> -#include <vector> +#include <chrono> namespace utils { -template <class T> -class SharableVector : public std::vector<T>, - public std::enable_shared_from_this<SharableVector<T>> { - public: - virtual ~SharableVector(){}; -}; +using SteadyClock = std::chrono::steady_clock; +using TimePoint = SteadyClock::time_point; +using Milliseconds = std::chrono::milliseconds; +using Microseconds = std::chrono::microseconds; } // namespace utils diff --git a/libtransport/src/hicn/transport/utils/content_store.cc b/libtransport/src/hicn/transport/utils/content_store.cc index 4c7637dad..d48e16daf 100644 --- a/libtransport/src/hicn/transport/utils/content_store.cc +++ b/libtransport/src/hicn/transport/utils/content_store.cc @@ -34,29 +34,29 @@ void ContentStore::insert( std::unique_lock<std::mutex> lock(cs_mutex_); if (TRANSPORT_EXPECT_FALSE(content_store_hash_table_.size() != - lru_list_.size())) { + fifo_list_.size())) { TRANSPORT_LOGW("Inconsistent size!!!!"); TRANSPORT_LOGW("Hash Table: %zu |||| FIFO List: %zu", - content_store_hash_table_.size(), lru_list_.size()); + content_store_hash_table_.size(), fifo_list_.size()); } // Check if the content can be cached if (content_object->getLifetime() > 0) { if (content_store_hash_table_.size() >= max_content_store_size_) { - content_store_hash_table_.erase(lru_list_.back()); - lru_list_.pop_back(); + content_store_hash_table_.erase(fifo_list_.back()); + fifo_list_.pop_back(); } // Insert new item auto it = content_store_hash_table_.find(content_object->getName()); if (it != content_store_hash_table_.end()) { - lru_list_.erase(it->second.second); + fifo_list_.erase(it->second.second); content_store_hash_table_.erase(content_object->getName()); } - lru_list_.push_front(std::cref(content_object->getName())); - auto pos = lru_list_.begin(); + fifo_list_.push_front(std::cref(content_object->getName())); + auto pos = fifo_list_.begin(); content_store_hash_table_[content_object->getName()] = ContentStoreEntry( ObjectTimeEntry(content_object, std::chrono::steady_clock::now()), pos); } @@ -67,13 +67,11 @@ const std::shared_ptr<ContentObject> &ContentStore::find( std::unique_lock<std::mutex> lock(cs_mutex_); auto it = content_store_hash_table_.find(interest.getName()); if (it != content_store_hash_table_.end()) { - // if (std::chrono::duration_cast<std::chrono::milliseconds>( - // std::chrono::steady_clock::now() - it->second.first.second).count() - // < it->second.first.first->getLifetime() || - // it->second.first.first->getLifetime() == - // default_values::never_expire_time) { - return it->second.first.first; - // } + if (std::chrono::duration_cast<std::chrono::milliseconds>( + std::chrono::steady_clock::now() - it->second.first.second) + .count() < it->second.first.first->getLifetime()) { + return it->second.first.first; + } } return empty_reference_; @@ -82,7 +80,7 @@ const std::shared_ptr<ContentObject> &ContentStore::find( void ContentStore::erase(const Name &exact_name) { std::unique_lock<std::mutex> lock(cs_mutex_); auto it = content_store_hash_table_.find(exact_name); - lru_list_.erase(it->second.second); + fifo_list_.erase(it->second.second); content_store_hash_table_.erase(exact_name); } diff --git a/libtransport/src/hicn/transport/utils/content_store.h b/libtransport/src/hicn/transport/utils/content_store.h index ab4963fff..39e87fb7d 100644 --- a/libtransport/src/hicn/transport/utils/content_store.h +++ b/libtransport/src/hicn/transport/utils/content_store.h @@ -41,7 +41,7 @@ typedef std::pair<std::shared_ptr<ContentObject>, typedef std::pair<ObjectTimeEntry, std::list<std::reference_wrapper<const Name>>::iterator> ContentStoreEntry; -typedef std::list<std::reference_wrapper<const Name>> LRUList; +typedef std::list<std::reference_wrapper<const Name>> FIFOList; typedef std::unordered_map<Name, ContentStoreEntry> ContentStoreHashTable; class ContentStore { @@ -66,7 +66,7 @@ class ContentStore { private: ContentStoreHashTable content_store_hash_table_; - LRUList lru_list_; + FIFOList fifo_list_; std::shared_ptr<ContentObject> empty_reference_; std::size_t max_content_store_size_; std::mutex cs_mutex_; diff --git a/libtransport/src/hicn/transport/utils/crypto_hash.h b/libtransport/src/hicn/transport/utils/crypto_hash.h index 0c15c8bda..945909d14 100644 --- a/libtransport/src/hicn/transport/utils/crypto_hash.h +++ b/libtransport/src/hicn/transport/utils/crypto_hash.h @@ -108,6 +108,10 @@ class CryptoHash { std::memcmp(digest1, digest2, hash_size_map[hash_type])); } + TRANSPORT_ALWAYS_INLINE void display() { + parcBuffer_Display(parcCryptoHash_GetDigest(hash_), 2); + } + private: PARCCryptoHash* hash_; }; diff --git a/libtransport/src/hicn/transport/utils/endianess.h b/libtransport/src/hicn/transport/utils/endianess.h deleted file mode 100644 index d86e764ab..000000000 --- a/libtransport/src/hicn/transport/utils/endianess.h +++ /dev/null @@ -1,141 +0,0 @@ -/* - * 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. - */ - -#pragma once - -#include <hicn/transport/portability/portability.h> - -#ifndef _WIN32 -#include <arpa/inet.h> -#else -#include <hicn/transport/portability/win_portability.h> -#endif - -#include <cstring> - -namespace utils { - -namespace { - -template <size_t Size> -struct uint_types_by_size; - -#define GENERATOR(sz, fn) \ - static TRANSPORT_ALWAYS_INLINE uint##sz##_t byteswap_gen(uint##sz##_t v) { \ - return fn(v); \ - } \ - template <> \ - struct uint_types_by_size<sz / 8> { \ - using type = uint##sz##_t; \ - }; - -GENERATOR(8, uint8_t) -#ifdef _MSC_VER -GENERATOR(64, _byteswap_uint64) -GENERATOR(32, _byteswap_ulong) -GENERATOR(16, _byteswap_ushort) -#else -GENERATOR(64, __builtin_bswap64) -GENERATOR(32, __builtin_bswap32) -GENERATOR(16, __builtin_bswap16) -#endif - -template <typename T> -struct EndianInt { - static_assert( - (std::is_integral<T>::value && !std::is_same<T, bool>::value) || - std::is_floating_point<T>::value, - "template type parameter must be non-bool integral or floating point"); - - static T swap(T x) { - // we implement this with memcpy because that is defined behavior in C++ - // we rely on compilers to optimize away the memcpy calls - constexpr auto s = sizeof(T); - using B = typename uint_types_by_size<s>::type; - B b; - std::memcpy(&b, &x, s); - b = byteswap_gen(b); - std::memcpy(&x, &b, s); - return x; - } - static T big(T x) { - return portability::little_endian_arch ? EndianInt::swap(x) : x; - } - static T little(T x) { - return portability::big_endian_arch ? EndianInt::swap(x) : x; - } -}; - -} // namespace - -// big* convert between native and big-endian representations -// little* convert between native and little-endian representations -// swap* convert between big-endian and little-endian representations -// -// ntohs, htons == big16 -// ntohl, htonl == big32 -#define GENERATOR1(fn, t, sz) \ - static t fn##sz(t x) { return fn<t>(x); } - -#define GENERATOR2(t, sz) \ - GENERATOR1(swap, t, sz) \ - GENERATOR1(big, t, sz) \ - GENERATOR1(little, t, sz) - -#define GENERATOR3(sz) \ - GENERATOR2(uint##sz##_t, sz) \ - GENERATOR2(int##sz##_t, sz) - -class Endian { - public: - enum class Order : uint8_t { LITTLE, BIG }; - - static constexpr Order order = - portability::little_endian_arch ? Order::LITTLE : Order::BIG; - - template <typename T> - static T swap(T x) { - return EndianInt<T>::swap(x); - } - - template <typename T> - static T big(T x) { - return EndianInt<T>::big(x); - } - - template <typename T> - static T little(T x) { - return EndianInt<T>::little(x); - } - -#if !defined(__ANDROID__) - GENERATOR3(64) - GENERATOR3(32) - GENERATOR3(16) - GENERATOR3(8) -#endif -}; - -template <typename T> -static TRANSPORT_ALWAYS_INLINE T ntoh(T x) { - return Endian::order == Endian::Order::LITTLE ? Endian::little(x) : x; -} - -template <typename T> -static TRANSPORT_ALWAYS_INLINE T hton(T x) { - return Endian::order == Endian::Order::LITTLE ? Endian::big(x) : x; -} - -} // namespace utils
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/utils/event_thread.h b/libtransport/src/hicn/transport/utils/event_thread.h index 3bf08c94b..e50ae9648 100644 --- a/libtransport/src/hicn/transport/utils/event_thread.h +++ b/libtransport/src/hicn/transport/utils/event_thread.h @@ -76,8 +76,6 @@ class EventThread { } void stop() { - TRANSPORT_LOGI("Stopping event thread!"); - io_service_.stop(); if (thread_ && thread_->joinable()) { diff --git a/libtransport/src/hicn/transport/utils/identity.h b/libtransport/src/hicn/transport/utils/identity.h index 018842ee3..349b38914 100644 --- a/libtransport/src/hicn/transport/utils/identity.h +++ b/libtransport/src/hicn/transport/utils/identity.h @@ -36,7 +36,6 @@ class Identity { unsigned int signature_length, unsigned int validity_days, const std::string &subject_name); - // No copies Identity(const Identity &other); Identity(std::string &file_name, std::string &password, diff --git a/libtransport/src/hicn/transport/utils/signer.cc b/libtransport/src/hicn/transport/utils/signer.cc index e1262ec94..981e5f02b 100644 --- a/libtransport/src/hicn/transport/utils/signer.cc +++ b/libtransport/src/hicn/transport/utils/signer.cc @@ -16,7 +16,6 @@ */ #include <hicn/transport/errors/malformed_ahpacket_exception.h> -#include <hicn/transport/utils/endianess.h> #include <hicn/transport/utils/key_id.h> #include <hicn/transport/utils/membuf.h> #include <hicn/transport/utils/signer.h> @@ -91,14 +90,8 @@ void Signer::sign(Packet &packet) { // Copy IP+TCP/ICMP header before zeroing them hicn_header_t header_copy; - if (format == HF_INET_TCP_AH) { - memcpy(&header_copy, hicn_packet, HICN_V4_TCP_HDRLEN); - } else if (format == HF_INET6_TCP_AH) { - memcpy(&header_copy, hicn_packet, HICN_V6_TCP_HDRLEN); - } else { - throw errors::RuntimeException( - "Signer::sign -- Packet format not expected."); - } + hicn_packet_copy_header(format, (const hicn_header_t *)packet.packet_start_, + &header_copy, false); std::size_t header_len = Packet::getHeaderSizeFromFormat(format); @@ -130,7 +123,8 @@ void Signer::sign(Packet &packet) { utils::CryptoHash hash = hasher.finalize(); PARCSignature *signature = parcSigner_SignDigestNoAlloc( - this->signer_, hash.hash_, packet.getSignature(), (uint32_t)sign_len_bytes); + this->signer_, hash.hash_, packet.getSignature(), + (uint32_t)sign_len_bytes); PARCBuffer *buffer = parcSignature_GetSignature(signature); size_t bytes_len = parcBuffer_Remaining(buffer); @@ -139,12 +133,8 @@ void Signer::sign(Packet &packet) { throw errors::MalformedAHPacketException(); } - /* Restore the resetted fields */ - if (format & HFO_INET) { - memcpy(hicn_packet, &header_copy, HICN_V4_TCP_HDRLEN); - } else if (format & HFO_INET6) { - memcpy(hicn_packet, &header_copy, HICN_V6_TCP_HDRLEN); - } + hicn_packet_copy_header(format, &header_copy, + (hicn_header_t *)packet.packet_start_, false); } PARCKeyStore *Signer::getKeyStore() { diff --git a/libtransport/src/hicn/transport/utils/socket.h b/libtransport/src/hicn/transport/utils/socket.h deleted file mode 100644 index ab8578f91..000000000 --- a/libtransport/src/hicn/transport/utils/socket.h +++ /dev/null @@ -1,267 +0,0 @@ -/* - * 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. - */ - -#pragma once - -#include <hicn/transport/config.h> -#include <hicn/transport/core/content_object.h> -#include <hicn/transport/core/facade.h> -#include <hicn/transport/core/interest.h> -#include <hicn/transport/core/manifest_format_fixed.h> -#include <hicn/transport/core/manifest_inline.h> -#include <hicn/transport/core/name.h> -#include <hicn/transport/transport/download_observer.h> -#include <hicn/transport/transport/socket_options_default_values.h> -#include <hicn/transport/transport/socket_options_keys.h> -#include <hicn/transport/utils/crypto_suite.h> -#include <hicn/transport/utils/identity.h> -#include <hicn/transport/utils/verifier.h> - -#define SOCKET_OPTION_GET 0 -#define SOCKET_OPTION_NOT_GET 1 -#define SOCKET_OPTION_SET 2 -#define SOCKET_OPTION_NOT_SET 3 -#define SOCKET_OPTION_DEFAULT 12345 - -#define VOID_HANDLER 0 - -namespace transport { - -namespace transport { - -template <typename PortalType> -class Socket; -class ConsumerSocket; -class ProducerSocket; - -using Interest = core::Interest; -using ContentObject = core::ContentObject; -using Name = core::Name; -using ContentObjectManifest = core::ManifestInline<ContentObject, core::Fixed>; -using InterestManifest = core::ManifestInline<Interest, core::Fixed>; -using HashAlgorithm = core::HashAlgorithm; -using CryptoSuite = utils::CryptoSuite; -using Identity = utils::Identity; -using Verifier = utils::Verifier; - -using HicnForwarderPortal = core::HicnForwarderPortal; - -#ifdef __linux__ -#ifndef __ANDROID__ -using RawSocketPortal = core::RawSocketPortal; -#endif -#endif - -#ifdef __vpp__ -using VPPForwarderPortal = core::VPPForwarderPortal; -using BaseSocket = Socket<VPPForwarderPortal>; -using BasePortal = VPPForwarderPortal; -#else -using BaseSocket = Socket<HicnForwarderPortal>; -using BasePortal = HicnForwarderPortal; -#endif - -using PayloadType = core::PayloadType; -using Prefix = core::Prefix; -using Array = utils::Array<uint8_t>; - -using ConsumerInterestCallback = - std::function<void(ConsumerSocket &, const Interest &)>; - -using ConsumerContentCallback = - std::function<void(ConsumerSocket &, std::size_t, const std::error_code &)>; - -using ConsumerTimerCallback = - std::function<void(ConsumerSocket &, std::size_t, - std::chrono::milliseconds &, float, uint32_t, uint32_t)>; - -using ProducerContentCallback = std::function<void( - ProducerSocket &, const std::error_code &, uint64_t bytes_written)>; - -using ConsumerContentObjectCallback = - std::function<void(ConsumerSocket &, const ContentObject &)>; - -using ConsumerContentObjectVerificationCallback = - std::function<bool(ConsumerSocket &, const ContentObject &)>; - -using ConsumerManifestCallback = - std::function<void(ConsumerSocket &, const ContentObjectManifest &)>; - -using ProducerContentObjectCallback = - std::function<void(ProducerSocket &, ContentObject &)>; - -using ProducerInterestCallback = - std::function<void(ProducerSocket &, const Interest &)>; - -using ProducerInterestCallback = - std::function<void(ProducerSocket &, const Interest &)>; - -template <typename PortalType> -class Socket { - static_assert(std::is_same<PortalType, HicnForwarderPortal>::value -#ifdef __linux__ -#ifndef __ANDROID__ - || std::is_same<PortalType, RawSocketPortal>::value -#ifdef __vpp__ - || std::is_same<PortalType, VPPForwarderPortal>::value -#endif -#endif - , -#else - , - -#endif - "This class is not allowed as Portal"); - - public: - typedef PortalType Portal; - - virtual asio::io_service &getIoService() = 0; - - virtual void connect() = 0; - - virtual int setSocketOption(int socket_option_key, - uint32_t socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - double socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - bool socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - Name socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - std::list<Prefix> socket_option_value) = 0; - - virtual int setSocketOption( - int socket_option_key, - ProducerContentObjectCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ProducerInterestCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ProducerContentCallback socket_option_value) = 0; - - virtual int setSocketOption( - int socket_option_key, - ConsumerContentObjectVerificationCallback socket_option_value) = 0; - - virtual int setSocketOption( - int socket_option_key, - ConsumerContentObjectCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerInterestCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerContentCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerManifestCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - IcnObserver *socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - HashAlgorithm socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - CryptoSuite socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - const Identity &socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - ConsumerTimerCallback socket_option_value) = 0; - - virtual int setSocketOption(int socket_option_key, - const std::string &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - uint32_t &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - double &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - bool &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - Name &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - std::list<Prefix> &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, - ProducerContentObjectCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, ProducerInterestCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, - ConsumerContentObjectVerificationCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, - ConsumerContentObjectCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, ConsumerInterestCallback &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - ConsumerContentCallback &socket_option_value) = 0; - - virtual int getSocketOption( - int socket_option_key, ConsumerManifestCallback &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - ProducerContentCallback &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - std::shared_ptr<Portal> &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - IcnObserver **socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - HashAlgorithm &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - CryptoSuite &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - Identity &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - std::string &socket_option_value) = 0; - - virtual int getSocketOption(int socket_option_key, - ConsumerTimerCallback &socket_option_value) = 0; - - protected: - virtual ~Socket(){}; - - protected: - std::string output_interface_; -}; - -} // namespace transport - -} // namespace transport diff --git a/libtransport/src/hicn/transport/utils/verifier.cc b/libtransport/src/hicn/transport/utils/verifier.cc index 4295aaab7..af19d8b5e 100644 --- a/libtransport/src/hicn/transport/utils/verifier.cc +++ b/libtransport/src/hicn/transport/utils/verifier.cc @@ -109,14 +109,8 @@ int Verifier::verify(const Packet &packet) { // Copy IP+TCP/ICMP header before zeroing them hicn_header_t header_copy; - if (format == HF_INET_TCP_AH) { - memcpy(&header_copy, hicn_packet, HICN_V4_TCP_HDRLEN); - } else if (format == HF_INET6_TCP_AH) { - memcpy(&header_copy, hicn_packet, HICN_V6_TCP_HDRLEN); - } else { - throw errors::RuntimeException( - "Verifier::verify -- Packet format not expected."); - } + hicn_packet_copy_header(format, (const hicn_header_t *)packet.packet_start_, + &header_copy, false); std::size_t header_len = Packet::getHeaderSizeFromFormat(format); @@ -130,7 +124,7 @@ int Verifier::verify(const Packet &packet) { int ah_payload_len = (int)packet.getSignatureSize(); uint8_t *_signature = packet.getSignature(); - uint8_t * signature = new uint8_t[ah_payload_len]; + uint8_t *signature = new uint8_t[ah_payload_len]; // TODO Remove signature copy at this point, by not setting to zero // the validation payload. @@ -185,11 +179,8 @@ int Verifier::verify(const Packet &packet) { verifier_, key_id, hash_computed_locally, suite, signatureToVerify); /* Restore the resetted fields */ - if (format & HFO_INET) { - memcpy(hicn_packet, &header_copy, HICN_V4_TCP_HDRLEN); - } else if (format & HFO_INET6) { - memcpy(hicn_packet, &header_copy, HICN_V6_TCP_HDRLEN); - } + hicn_packet_copy_header(format, &header_copy, + (hicn_header_t *)packet.packet_start_, false); delete[] signature; diff --git a/utils/CMakeLists.txt b/utils/CMakeLists.txt index 8b816383c..d9dafe7ec 100644 --- a/utils/CMakeLists.txt +++ b/utils/CMakeLists.txt @@ -14,7 +14,7 @@ cmake_minimum_required(VERSION 3.5 FATAL_ERROR) set(CMAKE_CXX_STANDARD 14) -project(Utils) +project(utils) set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} diff --git a/utils/src/hiperf.cc b/utils/src/hiperf.cc index f2387b57c..fbd9d1e26 100644 --- a/utils/src/hiperf.cc +++ b/utils/src/hiperf.cc @@ -52,7 +52,7 @@ struct ClientConfiguration { window(-1), virtual_download(true), producer_certificate("/tmp/rsa_certificate.pem"), - receive_buffer(std::make_shared<utils::SharableVector<uint8_t>>()), + receive_buffer(std::make_shared<std::vector<uint8_t>>()), download_size(0), report_interval_milliseconds_(1000), rtc_(false) {} @@ -64,7 +64,7 @@ struct ClientConfiguration { double window; bool virtual_download; std::string producer_certificate; - std::shared_ptr<utils::SharableVector<uint8_t>> receive_buffer; + std::shared_ptr<std::vector<uint8_t>> receive_buffer; std::size_t download_size; std::uint32_t report_interval_milliseconds_; TransportProtocolAlgorithms transport_protocol_; @@ -114,7 +114,7 @@ class HIperfClient { void processPayload(ConsumerSocket &c, std::size_t bytes_transferred, const std::error_code &ec) { Time t2 = std::chrono::steady_clock::now(); - TimeDuration dt = std::chrono::duration_cast<TimeDuration>(t2 - t1_); + TimeDuration dt = std::chrono::duration_cast<TimeDuration>(t2 - t_download_); long usec = (long)dt.count(); std::cout << "Content retrieved. Size: " << bytes_transferred << " [Bytes]" @@ -137,37 +137,34 @@ class HIperfClient { return true; } - void processLeavingInterest(ConsumerSocket &c, const Interest &interest) { - // std::cout << "LEAVES " << interest.getName().toUri() << std::endl; - } + void processLeavingInterest(ConsumerSocket &c, const Interest &interest) {} - void handleTimerExpiration(ConsumerSocket &c, std::size_t byte_count, - std::chrono::milliseconds &exact_duration, - float c_window, uint32_t retransmissions, - uint32_t average_rtt) { + void handleTimerExpiration(ConsumerSocket &c, const protocol::TransportStatistics &stats) { const char separator = ' '; const int width = 20; + utils::TimePoint t2 = utils::SteadyClock::now(); + auto exact_duration = std::chrono::duration_cast<utils::Milliseconds>(t2 - t_stats_); + std::stringstream interval; interval << total_duration_milliseconds_ / 1000 << "-" - << total_duration_milliseconds_ / 1000 + - exact_duration.count() / 1000; + << total_duration_milliseconds_ / 1000 + exact_duration.count() / 1000; std::stringstream bytes_transferred; bytes_transferred << std::fixed << std::setprecision(3) - << (byte_count - old_bytes_value_) / 1000000.0 + << (stats.getBytesRecv() - old_bytes_value_) / 1000000.0 << std::setfill(separator) << "[MBytes]"; std::stringstream bandwidth; - bandwidth << ((byte_count - old_bytes_value_) * 8) / + bandwidth << ((stats.getBytesRecv() - old_bytes_value_) * 8) / (exact_duration.count()) / 1000.0 << std::setfill(separator) << "[Mbps]"; std::stringstream window; - window << c_window << std::setfill(separator) << "[Interest]"; + window << stats.getAverageWindowSize() << std::setfill(separator) << "[Interest]"; std::stringstream avg_rtt; - avg_rtt << average_rtt << std::setfill(separator) << "[us]"; + avg_rtt << stats.getAverageRtt() << std::setfill(separator) << "[us]"; std::cout << std::left << std::setw(width) << "Interval"; std::cout << std::left << std::setw(width) << "Transfer"; @@ -179,13 +176,14 @@ class HIperfClient { std::cout << std::left << std::setw(width) << interval.str(); std::cout << std::left << std::setw(width) << bytes_transferred.str(); std::cout << std::left << std::setw(width) << bandwidth.str(); - std::cout << std::left << std::setw(width) << retransmissions; + std::cout << std::left << std::setw(width) << stats.getRetxCount(); std::cout << std::left << std::setw(width) << window.str(); std::cout << std::left << std::setw(width) << avg_rtt.str() << std::endl; std::cout << std::endl; total_duration_milliseconds_ += (uint32_t)exact_duration.count(); - old_bytes_value_ = byte_count; + old_bytes_value_ = stats.getBytesRecv(); + t_stats_ = utils::SteadyClock::now(); } int setup() { @@ -280,18 +278,17 @@ class HIperfClient { } ret = consumer_socket_->setSocketOption( - ConsumerCallbacksOptions::TIMER_EXPIRES, + ConsumerCallbacksOptions::STATS_SUMMARY, (ConsumerTimerCallback)std::bind( &HIperfClient::handleTimerExpiration, this, std::placeholders::_1, - std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, - std::placeholders::_5, std::placeholders::_6)); + std::placeholders::_2)); if (ret == SOCKET_OPTION_NOT_SET) { return ERROR_SETUP; } if (consumer_socket_->setSocketOption( - GeneralTransportOptions::TIMER_INTERVAL, + GeneralTransportOptions::STATS_INTERVAL, configuration_.report_interval_milliseconds_) == SOCKET_OPTION_NOT_SET) { return ERROR_SETUP; @@ -310,7 +307,7 @@ class HIperfClient { io_service_.stop(); }); - t1_ = std::chrono::steady_clock::now(); + t_download_ = std::chrono::steady_clock::now(); consumer_socket_->asyncConsume(configuration_.name, configuration_.receive_buffer); io_service_.run(); @@ -321,7 +318,8 @@ class HIperfClient { private: ClientConfiguration configuration_; std::unique_ptr<ConsumerSocket> consumer_socket_; - Time t1_; + Time t_stats_; + Time t_download_; uint32_t total_duration_milliseconds_; uint64_t old_bytes_value_; asio::io_service io_service_; @@ -338,8 +336,7 @@ class HIperfServer { content_objects_((std::uint16_t)(1 << log2_content_object_buffer_size)), content_objects_index_(0), mask_((std::uint16_t)(1 << log2_content_object_buffer_size) - 1) { - std::string buffer(1440, 'X'); - + std::string buffer(1200, 'X'); std::cout << "Producing contents under name " << conf.name.getName() << std::endl; @@ -396,13 +393,13 @@ class HIperfServer { << std::endl; } - utils::Identity setProducerIdentity(std::string &keystore_name, + std::shared_ptr<utils::Identity> setProducerIdentity(std::string &keystore_name, std::string &keystore_password, HashAlgorithm &hash_algorithm) { if (access(keystore_name.c_str(), F_OK) != -1) { - return utils::Identity(keystore_name, keystore_password, hash_algorithm); + return std::make_shared<utils::Identity>(keystore_name, keystore_password, hash_algorithm); } else { - return utils::Identity(keystore_name, keystore_password, + return std::make_shared<utils::Identity>(keystore_name, keystore_password, CryptoSuite::RSA_SHA256, 1024, 365, "producer-test"); } @@ -414,7 +411,7 @@ class HIperfServer { producer_socket_ = std::make_unique<ProducerSocket>(); if (configuration_.sign) { - Identity identity = setProducerIdentity(configuration_.keystore_name, + auto identity = setProducerIdentity(configuration_.keystore_name, configuration_.keystore_password, configuration_.hash_algorithm); @@ -426,6 +423,7 @@ class HIperfServer { } producer_socket_->registerPrefix(configuration_.name); + producer_socket_->connect(); if (!configuration_.virtual_producer) { if (producer_socket_->setSocketOption( @@ -478,8 +476,6 @@ class HIperfServer { } } - producer_socket_->connect(); - return ERROR_SUCCESS; } @@ -548,6 +544,9 @@ void usage() { << "RAAQM drop factor " "parameter" << std::endl; + std::cerr << "-M\t<Download for real>\t\t" + << "Store the content downloaded." + << std::endl; std::cerr << "-W\t<window_size>\t\t\t" << "Use a fixed congestion window " "for retrieving the data." @@ -584,14 +583,14 @@ int main(int argc, char *argv[]) { int opt; #ifndef _WIN32 - while ((opt = getopt(argc, argv, "DSCf:b:d:W:c:vs:rmlk:y:p:hi:x")) != -1) { + while ((opt = getopt(argc, argv, "DSCf:b:d:W:RMc:vs:rmlk:y:p:hi:x")) != -1) { switch (opt) { // Common case 'D': daemon = true; break; #else - while ((opt = getopt(argc, argv, "SCf:b:d:W:c:vs:rmlk:y:p:hi:x")) != -1) { + while ((opt = getopt(argc, argv, "SCf:b:d:W:RMc:vs:rmlk:y:p:hi:x")) != -1) { switch (opt) { #endif case 'f': @@ -619,6 +618,10 @@ int main(int argc, char *argv[]) { client_configuration.window = std::stod(optarg); options = 1; break; + case 'M': + client_configuration.virtual_download = false; + options = 1; + break; case 'c': client_configuration.producer_certificate = std::string(optarg); options = 1; diff --git a/utils/src/ping_server.cc b/utils/src/ping_server.cc index 8e3c6a0ed..faf86fb64 100644 --- a/utils/src/ping_server.cc +++ b/utils/src/ping_server.cc @@ -203,7 +203,7 @@ int main(int argc, char **argv) { bool quite = false; bool flags = false; bool reset = false; - uint32_t object_size = 1350; + uint32_t object_size = 1250; uint8_t ttl = 64; std::string keystore_path = "./rsa_crypto_material.p12"; std::string keystore_password = "cisco"; |