diff options
| -rw-r--r-- | libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc | 24 | ||||
| -rw-r--r-- | libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h | 4 | ||||
| -rw-r--r-- | libtransport/src/hicn/transport/interfaces/socket_producer.h | 8 | ||||
| -rw-r--r-- | pom.xml | 4 | ||||
| -rw-r--r-- | utils/src/hiperf.cc | 201 |
5 files changed, 186 insertions, 55 deletions
diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc index c07ca7989..5a432a99f 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc @@ -74,22 +74,22 @@ RTCProducerSocket::RTCProducerSocket() RTCProducerSocket::~RTCProducerSocket() {} -void RTCProducerSocket::registerName(Prefix &producer_namespace) { +void RTCProducerSocket::registerPrefix(const Prefix &producer_namespace) { ProducerSocket::registerPrefix(producer_namespace); flowName_ = producer_namespace.getName(); - - if (flowName_.getType() == HNT_CONTIGUOUS_V4 || - flowName_.getType() == HNT_IOV_V4) { - headerSize_ = sizeof(hicn_v6_hdr_t::ip); - } else if (flowName_.getType() == HNT_CONTIGUOUS_V6 || - flowName_.getType() == HNT_IOV_V6) { - headerSize_ = sizeof(hicn_v4_hdr_t::ip); - } else { - throw errors::RuntimeException("Unknown name format."); + auto family = flowName_.getAddressFamily(); + + switch (family) { + case AF_INET6: + headerSize_ = Packet::getHeaderSizeFromFormat(HF_INET6_TCP); + break; + case AF_INET: + headerSize_ = Packet::getHeaderSizeFromFormat(HF_INET_TCP); + break; + default: + throw errors::RuntimeException("Unknown name format."); } - - headerSize_ += TCP_HEADER_SIZE; } void RTCProducerSocket::updateStats(uint32_t packet_size) { diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h index f1bcaa9e8..bc54be4bb 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h @@ -33,9 +33,9 @@ class RTCProducerSocket : public ProducerSocket { ~RTCProducerSocket(); - void registerName(Prefix &producer_namespace); + void registerPrefix(const Prefix &producer_namespace) override; - void produce(const uint8_t *buffer, size_t buffer_size); + void produce(const uint8_t *buffer, size_t buffer_size) override; void onInterest(Interest::Ptr &&interest) override; diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h index d3738dc59..6ba5671cc 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h @@ -51,13 +51,19 @@ class ProducerSocket : public Socket<BasePortal>, void produce(ContentObject &content_object); + virtual void produce(const uint8_t *buffer, size_t buffer_size) { + // This API is meant to be used just with the RTC producer. + // Here it cannot be used since no name for the content is specified. + throw errors::NotImplementedException(); + } + void asyncProduce(const Name &suffix, const uint8_t *buf, size_t buffer_size); void asyncProduce(const Name &suffix, ContentBuffer &&output_buffer); void asyncProduce(ContentObject &content_object); - void registerPrefix(const Prefix &producer_namespace); + virtual void registerPrefix(const Prefix &producer_namespace); void serveForever(); @@ -12,7 +12,7 @@ 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. ---><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> +<!-- --><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <parent> <groupId>io.fd.hicn.common</groupId> @@ -40,4 +40,4 @@ <module>hicn-plugin</module> <module>libtransport</module> </modules> -</project>
\ No newline at end of file +</project> -->
\ No newline at end of file diff --git a/utils/src/hiperf.cc b/utils/src/hiperf.cc index 860e7a000..e10907ccc 100644 --- a/utils/src/hiperf.cc +++ b/utils/src/hiperf.cc @@ -13,6 +13,7 @@ * limitations under the License. */ +#include <hicn/transport/interfaces/rtc_socket_producer.h> #include <hicn/transport/interfaces/socket_consumer.h> #include <hicn/transport/interfaces/socket_producer.h> #ifndef _WIN32 @@ -71,6 +72,41 @@ struct ClientConfiguration { bool rtc_; }; +class Rate { + public: + Rate() : rate_kbps_(0) {} + + Rate(const std::string &rate) { + std::size_t found = rate.find("kbps"); + if (found != std::string::npos) { + rate_kbps_ = std::stof(rate.substr(0, found)); + } else { + throw std::runtime_error("Format " + rate + " not correct"); + } + } + + Rate(const Rate &other) : rate_kbps_(other.rate_kbps_) {} + + Rate &operator=(const std::string &rate) { + std::size_t found = rate.find("kbps"); + if (found != std::string::npos) { + rate_kbps_ = std::stof(rate.substr(0, found)); + } else { + throw std::runtime_error("Format " + rate + " not correct"); + } + + return *this; + } + + std::chrono::microseconds getMicrosecondsForPacket(std::size_t packet_size) { + return std::chrono::microseconds( + packet_size * long(std::round(1000.0 * 8.0 / rate_kbps_))); + } + + private: + float rate_kbps_; +}; + struct ServerConfiguration { ServerConfiguration() : name("b001::abcd/64"), @@ -84,7 +120,10 @@ struct ServerConfiguration { hash_algorithm(HashAlgorithm::SHA_256), keystore_name("/tmp/rsa_crypto_material.p12"), keystore_password("cisco"), - multiphase_produce_(false) {} + multiphase_produce_(false), + rtc_(false), + production_rate_(std::string("2048kbps")), + payload_size_(1400) {} Prefix name; bool virtual_producer; @@ -98,6 +137,9 @@ struct ServerConfiguration { std::string keystore_name; std::string keystore_password; bool multiphase_produce_; + bool rtc_; + Rate production_rate_; + std::size_t payload_size_; }; class HIperfClient { @@ -338,10 +380,11 @@ class HIperfServer { HIperfServer(ServerConfiguration &conf) : configuration_(conf), signals_(io_service_, SIGINT), + rtc_timer_(io_service_), 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(configuration_.payload_size_, 'X'); std::cout << "Producing contents under name " << conf.name.getName() << std::endl; @@ -357,16 +400,6 @@ class HIperfServer { void processInterest(ProducerSocket &p, const Interest &interest) { content_objects_[content_objects_index_ & mask_]->setName( interest.getName()); - - // if (final_chunk_number_ > 0 && interest.getName().getSuffix() == 0) { - // auto name = interest.getName(); - // manifest_ = std::make_shared<ContentObjectManifest>(name); - // // manifest_->setFinalChunkNumber(final_chunk_number_); - // manifest_->encode(); - // p.produce(*manifest_); - // return; - // } - producer_socket_->produce( *content_objects_[content_objects_index_++ & mask_]); } @@ -414,7 +447,11 @@ class HIperfServer { int setup() { int ret; - producer_socket_ = std::make_unique<ProducerSocket>(); + if (configuration_.rtc_) { + producer_socket_ = std::make_unique<RTCProducerSocket>(); + } else { + producer_socket_ = std::make_unique<ProducerSocket>(); + } if (configuration_.sign) { auto identity = setProducerIdentity(configuration_.keystore_name, @@ -431,6 +468,13 @@ class HIperfServer { producer_socket_->registerPrefix(configuration_.name); producer_socket_->connect(); + if (configuration_.rtc_) { + std::cout << "Running RTC producer: all other options (with the " + "exception of the bitrate) will be ignored." + << std::endl; + return ERROR_SUCCESS; + } + if (!configuration_.virtual_producer) { if (producer_socket_->setSocketOption( GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, @@ -485,6 +529,20 @@ class HIperfServer { return ERROR_SUCCESS; } + void sendRTCContentObjectCallback(std::error_code ec) { + if (!ec) { + auto payload = + content_objects_[content_objects_index_++ & mask_]->getPayload(); + producer_socket_->produce(payload.data(), payload.length()); + rtc_timer_.expires_from_now( + configuration_.production_rate_.getMicrosecondsForPacket( + configuration_.payload_size_)); + rtc_timer_.async_wait( + std::bind(&HIperfServer::sendRTCContentObjectCallback, this, + std::placeholders::_1)); + } + } + int run() { std::cerr << "Starting to serve consumers" << std::endl; @@ -494,6 +552,15 @@ class HIperfServer { io_service_.stop(); }); + if (configuration_.rtc_) { + rtc_timer_.expires_from_now( + configuration_.production_rate_.getMicrosecondsForPacket( + configuration_.payload_size_)); + rtc_timer_.async_wait( + std::bind(&HIperfServer::sendRTCContentObjectCallback, this, + std::placeholders::_1)); + } + io_service_.run(); return ERROR_SUCCESS; @@ -503,6 +570,7 @@ class HIperfServer { ServerConfiguration configuration_; asio::io_service io_service_; asio::signal_set signals_; + asio::steady_timer rtc_timer_; std::vector<std::shared_ptr<ContentObject>> content_objects_; std::uint16_t content_objects_index_; std::uint16_t mask_; @@ -519,10 +587,17 @@ void usage() { #ifndef _WIN32 std::cerr << "-D\t\t\t\t\t" << "Run as a daemon" << std::endl; + std::cerr << "-R\t\t\t\t\t" + << "Run RTC protocol (client or server)" << std::endl; + std::cerr << "-f\t<filename>\t\t\t" + << "Log file" << std::endl; #endif std::cerr << std::endl; std::cerr << "Server specific:" << std::endl; - std::cerr << "-s\t<content_size>\t\t\tSize of the content to publish" + std::cerr << "-A\t<download_size>\t\t\tSize of the content to publish. This " + "is not the size of the packet (see -s for it)." + << std::endl; + std::cerr << "-s\t<packet_size>\t\t\tSize of the payload of each data packet." << std::endl; std::cerr << "-r\t\t\t\t\t" << "Produce real content of content_size bytes" << std::endl; @@ -542,6 +617,15 @@ void usage() { << std::endl; std::cerr << "-p\t<password>\t\t\t" << "Password for p12 keystore" << std::endl; + std::cerr << "-x\t\t\t\t\t" + << "Produce a content of <download_size>, then after downloading " + "it produce a new content of" + << std::endl + << "\t\t\t\t\t<download_size> without resetting " + "the suffix to 0." + << std::endl; + std::cerr << "-B\t<bitrate>\t\t\t" + << "Bitrate for RTC producer, to be used with the -R option." << std::endl; std::cerr << std::endl; std::cerr << "Client specific:" << std::endl; std::cerr << "-b\t<beta_parameter>\t\t" @@ -561,6 +645,9 @@ void usage() { "to be used for verifying the " "origin of the packets received" << std::endl; + std::cerr << "-i\t<stats_interval>\t\t" + << "Show the statistics every <stats_interval> milliseconds." + << std::endl; std::cout << "-v\t\t\t\t\t" << "Enable verification of received data" << std::endl; } @@ -588,84 +675,109 @@ int main(int argc, char *argv[]) { int opt; #ifndef _WIN32 - while ((opt = getopt(argc, argv, "DSCf:b:d:W:RMc:vs:rmlk:y:p:hi:x")) != -1) { + while ((opt = getopt(argc, argv, "DSCf:b:d:W:RMc:vA:s:rmlk:y:p:hi:xB:")) != + -1) { switch (opt) { // Common - case 'D': + case 'D': { daemon = true; break; + } #else - while ((opt = getopt(argc, argv, "SCf:b:d:W:RMc:vs:rmlk:y:p:hi:x")) != -1) { + while ((opt = getopt(argc, argv, "SCf:b:d:W:RMc:vA:s:rmlk:y:p:hi:xB:")) != + -1) { switch (opt) { #endif - case 'f': + case 'f': { log_file = optarg; break; + } + case 'R': { + client_configuration.rtc_ = true; + server_configuration.rtc_ = true; + break; + } // Server or Client - case 'S': + case 'S': { role -= 1; break; - case 'C': + } + case 'C': { role += 1; break; + } // Client specifc - case 'b': + case 'b': { client_configuration.beta = std::stod(optarg); options = 1; break; - case 'd': + } + case 'd': { client_configuration.drop_factor = std::stod(optarg); options = 1; break; - case 'W': + } + case 'W': { client_configuration.window = std::stod(optarg); options = 1; break; - case 'M': + } + case 'M': { client_configuration.virtual_download = false; options = 1; break; - case 'c': + } + case 'c': { client_configuration.producer_certificate = std::string(optarg); options = 1; break; - case 'v': + } + case 'v': { client_configuration.verify = true; options = 1; break; - case 'i': + } + case 'i': { client_configuration.report_interval_milliseconds_ = std::stoul(optarg); options = 1; break; - case 'R': - client_configuration.rtc_ = true; - break; + } // Server specific - case 's': + case 'A': { server_configuration.download_size = std::stoul(optarg); options = -1; break; - case 'r': + } + case 's': { + server_configuration.payload_size_ = std::stoul(optarg); + options = -1; + break; + } + case 'r': { server_configuration.virtual_producer = false; options = -1; break; - case 'm': + } + case 'm': { server_configuration.manifest = true; options = -1; break; - case 'l': + } + case 'l': { server_configuration.live_production = true; options = -1; break; - case 'k': + } + case 'k': { server_configuration.keystore_name = std::string(optarg); server_configuration.sign = true; options = -1; break; - case 'y': + } + case 'y': { if (strncasecmp(optarg, "sha256", 6) == 0) { server_configuration.hash_algorithm = HashAlgorithm::SHA_256; } else if (strncasecmp(optarg, "sha512", 6) == 0) { @@ -678,14 +790,27 @@ int main(int argc, char *argv[]) { } options = -1; break; - case 'p': + } + case 'p': { server_configuration.keystore_password = std::string(optarg); options = -1; break; - case 'x': + } + case 'x': { server_configuration.multiphase_produce_ = true; options = -1; break; + } + case 'B': { + auto str = std::string(optarg); + std::transform(str.begin(), str.end(), str.begin(), ::tolower); + std::cout << "---------------------------------------------------------" + "---------------------->" + << str << std::endl; + server_configuration.production_rate_ = str; + options = -1; + break; + } case 'h': default: usage(); |