diff options
Diffstat (limited to 'apps/hiperf/src/server.cc')
| -rw-r--r-- | apps/hiperf/src/server.cc | 802 |
1 files changed, 483 insertions, 319 deletions
diff --git a/apps/hiperf/src/server.cc b/apps/hiperf/src/server.cc index 968d42e2c..3f6c335f9 100644 --- a/apps/hiperf/src/server.cc +++ b/apps/hiperf/src/server.cc @@ -1,5 +1,5 @@ /* - * Copyright (c) 2021 Cisco and/or its affiliates. + * Copyright (c) 2021-2022 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: @@ -17,167 +17,92 @@ namespace hiperf { +using transport::core::ContentObject; +using transport::core::Interest; +using transport::core::Name; +using transport::interface::GeneralTransportOptions; +using transport::interface::ProducerCallbacksOptions; +using transport::interface::ProducerInterestCallback; +using transport::interface::ProducerSocket; +using transport::interface::ProductionProtocolAlgorithms; + /** * Hiperf server class: configure and setup an hicn producer following the * ServerConfiguration. */ class HIperfServer::Impl { - const std::size_t log2_content_object_buffer_size = 8; - - public: - Impl(const hiperf::ServerConfiguration &conf) - : configuration_(conf), - signals_(io_service_), - rtc_timer_(io_service_), - unsatisfied_interests_(), - 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), - last_segment_(0), -#ifndef _WIN32 - ptr_last_segment_(&last_segment_), - input_(io_service_), - rtc_running_(false), -#else - ptr_last_segment_(&last_segment_), -#endif - flow_name_(configuration_.name.getName()), - socket_(io_service_), - recv_buffer_(nullptr, 0) { - std::string buffer(configuration_.payload_size_, 'X'); - std::cout << "Producing contents under name " << conf.name.getName() - << std::endl; -#ifndef _WIN32 - if (configuration_.interactive_) { - input_.assign(::dup(STDIN_FILENO)); - } -#endif - - for (int i = 0; i < (1 << log2_content_object_buffer_size); i++) { - content_objects_[i] = ContentObject::Ptr( - new ContentObject(conf.name.getName(), HF_INET6_TCP, 0, - (const uint8_t *)buffer.data(), buffer.size())); - content_objects_[i]->setLifetime( - default_values::content_object_expiry_time); - } - } - - void virtualProcessInterest(ProducerSocket &p, const Interest &interest) { - content_objects_[content_objects_index_ & mask_]->setName( - interest.getName()); - producer_socket_->produce( - *content_objects_[content_objects_index_++ & mask_]); + static constexpr std::size_t klog2_content_object_buffer_size() { return 8; } + static constexpr std::size_t kcontent_object_buffer_size() { + return (1 << klog2_content_object_buffer_size()); } - - void processInterest(ProducerSocket &p, const Interest &interest) { - p.setSocketOption(ProducerCallbacksOptions::CACHE_MISS, - (ProducerInterestCallback)VOID_HANDLER); - p.setSocketOption(GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, - 5000000_U32); - - produceContent(p, interest.getName(), interest.getName().getSuffix()); - std::cout << "Received interest " << interest.getName().getSuffix() - << std::endl; + static constexpr std::size_t kmask() { + return (kcontent_object_buffer_size() - 1); } - void asyncProcessInterest(ProducerSocket &p, const Interest &interest) { - p.setSocketOption(ProducerCallbacksOptions::CACHE_MISS, - (ProducerInterestCallback)bind(&Impl::cacheMiss, this, - std::placeholders::_1, - std::placeholders::_2)); - p.setSocketOption(GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, - 5000000_U32); - uint32_t suffix = interest.getName().getSuffix(); - - if (suffix == 0) { - last_segment_ = 0; - ptr_last_segment_ = &last_segment_; - unsatisfied_interests_.clear(); - } - - // The suffix will either be the one from the received interest or the - // smallest suffix of a previous interest not satisfed - if (!unsatisfied_interests_.empty()) { - auto it = - std::lower_bound(unsatisfied_interests_.begin(), - unsatisfied_interests_.end(), *ptr_last_segment_); - if (it != unsatisfied_interests_.end()) { - suffix = *it; + /** + * @brief As we can (potentially) setup many producer sockets, we need to keep + * a separate context for each one of them. The context contains parameters + * and variable that are specific to a single producer socket. + */ + class ProducerContext + : public Base<ProducerContext, ServerConfiguration, Impl>, + public ProducerSocket::Callback { + public: + using ConfType = ServerConfiguration; + using ParentType = typename HIperfServer::Impl; + static inline const auto getContextType() { return "ProducerContext"; } + + ProducerContext(HIperfServer::Impl &server, int producer_identifier) + : Base(server, server.io_service_, producer_identifier) { + // Allocate buffer to copy as content objects payload + std::string buffer(configuration_.payload_size_, 'X'); + + // Allocate array of content objects. They are share_ptr so that the + // transport will only capture a reference to them instead of performing + // an hard copy. + for (std::size_t i = 0; i < kcontent_object_buffer_size(); i++) { + const auto &element = + content_objects_.emplace_back(std::make_shared<ContentObject>( + configuration_.name_.makeName(), configuration_.packet_format_, + 0, (const uint8_t *)buffer.data(), buffer.size())); + element->setLifetime( + transport::interface::default_values::content_object_expiry_time); } - unsatisfied_interests_.erase(unsatisfied_interests_.begin(), it); } - std::cout << "Received interest " << interest.getName().getSuffix() - << ", starting production at " << suffix << std::endl; - std::cout << unsatisfied_interests_.size() << " interests still unsatisfied" - << std::endl; - produceContentAsync(p, interest.getName(), suffix); - } - - void produceContent(ProducerSocket &p, const Name &content_name, - uint32_t suffix) { - auto b = utils::MemBuf::create(configuration_.download_size); - std::memset(b->writableData(), '?', configuration_.download_size); - b->append(configuration_.download_size); - uint32_t total; - - utils::TimePoint t0 = utils::SteadyClock::now(); - total = p.produceStream(content_name, std::move(b), - !configuration_.multiphase_produce_, suffix); - utils::TimePoint t1 = utils::SteadyClock::now(); - - std::cout - << "Written " << total - << " data packets in output buffer (Segmentation time: " - << std::chrono::duration_cast<utils::Microseconds>(t1 - t0).count() - << " us)" << std::endl; - } - - void produceContentAsync(ProducerSocket &p, Name content_name, - uint32_t suffix) { - auto b = utils::MemBuf::create(configuration_.download_size); - std::memset(b->writableData(), '?', configuration_.download_size); - b->append(configuration_.download_size); + // To make vector happy (move or copy constructor is needed when vector + // resizes) + ProducerContext(ProducerContext &&other) noexcept + : Base(std::move(other)), + content_objects_(std::move(other.content_objects_)), + unsatisfied_interests_(std::move(other.unsatisfied_interests_)), + last_segment_(other.last_segment_), + producer_socket_(std::move(other.producer_socket_)), + content_objects_index_(other.content_objects_index_), + payload_size_max_(other.payload_size_max_) {} - p.asyncProduce(content_name, std::move(b), - !configuration_.multiphase_produce_, suffix, - &ptr_last_segment_); - } + virtual ~ProducerContext() = default; - void cacheMiss(ProducerSocket &p, const Interest &interest) { - unsatisfied_interests_.push_back(interest.getName().getSuffix()); - } + /** + * @brief Produce datagram + */ + void produceDatagram(const uint8_t *buffer, std::size_t buffer_size) const { + assert(producer_socket_); - void onContentProduced(ProducerSocket &p, const std::error_code &err, - uint64_t bytes_written) { - p.setSocketOption(ProducerCallbacksOptions::CACHE_MISS, - (ProducerInterestCallback)bind( - &Impl::asyncProcessInterest, this, - std::placeholders::_1, std::placeholders::_2)); - } + auto size = std::min(buffer_size, payload_size_max_); - std::shared_ptr<Identity> getProducerIdentity(std::string &keystore_path, - std::string &keystore_pwd, - CryptoHashType &hash_type) { - if (access(keystore_path.c_str(), F_OK) != -1) { - return std::make_shared<Identity>(keystore_path, keystore_pwd, hash_type); + producer_socket_->produceDatagram(flow_name_, buffer, size); } - return std::make_shared<Identity>(keystore_path, keystore_pwd, - CryptoSuite::RSA_SHA256, 1024, 365, - "producer-test"); - } - int setup() { - int ret; - int production_protocol; - - if (configuration_.secure_) { - auto identity = getProducerIdentity(configuration_.keystore_name, - configuration_.keystore_password, - configuration_.hash_algorithm); - producer_socket_ = std::make_unique<P2PSecureProducerSocket>( - configuration_.rtc_, identity); - } else { + /** + * @brief Create and setup the producer socket + */ + int setup() { + int ret; + int production_protocol; + std::shared_ptr<transport::auth::Signer> signer = + std::make_shared<transport::auth::VoidSigner>(); + if (!configuration_.rtc_) { production_protocol = ProductionProtocolAlgorithms::BYTE_STREAM; } else { @@ -185,210 +110,470 @@ class HIperfServer::Impl { } producer_socket_ = std::make_unique<ProducerSocket>(production_protocol); - } - if (producer_socket_->setSocketOption( - GeneralTransportOptions::MAKE_MANIFEST, configuration_.manifest) == - SOCKET_OPTION_NOT_SET) { - return ERROR_SETUP; - } + if (producer_socket_->setSocketOption( + ProducerCallbacksOptions::PRODUCER_CALLBACK, this) == + SOCKET_OPTION_NOT_SET) { + getOutputStream() << "Failed to set producer callback." << std::endl; + return ERROR_SETUP; + } - if (!configuration_.passphrase.empty()) { - std::shared_ptr<Signer> signer = std::make_shared<SymmetricSigner>( - CryptoSuite::HMAC_SHA256, configuration_.passphrase); - producer_socket_->setSocketOption(GeneralTransportOptions::SIGNER, - signer); - } + if (producer_socket_->setSocketOption( + GeneralTransportOptions::HASH_ALGORITHM, + configuration_.hash_algorithm_) == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (producer_socket_->setSocketOption( + GeneralTransportOptions::MANIFEST_MAX_CAPACITY, + configuration_.manifest_max_capacity_) == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (producer_socket_->setSocketOption(transport::interface::PACKET_FORMAT, + configuration_.packet_format_) == + SOCKET_OPTION_NOT_SET) { + getOutputStream() << "ERROR -- Impossible to set the packet format." + << std::endl; + return ERROR_SETUP; + } + + if (!configuration_.passphrase_.empty()) { + signer = std::make_shared<transport::auth::SymmetricSigner>( + transport::auth::CryptoSuite::HMAC_SHA256, + configuration_.passphrase_); + } + + if (!configuration_.keystore_name_.empty()) { + signer = std::make_shared<transport::auth::AsymmetricSigner>( + configuration_.keystore_name_, configuration_.keystore_password_); + } - if (!configuration_.keystore_name.empty()) { - auto identity = getProducerIdentity(configuration_.keystore_name, - configuration_.keystore_password, - configuration_.hash_algorithm); - std::shared_ptr<Signer> signer = identity->getSigner(); producer_socket_->setSocketOption(GeneralTransportOptions::SIGNER, signer); - } - uint32_t rtc_header_size = 0; - if (configuration_.rtc_) rtc_header_size = 12; - producer_socket_->setSocketOption( - GeneralTransportOptions::DATA_PACKET_SIZE, - (uint32_t)( - configuration_.payload_size_ + rtc_header_size + - (configuration_.name.getAddressFamily() == AF_INET ? 40 : 60))); - producer_socket_->registerPrefix(configuration_.name); - producer_socket_->connect(); - - if (configuration_.rtc_) { - std::cout << "Running RTC producer: the prefix length will be ignored." - " Use /128 by default in RTC mode" - << std::endl; - return ERROR_SUCCESS; - } + // Compute maximum payload size + Packet::Format format = configuration_.packet_format_; + if (!configuration_.manifest_max_capacity_) + format = Packet::toAHFormat(format); + payload_size_max_ = PayloadSize(format).getPayloadSizeMax( + configuration_.rtc_ ? RTC_HEADER_SIZE : 0, + configuration_.fec_type_.empty() ? 0 : FEC_HEADER_MAX_SIZE, + !configuration_.manifest_max_capacity_ + ? signer->getSignatureFieldSize() + : 0); + + if (configuration_.payload_size_ > payload_size_max_) { + getOutputStream() << "WARNING: Payload has size " + << configuration_.payload_size_ << ", maximum is " + << payload_size_max_ + << ". Payload will be truncated to fit." << std::endl; + } + + // Verifier for aggregated interests + std::shared_ptr<transport::auth::Verifier> verifier = + std::make_shared<transport::auth::VoidVerifier>(); + if (!configuration_.aggr_interest_passphrase_.empty()) { + verifier = std::make_unique<transport::auth::SymmetricVerifier>( + configuration_.aggr_interest_passphrase_); + } + ret = producer_socket_->setSocketOption(GeneralTransportOptions::VERIFIER, + verifier); + if (ret == SOCKET_OPTION_NOT_SET) return ERROR_SETUP; + + if (configuration_.rtc_) { + ret = producer_socket_->setSocketOption( + transport::interface::RtcTransportOptions::AGGREGATED_DATA, + configuration_.aggregated_data_); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + ret = producer_socket_->setSocketOption( + GeneralTransportOptions::FEC_TYPE, configuration_.fec_type_); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } - if (!configuration_.virtual_producer) { if (producer_socket_->setSocketOption( GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, - configuration_.content_lifetime) == SOCKET_OPTION_NOT_SET) { + configuration_.content_lifetime_) == SOCKET_OPTION_NOT_SET) { return ERROR_SETUP; } - if (producer_socket_->setSocketOption( - GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 200000U) == - SOCKET_OPTION_NOT_SET) { - return ERROR_SETUP; + producer_socket_->registerPrefix(Prefix(flow_name_, 128)); + producer_socket_->connect(); + producer_socket_->start(); + + if (configuration_.rtc_) { + return ERROR_SUCCESS; } - if (!configuration_.live_production) { - produceContent(*producer_socket_, configuration_.name.getName(), 0); + if (!configuration_.virtual_producer_) { + if (producer_socket_->setSocketOption( + GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 200000U) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (producer_socket_->setSocketOption( + GeneralTransportOptions::MAX_SEGMENT_SIZE, + static_cast<uint32_t>(configuration_.payload_size_)) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (!configuration_.live_production_) { + produceContent(*producer_socket_, configuration_.name_.makeName(), 0); + } else { + ret = producer_socket_->setSocketOption( + ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)bind( + &ProducerContext::asyncProcessInterest, this, + std::placeholders::_1, std::placeholders::_2)); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } } else { ret = producer_socket_->setSocketOption( + GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 0U); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + ret = producer_socket_->setSocketOption( ProducerCallbacksOptions::CACHE_MISS, - (ProducerInterestCallback)bind(&Impl::asyncProcessInterest, this, - std::placeholders::_1, - std::placeholders::_2)); + (ProducerInterestCallback)bind( + &ProducerContext::virtualProcessInterest, this, + std::placeholders::_1, std::placeholders::_2)); if (ret == SOCKET_OPTION_NOT_SET) { return ERROR_SETUP; } } - } else { - ret = producer_socket_->setSocketOption( - GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 0U); + ret = producer_socket_->setSocketOption( + ProducerCallbacksOptions::CONTENT_PRODUCED, + (transport::interface::ProducerContentCallback)bind( + &ProducerContext::onContentProduced, this, std::placeholders::_1, + std::placeholders::_2, std::placeholders::_3)); if (ret == SOCKET_OPTION_NOT_SET) { return ERROR_SETUP; } - ret = producer_socket_->setSocketOption( + return ERROR_SUCCESS; + } + + int run() { + getOutputStream() << "started to serve consumers with name " << flow_name_ + << std::endl; + return ERROR_SUCCESS; + } + + void stop() { + getOutputStream() << "stopped to serve consumers" << std::endl; + producer_socket_->stop(); + } + + private: + /** + * @brief Produce an existing content object. Set the name as the + * interest. + */ + void virtualProcessInterest(ProducerSocket &p, const Interest &interest) { + content_objects_[content_objects_index_ & kmask()]->setName( + interest.getName()); + p.produce(*content_objects_[content_objects_index_++ & kmask()]); + } + + /** + * @brief Create and produce a buffer of configuration_.download_size_ + * length. + */ + void produceContent(ProducerSocket &p, const Name &content_name, + uint32_t suffix) const { + uint32_t total; + + auto b = utils::MemBuf::create(configuration_.download_size_); + std::memset(b->writableData(), '?', configuration_.download_size_); + b->append(configuration_.download_size_); + + utils::SteadyTime::TimePoint t0 = utils::SteadyTime::Clock::now(); + total = p.produceStream(content_name, std::move(b), + !configuration_.multiphase_produce_, suffix); + utils::SteadyTime::TimePoint t1 = utils::SteadyTime::Clock::now(); + + LoggerInfo() << "Written " << total + << " data packets in output buffer (Segmentation time: " + << utils::SteadyTime::getDurationUs(t0, t1).count() << " us)" + << std::endl; + } + + /** + * @brief Synchronously produce content upon reception of one interest + */ + void processInterest(ProducerSocket &p, const Interest &interest) const { + p.setSocketOption( ProducerCallbacksOptions::CACHE_MISS, - (ProducerInterestCallback)bind(&Impl::virtualProcessInterest, this, - std::placeholders::_1, - std::placeholders::_2)); + (ProducerInterestCallback)transport::interface::VOID_HANDLER); + p.setSocketOption(GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, + configuration_.content_lifetime_); - if (ret == SOCKET_OPTION_NOT_SET) { - return ERROR_SETUP; + produceContent(p, interest.getName(), interest.getName().getSuffix()); + LoggerInfo() << "Received interest " << interest.getName().getSuffix() + << std::endl; + } + + /** + * @brief Async create and produce a buffer of + * configuration_.download_size_ length. + */ + void produceContentAsync(ProducerSocket &p, Name content_name, + uint32_t suffix) { + parent_.produce_thread_.add([this, suffix, content_name, &p]() { + auto b = utils::MemBuf::create(configuration_.download_size_); + std::memset(b->writableData(), '?', configuration_.download_size_); + b->append(configuration_.download_size_); + + last_segment_ = + suffix + p.produceStream(content_name, std::move(b), + !configuration_.multiphase_produce_, + suffix); + }); + } + + /** + * @brief Asynchronously produce content upon reception of one interest + */ + void asyncProcessInterest(ProducerSocket &p, const Interest &interest) { + p.setSocketOption(ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)bind( + &ProducerContext::cacheMiss, this, + std::placeholders::_1, std::placeholders::_2)); + p.setSocketOption(GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, + configuration_.content_lifetime_); + uint32_t suffix = interest.getName().getSuffix(); + + if (suffix == 0) { + last_segment_ = 0; + unsatisfied_interests_.clear(); + } + + // The suffix will either come from the received interest or will be set + // to the smallest suffix of a previous interest not satisfied + if (!unsatisfied_interests_.empty()) { + auto it = std::lower_bound(unsatisfied_interests_.begin(), + unsatisfied_interests_.end(), last_segment_); + if (it != unsatisfied_interests_.end()) { + suffix = *it; + } + unsatisfied_interests_.erase(unsatisfied_interests_.begin(), it); } + + getOutputStream() << " Received interest " + << interest.getName().getSuffix() + << ", starting production at " << suffix << end_mod_ + << std::endl; + getOutputStream() << unsatisfied_interests_.size() + << " interests still unsatisfied" << end_mod_ + << std::endl; + produceContentAsync(p, interest.getName(), suffix); + } + + /** + * @brief Register cache miss events + */ + void cacheMiss([[maybe_unused]] const ProducerSocket &p, + const Interest &interest) { + unsatisfied_interests_.push_back(interest.getName().getSuffix()); } - ret = producer_socket_->setSocketOption( - ProducerCallbacksOptions::CONTENT_PRODUCED, - (ProducerContentCallback)bind( - &Impl::onContentProduced, this, std::placeholders::_1, - std::placeholders::_2, std::placeholders::_3)); + /** + * @brief When content is produced, set cache miss callback so that we can + * register any cache miss happening after the production. + */ + void onContentProduced(ProducerSocket &p, + [[maybe_unused]] const std::error_code &err, + [[maybe_unused]] uint64_t bytes_written) { + p.setSocketOption(ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)bind( + &ProducerContext::asyncProcessInterest, this, + std::placeholders::_1, std::placeholders::_2)); + } - return ERROR_SUCCESS; + /** + * @brief Internal producer error. When this callback is triggered + * something important happened. Here we stop the program. + */ + void produceError(const std::error_code &err) noexcept override { + getOutputStream() << "Error from producer transport: " << err.message() + << std::endl; + parent_.stop(); + } + + // Members initialized in constructor + std::vector<ContentObject::Ptr> content_objects_; + + // Members initialized by in-class initializer + std::vector<uint32_t> unsatisfied_interests_; + std::uint32_t last_segment_{0}; + std::unique_ptr<ProducerSocket> producer_socket_{nullptr}; + std::uint16_t content_objects_index_{0}; + std::size_t payload_size_max_{0}; + }; + + public: + explicit Impl(const hiperf::ServerConfiguration &conf) : config_(conf) { +#ifndef _WIN32 + if (config_.interactive_) { + input_.assign(::dup(STDIN_FILENO)); + } +#endif + + std::memset(rtc_payload_.data(), 'X', rtc_payload_.size()); + } + + ~Impl() = default; + + int setup() { + int ret = ensureFlows(config_.name_, config_.parallel_flows_); + if (ret != ERROR_SUCCESS) { + return ret; + } + + producer_contexts_.reserve(config_.parallel_flows_); + for (uint32_t i = 0; i < config_.parallel_flows_; i++) { + auto &ctx = producer_contexts_.emplace_back(*this, i); + ret = ctx.setup(); + + if (ret) { + break; + } + } + + return ret; } void receiveStream() { socket_.async_receive_from( - asio::buffer(recv_buffer_.first, recv_buffer_.second), remote_, - [this](std::error_code ec, std::size_t length) { + asio::buffer(recv_buffer_.writableData(), recv_buffer_.capacity()), + remote_, [this](const std::error_code &ec, std::size_t length) { if (ec) return; - sendRTCContentFromStream(recv_buffer_.first, length); + sendRTCContentFromStream(recv_buffer_.writableData(), length); receiveStream(); }); } - void sendRTCContentFromStream(uint8_t *buff, std::size_t len) { - auto payload = - content_objects_[content_objects_index_++ & mask_]->getPayload(); + void sendRTCContentFromStream(const uint8_t *buff, std::size_t len) { // this is used to compute the data packet delay // Used only for performance evaluation // It requires clock synchronization between producer and consumer - uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>( - std::chrono::system_clock::now().time_since_epoch()) - .count(); - uint8_t *start = (uint8_t *)payload->writableData(); + auto now = utils::SystemTime::nowMs().count(); + + auto start = rtc_payload_.data(); std::memcpy(start, &now, sizeof(uint64_t)); std::memcpy(start + sizeof(uint64_t), buff, len); - producer_socket_->produceDatagram(flow_name_, start, - len + sizeof(uint64_t)); + + for (const auto &producer_context : producer_contexts_) { + producer_context.produceDatagram(start, len + sizeof(uint64_t)); + } } - void sendRTCContentObjectCallback(std::error_code ec) { + void sendRTCContentObjectCallback(const std::error_code &ec) { if (ec) return; rtc_timer_.expires_from_now( - configuration_.production_rate_.getMicrosecondsForPacket( - configuration_.payload_size_)); + config_.production_rate_.getMicrosecondsForPacket( + config_.payload_size_)); rtc_timer_.async_wait(std::bind(&Impl::sendRTCContentObjectCallback, this, std::placeholders::_1)); - auto payload = - content_objects_[content_objects_index_++ & mask_]->getPayload(); + + auto start = rtc_payload_.data(); // this is used to compute the data packet delay // Used only for performance evaluation // It requires clock synchronization between producer and consumer - uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>( - std::chrono::system_clock::now().time_since_epoch()) - .count(); - - std::memcpy(payload->writableData(), &now, sizeof(uint64_t)); + auto now = utils::SystemTime::nowMs().count(); + std::memcpy(start, &now, sizeof(uint64_t)); - producer_socket_->produceDatagram( - flow_name_, payload->data(), - payload->length() < 1400 ? payload->length() : 1400); + for (const auto &producer_context : producer_contexts_) { + producer_context.produceDatagram(start, config_.payload_size_); + } } - void sendRTCContentObjectCallbackWithTrace(std::error_code ec) { + void sendRTCContentObjectCallbackWithTrace(const std::error_code &ec) { if (ec) return; - auto payload = - content_objects_[content_objects_index_++ & mask_]->getPayload(); - - uint32_t packet_len = - configuration_.trace_[configuration_.trace_index_].size; + std::size_t packet_len = config_.trace_[config_.trace_index_].size; // this is used to compute the data packet delay // used only for performance evaluation // it requires clock synchronization between producer and consumer - uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>( - std::chrono::system_clock::now().time_since_epoch()) - .count(); - - std::memcpy(payload->writableData(), &now, sizeof(uint64_t)); + auto now = utils::SystemTime::nowMs().count(); + auto start = rtc_payload_.data(); + std::memcpy(start, &now, sizeof(uint64_t)); - if (packet_len > payload->length()) packet_len = payload->length(); - if (packet_len > 1400) packet_len = 1400; + if (packet_len > config_.payload_size_) { + packet_len = config_.payload_size_; + } - producer_socket_->produceDatagram(flow_name_, payload->data(), packet_len); + for (const auto &producer_context : producer_contexts_) { + producer_context.produceDatagram(start, packet_len); + } - uint32_t next_index = configuration_.trace_index_ + 1; + uint32_t next_index = config_.trace_index_ + 1; uint64_t schedule_next; - if (next_index < configuration_.trace_.size()) { - schedule_next = - configuration_.trace_[next_index].timestamp - - configuration_.trace_[configuration_.trace_index_].timestamp; + if (next_index < config_.trace_.size()) { + schedule_next = config_.trace_[next_index].timestamp - + config_.trace_[config_.trace_index_].timestamp; } else { // here we need to loop, schedule in a random time schedule_next = 1000; } - configuration_.trace_index_ = - (configuration_.trace_index_ + 1) % configuration_.trace_.size(); + config_.trace_index_ = (config_.trace_index_ + 1) % config_.trace_.size(); rtc_timer_.expires_from_now(std::chrono::microseconds(schedule_next)); rtc_timer_.async_wait( std::bind(&Impl::sendRTCContentObjectCallbackWithTrace, this, std::placeholders::_1)); } + int parseTraceFile() { + std::ifstream trace(config_.trace_file_); + if (trace.fail()) { + return -1; + } + std::string line; + while (std::getline(trace, line)) { + std::istringstream iss(line); + hiperf::packet_t packet; + iss >> packet.timestamp >> packet.size; + config_.trace_.push_back(packet); + } + return 0; + } + #ifndef _WIN32 void handleInput(const std::error_code &error, std::size_t length) { if (error) { - producer_socket_->stop(); - io_service_.stop(); + stop(); } if (rtc_running_) { - std::cout << "stop real time content production" << std::endl; + LoggerInfo() << "stop real time content production" << std::endl; rtc_running_ = false; rtc_timer_.cancel(); } else { - std::cout << "start real time content production" << std::endl; + LoggerInfo() << "start real time content production" << std::endl; rtc_running_ = true; rtc_timer_.expires_from_now( - configuration_.production_rate_.getMicrosecondsForPacket( - configuration_.payload_size_)); + config_.production_rate_.getMicrosecondsForPacket( + config_.payload_size_)); rtc_timer_.async_wait(std::bind(&Impl::sendRTCContentObjectCallback, this, std::placeholders::_1)); } @@ -401,46 +586,33 @@ class HIperfServer::Impl { } #endif - int parseTraceFile() { - std::ifstream trace(configuration_.trace_file_); - if (trace.fail()) { - return -1; - } - std::string line; - while (std::getline(trace, line)) { - std::istringstream iss(line); - hiperf::packet_t packet; - iss >> packet.timestamp >> packet.size; - configuration_.trace_.push_back(packet); + void stop() { + for (auto &producer_context : producer_contexts_) { + producer_context.stop(); } - return 0; + + io_service_.stop(); } int run() { - std::cerr << "Starting to serve consumers" << std::endl; - signals_.add(SIGINT); - signals_.async_wait([this](const std::error_code &, const int &) { - std::cout << "STOPPING!!" << std::endl; - producer_socket_->stop(); - io_service_.stop(); - }); + signals_.async_wait( + [this](const std::error_code &, const int &) { stop(); }); - if (configuration_.rtc_) { -#ifndef _WIN32 - if (configuration_.interactive_) { + if (config_.rtc_) { + if (config_.interactive_) { asio::async_read_until( input_, input_buffer_, '\n', std::bind(&Impl::handleInput, this, std::placeholders::_1, std::placeholders::_2)); - } else if (configuration_.trace_based_) { - std::cout << "trace-based mode enabled" << std::endl; - if (configuration_.trace_file_ == nullptr) { - std::cout << "cannot find the trace file" << std::endl; + } else if (config_.trace_based_) { + LoggerInfo() << "trace-based mode enabled" << std::endl; + if (config_.trace_file_ == nullptr) { + LoggerErr() << "cannot find the trace file" << std::endl; return ERROR_SETUP; } if (parseTraceFile() < 0) { - std::cout << "cannot parse the trace file" << std::endl; + LoggerErr() << "cannot parse the trace file" << std::endl; return ERROR_SETUP; } rtc_running_ = true; @@ -448,31 +620,26 @@ class HIperfServer::Impl { rtc_timer_.async_wait( std::bind(&Impl::sendRTCContentObjectCallbackWithTrace, this, std::placeholders::_1)); - } else if (configuration_.input_stream_mode_) { + } else if (config_.input_stream_mode_) { rtc_running_ = true; - // crate socket + // create socket remote_ = asio::ip::udp::endpoint( - asio::ip::address::from_string("127.0.0.1"), configuration_.port_); + asio::ip::address::from_string("127.0.0.1"), config_.port_); socket_.open(asio::ip::udp::v4()); socket_.bind(remote_); - recv_buffer_.first = (uint8_t *)malloc(1500); - recv_buffer_.second = 1500; receiveStream(); } else { rtc_running_ = true; rtc_timer_.expires_from_now( - configuration_.production_rate_.getMicrosecondsForPacket( - configuration_.payload_size_)); + config_.production_rate_.getMicrosecondsForPacket( + config_.payload_size_)); rtc_timer_.async_wait(std::bind(&Impl::sendRTCContentObjectCallback, this, std::placeholders::_1)); } -#else - rtc_timer_.expires_from_now( - configuration_.production_rate_.getMicrosecondsForPacket( - configuration_.payload_size_)); - rtc_timer_.async_wait(std::bind(&Impl::sendRTCContentObjectCallback, this, - std::placeholders::_1)); -#endif + } + + for (auto &producer_context : producer_contexts_) { + producer_context.run(); } io_service_.run(); @@ -480,34 +647,31 @@ class HIperfServer::Impl { return ERROR_SUCCESS; } + ServerConfiguration &getConfig() { return config_; } + private: - hiperf::ServerConfiguration configuration_; + // Variables initialized by the constructor. + ServerConfiguration config_; + + // Variable initialized in the in-class initializer list. asio::io_service io_service_; - asio::signal_set signals_; - asio::steady_timer rtc_timer_; - std::vector<uint32_t> unsatisfied_interests_; - std::vector<std::shared_ptr<ContentObject>> content_objects_; - std::uint16_t content_objects_index_; - std::uint16_t mask_; - std::uint32_t last_segment_; - std::uint32_t *ptr_last_segment_; - std::unique_ptr<ProducerSocket> producer_socket_; -#ifndef _WIN32 - asio::posix::stream_descriptor input_; + asio::signal_set signals_{io_service_}; + asio::steady_timer rtc_timer_{io_service_}; + asio::posix::stream_descriptor input_{io_service_}; + asio::ip::udp::socket socket_{io_service_}; + std::vector<ProducerContext> producer_contexts_; + ::utils::EventThread produce_thread_; asio::streambuf input_buffer_; - bool rtc_running_; - Name flow_name_; - asio::ip::udp::socket socket_; + bool rtc_running_{false}; asio::ip::udp::endpoint remote_; - std::pair<uint8_t *, std::size_t> recv_buffer_; -#endif + utils::MemBuf recv_buffer_{utils::MemBuf::CREATE, HIPERF_MTU}; + std::array<uint8_t, HIPERF_MTU> rtc_payload_; }; -HIperfServer::HIperfServer(const ServerConfiguration &conf) { - impl_ = new Impl(conf); -} +HIperfServer::HIperfServer(const ServerConfiguration &conf) + : impl_(std::make_unique<Impl>(conf)) {} -HIperfServer::~HIperfServer() { delete impl_; } +HIperfServer::~HIperfServer() = default; int HIperfServer::setup() { return impl_->setup(); } |