aboutsummaryrefslogtreecommitdiffstats
path: root/utils
diff options
context:
space:
mode:
Diffstat (limited to 'utils')
-rw-r--r--utils/src/hiperf.cc93
1 files changed, 86 insertions, 7 deletions
diff --git a/utils/src/hiperf.cc b/utils/src/hiperf.cc
index d546cfca9..7594d1f94 100644
--- a/utils/src/hiperf.cc
+++ b/utils/src/hiperf.cc
@@ -21,6 +21,7 @@
#endif
#include <hicn/transport/utils/literals.h>
+#include <unordered_set>
#include <fstream>
#include <iomanip>
@@ -56,7 +57,8 @@ struct ClientConfiguration {
receive_buffer(std::make_shared<std::vector<uint8_t>>()),
download_size(0),
report_interval_milliseconds_(1000),
- rtc_(false) {}
+ rtc_(false),
+ test_mode_(false) {}
Name name;
bool verify;
@@ -70,6 +72,7 @@ struct ClientConfiguration {
std::uint32_t report_interval_milliseconds_;
TransportProtocolAlgorithms transport_protocol_;
bool rtc_;
+ bool test_mode_;
};
class Rate {
@@ -153,7 +156,9 @@ class HIperfClient {
: configuration_(conf),
total_duration_milliseconds_(0),
old_bytes_value_(0),
- signals_(io_service_, SIGINT) {}
+ signals_(io_service_, SIGINT),
+ expected_seg_(0),
+ lost_packets_(std::unordered_set<uint32_t>()) {}
void processPayload(ConsumerSocket &c, std::size_t bytes_transferred,
const std::error_code &ec) {
@@ -177,6 +182,57 @@ class HIperfClient {
configuration_.receive_buffer->clear();
}
+ void checkReceivedRtcContent(ConsumerSocket &c,
+ const ContentObject &contentObject) {
+ if(!configuration_.test_mode_)
+ return;
+
+ uint32_t receivedSeg = contentObject.getName().getSuffix();
+ auto payload = contentObject.getPayload();
+
+ if((uint32_t)payload->length() == 8){ //8 is the size of the NACK payload
+ uint32_t *payloadPtr = (uint32_t *)payload->data();
+ uint32_t productionSeg = *(payloadPtr);
+ uint32_t productionRate = *(++payloadPtr);
+
+ if(productionRate == 0){
+ std::cout << "[STOP] producer is not producing content"
+ << std::endl;
+ return;
+ }
+
+ if(receivedSeg < productionSeg){
+ std::cout << "[OUT OF SYNCH] received NACK for " << receivedSeg <<
+ ". Next expected packet " << productionSeg + 1 << std::endl;
+ expected_seg_ = productionSeg;
+ } else if(receivedSeg > productionSeg){
+ std::cout << "[WINDOW TO LARGE] received NACK for " << receivedSeg <<
+ ". Next expected packet " << productionSeg << std::endl;
+ }
+ return;
+ }
+
+ if(receivedSeg > expected_seg_){
+ for(uint32_t i = expected_seg_; i < receivedSeg; i++){
+ std::cout << "[LOSS] lost packet " << i << std::endl;
+ lost_packets_.insert(i);
+ }
+ expected_seg_ = receivedSeg + 1;
+ return;
+ }else if (receivedSeg < expected_seg_){
+ auto it = lost_packets_.find(receivedSeg);
+ if(it != lost_packets_.end()){
+ std::cout << "[RECOVER] recovered packet " << receivedSeg << std::endl;
+ lost_packets_.erase(it);
+ }else{
+ std::cout << "[OUT OF ORDER] recevied " << receivedSeg << " expedted " <<
+ expected_seg_ << std::endl;
+ }
+ return;
+ }
+ expected_seg_ = receivedSeg + 1;
+ }
+
bool verifyData(ConsumerSocket &c, const ContentObject &contentObject) {
if (contentObject.getPayloadType() == PayloadType::CONTENT_OBJECT) {
std::cout << "VERIFY CONTENT" << std::endl;
@@ -339,6 +395,17 @@ class HIperfClient {
return ERROR_SETUP;
}
+ if(configuration_.rtc_){
+ ret = consumer_socket_->setSocketOption(
+ ConsumerCallbacksOptions::CONTENT_OBJECT_INPUT,
+ (ConsumerContentObjectCallback)std::bind(
+ &HIperfClient::checkReceivedRtcContent, this,
+ std::placeholders::_1, std::placeholders::_2));
+ if (ret == SOCKET_OPTION_NOT_SET) {
+ return ERROR_SETUP;
+ }
+ }
+
ret = consumer_socket_->setSocketOption(
ConsumerCallbacksOptions::STATS_SUMMARY,
(ConsumerTimerCallback)std::bind(&HIperfClient::handleTimerExpiration,
@@ -386,6 +453,8 @@ class HIperfClient {
asio::io_service io_service_;
asio::signal_set signals_;
std::unique_ptr<ConsumerSocket> consumer_socket_;
+ uint32_t expected_seg_;
+ std::unordered_set<uint32_t> lost_packets_;
};
class HIperfServer {
@@ -489,8 +558,8 @@ class HIperfServer {
producer_socket_->connect();
if (configuration_.rtc_) {
- std::cout << "Running RTC producer: all other options (with the "
- "exception of the bitrate) will be ignored."
+ std::cout << "Running RTC producer: the prefix length will be ignored."
+ " Use /128 by default in RTC mode"
<< std::endl;
return ERROR_SUCCESS;
}
@@ -726,6 +795,12 @@ void usage() {
std::cout
<< "-v = Enable verification of received data"
<< std::endl;
+ std::cout
+ << "-t = Test mode, check if the client is "
+ "receiving the correct data. This is an RTC specific option, to be "
+ "used with the -R (default false)"
+ << std::endl;
+
}
int main(int argc, char *argv[]) {
@@ -751,7 +826,7 @@ int main(int argc, char *argv[]) {
int opt;
#ifndef _WIN32
- while ((opt = getopt(argc, argv, "DSCf:b:d:W:RMc:vA:s:rmlk:y:p:hi:xB:I")) !=
+ while ((opt = getopt(argc, argv, "DSCf:b:d:W:RMc:vA:s:rmlk:y:p:hi:xB:It")) !=
-1) {
switch (opt) {
// Common
@@ -764,7 +839,7 @@ int main(int argc, char *argv[]) {
break;
}
#else
- while ((opt = getopt(argc, argv, "SCf:b:d:W:RMc:vA:s:rmlk:y:p:hi:xB:")) !=
+ while ((opt = getopt(argc, argv, "SCf:b:d:W:RMc:vA:s:rmlk:y:p:hi:xB:t")) !=
-1) {
switch (opt) {
#endif
@@ -824,7 +899,11 @@ int main(int argc, char *argv[]) {
options = 1;
break;
}
-
+ case 't':{
+ client_configuration.test_mode_ = true;
+ options = 1;
+ break;
+ }
// Server specific
case 'A': {
server_configuration.download_size = std::stoul(optarg);