feat: boostrap hicn 22.02

The current patch provides several new features, improvements,
bug fixes and also complete rewrite of entire components.

- lib
The hicn packet parser has been improved with a new packet
format fully based on UDP. The TCP header is still temporarily
supported but the UDP header will replace completely the new hicn
packet format. Improvements have been made to make sure every
packet parsing operation is made via this library. The current
new header can be used as header between the payload and the
UDP header or as trailer in the UDP surplus area to be tested
when UDP options will start to be used.

- hicn-light
The portable packet forwarder has been completely rewritten from
scratch with the twofold objective to improve performance and
code size but  also to drop dependencies such as libparc which is
now removed by the current implementation.

- hicn control
the control library is the agent that is used to program the
packet forwarders via their binary API. This component has
benefited from significant improvements in terms of interaction
model which is now event driven and more robust to failures.

- VPP plugin has been updated to support VPP 22.02

- transport
Major improvement have been made to the RTC protocol, to the
support of IO modules and to the security sub system. Signed
manifests are the default data authenticity and integrity framework.
Confidentiality can be enabled by sharing the encryption key to the
prod/cons layer. The library has been tested with group key based
applications such as broadcast/multicast and real-time on-line
meetings with trusted server keys or MLS.

- testing
Unit testing has been introduced using GoogleTest. One third of
the code base is covered by unit testing with priority on
critical features. Functional testing has also been introduce
using Docker, linux bridging and Robot Framework to define
test with Less Code techniques to facilitate the extension
of the coverage.

Co-authored-by: Mauro Sardara <msardara@cisco.com>
Co-authored-by: Jordan Augé <jordan.auge+fdio@cisco.com>
Co-authored-by: Michele Papalini <micpapal@cisco.com>
Co-authored-by: Angelo Mantellini <manangel@cisco.com>
Co-authored-by: Jacques Samain <jsamain@cisco.com>
Co-authored-by: Olivier Roques <oroques+fdio@cisco.com>
Co-authored-by: Enrico Loparco <eloparco@cisco.com>
Co-authored-by: Giulio Grassi <gigrassi@cisco.com>

Change-Id: I75d0ef70f86d921e3ef503c99271216ff583c215
Signed-off-by: Luca Muscariello <muscariello@ieee.org>
Signed-off-by: Mauro Sardara <msardara@cisco.com>

1 files changed, 362 insertions, 124 deletions

diff --git a/libtransport/src/protocols/rtc/rtc_state.cc b/libtransport/src/protocols/rtc/rtc_state.cc
index c99205a26..6a21531f8 100644
--- a/libtransport/src/protocols/rtc/rtc_state.cc
+++ b/libtransport/src/protocols/rtc/rtc_state.cc
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2021 Cisco and/or its affiliates.
+ * Copyright (c) 2021 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:
@@ -24,15 +24,15 @@ namespace protocol {
 namespace rtc {
 
 RTCState::RTCState(Indexer *indexer,
-                   ProbeHandler::SendProbeCallback &&rtt_probes_callback,
+                   ProbeHandler::SendProbeCallback &&probe_callback,
                    DiscoveredRttCallback &&discovered_rtt_callback,
                    asio::io_service &io_service)
-    : indexer_(indexer),
-      rtt_probes_(std::make_shared<ProbeHandler>(std::move(rtt_probes_callback),
-                                                 io_service)),
+    : loss_history_(10),  // log 10sec history
+      indexer_(indexer),
+      probe_handler_(std::make_shared<ProbeHandler>(std::move(probe_callback),
+                                                    io_service)),
       discovered_rtt_callback_(std::move(discovered_rtt_callback)) {
   init_rtt_timer_ = std::make_unique<asio::steady_timer>(io_service);
-  initParams();
 }
 
 RTCState::~RTCState() {}
@@ -55,9 +55,19 @@ void RTCState::initParams() {
   highest_seq_received_in_order_ = 0;
   last_seq_nacked_ = 0;
   loss_rate_ = 0.0;
-  avg_loss_rate_ = 0.0;
+  avg_loss_rate_ = -1.0;
   max_loss_rate_ = 0.0;
   last_round_loss_rate_ = 0.0;
+
+  // loss rate per sec
+  lost_per_sec_ = 0;
+  total_expected_packets_ = 0;
+  per_sec_loss_rate_ = 0.0;
+
+  // residual losses counters
+  expected_packets_ = 0;
+  packets_sent_to_app_ = 0;
+  rounds_from_last_compute_ = 0;
   residual_loss_rate_ = 0.0;
 
   // fec counters
@@ -66,9 +76,13 @@ void RTCState::initParams() {
 
   // bw counters
   received_bytes_ = 0;
+  received_fec_bytes_ = 0;
+  recovered_bytes_with_fec_ = 0;
+
   avg_packet_size_ = INIT_PACKET_SIZE;
   production_rate_ = 0.0;
   received_rate_ = 0.0;
+  fec_recovered_rate_ = 0.0;
 
   // nack counter
   nack_on_last_round_ = false;
@@ -95,31 +109,30 @@ void RTCState::initParams() {
   path_table_.clear();
   main_path_ = nullptr;
 
-  // packet received
-  received_or_lost_packets_.clear();
+  // packet cache (not pending anymore)
+  packet_cache_.clear();
 
   // pending interests
   pending_interests_.clear();
 
-  // skipped interest
+  // used to keep track of the skipped interest
   last_interest_sent_ = 0;
-  skipped_interests_.clear();
 
   // init rtt
   first_interest_sent_time_ = ~0;
   first_interest_sent_seq_ = 0;
 
+  // start probing the producer
   init_rtt_ = false;
-  rtt_probes_->setProbes(INIT_RTT_PROBE_INTERVAL, INIT_RTT_PROBES);
-  rtt_probes_->sendProbes();
+  probe_handler_->setSuffixRange(MIN_INIT_PROBE_SEQ, MAX_INIT_PROBE_SEQ);
+  probe_handler_->setProbes(INIT_RTT_PROBE_INTERVAL, INIT_RTT_PROBES);
+  probe_handler_->sendProbes();
   setInitRttTimer(INIT_RTT_PROBE_RESTART);
 }
 
 // packet events
 void RTCState::onSendNewInterest(const core::Name *interest_name) {
-  uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
-                     std::chrono::steady_clock::now().time_since_epoch())
-                     .count();
+  uint64_t now = utils::SteadyTime::nowMs().count();
   uint32_t seq = interest_name->getSuffix();
   pending_interests_.insert(std::pair<uint32_t, uint64_t>(seq, now));
 
@@ -137,11 +150,12 @@ void RTCState::onSendNewInterest(const core::Name *interest_name) {
   }
 
   // TODO what happen in case of jumps?
-  // look for skipped interests
-  skipped_interests_.erase(seq);  // remove seq if it is there
+  eraseFromPacketCache(
+      seq);  // if we send this interest we don't know its state
   for (uint32_t i = last_interest_sent_ + 1; i < seq; i++) {
     if (indexer_->isFec(i)) {
-      skipped_interests_.insert(i);
+      // only fec packets can be skipped
+      addToPacketCache(i, PacketState::SKIPPED);
     }
   }
 
@@ -155,6 +169,7 @@ void RTCState::onTimeout(uint32_t seq, bool lost) {
   auto it = pending_interests_.find(seq);
   if (it != pending_interests_.end()) {
     pending_interests_.erase(it);
+    if (indexer_->isFec(seq)) pending_fec_pkt_--;
   }
   received_timeouts_++;
 
@@ -162,11 +177,12 @@ void RTCState::onTimeout(uint32_t seq, bool lost) {
 }
 
 void RTCState::onLossDetected(uint32_t seq) {
-  if (!indexer_->isFec(seq)) {
-    packets_lost_++;
-  } else if (skipped_interests_.find(seq) == skipped_interests_.end() &&
-             seq >= first_interest_sent_seq_) {
+  PacketState state = getPacketState(seq);
+
+  // if the packet is already marked with a state, do nothing
+  if (state == PacketState::UNKNOWN) {
     packets_lost_++;
+    addToPacketCache(seq, PacketState::LOST);
   }
 }
 
@@ -178,30 +194,40 @@ void RTCState::onRetransmission(uint32_t seq) {
   auto it = pending_interests_.find(seq);
   if (it != pending_interests_.end()) {
     pending_interests_.erase(it);
-#if 0
-    packets_lost_++;
-#endif
+    if (indexer_->isFec(seq)) pending_fec_pkt_--;
   }
   sent_rtx_++;
   sent_rtx_last_round_++;
 }
 
+void RTCState::onPossibleLossWithNoRtx(uint32_t seq) {
+  // if fec is on or rtx is disable we don't need to do anything to recover a
+  // packet. however in both cases we need to remove possible missing packets
+  // from the window of pendinig interest in order to free space without wating
+  // for the timeout.
+  auto it = pending_interests_.find(seq);
+  if (it != pending_interests_.end()) {
+    pending_interests_.erase(it);
+    if (indexer_->isFec(seq)) pending_fec_pkt_--;
+  }
+}
+
 void RTCState::onDataPacketReceived(const core::ContentObject &content_object,
                                     bool compute_stats) {
   uint32_t seq = content_object.getName().getSuffix();
+
   if (compute_stats) {
     updatePathStats(content_object, false);
     received_data_last_round_++;
   }
   received_data_++;
+  packets_sent_to_app_++;
 
-  struct data_packet_t *data_pkt =
-      (struct data_packet_t *)content_object.getPayload()->data();
-  uint64_t production_time = data_pkt->getTimestamp();
-  if (last_prod_update_ < production_time) {
-    last_prod_update_ = production_time;
-    uint32_t production_rate = data_pkt->getProductionRate();
-    production_rate_ = (double)production_rate;
+  core::ParamsRTC params = RTCState::getDataParams(content_object);
+
+  if (last_prod_update_ < params.timestamp) {
+    last_prod_update_ = params.timestamp;
+    production_rate_ = (double)params.prod_rate;
   }
 
   updatePacketSize(content_object);
@@ -219,9 +245,18 @@ void RTCState::onDataPacketReceived(const core::ContentObject &content_object,
 
 void RTCState::onFecPacketReceived(const core::ContentObject &content_object) {
   uint32_t seq = content_object.getName().getSuffix();
-  updateReceivedBytes(content_object);
+  // updateReceivedBytes(content_object);
+  received_fec_bytes_ +=
+      (uint32_t)(content_object.headerSize() + content_object.payloadSize());
+
+  if (seq > highest_seq_received_) highest_seq_received_ = seq;
+
+  PacketState state = getPacketState(seq);
+  if (state != PacketState::LOST) {
+    // increase only for not lost packets
+    received_fec_pkt_++;
+  }
   addRecvOrLost(seq, PacketState::RECEIVED);
-  received_fec_pkt_++;
   // the producer is responding
   // it is generating valid data packets so we consider it active
   producer_is_active_ = true;
@@ -233,7 +268,7 @@ void RTCState::onNackPacketReceived(const core::ContentObject &nack,
   struct nack_packet_t *nack_pkt =
       (struct nack_packet_t *)nack.getPayload()->data();
   uint64_t production_time = nack_pkt->getTimestamp();
-  uint32_t production_seq = nack_pkt->getProductionSegement();
+  uint32_t production_seq = nack_pkt->getProductionSegment();
   uint32_t production_rate = nack_pkt->getProductionRate();
 
   if (TRANSPORT_EXPECT_FALSE(main_path_ == nullptr) ||
@@ -255,6 +290,7 @@ void RTCState::onNackPacketReceived(const core::ContentObject &nack,
   received_nacks_++;
   received_nacks_last_round_++;
 
+  bool to_delete = false;
   if (production_seq > seq) {
     // old nack, seq is lost
     // update last nacked
@@ -266,12 +302,19 @@ void RTCState::onNackPacketReceived(const core::ContentObject &nack,
     // future nack
     // remove the nack from the pending interest map
     // (the packet is not received/lost yet)
-    if (indexer_->isFec(seq)) pending_fec_pkt_--;
-    pending_interests_.erase(seq);
+    to_delete = true;
   } else {
     // this should be a quite rear event. simply remove the
     // packet from the pending interest list
-    pending_interests_.erase(seq);
+    to_delete = true;
+  }
+
+  if (to_delete) {
+    auto it = pending_interests_.find(seq);
+    if (it != pending_interests_.end()) {
+      pending_interests_.erase(it);
+      if (indexer_->isFec(seq)) pending_fec_pkt_--;
+    }
   }
 
   // the producer is responding
@@ -295,44 +338,58 @@ void RTCState::onPacketLost(uint32_t seq) {
   }
 #endif
   if (!indexer_->isFec(seq)) {
-    definitely_lost_pkt_++;
-    DLOG_IF(INFO, VLOG_IS_ON(4)) << "packet " << seq << " is lost";
+    PacketState state = getPacketState(seq);
+    if (state == PacketState::LOST || state == PacketState::UNKNOWN) {
+      definitely_lost_pkt_++;
+      DLOG_IF(INFO, VLOG_IS_ON(4)) << "packet " << seq << " is lost";
+    }
   }
-  addRecvOrLost(seq, PacketState::LOST);
+  addRecvOrLost(seq, PacketState::DEFINITELY_LOST);
 }
 
 void RTCState::onPacketRecoveredRtx(uint32_t seq) {
+  packets_sent_to_app_++;
+  if (seq > highest_seq_received_) highest_seq_received_ = seq;
   losses_recovered_++;
   addRecvOrLost(seq, PacketState::RECEIVED);
 }
 
-void RTCState::onPacketRecoveredFec(uint32_t seq) {
+void RTCState::onFecPacketRecoveredRtx(uint32_t seq) {
+  // This is the same as onPacketRecoveredRtx, but in this is case the
+  // pkt is also a FEC pkt, the addRecvOrLost will be called afterwards
+  if (seq > highest_seq_received_) highest_seq_received_ = seq;
+  losses_recovered_++;
+}
+
+void RTCState::onPacketRecoveredFec(uint32_t seq, uint32_t size) {
   losses_recovered_++;
+  packets_sent_to_app_++;
+  recovered_bytes_with_fec_ += size;
+
+  if (seq > highest_seq_received_) highest_seq_received_ = seq;
+
+  // adding header to the count
+  recovered_bytes_with_fec_ += 60;  // XXX get header size some where
+
+  if (getPacketState(seq) == PacketState::UNKNOWN)
+    onLossDetected(seq);  // the pkt was lost but didn't account for it yet
+
   addRecvOrLost(seq, PacketState::RECEIVED);
 }
 
 bool RTCState::onProbePacketReceived(const core::ContentObject &probe) {
   uint32_t seq = probe.getName().getSuffix();
-  uint64_t rtt;
-
-  rtt = rtt_probes_->getRtt(seq);
 
+  uint64_t rtt;
+  rtt = probe_handler_->getRtt(seq);
   if (rtt == 0) return false;  // this is not a valid probe
 
-  // like for data and nacks update the path stats. Here the RTT is computed
-  // by the probe handler. Both probes for rtt and bw are good to esimate
-  // info on the path
+  // Like for data and nacks update the path stats. Here the RTT is computed
+  // by the probe handler. Both probes for rtt and bw are good to estimate
+  // info on the path.
   uint32_t path_label = probe.getPathLabel();
-
   auto path_it = path_table_.find(path_label);
 
-  // update production rate and last_seq_nacked like in case of a nack
-  struct nack_packet_t *probe_pkt =
-      (struct nack_packet_t *)probe.getPayload()->data();
-  uint64_t sender_timestamp = probe_pkt->getTimestamp();
-  uint32_t production_seq = probe_pkt->getProductionSegement();
-  uint32_t production_rate = probe_pkt->getProductionRate();
-
   if (path_it == path_table_.end()) {
     // found a new path
     std::shared_ptr<RTCDataPath> newPath =
@@ -344,26 +401,26 @@ bool RTCState::onProbePacketReceived(const core::ContentObject &probe) {
 
   auto path = path_it->second;
 
-  path->insertRttSample(rtt);
+  path->insertRttSample(utils::SteadyTime::Milliseconds(rtt), true);
   path->receivedNack();
 
-  uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
-                     std::chrono::steady_clock::now().time_since_epoch())
-                     .count();
+  uint64_t now = utils::SteadyTime::nowMs().count();
+
+  core::ParamsRTC params = RTCState::getProbeParams(probe);
 
-  int64_t OWD = now - sender_timestamp;
+  int64_t OWD = now - params.timestamp;
   path->insertOwdSample(OWD);
 
-  if (last_prod_update_ < sender_timestamp) {
-    last_production_seq_ = production_seq;
-    last_prod_update_ = sender_timestamp;
-    production_rate_ = (double)production_rate;
+  if (last_prod_update_ < params.timestamp) {
+    last_production_seq_ = params.prod_seg;
+    last_prod_update_ = params.timestamp;
+    production_rate_ = (double)params.prod_rate;
   }
 
   // the producer is responding
   // we consider it active only if the production rate is not 0
   // or the production sequence numner is not 1
-  if (production_rate_ != 0 || production_seq != 1) {
+  if (production_rate_ != 0 || params.prod_seg != 1) {
     producer_is_active_ = true;
   }
 
@@ -375,7 +432,7 @@ bool RTCState::onProbePacketReceived(const core::ContentObject &probe) {
   if (!init_rtt_ && received_probes_ <= INIT_RTT_PROBES) {
     if (received_probes_ == 1) {
       // we got the first probe, wait at most INIT_RTT_PROBE_WAIT sec for the
-      // others
+      // others.
       main_path_ = path;
       setInitRttTimer(INIT_RTT_PROBE_WAIT);
     }
@@ -393,11 +450,21 @@ bool RTCState::onProbePacketReceived(const core::ContentObject &probe) {
   return true;
 }
 
-void RTCState::onNewRound(double round_len, bool in_sync) {
-  // XXX
-  // here we take into account only the single path case so we assume that we
-  // don't use two paths in parellel for this single flow
+void RTCState::onJumpForward(uint32_t next_seq) {
+  for (uint32_t seq = highest_seq_received_in_order_ + 1; seq < next_seq;
+       seq++) {
+    auto it = pending_interests_.find(seq);
+    PacketState packet_state = getPacketState(seq);
+    if (it == pending_interests_.end() &&
+        packet_state != PacketState::RECEIVED &&
+        packet_state != PacketState::DEFINITELY_LOST) {
+      onLossDetected(seq);
+      onPacketLost(seq);
+    }
+  }
+}
 
+void RTCState::onNewRound(double round_len, bool in_sync) {
   if (path_table_.empty()) return;
 
   double bytes_per_sec =
@@ -407,7 +474,24 @@ void RTCState::onNewRound(double round_len, bool in_sync) {
   else
     received_rate_ = (received_rate_ * MOVING_AVG_ALPHA) +
                      ((1 - MOVING_AVG_ALPHA) * bytes_per_sec);
+  double fec_bytes_per_sec =
+      ((double)received_fec_bytes_ * (MILLI_IN_A_SEC / round_len));
+
+  if (fec_received_rate_ == 0)
+    fec_received_rate_ = fec_bytes_per_sec;
+  else
+    fec_received_rate_ = (fec_received_rate_ * 0.8) + (0.2 * fec_bytes_per_sec);
+
+  double fec_recovered_bytes_per_sec =
+      ((double)recovered_bytes_with_fec_ * (MILLI_IN_A_SEC / round_len));
 
+  if (fec_recovered_rate_ == 0)
+    fec_recovered_rate_ = fec_recovered_bytes_per_sec;
+  else
+    fec_recovered_rate_ =
+        (fec_recovered_rate_ * 0.8) + (0.2 * fec_recovered_bytes_per_sec);
+
+#if 0
   // search for an active path. There should be only one active path (meaning a
   // path that leads to the producer socket -no cache- and from which we are
   // currently getting data packets) at any time. However it may happen that
@@ -428,9 +512,36 @@ void RTCState::onNewRound(double round_len, bool in_sync) {
       }
     }
   }
+#endif
+
+  // search for an active path. Is it possible to have multiple path that are
+  // used at the same time. We use as reference path the one from where we gets
+  // more packets. This means that the path should have better lantecy or less
+  // channel losses
+
+  uint32_t last_round_packets = 0;
+  std::shared_ptr<RTCDataPath> old_main_path = main_path_;
+  main_path_ = nullptr;
+
+  for (auto it = path_table_.begin(); it != path_table_.end(); it++) {
+    if (it->second->isActive()) {
+      uint32_t pkt = it->second->getPacketsLastRound();
+      if (pkt > last_round_packets) {
+        last_round_packets = pkt;
+        main_path_ = it->second;
+      }
+    }
+    it->second->roundEnd();
+  }
 
-  // if (in_sync) updateLossRate();
-  updateLossRate();
+  if (main_path_ == nullptr) main_path_ = old_main_path;
+
+  // in case we get a new main path we reset the stats of the old one. this is
+  // beacuse, in case we need to switch back we don't what to take decisions on
+  // old stats that may be outdated.
+  if (main_path_ != old_main_path) old_main_path->clearRtt();
+
+  updateLossRate(in_sync);
 
   // handle nacks
   if (!nack_on_last_round_ && received_bytes_ > 0) {
@@ -460,6 +571,8 @@ void RTCState::onNewRound(double round_len, bool in_sync) {
 
   // reset counters
   received_bytes_ = 0;
+  received_fec_bytes_ = 0;
+  recovered_bytes_with_fec_ = 0;
   packets_lost_ = 0;
   definitely_lost_pkt_ = 0;
   losses_recovered_ = 0;
@@ -516,20 +629,16 @@ void RTCState::updatePathStats(const core::ContentObject &content_object,
              // it means that we are processing an interest
              // that is not pending
 
-  uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
-                     std::chrono::steady_clock::now().time_since_epoch())
-                     .count();
+  uint64_t now = utils::SteadyTime::nowMs().count();
 
   uint64_t RTT = now - interest_sent_time;
 
-  path->insertRttSample(RTT);
+  path->insertRttSample(utils::SteadyTime::Milliseconds(RTT), false);
 
   // compute OWD (the first part of the nack and data packet header are the
   // same, so we cast to data data packet)
-  struct data_packet_t *packet =
-      (struct data_packet_t *)content_object.getPayload()->data();
-  uint64_t sender_timestamp = packet->getTimestamp();
-  int64_t OWD = now - sender_timestamp;
+  core::ParamsRTC params = RTCState::getDataParams(content_object);
+  int64_t OWD = now - params.timestamp;
   path->insertOwdSample(OWD);
 
   // compute IAT or set path to producer
@@ -543,59 +652,106 @@ void RTCState::updatePathStats(const core::ContentObject &content_object,
   }
 }
 
-void RTCState::updateLossRate() {
+void RTCState::updateLossRate(bool in_sync) {
   last_round_loss_rate_ = loss_rate_;
   loss_rate_ = 0.0;
-  residual_loss_rate_ = 0.0;
 
   uint32_t number_theorically_received_packets_ =
       highest_seq_received_ - first_seq_in_round_;
 
-  // in this case no new packet was recevied after the previuos round, avoid
-  // division by 0
-  if (number_theorically_received_packets_ == 0) return;
-
   // XXX this may be quite inefficient if the rate is high
   // maybe is better to iterate over the set?
-  for (uint32_t i = first_seq_in_round_; i < highest_seq_received_; i++) {
-    auto it = skipped_interests_.find(i);
-    if (it != skipped_interests_.end()) {
+
+  uint32_t fec_packets = 0;
+  for (uint32_t i = (first_seq_in_round_ + 1); i < highest_seq_received_; i++) {
+    PacketState state = getPacketState(i);
+    if (state == PacketState::SKIPPED) {
       if (number_theorically_received_packets_ > 0)
         number_theorically_received_packets_--;
-      skipped_interests_.erase(it);
     }
+    if (indexer_->isFec(i)) fec_packets++;
   }
+  if (indexer_->isFec(highest_seq_received_)) fec_packets++;
 
-  loss_rate_ = (double)((double)(packets_lost_) /
-                        (double)number_theorically_received_packets_);
+  // in this case no new packet was received after the previous round, avoid
+  // division by 0
+  if (number_theorically_received_packets_ == 0 && packets_lost_ == 0) return;
 
-  if (rounds_ % 15 == 0) max_loss_rate_ = 0;  // reset every 3 sec
-  if (loss_rate_ > max_loss_rate_) max_loss_rate_ = loss_rate_;
+  if (number_theorically_received_packets_ != 0)
+    loss_rate_ = (double)((double)(packets_lost_) /
+                          (double)number_theorically_received_packets_);
+  else
+    // we didn't receive anything except NACKs that triggered losses
+    loss_rate_ = 1.0;
 
-  if (avg_loss_rate_ == 0)
+  if (avg_loss_rate_ == -1.0)
     avg_loss_rate_ = loss_rate_;
   else
     avg_loss_rate_ =
         avg_loss_rate_ * MOVING_AVG_ALPHA + loss_rate_ * (1 - MOVING_AVG_ALPHA);
 
-  residual_loss_rate_ = (double)((double)(packets_lost_ - losses_recovered_) /
-                                 (double)number_theorically_received_packets_);
+  // update counters for loss rate per second
+  total_expected_packets_ += number_theorically_received_packets_;
+  lost_per_sec_ += packets_lost_;
+
+  if (in_sync) {
+    // update counters for residual losses
+    // fec packets are not sent to the app so we don't want to count them here
+    expected_packets_ +=
+        ((highest_seq_received_ - first_seq_in_round_) - fec_packets);
+  } else {
+    packets_sent_to_app_ = 0;
+  }
+
+  if (rounds_from_last_compute_ >= (MILLI_IN_A_SEC / ROUND_LEN)) {
+    // compute loss rate per second
+    if (lost_per_sec_ > total_expected_packets_)
+      lost_per_sec_ = total_expected_packets_;
+
+    if (total_expected_packets_ == 0)
+      per_sec_loss_rate_ = 0;
+    else
+      per_sec_loss_rate_ =
+          (double)((double)(lost_per_sec_) / (double)total_expected_packets_);
+
+    loss_history_.pushBack(per_sec_loss_rate_);
+    max_loss_rate_ = getMaxLoss();
+
+    if (in_sync && expected_packets_ != 0) {
+      // compute residual loss rate
+      if (packets_sent_to_app_ > expected_packets_) {
+        // this may happen if we get packet from the prev bin that get recovered
+        // on the current one
+        packets_sent_to_app_ = expected_packets_;
+      }
+
+      residual_loss_rate_ =
+          1.0 - ((double)packets_sent_to_app_ / (double)expected_packets_);
+      if (residual_loss_rate_ < 0.0) residual_loss_rate_ = 0.0;
+    }
+
+    lost_per_sec_ = 0;
+    total_expected_packets_ = 0;
+    expected_packets_ = 0;
+    packets_sent_to_app_ = 0;
+    rounds_from_last_compute_ = 0;
+  }
+
+  rounds_from_last_compute_++;
+}
 
-  if (residual_loss_rate_ < 0) residual_loss_rate_ = 0;
+void RTCState::dataToBeReceived(uint32_t seq) {
+  addToPacketCache(seq, PacketState::TO_BE_RECEIVED);
 }
 
 void RTCState::addRecvOrLost(uint32_t seq, PacketState state) {
-  if (indexer_->isFec(seq)) {
-    pending_fec_pkt_--;
+  auto it = pending_interests_.find(seq);
+  if (it != pending_interests_.end()) {
+    pending_interests_.erase(it);
+    if (indexer_->isFec(seq)) pending_fec_pkt_--;
   }
 
-  pending_interests_.erase(seq);
-  if (received_or_lost_packets_.size() >= MAX_CACHED_PACKETS) {
-    received_or_lost_packets_.erase(received_or_lost_packets_.begin());
-  }
-  // notice that it may happen that a packet that we consider lost arrives after
-  // some time, in this case we simply overwrite the packet state.
-  received_or_lost_packets_[seq] = state;
+  addToPacketCache(seq, state);
 
   // keep track of the last packet received/lost
   // without holes.
@@ -608,16 +764,25 @@ void RTCState::addRecvOrLost(uint32_t seq, PacketState state) {
   } else if (seq <= highest_seq_received_in_order_) {
     // here we do nothing
   } else if (seq > highest_seq_received_in_order_) {
-    // 1) there is a gap in the sequence so we do not update largest_in_seq_
-    // 2) all the packets from largest_in_seq_ to seq are in
-    //    received_or_lost_packets_ an we upate largest_in_seq_
-    //    or are FEC packets
+    // 1) there is a gap in the sequence so we do not update
+    // highest_seq_received_in_order_
+    // 2) all the packets from highest_seq_received_in_order_ to seq are
+    // received or lost or are fec packetis. In this case we increase
+    // highest_seq_received_in_order_ until we find an hole in the sequence
 
     for (uint32_t i = highest_seq_received_in_order_ + 1; i <= seq; i++) {
-      if (received_or_lost_packets_.find(i) ==
-              received_or_lost_packets_.end() &&
-          !indexer_->isFec(i)) {
-        break;
+      PacketState state = getPacketState(i);
+      if ((state == PacketState::UNKNOWN || state == PacketState::LOST)) {
+        if (indexer_->isFec(i)) {
+          // this is a fec packet and we don't care to receive it
+          // however we may need to increse the number or lost packets
+          // XXX: in case we want to use rtx to recover fec packets,
+          // this may prevent to detect a packet loss and no rtx will be sent
+          onLossDetected(i);
+        } else {
+          // this is a data packet and we need to get it
+          break;
+        }
       }
       // this packet is in order so we can update the
       // highest_seq_received_in_order_
@@ -629,9 +794,14 @@ void RTCState::addRecvOrLost(uint32_t seq, PacketState state) {
 void RTCState::setInitRttTimer(uint32_t wait) {
   init_rtt_timer_->cancel();
   init_rtt_timer_->expires_from_now(std::chrono::milliseconds(wait));
-  init_rtt_timer_->async_wait([this](std::error_code ec) {
+
+  std::weak_ptr<RTCState> self = shared_from_this();
+  init_rtt_timer_->async_wait([self](const std::error_code &ec) {
     if (ec) return;
-    checkInitRttTimer();
+
+    if (auto ptr = self.lock()) {
+      ptr->checkInitRttTimer();
+    }
   });
 }
 
@@ -639,19 +809,25 @@ void RTCState::checkInitRttTimer() {
   if (received_probes_ < INIT_RTT_MIN_PROBES_TO_RECV) {
     // we didn't received enough probes, restart
     received_probes_ = 0;
-    rtt_probes_->setProbes(INIT_RTT_PROBE_INTERVAL, INIT_RTT_PROBES);
-    rtt_probes_->sendProbes();
+    probe_handler_->setSuffixRange(MIN_INIT_PROBE_SEQ, MAX_INIT_PROBE_SEQ);
+    probe_handler_->setProbes(INIT_RTT_PROBE_INTERVAL, INIT_RTT_PROBES);
+    probe_handler_->sendProbes();
     setInitRttTimer(INIT_RTT_PROBE_RESTART);
     return;
   }
+
   init_rtt_ = true;
   main_path_->roundEnd();
-  rtt_probes_->setProbes(RTT_PROBE_INTERVAL, 0);
-  rtt_probes_->sendProbes();
+  loss_history_.pushBack(probe_handler_->getProbeLossRate());
+  max_loss_rate_ = getMaxLoss();
+
+  probe_handler_->setSuffixRange(MIN_RTT_PROBE_SEQ, MAX_RTT_PROBE_SEQ);
+  probe_handler_->setProbes(RTT_PROBE_INTERVAL, 0);
+  probe_handler_->sendProbes();
 
   // init last_seq_nacked_. skip packets that may come from the cache
   double prod_rate = getProducerRate();
-  double rtt = (double)getRTT() / MILLI_IN_A_SEC;
+  double rtt = (double)getMinRTT() / MILLI_IN_A_SEC;
   double packet_size = getAveragePacketSize();
   uint32_t pkt_in_rtt_ = std::floor(((prod_rate / packet_size) * rtt) * 0.8);
   last_seq_nacked_ = last_production_seq_ + pkt_in_rtt_;
@@ -659,6 +835,68 @@ void RTCState::checkInitRttTimer() {
   discovered_rtt_callback_();
 }
 
+double RTCState::getMaxLoss() {
+  if (loss_history_.size() != 0) return loss_history_.begin();
+  return 0;
+}
+
+core::ParamsRTC RTCState::getProbeParams(const core::ContentObject &probe) {
+  uint32_t seq = probe.getName().getSuffix();
+  core::ParamsRTC params;
+
+  switch (ProbeHandler::getProbeType(seq)) {
+    case ProbeType::INIT: {
+      core::ContentObjectManifest manifest(
+          const_cast<core::ContentObject &>(probe));
+      manifest.decode();
+      params = manifest.getParamsRTC();
+      break;
+    }
+    case ProbeType::RTT: {
+      struct nack_packet_t *probe_pkt =
+          (struct nack_packet_t *)probe.getPayload()->data();
+      params = core::ParamsRTC{
+          .timestamp = probe_pkt->getTimestamp(),
+          .prod_rate = probe_pkt->getProductionRate(),
+          .prod_seg = probe_pkt->getProductionSegment(),
+      };
+      break;
+    }
+    default:
+      break;
+  }
+
+  return params;
+}
+
+core::ParamsRTC RTCState::getDataParams(const core::ContentObject &data) {
+  core::ParamsRTC params;
+
+  switch (data.getPayloadType()) {
+    case core::PayloadType::DATA: {
+      struct data_packet_t *data_pkt =
+          (struct data_packet_t *)data.getPayload()->data();
+      params = core::ParamsRTC{
+          .timestamp = data_pkt->getTimestamp(),
+          .prod_rate = data_pkt->getProductionRate(),
+          .prod_seg = data.getName().getSuffix(),
+      };
+      break;
+    }
+    case core::PayloadType::MANIFEST: {
+      core::ContentObjectManifest manifest(
+          const_cast<core::ContentObject &>(data));
+      manifest.decode();
+      params = manifest.getParamsRTC();
+      break;
+    }
+    default:
+      break;
+  }
+
+  return params;
+}
+
 }  // namespace rtc
 
 }  // namespace protocol