diff options
Diffstat (limited to 'utils/src')
| -rwxr-xr-x | utils/src/hiperf.cc | 724 | ||||
| -rwxr-xr-x | utils/src/ping_client.cc | 428 | ||||
| -rwxr-xr-x | utils/src/ping_server.cc | 300 |
3 files changed, 1452 insertions, 0 deletions
diff --git a/utils/src/hiperf.cc b/utils/src/hiperf.cc new file mode 100755 index 000000000..d8953b36a --- /dev/null +++ b/utils/src/hiperf.cc @@ -0,0 +1,724 @@ +/* + * 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/interfaces/socket_producer.h> +#include <hicn/transport/utils/daemonizator.h> +#include <hicn/transport/utils/literals.h> + +#include <fstream> +#include <iomanip> + +#ifdef __linux__ +#include <mcheck.h> +#endif + +namespace transport { + +namespace interface { + +#define ERROR_SUCCESS 0 +#define ERROR_SETUP -5 + +using CryptoSuite = utils::CryptoSuite; +using Identity = utils::Identity; + +struct ClientConfiguration { + ClientConfiguration() + : name("b001::abcd", 0), + verify(false), + beta(-1.f), + drop_factor(-1.f), + window(-1), + virtual_download(true), + producer_certificate("/tmp/rsa_certificate.pem"), + receive_buffer(std::make_shared<utils::SharableVector<uint8_t>>()), + download_size(0), + report_interval_milliseconds_(1000), + rtc_(false) {} + + Name name; + bool verify; + double beta; + double drop_factor; + double window; + bool virtual_download; + std::string producer_certificate; + std::shared_ptr<utils::SharableVector<uint8_t>> receive_buffer; + std::size_t download_size; + std::uint32_t report_interval_milliseconds_; + TransportProtocolAlgorithms transport_protocol_; + bool rtc_; +}; + +struct ServerConfiguration { + ServerConfiguration() + : name("b001::abcd/64"), + virtual_producer(true), + manifest(false), + live_production(false), + sign(false), + content_lifetime(600000000_U32), + content_object_size(1440), + download_size(20 * 1024 * 1024), + hash_algorithm(HashAlgorithm::SHA_256), + keystore_name("/tmp/rsa_crypto_material.p12"), + keystore_password("cisco"), + multiphase_produce_(false) {} + + Prefix name; + bool virtual_producer; + bool manifest; + bool live_production; + bool sign; + std::uint32_t content_lifetime; + std::uint16_t content_object_size; + std::uint32_t download_size; + HashAlgorithm hash_algorithm; + std::string keystore_name; + std::string keystore_password; + bool multiphase_produce_; +}; + +class HIperfClient { + typedef std::chrono::time_point<std::chrono::steady_clock> Time; + typedef std::chrono::microseconds TimeDuration; + + public: + HIperfClient(const ClientConfiguration &conf) + : configuration_(conf), + total_duration_milliseconds_(0), + old_bytes_value_(0) {} + + 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_); + long usec = dt.count(); + + std::cout << "Content retrieved. Size: " << bytes_transferred << " [Bytes]" + << std::endl; + + std::cerr << "Elapsed Time: " << usec / 1000000.0 << " seconds -- " + << (bytes_transferred * 8) * 1.0 / usec * 1.0 << " [Mbps]" + << std::endl; + } + + bool verifyData(ConsumerSocket &c, const ContentObject &contentObject) { + if (contentObject.getPayloadType() == PayloadType::CONTENT_OBJECT) { + std::cout << "VERIFY CONTENT" << std::endl; + } else if (contentObject.getPayloadType() == PayloadType::MANIFEST) { + std::cout << "VERIFY MANIFEST" << std::endl; + } + + return true; + } + + void processLeavingInterest(ConsumerSocket &c, const Interest &interest) { + // std::cout << "LEAVES " << interest.getName().toUri() << std::endl; + } + + void handleTimerExpiration(ConsumerSocket &c, std::size_t byte_count, + std::chrono::milliseconds &exact_duration, + float c_window, uint32_t retransmissions, + uint32_t average_rtt) { + const char separator = ' '; + const int width = 20; + + std::stringstream interval; + interval << total_duration_milliseconds_ / 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 + << std::setfill(separator) << "[MBytes]"; + + std::stringstream bandwidth; + bandwidth << ((byte_count - old_bytes_value_) * 8) / + (exact_duration.count()) / 1000.0 + << std::setfill(separator) << "[Mbps]"; + + std::stringstream window; + window << c_window << std::setfill(separator) << "[Interest]"; + + std::stringstream avg_rtt; + avg_rtt << average_rtt << std::setfill(separator) << "[us]"; + + std::cout << std::left << std::setw(width) << "Interval"; + std::cout << std::left << std::setw(width) << "Transfer"; + std::cout << std::left << std::setw(width) << "Bandwidth"; + std::cout << std::left << std::setw(width) << "Retr"; + std::cout << std::left << std::setw(width) << "Cwnd"; + std::cout << std::left << std::setw(width) << "AvgRtt" << std::endl; + + 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) << window.str(); + std::cout << std::left << std::setw(width) << avg_rtt.str() << std::endl; + std::cout << std::endl; + + total_duration_milliseconds_ += exact_duration.count(); + old_bytes_value_ = byte_count; + } + + int setup() { + int ret; + + // Set the transport algorithm + TransportProtocolAlgorithms transport_protocol; + + if (configuration_.rtc_) { + transport_protocol = RTC; + } else if (configuration_.window < 0) { + transport_protocol = RAAQM; + } else { + transport_protocol = CBR; + } + + consumer_socket_ = std::make_unique<ConsumerSocket>(transport_protocol); + +#if defined(DEBUG) && defined(__linux__) + std::shared_ptr<transport::BasePortal> portal; + consumer_socket_->getSocketOption(GeneralTransportOptions::PORTAL, portal); + signals_ = + std::make_unique<asio::signal_set>(portal->getIoService(), SIGUSR1); + signals_->async_wait([this](const std::error_code &, const int &) { + std::cout << "Signal SIGUSR1!" << std::endl; + mtrace(); + }); +#endif + + if (consumer_socket_->setSocketOption(CURRENT_WINDOW_SIZE, + configuration_.window) == + SOCKET_OPTION_NOT_SET) { + std::cerr << "ERROR -- Impossible to set the size of the window." + << std::endl; + return ERROR_SETUP; + } + + if (transport_protocol == RAAQM && configuration_.beta != -1.f) { + if (consumer_socket_->setSocketOption(RaaqmTransportOptions::BETA_VALUE, + configuration_.beta) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } + + if (transport_protocol == RAAQM && configuration_.drop_factor != -1.f) { + if (consumer_socket_->setSocketOption(RaaqmTransportOptions::DROP_FACTOR, + configuration_.drop_factor) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } + + if (consumer_socket_->setSocketOption(OtherOptions::VIRTUAL_DOWNLOAD, + false) == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (configuration_.verify) { + if (consumer_socket_->setSocketOption( + GeneralTransportOptions::CERTIFICATE, + configuration_.producer_certificate) == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } + + if (consumer_socket_->setSocketOption( + GeneralTransportOptions::VERIFY_SIGNATURE, configuration_.verify) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + ret = consumer_socket_->setSocketOption( + ConsumerCallbacksOptions::INTEREST_OUTPUT, + (ConsumerInterestCallback)std::bind( + &HIperfClient::processLeavingInterest, this, std::placeholders::_1, + std::placeholders::_2)); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + ret = consumer_socket_->setSocketOption( + ConsumerCallbacksOptions::CONTENT_RETRIEVED, + (ConsumerContentCallback)std::bind( + &HIperfClient::processPayload, this, std::placeholders::_1, + std::placeholders::_2, std::placeholders::_3)); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + ret = consumer_socket_->setSocketOption( + ConsumerCallbacksOptions::TIMER_EXPIRES, + (ConsumerTimerCallback)std::bind( + &HIperfClient::handleTimerExpiration, this, std::placeholders::_1, + std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, + std::placeholders::_5, std::placeholders::_6)); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (consumer_socket_->setSocketOption( + GeneralTransportOptions::TIMER_INTERVAL, + configuration_.report_interval_milliseconds_) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + consumer_socket_->connect(); + + return ERROR_SUCCESS; + } + + int run() { + std::cout << "Starting download of " << configuration_.name << std::endl; + + do { + t1_ = std::chrono::steady_clock::now(); + consumer_socket_->consume(configuration_.name, + *configuration_.receive_buffer); + } while (configuration_.virtual_download); + + return ERROR_SUCCESS; + } + + private: + ClientConfiguration configuration_; + std::unique_ptr<ConsumerSocket> consumer_socket_; + Time t1_; + uint32_t total_duration_milliseconds_; + uint64_t old_bytes_value_; + // std::unique_ptr<asio::signal_set> signals_; +}; + +class HIperfServer { + const std::size_t log2_content_object_buffer_size = 8; + + public: + HIperfServer(ServerConfiguration &conf) + : configuration_(conf), + // signals_(io_service_, SIGINT, SIGQUIT), + content_objects_((1 << log2_content_object_buffer_size)), + content_objects_index_(0), + mask_((1 << log2_content_object_buffer_size) - 1) { + // signals_.async_wait([this] (const std::error_code&, const int&) + // {std::cout << "STOPPING!!" << std::endl; io_service_.stop();}); + + std::string buffer(1440, 'X'); + + std::cout << "Producing contents under name " << conf.name.getName() + << std::endl; + + for (int i = 0; i < (1 << log2_content_object_buffer_size); i++) { + content_objects_[i] = std::make_shared<ContentObject>( + conf.name.getName(), HF_INET6_TCP, (const uint8_t *)buffer.data(), + buffer.size()); + content_objects_[i]->setLifetime( + default_values::content_object_expiry_time); + } + } + + 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_]); + } + + void processInterest2(ProducerSocket &p, const Interest &interest) { + producer_socket_->setSocketOption(ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)VOID_HANDLER); + producer_socket_->setSocketOption( + GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, 5000_U32); + produceContent(interest.getName().getSuffix()); + producer_socket_->setSocketOption( + ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)bind(&HIperfServer::processInterest2, this, + std::placeholders::_1, + std::placeholders::_2)); + } + + void produceContent(uint32_t suffix) { + core::Name name = configuration_.name.getName(); + + std::string content(configuration_.download_size, '?'); + uint32_t total; + + total = producer_socket_->produce( + name, reinterpret_cast<const uint8_t *>(content.data()), content.size(), + !configuration_.multiphase_produce_); + + std::cout << "Written " << total << "pieces of data in output buffer" + << std::endl; + } + + 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); + } else { + return utils::Identity(keystore_name, keystore_password, + CryptoSuite::RSA_SHA256, 1024, 365, + "producer-test"); + } + } + + int setup() { + int ret; + + producer_socket_ = std::make_unique<ProducerSocket>(); + + if (configuration_.sign) { + Identity identity = setProducerIdentity(configuration_.keystore_name, + configuration_.keystore_password, + configuration_.hash_algorithm); + + if (producer_socket_->setSocketOption(GeneralTransportOptions::IDENTITY, + identity) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } + + producer_socket_->registerPrefix(configuration_.name); + + if (!configuration_.virtual_producer) { + if (producer_socket_->setSocketOption( + GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME, + configuration_.content_lifetime) == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (producer_socket_->setSocketOption( + GeneralTransportOptions::MAKE_MANIFEST, + configuration_.manifest) == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (producer_socket_->setSocketOption( + GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 200000U) == + SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + + if (!configuration_.live_production) { + produceContent(0); + } else { + ret = producer_socket_->setSocketOption( + ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)bind(&HIperfServer::processInterest2, + 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(&HIperfServer::processInterest, this, + std::placeholders::_1, + std::placeholders::_2)); + + if (ret == SOCKET_OPTION_NOT_SET) { + return ERROR_SETUP; + } + } + + producer_socket_->connect(); + + return ERROR_SUCCESS; + } + + int run() { + std::cerr << "Starting to serve consumers" << std::endl; + producer_socket_->serveForever(); + + return ERROR_SUCCESS; + } + + private: + ServerConfiguration configuration_; + std::unique_ptr<ProducerSocket> producer_socket_; + // asio::signal_set signals_; + std::vector<std::shared_ptr<ContentObject>> content_objects_; + std::uint16_t content_objects_index_; + std::uint16_t mask_; +}; + +void usage() { + std::cerr << std::endl; + std::cerr << "HIPERF - A tool for performing network throughput " + "measurements with hICN" + << std::endl; + std::cerr << "usage: hiperf [-S|-C] [options] [prefix|name]" << std::endl; + std::cerr << "Server or Client:" << std::endl; + std::cerr << "-D\t\t\t\t\t" + << "Run as a daemon" << std::endl; + 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::endl; + std::cerr << "-r\t\t\t\t\t" + << "Produce real content of content_size bytes" << std::endl; + std::cerr << "-m\t\t\t\t\t" + << "Produce transport manifest" << std::endl; + std::cerr << "-l\t\t\t\t\t" + << "Start producing content upon the reception of the " + "first interest" + << std::endl; + std::cerr << "-k\t<keystore_path>\t\t\t" + << "Path of p12 file containing the " + "crypto material used for signing the packets" + << std::endl; + std::cerr << "-y\t<hash_algorithm>\t\t" + << "Use the selected hash algorithm for " + "calculating manifest digests" + << std::endl; + std::cerr << "-p\t<password>\t\t\t" + << "Password for p12 keystore" << std::endl; + std::cerr << std::endl; + std::cerr << "Client specific:" << std::endl; + std::cerr << "-b\t<beta_parameter>\t\t" + << "RAAQM beta parameter" << std::endl; + std::cerr << "-d\t<drop_factor_parameter>\t\t" + << "RAAQM drop factor " + "parameter" + << std::endl; + std::cerr << "-W\t<window_size>\t\t\t" + << "Use a fixed congestion window " + "for retrieving the data." + << std::endl; + std::cerr << "-c\t<certificate_path>\t\t" + << "Path of the producer certificate " + "to be used for verifying the " + "origin of the packets received" + << std::endl; + std::cout << "-v\t\t\t\t\t" + << "Enable verification of received data" << std::endl; +} + +int main(int argc, char *argv[]) { + // Common + bool daemon = false; + + // -1 server, 0 undefined, 1 client + int role = 0; + int options = 0; + + char *log_file = nullptr; + + // Consumer + ClientConfiguration client_configuration; + + // Producer + ServerConfiguration server_configuration; + + int opt; + while ((opt = getopt(argc, argv, "DSCf:b:d:W:c:vs:rmlk:y:p:hi:x")) != -1) { + switch (opt) { + // Common + case 'D': + daemon = true; + break; + case 'f': + log_file = optarg; + break; + + // Server or Client + case 'S': + role -= 1; + break; + case 'C': + role += 1; + break; + + // Client specifc + case 'b': + client_configuration.beta = std::stod(optarg); + options = 1; + break; + case 'd': + client_configuration.drop_factor = std::stod(optarg); + options = 1; + break; + case 'W': + client_configuration.window = std::stod(optarg); + options = 1; + break; + case 'c': + client_configuration.producer_certificate = std::string(optarg); + options = 1; + break; + case 'v': + client_configuration.verify = true; + options = 1; + break; + case 'i': + client_configuration.report_interval_milliseconds_ = std::stoul(optarg); + options = 1; + break; + case 'R': + client_configuration.rtc_ = true; + break; + + // Server specific + case 's': + server_configuration.download_size = std::stoul(optarg); + options = -1; + break; + case 'r': + server_configuration.virtual_producer = false; + options = -1; + break; + case 'm': + server_configuration.manifest = true; + options = -1; + break; + case 'l': + server_configuration.live_production = true; + options = -1; + break; + case 'k': + server_configuration.keystore_name = std::string(optarg); + server_configuration.sign = true; + options = -1; + break; + case 'y': + if (strncasecmp(optarg, "sha256", 6) == 0) { + server_configuration.hash_algorithm = HashAlgorithm::SHA_256; + } else if (strncasecmp(optarg, "sha512", 6) == 0) { + server_configuration.hash_algorithm = HashAlgorithm::SHA_512; + } else if (strncasecmp(optarg, "crc32", 5) == 0) { + server_configuration.hash_algorithm = HashAlgorithm::CRC32C; + } else { + std::cerr << "Ignored unknown hash algorithm. Using SHA 256." + << std::endl; + } + options = -1; + break; + case 'p': + server_configuration.keystore_password = std::string(optarg); + options = -1; + break; + case 'x': + server_configuration.multiphase_produce_ = true; + options = -1; + break; + case 'h': + default: + usage(); + return EXIT_FAILURE; + } + } + + if (options > 0 && role < 0) { + std::cerr << "Client options cannot be used when using the " + "software in server mode" + << std::endl; + usage(); + return EXIT_FAILURE; + + } else if (options < 0 && role > 0) { + std::cerr << "Server options cannot be used when using the " + "software in client mode" + << std::endl; + usage(); + return EXIT_FAILURE; + } else if (!role) { + std::cerr << "Please specify if running hiperf as client " + "or server." + << std::endl; + usage(); + return EXIT_FAILURE; + } + + if (argv[optind] == 0) { + std::cerr << "Please specify the name/prefix to use." << std::endl; + usage(); + return EXIT_FAILURE; + } else { + if (role > 0) { + client_configuration.name = Name(argv[optind]); + } else { + server_configuration.name = Prefix(argv[optind]); + } + } + + if (log_file) { + int fd = open(log_file, O_WRONLY | O_APPEND | O_CREAT, S_IWUSR | S_IRUSR); + dup2(fd, STDOUT_FILENO); + dup2(STDOUT_FILENO, STDERR_FILENO); + close(fd); + } + + if (daemon) { + utils::Daemonizator::daemonize(false); + } + + if (role > 0) { + HIperfClient c(client_configuration); + if (c.setup() != ERROR_SETUP) { + c.run(); + } + } else if (role < 0) { + HIperfServer s(server_configuration); + if (s.setup() != ERROR_SETUP) { + s.run(); + } + } else { + usage(); + return EXIT_FAILURE; + } + + std::cout << "Bye bye" << std::endl; + + return 0; +} + +} // end namespace interface + +} // end namespace transport + +int main(int argc, char *argv[]) { + return transport::interface::main(argc, argv); +} diff --git a/utils/src/ping_client.cc b/utils/src/ping_client.cc new file mode 100755 index 000000000..178bd8bac --- /dev/null +++ b/utils/src/ping_client.cc @@ -0,0 +1,428 @@ +/* + * 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.h> +#include <hicn/transport/utils/verifier.h> + +#include <asio/steady_timer.hpp> +#include <chrono> +#include <map> + +#define SYN_STATE 1 +#define ACK_STATE 2 + +namespace transport { + +namespace core { + +namespace ping { + +typedef std::map<uint64_t, uint64_t> SendTimeMap; +typedef utils::Verifier Verifier; + +class Configuration { + public: + uint64_t interestLifetime_; + uint64_t pingInterval_; + uint64_t maxPing_; + uint64_t first_suffix_; + std::string name_; + std::string certificate_; + uint16_t srcPort_; + uint16_t dstPort_; + bool verbose_; + bool dump_; + bool jump_; + bool open_; + bool always_syn_; + bool always_ack_; + bool quiet_; + uint32_t jump_freq_; + uint32_t jump_size_; + uint8_t ttl_; + + Configuration() { + interestLifetime_ = 500; // ms + pingInterval_ = 1000000; // us + maxPing_ = 10; // number of interests + first_suffix_ = 0; + name_ = "b001::1"; // string + srcPort_ = 9695; + dstPort_ = 8080; + verbose_ = false; + dump_ = false; + jump_ = false; + open_ = false; + always_syn_ = false; + always_ack_ = false; + quiet_ = false; + jump_freq_ = 0; + jump_size_ = 0; + ttl_ = 64; + } +}; + +class Client : interface::BasePortal::ConsumerCallback { + public: + Client(Configuration *c) : portal_() { + // Let the main thread to catch SIGINT and SIGQUIT + // asio::signal_set signals(io_service, SIGINT, SIGQUIT); + // signals.async_wait(std::bind(&Client::afterSignal, this)); + + portal_.connect(); + portal_.setConsumerCallback(this); + timer_.reset(new asio::steady_timer(portal_.getIoService())); + config_ = c; + sequence_number_ = config_->first_suffix_; + last_jump_ = 0; + processed_ = 0; + state_ = SYN_STATE; + sent_ = 0; + received_ = 0; + timedout_ = 0; + if (!c->certificate_.empty()) { + key_id_ = verifier_.addKeyFromCertificate(c->certificate_); + } + } + + void ping() { + std::cout << "start ping" << std::endl; + doPing(); + portal_.runEventsLoop(); + } + + void onContentObject(Interest::Ptr &&interest, + ContentObject::Ptr &&object) override { + uint64_t rtt = 0; + + if (!config_->certificate_.empty()) { + auto t0 = std::chrono::steady_clock::now(); + if (verifier_.verify(*object)) { + auto t1 = std::chrono::steady_clock::now(); + auto dt = + std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0); + std::cout << "Verification time: " << dt.count() << std::endl; + std::cout << "<<<<<< Signature OK!!!" << std::endl; + } else { + std::cout << "<<<<<< Signature verification failed!" << std::endl; + } + } + + auto it = send_timestamps_.find(interest->getName().getSuffix()); + if (it != send_timestamps_.end()) { + rtt = std::chrono::duration_cast<std::chrono::microseconds>( + std::chrono::steady_clock::now().time_since_epoch()) + .count() - + it->second; + send_timestamps_.erase(it); + } + + if (config_->verbose_) { + std::cout << "<<< recevied object. " << std::endl; + std::cout << "<<< interest name: " << interest->getName() + << " src port: " << interest->getSrcPort() + << " dst port: " << interest->getDstPort() + << " flags: " << interest->printFlags() << std::endl; + std::cout << "<<< object name: " << object->getName() + << " src port: " << object->getSrcPort() + << " dst port: " << object->getDstPort() + << " flags: " << object->printFlags() << " path label " + << object->getPathLabel() << " (" + << (object->getPathLabel() >> 24) << ")" + << " TTL: " << (int)object->getTTL() << std::endl; + } else if (!config_->quiet_) { + std::cout << "<<< received object. " << std::endl; + std::cout << "<<< round trip: " << rtt << " [us]" << std::endl; + std::cout << "<<< interest name: " << interest->getName() << std::endl; + std::cout << "<<< object name: " << object->getName() << std::endl; + std::cout << "<<< content object size: " + << object->payloadSize() + object->headerSize() << " [bytes]" + << std::endl; + } + + if (config_->dump_) { + std::cout << "----- interest dump -----" << std::endl; + interest->dump(); + std::cout << "-------------------------" << std::endl; + std::cout << "----- object dump -------" << std::endl; + object->dump(); + std::cout << "-------------------------" << std::endl; + } + + if (!config_->quiet_) std::cout << std::endl; + + if (!config_->always_syn_) { + if (object->testSyn() && object->testAck() && state_ == SYN_STATE) { + state_ = ACK_STATE; + } + } + + received_++; + processed_++; + if (processed_ >= config_->maxPing_) { + afterSignal(); + } + } + + void onTimeout(Interest::Ptr &&interest) override { + if (config_->verbose_) { + std::cout << "### timeout for " << interest->getName() + << " src port: " << interest->getSrcPort() + << " dst port: " << interest->getDstPort() + << " flags: " << interest->printFlags() << std::endl; + } else if (!config_->quiet_) { + std::cout << "### timeout for " << interest->getName() << std::endl; + } + + if (config_->dump_) { + std::cout << "----- interest dump -----" << std::endl; + interest->dump(); + std::cout << "-------------------------" << std::endl; + } + + if (!config_->quiet_) std::cout << std::endl; + + timedout_++; + processed_++; + if (processed_ >= config_->maxPing_) { + afterSignal(); + } + } + + void doPing() { + Name interest_name(config_->name_, sequence_number_); + hicn_format_t format; + if (interest_name.getAddressFamily() == AF_INET) { + format = HF_INET_TCP; + } else { + format = HF_INET6_TCP; + } + + Interest::Ptr interest(new Interest(std::move(interest_name), format), + nullptr); + + interest->setLifetime(uint32_t(config_->interestLifetime_)); + + interest->resetFlags(); + + if (config_->open_ || config_->always_syn_) { + if (state_ == SYN_STATE) { + interest->setSyn(); + } else if (state_ == ACK_STATE) { + interest->setAck(); + } + } else if (config_->always_ack_) { + interest->setAck(); + } + + interest->setSrcPort(config_->srcPort_); + interest->setDstPort(config_->dstPort_); + interest->setTTL(config_->ttl_); + + if (config_->verbose_) { + std::cout << ">>> send interest " << interest->getName() + << " src port: " << interest->getSrcPort() + << " dst port: " << interest->getDstPort() + << " flags: " << interest->printFlags() + << " TTL: " << (int)interest->getTTL() << std::endl; + } else if (!config_->quiet_) { + std::cout << ">>> send interest " << interest->getName() << std::endl; + } + + if (config_->dump_) { + std::cout << "----- interest dump -----" << std::endl; + interest->dump(); + std::cout << "-------------------------" << std::endl; + } + + if (!config_->quiet_) std::cout << std::endl; + + send_timestamps_[sequence_number_] = + std::chrono::duration_cast<std::chrono::microseconds>( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + + portal_.sendInterest(std::move(interest)); + + sequence_number_++; + sent_++; + + if (sent_ < config_->maxPing_) { + this->timer_->expires_from_now( + std::chrono::microseconds(config_->pingInterval_)); + this->timer_->async_wait([this](const std::error_code e) { doPing(); }); + } + } + + void afterSignal() { + std::cout << "Stop ping" << std::endl; + std::cout << "Sent: " << sent_ << " Received: " << received_ + << " Timeouts: " << timedout_ << std::endl; + portal_.stopEventsLoop(); + } + + void reset() { + timer_.reset(new asio::steady_timer(portal_.getIoService())); + sequence_number_ = config_->first_suffix_; + last_jump_ = 0; + processed_ = 0; + state_ = SYN_STATE; + sent_ = 0; + received_ = 0; + timedout_ = 0; + } + + private: + SendTimeMap send_timestamps_; + interface::BasePortal portal_; + uint64_t sequence_number_; + uint64_t last_jump_; + uint64_t processed_; + uint32_t state_; + uint32_t sent_; + uint32_t received_; + uint32_t timedout_; + std::unique_ptr<asio::steady_timer> timer_; + Configuration *config_; + Verifier verifier_; + PARCKeyId *key_id_; +}; + +void help() { + std::cout << "usage: hicn-consumer-ping [options]" << std::endl; + std::cout << "PING options" << std::endl; + std::cout + << "-i <val> ping interval in microseconds (default 1000000ms)" + << std::endl; + std::cout << "-m <val> maximum number of pings to send (default 10)" + << std::endl; + std::cout << "-s <val> sorce port (default 9695)" << std::endl; + std::cout << "-d <val> destination port (default 8080)" << std::endl; + std::cout << "-t <val> set packet ttl (default 64)" << std::endl; + std::cout << "-O open tcp connection (three way handshake) " + "(default false)" + << std::endl; + std::cout << "-S send always syn messages (default false)" + << std::endl; + std::cout << "-A send always ack messages (default false)" + << std::endl; + std::cout << "HICN options" << std::endl; + std::cout << "-n <val> hicn name (default b001::1)" << std::endl; + std::cout + << "-l <val> interest lifetime in milliseconds (default 500ms)" + << std::endl; + std::cout << "OUTPUT options" << std::endl; + std::cout << "-V verbose, prints statistics about the " + "messagges sent and received (default false)" + << std::endl; + std::cout << "-D dump, dumps sent and received packets " + "(default false)" + << std::endl; + std::cout << "-q quiet, not prints (default false)" + << std::endl; + std::cout << "-H prints this message" << std::endl; +} + +int main(int argc, char *argv[]) { + Configuration *c = new Configuration(); + int opt; + std::string producer_certificate = ""; + + while ((opt = getopt(argc, argv, "j::t:i:m:s:d:n:l:f:c:SAOqVDH")) != -1) { + switch (opt) { + case 't': + c->ttl_ = (uint8_t)std::stoi(optarg); + break; + case 'i': + c->pingInterval_ = std::stoi(optarg); + break; + case 'm': + c->maxPing_ = std::stoi(optarg); + break; + case 'f': + c->first_suffix_ = std::stoul(optarg); + break; + case 's': + c->srcPort_ = std::stoi(optarg); + break; + case 'd': + c->dstPort_ = std::stoi(optarg); + break; + case 'n': + c->name_ = optarg; + break; + case 'l': + c->interestLifetime_ = std::stoi(optarg); + break; + case 'V': + c->verbose_ = true; + ; + break; + case 'D': + c->dump_ = true; + break; + case 'O': + c->always_syn_ = false; + c->always_ack_ = false; + c->open_ = true; + break; + case 'S': + c->always_syn_ = true; + c->always_ack_ = false; + c->open_ = false; + break; + case 'A': + c->always_syn_ = false; + c->always_ack_ = true; + c->open_ = false; + break; + case 'q': + c->quiet_ = true; + c->verbose_ = false; + c->dump_ = false; + break; + case 'c': + c->certificate_ = std::string(optarg); + break; + case 'H': + default: + help(); + exit(EXIT_FAILURE); + } + } + + Client *ping = new Client(c); + + auto t0 = std::chrono::steady_clock::now(); + ping->ping(); + auto t1 = std::chrono::steady_clock::now(); + + std::cout + << "Elapsed time: " + << std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count() + << std::endl; + + return 0; +} + +} // namespace ping + +} // namespace core + +} // namespace transport + +int main(int argc, char *argv[]) { + return transport::core::ping::main(argc, argv); +} diff --git a/utils/src/ping_server.cc b/utils/src/ping_server.cc new file mode 100755 index 000000000..19de34fec --- /dev/null +++ b/utils/src/ping_server.cc @@ -0,0 +1,300 @@ +/* + * 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_producer.h> +#include <hicn/transport/utils/daemonizator.h> +#include <hicn/transport/utils/signer.h> +#include <hicn/transport/utils/string_tokenizer.h> + +namespace transport { + +namespace interface { + +using HashAlgorithm = core::HashAlgorithm; +using CryptoSuite = utils::CryptoSuite; + +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); + } else { + return utils::Identity(keystore_name, keystore_password, + CryptoSuite::RSA_SHA256, 1024, 365, "producer-test"); + } +} + +class CallbackContainer { + const std::size_t log2_content_object_buffer_size = 12; + + public: + CallbackContainer(const Name &prefix, uint32_t object_size, bool verbose, + bool dump, bool quite, bool flags, bool reset, uint8_t ttl, + utils::Identity *identity, bool sign) + : buffer_(object_size, 'X'), + content_objects_(1 << log2_content_object_buffer_size), + mask_((1 << log2_content_object_buffer_size) - 1), + content_objects_index_(0), + verbose_(verbose), + dump_(dump), + quite_(quite), + flags_(flags), + reset_(reset), + ttl_(ttl), + identity_(identity), + sign_(sign) { + core::Packet::Format format; + + if (prefix.getAddressFamily() == AF_INET) { + format = core::Packet::Format::HF_INET_TCP; + if (sign_) { + format = core::Packet::Format::HF_INET_TCP_AH; + } + } else { + format = core::Packet::Format::HF_INET6_TCP; + if (sign_) { + format = core::Packet::Format::HF_INET6_TCP_AH; + } + } + + for (int i = 0; i < (1 << log2_content_object_buffer_size); i++) { + content_objects_[i] = std::make_shared<ContentObject>( + prefix, format, (const uint8_t *)buffer_.data(), buffer_.size()); + content_objects_[i]->setLifetime( + default_values::content_object_expiry_time); + } + } + + void processInterest(ProducerSocket &p, const Interest &interest) { + if (verbose_) { + std::cout << "<<< received interest " << interest.getName() + << " src port: " << interest.getSrcPort() + << " dst port: " << interest.getDstPort() + << " flags: " << interest.printFlags() + << "TTL: " << (int)interest.getTTL() << std::endl; + } else if (!quite_) { + std::cout << "<<< received interest " << interest.getName() << std::endl; + } + + if (dump_) { + std::cout << "----- interest dump -----" << std::endl; + interest.dump(); + std::cout << "-------------------------" << std::endl; + } + + if (interest.testRst()) { + std::cout << "!!!got a reset, I don't reply" << std::endl; + } else { + auto &content_object = content_objects_[content_objects_index_++ & mask_]; + + content_object->setName(interest.getName()); + content_object->setLifetime(default_values::content_object_expiry_time); + content_object->setLocator(interest.getLocator()); + content_object->setSrcPort(interest.getDstPort()); + content_object->setDstPort(interest.getSrcPort()); + content_object->setTTL(ttl_); + + if (sign_) { + content_object->setSignatureSize(identity_->getSignatureLength()); + } else { + content_object->resetFlags(); + } + + if (flags_) { + if (interest.testSyn()) { + content_object->setSyn(); + content_object->setAck(); + } else if (interest.testAck()) { + content_object->setAck(); + } // here I may need to handle the FIN flag; + } else if (reset_) { + content_object->setRst(); + } + + if (verbose_) { + std::cout << ">>> send object " << content_object->getName() + << " src port: " << content_object->getSrcPort() + << " dst port: " << content_object->getDstPort() + << " flags: " << content_object->printFlags() + << " TTL: " << (int)content_object->getTTL() << std::endl; + } else if (!quite_) { + std::cout << ">>> send object " << content_object->getName() + << std::endl; + } + + if (dump_) { + std::cout << "----- object dump -----" << std::endl; + content_object->dump(); + std::cout << "-----------------------" << std::endl; + } + + if (!quite_) std::cout << std::endl; + + if (sign_) { + identity_->getSigner().sign(*content_object); + } + + p.produce(*content_object); + } + } + + private: + std::string buffer_; + std::vector<std::shared_ptr<ContentObject>> content_objects_; + std::uint16_t mask_; + std::uint16_t content_objects_index_; + bool verbose_; + bool dump_; + bool quite_; + bool flags_; + bool reset_; + uint8_t ttl_; + utils::Identity *identity_; + bool sign_; +}; + +void help() { + std::cout << "usage: hicn-preoducer-ping [options]" << std::endl; + std::cout << "PING options" << std::endl; + std::cout << "-s <val> object content size (default 1350B)" << std::endl; + std::cout << "-n <val> hicn name (default b001::/64)" << std::endl; + std::cout << "-f set tcp flags according to the flag received " + "(default false)" + << std::endl; + std::cout << "-r always reply with a reset flag (default false)" + << std::endl; + std::cout << "-t set ttl (default 64)" << std::endl; + std::cout << "OUTPUT options" << std::endl; + std::cout << "-V verbose, prints statistics about the messagges sent " + "and received (default false)" + << std::endl; + std::cout << "-D dump, dumps sent and received packets (default false)" + << std::endl; + std::cout << "-q quite, not prints (default false)" << std::endl; + std::cout << "-d daemon mode" << std::endl; + std::cout << "-H prints this message" << std::endl; +} + +int main(int argc, char **argv) { + std::string name_prefix = "b001::0/64"; + std::string delimiter = "/"; + bool daemon = false; + bool verbose = false; + bool dump = false; + bool quite = false; + bool flags = false; + bool reset = false; + uint32_t object_size = 1350; + uint8_t ttl = 64; + std::string keystore_path = "./rsa_crypto_material.p12"; + std::string keystore_password = "cisco"; + bool sign = false; + + int opt; + while ((opt = getopt(argc, argv, "s:n:t:qfrVDdHk:p:")) != -1) { + switch (opt) { + case 's': + object_size = std::stoi(optarg); + break; + case 'n': + name_prefix = optarg; + break; + case 't': + ttl = (uint8_t)std::stoi(optarg); + break; + case 'V': + verbose = true; + break; + case 'D': + dump = true; + break; + case 'q': + verbose = false; + dump = false; + quite = true; + break; + case 'd': + daemon = true; + break; + case 'f': + flags = true; + break; + case 'r': + reset = true; + break; + case 'k': + keystore_path = optarg; + sign = true; + break; + case 'p': + keystore_password = optarg; + break; + case 'H': + default: + help(); + exit(EXIT_FAILURE); + } + } + + if (daemon) { + utils::Daemonizator::daemonize(); + } + + core::Prefix producer_namespace(name_prefix); + + utils::StringTokenizer tokenizer(name_prefix, delimiter); + std::string ip_address = tokenizer.nextToken(); + Name n(ip_address); + + if (object_size > 1350) object_size = 1350; + + CallbackContainer *stubs; + utils::Identity identity = setProducerIdentity( + keystore_path, keystore_password, HashAlgorithm::SHA_256); + + if (sign) { + stubs = new CallbackContainer(n, object_size, verbose, dump, quite, flags, + reset, ttl, &identity, sign); + } else { + utils::Identity *identity = nullptr; + stubs = new CallbackContainer(n, object_size, verbose, dump, quite, flags, + reset, ttl, identity, sign); + } + + asio::io_service io_service; + + ProducerSocket p(io_service); // , setProducerIdentity()); + p.registerPrefix(producer_namespace); + + p.setSocketOption(GeneralTransportOptions::OUTPUT_BUFFER_SIZE, 0U); + p.setSocketOption(ProducerCallbacksOptions::CACHE_MISS, + (ProducerInterestCallback)bind( + &CallbackContainer::processInterest, stubs, + std::placeholders::_1, std::placeholders::_2)); + + p.connect(); + + p.serveForever(); + + return 0; +} + +} // namespace interface + +} // end namespace transport + +int main(int argc, char **argv) { + return transport::interface::main(argc, argv); +} |