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, 457 insertions, 159 deletions

diff --git a/libtransport/src/protocols/rtc/rtc.cc b/libtransport/src/protocols/rtc/rtc.cc
index 0cb4cda1d..df6522471 100644
--- a/libtransport/src/protocols/rtc/rtc.cc
+++ b/libtransport/src/protocols/rtc/rtc.cc
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2019 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:
@@ -22,7 +22,7 @@
 #include <protocols/rtc/rtc.h>
 #include <protocols/rtc/rtc_consts.h>
 #include <protocols/rtc/rtc_indexer.h>
-#include <protocols/rtc/rtc_rc_queue.h>
+#include <protocols/rtc/rtc_rc_congestion_detection.h>
 
 #include <algorithm>
 
@@ -37,13 +37,15 @@ using namespace interface;
 RTCTransportProtocol::RTCTransportProtocol(
     implementation::ConsumerSocket *icn_socket)
     : TransportProtocol(icn_socket, new RtcIndexer<>(icn_socket, this),
-                        new DatagramReassembly(icn_socket, this)),
+                        new RtcReassembly(icn_socket, this)),
       number_(0) {
   icn_socket->getSocketOption(PORTAL, portal_);
-  round_timer_ = std::make_unique<asio::steady_timer>(portal_->getIoService());
+  round_timer_ =
+      std::make_unique<asio::steady_timer>(portal_->getThread().getIoService());
   scheduler_timer_ =
-      std::make_unique<asio::steady_timer>(portal_->getIoService());
-  pacing_timer_ = std::make_unique<asio::steady_timer>(portal_->getIoService());
+      std::make_unique<asio::steady_timer>(portal_->getThread().getIoService());
+  pacing_timer_ =
+      std::make_unique<asio::steady_timer>(portal_->getThread().getIoService());
 }
 
 RTCTransportProtocol::~RTCTransportProtocol() {}
@@ -61,25 +63,52 @@ std::size_t RTCTransportProtocol::transportHeaderLength() {
 // private
 void RTCTransportProtocol::initParams() {
   TransportProtocol::reset();
+  fwd_strategy_.setCallback(on_fwd_strategy_);
 
-  rc_ = std::make_shared<RTCRateControlQueue>();
+  std::weak_ptr<RTCTransportProtocol> self = shared_from_this();
+
+  std::shared_ptr<auth::Verifier> verifier;
+  socket_->getSocketOption(GeneralTransportOptions::VERIFIER, verifier);
+
+  uint32_t max_unverified_delay;
+  socket_->getSocketOption(GeneralTransportOptions::MAX_UNVERIFIED_TIME,
+                           max_unverified_delay);
+
+  rc_ = std::make_shared<RTCRateControlCongestionDetection>();
   ldr_ = std::make_shared<RTCLossDetectionAndRecovery>(
-      indexer_verifier_.get(),
-      std::bind(&RTCTransportProtocol::sendRtxInterest, this,
-                std::placeholders::_1),
-      portal_->getIoService());
+      indexer_verifier_.get(), portal_->getThread().getIoService(),
+      interface::RtcTransportRecoveryStrategies::RTX_ONLY,
+      [self](uint32_t seq) {
+        auto ptr = self.lock();
+        if (ptr && ptr->isRunning()) {
+          ptr->sendRtxInterest(seq);
+        }
+      },
+      on_rec_strategy_);
+  verifier_ = std::make_shared<RTCVerifier>(verifier, max_unverified_delay);
 
   state_ = std::make_shared<RTCState>(
       indexer_verifier_.get(),
-      std::bind(&RTCTransportProtocol::sendProbeInterest, this,
-                std::placeholders::_1),
-      std::bind(&RTCTransportProtocol::discoveredRtt, this),
-      portal_->getIoService());
+      [self](uint32_t seq) {
+        auto ptr = self.lock();
+        if (ptr && ptr->isRunning()) {
+          ptr->sendProbeInterest(seq);
+        }
+      },
+      [self]() {
+        auto ptr = self.lock();
+        if (ptr && ptr->isRunning()) {
+          ptr->discoveredRtt();
+        }
+      },
+      portal_->getThread().getIoService());
+  state_->initParams();
 
   rc_->setState(state_);
-  // TODO: for the moment we keep the congestion control disabled
-  // rc_->tunrOnRateControl();
-  ldr_->setState(state_);
+  rc_->turnOnRateControl();
+  ldr_->setState(state_.get());
+  ldr_->setRateControl(rc_.get());
+  verifier_->setState(state_);
 
   // protocol state
   start_send_interest_ = false;
@@ -102,6 +131,10 @@ void RTCTransportProtocol::initParams() {
   }
 #else
   max_aggregated_interest_ = 1;
+  if (const char *max_aggr = std::getenv("MAX_AGGREGATED_INTERESTS")) {
+    LOG(INFO) << "Max Aggregated: " << max_aggr;
+    max_aggregated_interest_ = std::stoul(std::string(max_aggr));
+  }
 #endif
 
   max_sent_int_ =
@@ -131,6 +164,7 @@ void RTCTransportProtocol::initParams() {
     indexer_verifier_->setNFec(0);
     ldr_->setFecParams(fec::FECUtils::getBlockSymbols(fec_type_),
                        fec::FECUtils::getSourceSymbols(fec_type_));
+    fec_decoder_->setIOService(portal_->getThread().getIoService());
   } else {
     indexer_verifier_->disableFec();
   }
@@ -162,61 +196,97 @@ void RTCTransportProtocol::inactiveProducer() {
 
 void RTCTransportProtocol::newRound() {
   round_timer_->expires_from_now(std::chrono::milliseconds(ROUND_LEN));
-  // TODO pass weak_ptr here
-  round_timer_->async_wait([this, n{number_}](std::error_code ec) {
-    if (ec) return;
 
-    if (n != number_) {
+  std::weak_ptr<RTCTransportProtocol> self = shared_from_this();
+  round_timer_->async_wait([self](const std::error_code &ec) {
+    if (ec) {
       return;
     }
 
+    auto ptr = self.lock();
+
+    if (!ptr || !ptr->isRunning()) {
+      return;
+    }
+
+    auto &state = ptr->state_;
+
     // saving counters that will be reset on new round
-    uint32_t sent_retx = state_->getSentRtxInRound();
-    uint32_t received_bytes = state_->getReceivedBytesInRound();
-    uint32_t sent_interest = state_->getSentInterestInRound();
-    uint32_t lost_data = state_->getLostData();
-    uint32_t definitely_lost = state_->getDefinitelyLostPackets();
-    uint32_t recovered_losses = state_->getRecoveredLosses();
-    uint32_t received_nacks = state_->getReceivedNacksInRound();
-    uint32_t received_fec = state_->getReceivedFecPackets();
-
-    bool in_sync = (current_state_ == SyncState::in_sync);
-    ldr_->onNewRound(in_sync);
-    state_->onNewRound((double)ROUND_LEN, in_sync);
-    rc_->onNewRound((double)ROUND_LEN);
+    uint32_t sent_retx = state->getSentRtxInRound();
+    uint32_t received_bytes =
+        (state->getReceivedBytesInRound() +     // data packets received
+         state->getReceivedFecBytesInRound());  // fec packets received
+    uint32_t sent_interest = state->getSentInterestInRound();
+    uint32_t lost_data = state->getLostData();
+    uint32_t definitely_lost = state->getDefinitelyLostPackets();
+    uint32_t recovered_losses = state->getRecoveredLosses();
+    uint32_t received_nacks = state->getReceivedNacksInRound();
+    uint32_t received_fec = state->getReceivedFecPackets();
+
+    bool in_sync = (ptr->current_state_ == SyncState::in_sync);
+    ptr->ldr_->onNewRound(in_sync);
+    ptr->state_->onNewRound((double)ROUND_LEN, in_sync);
+    ptr->rc_->onNewRound((double)ROUND_LEN);
 
     // update sync state if needed
-    if (current_state_ == SyncState::in_sync) {
-      double cache_rate = state_->getPacketFromCacheRatio();
+    if (ptr->current_state_ == SyncState::in_sync) {
+      double cache_rate = state->getPacketFromCacheRatio();
       if (cache_rate > MAX_DATA_FROM_CACHE) {
-        current_state_ = SyncState::catch_up;
+        ptr->current_state_ = SyncState::catch_up;
       }
     } else {
-      double target_rate = state_->getProducerRate() * PRODUCTION_RATE_FRACTION;
-      double received_rate = state_->getReceivedRate();
-      uint32_t round_without_nacks = state_->getRoundsWithoutNacks();
-      double cache_ratio = state_->getPacketFromCacheRatio();
+      double target_rate = state->getProducerRate() * PRODUCTION_RATE_FRACTION;
+      double received_rate =
+          state->getReceivedRate() + state->getRecoveredFecRate();
+      uint32_t round_without_nacks = state->getRoundsWithoutNacks();
+      double cache_ratio = state->getPacketFromCacheRatio();
       if (round_without_nacks >= ROUNDS_IN_SYNC_BEFORE_SWITCH &&
           received_rate >= target_rate && cache_ratio < MAX_DATA_FROM_CACHE) {
-        current_state_ = SyncState::in_sync;
+        ptr->current_state_ = SyncState::in_sync;
       }
     }
 
     DLOG_IF(INFO, VLOG_IS_ON(3))
         << "Calling updateSyncWindow in newRound function";
-    updateSyncWindow();
+    ptr->updateSyncWindow();
 
-    sendStatsToApp(sent_retx, received_bytes, sent_interest, lost_data,
-                   definitely_lost, recovered_losses, received_nacks,
-                   received_fec);
-    newRound();
+    ptr->sendStatsToApp(sent_retx, received_bytes, sent_interest, lost_data,
+                        definitely_lost, recovered_losses, received_nacks,
+                        received_fec);
+    ptr->fwd_strategy_.checkStrategy();
+    ptr->newRound();
   });
 }
 
 void RTCTransportProtocol::discoveredRtt() {
   start_send_interest_ = true;
-  ldr_->turnOnRTX();
+  uint32_t strategy;
+  socket_->getSocketOption(RtcTransportOptions::RECOVERY_STRATEGY, strategy);
+  ldr_->changeRecoveryStrategy(
+      (interface::RtcTransportRecoveryStrategies)strategy);
+  ldr_->turnOnRecovery();
   ldr_->onNewRound(false);
+
+  // set forwarding strategy switch if selected
+  Name *name = nullptr;
+  socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name);
+  Prefix prefix(*name, 128);
+  if ((interface::RtcTransportRecoveryStrategies)strategy ==
+      interface::RtcTransportRecoveryStrategies::LOW_RATE_AND_BESTPATH) {
+    fwd_strategy_.initFwdStrategy(portal_, prefix, state_.get(),
+                                  RTCForwardingStrategy::BEST_PATH);
+  } else if ((interface::RtcTransportRecoveryStrategies)strategy ==
+             interface::RtcTransportRecoveryStrategies::
+                 LOW_RATE_AND_REPLICATION) {
+    fwd_strategy_.initFwdStrategy(portal_, prefix, state_.get(),
+                                  RTCForwardingStrategy::REPLICATION);
+  } else if ((interface::RtcTransportRecoveryStrategies)strategy ==
+             interface::RtcTransportRecoveryStrategies::
+                 LOW_RATE_AND_ALL_FWD_STRATEGIES) {
+    fwd_strategy_.initFwdStrategy(portal_, prefix, state_.get(),
+                                  RTCForwardingStrategy::BOTH);
+  }
+
   updateSyncWindow();
 }
 
@@ -244,7 +314,7 @@ void RTCTransportProtocol::computeMaxSyncWindow() {
       (production_rate * lifetime_ms * INTEREST_LIFETIME_REDUCTION_FACTOR) /
       packet_size);
 
-  max_sync_win_ = std::min(max_sync_win_, rc_->getCongesionWindow());
+  max_sync_win_ = std::min(max_sync_win_, rc_->getCongestionWindow());
 }
 
 void RTCTransportProtocol::updateSyncWindow() {
@@ -259,25 +329,14 @@ void RTCTransportProtocol::updateSyncWindow() {
   }
 
   double prod_rate = state_->getProducerRate();
-  double rtt = (double)state_->getRTT() / MILLI_IN_A_SEC;
+  double rtt = (double)state_->getMinRTT() / MILLI_IN_A_SEC;
   double packet_size = state_->getAveragePacketSize();
 
   // if some of the info are not available do not update the current win
   if (prod_rate != 0.0 && rtt != 0.0 && packet_size != 0.0) {
-    double fec_interest_overhead = (double)state_->getPendingFecPackets() /
-                                   (double)(state_->getPendingInterestNumber() -
-                                            state_->getPendingFecPackets());
-
-    double fec_overhead =
-        std::max(indexer_verifier_->getFecOverhead(), fec_interest_overhead);
-
-    prod_rate += (prod_rate * fec_overhead);
-
     current_sync_win_ = (uint32_t)ceil(prod_rate * rtt / packet_size);
     uint32_t buffer = PRODUCER_BUFFER_MS;
-    if (rtt > 150)
-      buffer = buffer * 2;  // if the RTT is too large we increase the
-                            // the size of the buffer
+
     current_sync_win_ +=
         ceil(prod_rate * (buffer / MILLI_IN_A_SEC) / packet_size);
 
@@ -285,8 +344,17 @@ void RTCTransportProtocol::updateSyncWindow() {
       current_sync_win_ = current_sync_win_ * CATCH_UP_WIN_INCREMENT;
     }
 
+    uint32_t min_win = WIN_MIN;
+    bool aggregated_data_on;
+    socket_->getSocketOption(RtcTransportOptions::AGGREGATED_DATA,
+                             aggregated_data_on);
+    if (aggregated_data_on) {
+      min_win = WIN_MIN_WITH_AGGREGARED_DATA;
+      min_win += (min_win * (1 - (std::max(0.3, rtt) - rtt) / 0.3));
+    }
+
     current_sync_win_ = std::min(current_sync_win_, max_sync_win_);
-    current_sync_win_ = std::max(current_sync_win_, WIN_MIN);
+    current_sync_win_ = std::max(current_sync_win_, min_win);
   }
 
   scheduleNextInterests();
@@ -322,7 +390,7 @@ void RTCTransportProtocol::sendProbeInterest(uint32_t seq) {
   socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME,
                            &interest_name);
 
-  DLOG_IF(INFO, VLOG_IS_ON(3)) << "send probe " << seq;
+  DLOG_IF(INFO, VLOG_IS_ON(3)) << "Send probe " << seq;
   interest_name->setSuffix(seq);
   sendInterest(*interest_name);
 }
@@ -330,13 +398,18 @@ void RTCTransportProtocol::sendProbeInterest(uint32_t seq) {
 void RTCTransportProtocol::scheduleNextInterests() {
   DLOG_IF(INFO, VLOG_IS_ON(3)) << "Schedule next interests";
 
-  if (!isRunning() && !is_first_) return;
+  if (!isRunning() && !is_first_) {
+    return;
+  }
 
-  if (pacing_timer_on_) return;  // wait pacing timer for the next send
+  if (pacing_timer_on_) {
+    return;  // wait pacing timer for the next send
+  }
 
-  if (!start_send_interest_)
+  if (!start_send_interest_) {
     return;  // RTT discovering phase is not finished so
              // do not start to send interests
+  }
 
   if (TRANSPORT_EXPECT_FALSE(!state_->isProducerActive())) {
     DLOG_IF(INFO, VLOG_IS_ON(3)) << "Inactive producer.";
@@ -352,7 +425,7 @@ void RTCTransportProtocol::scheduleNextInterests() {
                              &interest_name);
 
     uint32_t next_seg = 0;
-    DLOG_IF(INFO, VLOG_IS_ON(3)) << "send interest " << next_seg;
+    DLOG_IF(INFO, VLOG_IS_ON(3)) << "Send interest " << next_seg;
     interest_name->setSuffix(next_seg);
 
     if (portal_->interestIsPending(*interest_name)) {
@@ -370,28 +443,37 @@ void RTCTransportProtocol::scheduleNextInterests() {
       << " -- current_sync_win_: " << current_sync_win_;
 
   uint32_t pending = state_->getPendingInterestNumber();
-  if (pending >= current_sync_win_) return;  // no space in the window
+  uint32_t pending_fec = state_->getPendingFecPackets();
+
+  if ((pending - pending_fec) >= current_sync_win_)
+    return;  // no space in the window
 
-  if ((current_sync_win_ - pending) < max_aggregated_interest_) {
+  // XXX double check if aggregated interests are still working here
+  if ((current_sync_win_ - (pending - pending_fec)) <
+      max_aggregated_interest_) {
     if (scheduler_timer_on_) return;  // timer already scheduled
 
-    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 time = now - last_interest_sent_time_;
     if (time < WAIT_FOR_INTEREST_BATCH) {
       uint64_t next = WAIT_FOR_INTEREST_BATCH - time;
       scheduler_timer_on_ = true;
       scheduler_timer_->expires_from_now(std::chrono::milliseconds(next));
-      scheduler_timer_->async_wait([this](std::error_code ec) {
+
+      std::weak_ptr<RTCTransportProtocol> self = shared_from_this();
+      scheduler_timer_->async_wait([self](const std::error_code &ec) {
         if (ec) return;
-        if (!scheduler_timer_on_) return;
 
-        scheduler_timer_on_ = false;
-        scheduleNextInterests();
+        auto ptr = self.lock();
+        if (ptr && ptr->isRunning()) {
+          if (!ptr->scheduler_timer_on_) return;
+
+          ptr->scheduler_timer_on_ = false;
+          ptr->scheduleNextInterests();
+        }
       });
-      return;  // whait for the timer
+      return;  // wait for the timer
     }
   }
 
@@ -403,7 +485,7 @@ void RTCTransportProtocol::scheduleNextInterests() {
     indexer_verifier_->jumpToIndex(state_->getLastSeqNacked() + 1);
   }
 
-  // skipe received packets
+  // skip received packets
   if (indexer_verifier_->checkNextSuffix() <=
       state_->getHighestSeqReceivedInOrder()) {
     indexer_verifier_->jumpToIndex(state_->getHighestSeqReceivedInOrder() + 1);
@@ -417,7 +499,8 @@ void RTCTransportProtocol::scheduleNextInterests() {
   socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name);
   std::array<uint32_t, MAX_AGGREGATED_INTEREST> additional_suffixes;
 
-  while ((state_->getPendingInterestNumber() < current_sync_win_) &&
+  while (((state_->getPendingInterestNumber() -
+           state_->getPendingFecPackets()) < current_sync_win_) &&
          (sent_interests < max_sent_int_)) {
     DLOG_IF(INFO, VLOG_IS_ON(3))
         << "In while loop. Window size: " << current_sync_win_;
@@ -428,19 +511,20 @@ void RTCTransportProtocol::scheduleNextInterests() {
 
     // send the packet only if:
     // 1) it is not pending yet (not true for rtx)
-    // 2) the packet is not received or lost
+    // 2) the packet is not received or def lost
     // 3) is not in the rtx list
     // 4) is fec and is not in order (!= last sent + 1)
+    PacketState packet_state = state_->getPacketState(next_seg);
     if (portal_->interestIsPending(*name) ||
-        state_->isReceivedOrLost(next_seg) != PacketState::UNKNOWN ||
-        ldr_->isRtx(next_seg) ||
+        packet_state == PacketState::RECEIVED ||
+        packet_state == PacketState::DEFINITELY_LOST || ldr_->isRtx(next_seg) ||
         (indexer_verifier_->isFec(next_seg) &&
          next_seg != last_interest_sent_seq_ + 1)) {
       DLOG_IF(INFO, VLOG_IS_ON(3))
           << "skip interest " << next_seg << " because: pending "
-          << portal_->interestIsPending(*name) << ", recv "
-          << (state_->isReceivedOrLost(next_seg) != PacketState::UNKNOWN)
-          << ", rtx " << (ldr_->isRtx(next_seg)) << ", is old fec "
+          << portal_->interestIsPending(*name) << ", recv or lost"
+          << (int)packet_state << ", rtx " << (ldr_->isRtx(next_seg))
+          << ", is old fec "
           << ((indexer_verifier_->isFec(next_seg) &&
                next_seg != last_interest_sent_seq_ + 1));
       continue;
@@ -462,10 +546,7 @@ void RTCTransportProtocol::scheduleNextInterests() {
       sent_packets++;
       sent_interests++;
       sendInterest(interest_name, &additional_suffixes, aggregated_counter - 1);
-      last_interest_sent_time_ =
-          std::chrono::duration_cast<std::chrono::milliseconds>(
-              std::chrono::steady_clock::now().time_since_epoch())
-              .count();
+      last_interest_sent_time_ = utils::SteadyTime::nowMs().count();
       aggregated_counter = 0;
     }
   }
@@ -473,25 +554,29 @@ void RTCTransportProtocol::scheduleNextInterests() {
   // exiting the while we may have some pending interest to send
   if (aggregated_counter != 0) {
     sent_packets++;
-    last_interest_sent_time_ =
-        std::chrono::duration_cast<std::chrono::milliseconds>(
-            std::chrono::steady_clock::now().time_since_epoch())
-            .count();
+    last_interest_sent_time_ = utils::SteadyTime::nowMs().count();
     sendInterest(interest_name, &additional_suffixes, aggregated_counter - 1);
   }
 
-  if (state_->getPendingInterestNumber() < current_sync_win_) {
+  if ((state_->getPendingInterestNumber() - state_->getPendingFecPackets()) <
+      current_sync_win_) {
     // we still have space in the window but we already sent too many packets
     // wait PACING_WAIT to avoid drops in the kernel
 
     pacing_timer_on_ = true;
     pacing_timer_->expires_from_now(std::chrono::microseconds(PACING_WAIT));
-    scheduler_timer_->async_wait([this](std::error_code ec) {
+
+    std::weak_ptr<RTCTransportProtocol> self = shared_from_this();
+    scheduler_timer_->async_wait([self](const std::error_code &ec) {
       if (ec) return;
-      if (!pacing_timer_on_) return;
 
-      pacing_timer_on_ = false;
-      scheduleNextInterests();
+      auto ptr = self.lock();
+      if (ptr && ptr->isRunning()) {
+        if (!ptr->pacing_timer_on_) return;
+
+        ptr->pacing_timer_on_ = false;
+        ptr->scheduleNextInterests();
+      }
     });
   }
 }
@@ -500,13 +585,13 @@ void RTCTransportProtocol::onInterestTimeout(Interest::Ptr &interest,
                                              const Name &name) {
   uint32_t segment_number = name.getSuffix();
 
-  if (segment_number >= MIN_PROBE_SEQ) {
+  if (ProbeHandler::getProbeType(segment_number) != ProbeType::NOT_PROBE) {
     // this is a timeout on a probe, do nothing
     return;
   }
 
-  PacketState state = state_->isReceivedOrLost(segment_number);
-  if (state != PacketState::UNKNOWN) {
+  PacketState state = state_->getPacketState(segment_number);
+  if (state == PacketState::RECEIVED || state == PacketState::DEFINITELY_LOST) {
     // we may recover a packets using fec, ignore this timer
     return;
   }
@@ -524,13 +609,18 @@ void RTCTransportProtocol::onInterestTimeout(Interest::Ptr &interest,
     DLOG_IF(INFO, VLOG_IS_ON(3))
         << "handle timeout for packet  " << segment_number << " using rtx";
     if (ldr_->isRtxOn()) {
-      ldr_->onTimeout(segment_number);
-      if (indexer_verifier_->isFec(segment_number))
+      if (indexer_verifier_->isFec(segment_number)) {
+        // if this is a fec packet we do not recover it with rtx so we consider
+        // the packet to be lost
+        ldr_->onTimeout(segment_number, true);
         state_->onTimeout(segment_number, true);
-      else
+      } else {
+        ldr_->onTimeout(segment_number, false);
         state_->onTimeout(segment_number, false);
+      }
     } else {
       // in this case we wil never recover the timeout
+      ldr_->onTimeout(segment_number, true);
       state_->onTimeout(segment_number, true);
     }
     scheduleNextInterests();
@@ -559,7 +649,7 @@ void RTCTransportProtocol::onInterestTimeout(Interest::Ptr &interest,
 void RTCTransportProtocol::onNack(const ContentObject &content_object) {
   struct nack_packet_t *nack =
       (struct nack_packet_t *)content_object.getPayload()->data();
-  uint32_t production_seg = nack->getProductionSegement();
+  uint32_t production_seg = nack->getProductionSegment();
   uint32_t nack_segment = content_object.getName().getSuffix();
   bool is_rtx = ldr_->isRtx(nack_segment);
 
@@ -592,6 +682,8 @@ void RTCTransportProtocol::onNack(const ContentObject &content_object) {
     // remove the nack is it exists
     if (tn_it != timeouts_or_nacks_.end()) timeouts_or_nacks_.erase(tn_it);
 
+    state_->onJumpForward(production_seg);
+    verifier_->onJumpForward(production_seg);
     // the client is asking for content in the past
     // switch to catch up state and increase the window
     // this is true only if the packet is not an RTX
@@ -618,11 +710,9 @@ void RTCTransportProtocol::onProbe(const ContentObject &content_object) {
   bool valid = state_->onProbePacketReceived(content_object);
   if (!valid) return;
 
-  struct nack_packet_t *probe =
-      (struct nack_packet_t *)content_object.getPayload()->data();
-  uint32_t production_seg = probe->getProductionSegement();
+  uint32_t production_seg = RTCState::getProbeParams(content_object).prod_seg;
 
-  // as for the nacks set next_segment
+  // As for the nacks set next_segment
   DLOG_IF(INFO, VLOG_IS_ON(3))
       << "on probe next seg = " << indexer_verifier_->checkNextSuffix()
       << ", jump to " << production_seg;
@@ -636,12 +726,39 @@ void RTCTransportProtocol::onContentObjectReceived(
     Interest &interest, ContentObject &content_object, std::error_code &ec) {
   DLOG_IF(INFO, VLOG_IS_ON(3))
       << "Received content object of size: " << content_object.payloadSize();
-  uint32_t payload_size = content_object.payloadSize();
+
   uint32_t segment_number = content_object.getName().getSuffix();
+  PayloadType payload_type = content_object.getPayloadType();
+  PacketState state;
+
+  ContentObject *content_ptr = &content_object;
+  ContentObject::Ptr manifest_ptr = nullptr;
+
+  bool is_probe =
+      ProbeHandler::getProbeType(segment_number) != ProbeType::NOT_PROBE;
+  bool is_nack = !is_probe && content_object.payloadSize() == NACK_HEADER_SIZE;
+  bool is_fec = indexer_verifier_->isFec(segment_number);
+  bool is_manifest =
+      !is_probe && !is_nack && !is_fec && payload_type == PayloadType::MANIFEST;
+  bool is_data =
+      !is_probe && !is_nack && !is_fec && payload_type == PayloadType::DATA;
+  bool compute_stats = is_data || is_manifest;
 
   ec = make_error_code(protocol_error::not_reassemblable);
 
-  if (segment_number >= MIN_PROBE_SEQ) {
+  // A helper function to process manifests or data packets received
+  auto onDataPacketReceived = [this](ContentObject &content_object,
+                                     bool compute_stats) {
+    ldr_->onDataPacketReceived(content_object);
+    rc_->onDataPacketReceived(content_object, compute_stats);
+    updateSyncWindow();
+  };
+
+  // First verify the packet signature and apply the corresponding policy
+  auth::VerificationPolicy policy = verifier_->verify(content_object, is_fec);
+  indexer_verifier_->applyPolicy(interest, content_object, false, policy);
+
+  if (is_probe) {
     DLOG_IF(INFO, VLOG_IS_ON(3)) << "Received probe " << segment_number;
     if (*on_content_object_input_) {
       (*on_content_object_input_)(*socket_->getInterface(), content_object);
@@ -650,7 +767,7 @@ void RTCTransportProtocol::onContentObjectReceived(
     return;
   }
 
-  if (payload_size == NACK_HEADER_SIZE) {
+  if (is_nack) {
     DLOG_IF(INFO, VLOG_IS_ON(3)) << "Received nack " << segment_number;
     if (*on_content_object_input_) {
       (*on_content_object_input_)(*socket_->getInterface(), content_object);
@@ -659,57 +776,122 @@ void RTCTransportProtocol::onContentObjectReceived(
     return;
   }
 
+  // content_ptr will point either to the input data packet or to a manifest
+  // whose FEC header has been removed
+  if (is_manifest) {
+    manifest_ptr = removeFecHeader(content_object);
+    if (manifest_ptr) {
+      content_ptr = manifest_ptr.get();
+    }
+  }
+
+  // From there, the packet is either a FEC, a manifest or a data packet.
   DLOG_IF(INFO, VLOG_IS_ON(3)) << "Received content " << segment_number;
 
-  bool compute_stats = true;
+  // Do not count timed out packets in stats
   auto tn_it = timeouts_or_nacks_.find(segment_number);
   if (tn_it != timeouts_or_nacks_.end()) {
     compute_stats = false;
     timeouts_or_nacks_.erase(tn_it);
   }
-  if (ldr_->isRtx(segment_number)) {
+
+  // Do not count retransmissions or losses in stats
+  if (ldr_->isRtx(segment_number) ||
+      ldr_->isPossibleLossWithNoRtx(segment_number)) {
     compute_stats = false;
   }
 
-  // check if the packet was already received
-  PacketState state = state_->isReceivedOrLost(segment_number);
+  // Fetch packet state
+  state = state_->getPacketState(segment_number);
 
-  if (state != PacketState::RECEIVED) {
-    // send packet to decoder
-    if (fec_decoder_) {
-      DLOG_IF(INFO, VLOG_IS_ON(4))
-          << "send packet " << segment_number << " to FEC decoder";
-      fec_decoder_->onDataPacket(
-          content_object, content_object.headerSize() + rtc::DATA_HEADER_SIZE);
-    }
-    if (!indexer_verifier_->isFec(segment_number)) {
-      // the packet may be alredy sent to the ap by the decoder, check again if
-      // it is already received
-      state = state_->isReceivedOrLost(segment_number);
-      if (state != PacketState::RECEIVED) {
-        DLOG_IF(INFO, VLOG_IS_ON(4)) << "Received content " << segment_number;
+  // Check if the packet is a retransmission
+  if (ldr_->isRtx(segment_number) && state != PacketState::RECEIVED) {
+    if (is_data || is_manifest) {
+      state_->onPacketRecoveredRtx(segment_number);
 
-        state_->onDataPacketReceived(content_object, compute_stats);
+      if (*on_content_object_input_) {
+        (*on_content_object_input_)(*socket_->getInterface(), content_object);
+      }
 
-        if (*on_content_object_input_) {
-          (*on_content_object_input_)(*socket_->getInterface(), content_object);
-        }
-        ec = make_error_code(protocol_error::success);
+      if (is_manifest) {
+        processManifest(interest, *content_ptr);
       }
-    } else {
-      DLOG_IF(INFO, VLOG_IS_ON(4)) << "Received fec " << segment_number;
-      state_->onFecPacketReceived(content_object);
+
+      ec = is_manifest ? make_error_code(protocol_error::not_reassemblable)
+                       : make_error_code(protocol_error::success);
+
+      // The packet is considered received, return early
+      onDataPacketReceived(*content_ptr, compute_stats);
+      return;
     }
-  } else {
+
+    if (is_fec) {
+      state_->onFecPacketRecoveredRtx(segment_number);
+    }
+  }
+
+  // Fetch packet state again; it may have changed
+  state = state_->getPacketState(segment_number);
+
+  // Check if the packet was already received
+  if (state == PacketState::RECEIVED || state == PacketState::TO_BE_RECEIVED) {
     DLOG_IF(INFO, VLOG_IS_ON(3))
         << "Received duplicated content " << segment_number << ", drop it";
     ec = make_error_code(protocol_error::duplicated_content);
+    onDataPacketReceived(*content_ptr, compute_stats);
+    return;
   }
 
-  ldr_->onDataPacketReceived(content_object);
-  rc_->onDataPacketReceived(content_object);
+  if (!is_fec) {
+    state_->dataToBeReceived(segment_number);
+  }
 
-  updateSyncWindow();
+  // Send packet to FEC decoder
+  if (fec_decoder_) {
+    DLOG_IF(INFO, VLOG_IS_ON(4))
+        << "Send packet " << segment_number << " to FEC decoder";
+
+    uint32_t offset = is_manifest
+                          ? content_object.headerSize()
+                          : content_object.headerSize() + rtc::DATA_HEADER_SIZE;
+    uint32_t metadata = static_cast<uint32_t>(content_object.getPayloadType());
+
+    fec_decoder_->onDataPacket(content_object, offset, metadata);
+  }
+
+  // We can return early if FEC
+  if (is_fec) {
+    DLOG_IF(INFO, VLOG_IS_ON(4)) << "Received FEC " << segment_number;
+    state_->onFecPacketReceived(content_object);
+    onDataPacketReceived(*content_ptr, compute_stats);
+    return;
+  }
+
+  // The packet may have been already sent to the app by the decoder, check
+  // again if it is already received
+  state = state_->getPacketState(
+      segment_number);  // state == RECEIVED or TO_BE_RECEIVED
+
+  if (state != PacketState::RECEIVED) {
+    DLOG_IF(INFO, VLOG_IS_ON(4))
+        << (is_manifest ? "Received manifest " : "Received data ")
+        << segment_number;
+
+    if (is_manifest) {
+      processManifest(interest, *content_ptr);
+    }
+
+    state_->onDataPacketReceived(*content_ptr, compute_stats);
+
+    if (*on_content_object_input_) {
+      (*on_content_object_input_)(*socket_->getInterface(), content_object);
+    }
+
+    ec = is_manifest ? make_error_code(protocol_error::not_reassemblable)
+                     : make_error_code(protocol_error::success);
+  }
+
+  onDataPacketReceived(*content_ptr, compute_stats);
 }
 
 void RTCTransportProtocol::sendStatsToApp(
@@ -729,33 +911,149 @@ void RTCTransportProtocol::sendStatsToApp(
     stats_->updateReceivedNacks(received_nacks);
     stats_->updateReceivedFEC(received_fec);
 
-    stats_->updateAverageWindowSize(current_sync_win_);
-    stats_->updateLossRatio(state_->getLossRate());
-    stats_->updateAverageRtt(state_->getRTT());
+    stats_->updateAverageWindowSize(state_->getPendingInterestNumber());
+    stats_->updateLossRatio(state_->getPerSecondLossRate());
+    uint64_t rtt = state_->getAvgRTT();
+    stats_->updateAverageRtt(utils::SteadyTime::Milliseconds(rtt));
+
     stats_->updateQueuingDelay(state_->getQueuing());
     stats_->updateLostData(lost_data);
     stats_->updateDefinitelyLostData(definitely_lost);
     stats_->updateRecoveredData(recovered_losses);
     stats_->updateCCState((unsigned int)current_state_ ? 1 : 0);
     (*stats_summary_)(*socket_->getInterface(), *stats_);
+    bool in_congestion = rc_->inCongestionState();
+    stats_->updateCongestionState(in_congestion);
+    double residual_losses = state_->getResidualLossRate();
+    stats_->updateResidualLossRate(residual_losses);
+    stats_->updateQualityScore(state_->getQualityScore());
+
+    // set alerts
+    if (rtt > MAX_RTT)
+      stats_->setAlert(interface::TransportStatistics::statsAlerts::LATENCY);
+    else
+      stats_->clearAlert(interface::TransportStatistics::statsAlerts::LATENCY);
+
+    if (in_congestion)
+      stats_->setAlert(interface::TransportStatistics::statsAlerts::CONGESTION);
+    else
+      stats_->clearAlert(
+          interface::TransportStatistics::statsAlerts::CONGESTION);
+
+    if (residual_losses > MAX_RESIDUAL_LOSSES)
+      stats_->setAlert(interface::TransportStatistics::statsAlerts::LOSSES);
+    else
+      stats_->clearAlert(interface::TransportStatistics::statsAlerts::LOSSES);
   }
 }
 
-void RTCTransportProtocol::onFecPackets(
-    std::vector<std::pair<uint32_t, fec::buffer>> &packets) {
+void RTCTransportProtocol::onFecPackets(fec::BufferArray &packets) {
+  Packet::Format format;
+  socket_->getSocketOption(interface::GeneralTransportOptions::PACKET_FORMAT,
+                           format);
+
+  Name *name = nullptr;
+  socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, &name);
+
   for (auto &packet : packets) {
-    PacketState state = state_->isReceivedOrLost(packet.first);
-    if (state != PacketState::RECEIVED) {
-      state_->onPacketRecoveredFec(packet.first);
-      ldr_->onPacketRecoveredFec(packet.first);
-      DLOG_IF(INFO, VLOG_IS_ON(3))
-          << "Recovered packet " << packet.first << " through FEC.";
-      reassembly_->reassemble(*packet.second, packet.first);
-    } else {
-      DLOG_IF(INFO, VLOG_IS_ON(3))
-          << "Packet" << packet.first << "already received.";
+    uint32_t seq_number = packet.getIndex();
+    uint32_t metadata = packet.getMetadata();
+    fec::buffer buffer = packet.getBuffer();
+
+    PayloadType payload_type = static_cast<PayloadType>(metadata);
+    switch (payload_type) {
+      case PayloadType::DATA:
+      case PayloadType::MANIFEST:
+        break;
+      case PayloadType::UNSPECIFIED:
+      default:
+        payload_type = PayloadType::DATA;
+        break;
     }
+
+    switch (state_->getPacketState(seq_number)) {
+      case PacketState::RECEIVED:
+      case PacketState::TO_BE_RECEIVED: {
+        DLOG_IF(INFO, VLOG_IS_ON(3))
+            << "Packet " << seq_number << " already received";
+        break;
+      }
+      default: {
+        DLOG_IF(INFO, VLOG_IS_ON(3))
+            << "Recovered packet " << seq_number << " through FEC";
+
+        if (payload_type == PayloadType::MANIFEST) {
+          name->setSuffix(seq_number);
+
+          auto interest =
+              core::PacketManager<>::getInstance().getPacket<Interest>(format);
+          interest->setName(*name);
+
+          auto content_object = toContentObject(
+              *name, format, payload_type, buffer->data(), buffer->length());
+
+          processManifest(*interest, *content_object);
+        }
+
+        state_->onPacketRecoveredFec(seq_number, buffer->length());
+        ldr_->onPacketRecoveredFec(seq_number);
+
+        if (payload_type == PayloadType::DATA) {
+          verifier_->onDataRecoveredFec(seq_number);
+          reassembly_->reassemble(*buffer, seq_number);
+        }
+
+        break;
+      }
+    }
+  }
+}
+
+void RTCTransportProtocol::processManifest(Interest &interest,
+                                           ContentObject &manifest) {
+  auth::VerificationPolicy policy = verifier_->processManifest(manifest);
+  indexer_verifier_->applyPolicy(interest, manifest, false, policy);
+}
+
+ContentObject::Ptr RTCTransportProtocol::removeFecHeader(
+    const ContentObject &content_object) {
+  if (!fec_decoder_ || !fec_decoder_->getFecHeaderSize()) {
+    return nullptr;
   }
+
+  size_t fec_header_size = fec_decoder_->getFecHeaderSize();
+  const uint8_t *payload =
+      content_object.data() + content_object.headerSize() + fec_header_size;
+  size_t payload_size = content_object.payloadSize() - fec_header_size;
+
+  ContentObject::Ptr co =
+      toContentObject(content_object.getName(), content_object.getFormat(),
+                      content_object.getPayloadType(), payload, payload_size);
+
+  return co;
+}
+
+ContentObject::Ptr RTCTransportProtocol::toContentObject(
+    const Name &name, Packet::Format format, PayloadType payload_type,
+    const uint8_t *payload, std::size_t payload_size,
+    std::size_t additional_header_size) {
+  // Recreate ContentObject
+  ContentObject::Ptr co =
+      core::PacketManager<>::getInstance().getPacket<ContentObject>(
+          format, additional_header_size);
+  co->updateLength(payload_size);
+  co->append(payload_size);
+  co->trimStart(co->headerSize());
+
+  // Copy payload
+  std::memcpy(co->writableData(), payload, payload_size);
+
+  // Restore network headers and some fields
+  co->prepend(co->headerSize());
+  co->setName(name);
+  co->setPayloadType(payload_type);
+
+  return co;
 }
 
 }  // end namespace rtc