[HICN-690] Transport Library Major Refactory

The current patch provides a major refactory of the transportlibrary.
A summary of the different components that underwent major modifications is
reported below.

- Transport protocol updates
The hierarchy of classes has been optimized to have common transport services
across different transport protocols. This can allow to customize a transport
protocol with new features.

- A new real-time communication protocol
The RTC protocol has been optimized in terms of algorithms to reduce
consumer-producer synchronization latency.

- A novel socket API
The API has been reworked to be easier to consumer but also to have a more
efficient integration in L4 proxies.

- Several performance improvements
A large number of performance improvements have been included in
particular to make the entire stack zero-copy and optimize cache miss.

- New memory buffer framework
Memory management has been reworked entirely to provide a more efficient infra
with a richer API. Buffers are now allocated in blocks and a single buffer
holds the memory for (1) the shared_ptr control block, (2) the metadata of the
packet (e.g. name, pointer to other buffers if buffer is chained and relevant
offsets), and (3) the packet itself, as it is sent/received over the network.

- A new slab allocator
Dynamic memory allocation is now managed by a novel slab allocator that is
optimised for packet processing and connection management. Memory is organized
in pools of blocks all of the same size which are used during the processing of
outgoing/incoming packets. When a memory block Is allocated is always taken
from a global pool and when it is deallocated is returned to the pool, thus
avoiding the cost of any heap allocation in the data path.

- New transport connectors
Consumer and producer end-points can communication either using an hicn packet
forwarder or with direct connector based on shared memories or sockets.
The usage of transport connectors typically for unit and funcitonal
testing but may have additional usage.

- Support for FEC/ECC for transport services
FEC/ECC via reed solomon is supported by default and made available to
transport services as a modular component. Reed solomon block codes is a
default FEC model that can be replaced in a modular way by many other
codes including RLNC not avaiable in this distribution.
The current FEC framework support variable size padding and efficiently
makes use of the infra memory buffers to avoid additiona copies.

- Secure transport framework for signature computation and verification
Crypto support is nativelty used in hICN for integrity and authenticity.
Novel support that includes RTC has been implemented and made modular
and reusable acrosso different transport protocols.

- TLS - Transport layer security over hicn
Point to point confidentiality is provided by integrating TLS on top of
hICN reliable and non-reliable transport. The integration is common and
makes a different use of the TLS record.

- MLS - Messaging layer security over hicn
MLS integration on top of hICN is made by using the MLSPP implemetation
open sourced by Cisco. We have included instrumentation tools to deploy
performance and functional tests of groups of end-points.

- Android support
The overall code has been heavily tested in Android environments and
has received heavy lifting to better run natively in recent Android OS.

Co-authored-by: Mauro Sardara <msardara@cisco.com>
Co-authored-by: Michele Papalini <micpapal@cisco.com>
Co-authored-by: Olivier Roques <oroques+fdio@cisco.com>
Co-authored-by: Giulio Grassi <gigrassi@cisco.com>
Change-Id: If477ba2fa686e6f47bdf96307ac60938766aef69
Signed-off-by: Luca Muscariello <muscariello@ieee.org>

1 files changed, 427 insertions, 0 deletions

diff --git a/libtransport/src/protocols/rtc/rtc_ldr.cc b/libtransport/src/protocols/rtc/rtc_ldr.cc
new file mode 100644
index 000000000..e91b29c04
--- /dev/null
+++ b/libtransport/src/protocols/rtc/rtc_ldr.cc
@@ -0,0 +1,427 @@
+/*
+ * Copyright (c) 2017-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:
+ *
+ *     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 <protocols/rtc/rtc_consts.h>
+#include <protocols/rtc/rtc_ldr.h>
+
+#include <algorithm>
+#include <unordered_set>
+
+namespace transport {
+
+namespace protocol {
+
+namespace rtc {
+
+RTCLossDetectionAndRecovery::RTCLossDetectionAndRecovery(
+    SendRtxCallback &&callback, asio::io_service &io_service)
+    : rtx_on_(false),
+      next_rtx_timer_(MAX_TIMER_RTX),
+      last_event_(0),
+      sentinel_timer_interval_(MAX_TIMER_RTX),
+      send_rtx_callback_(std::move(callback)) {
+  timer_ = std::make_unique<asio::steady_timer>(io_service);
+  sentinel_timer_ = std::make_unique<asio::steady_timer>(io_service);
+}
+
+RTCLossDetectionAndRecovery::~RTCLossDetectionAndRecovery() {}
+
+void RTCLossDetectionAndRecovery::turnOnRTX() {
+  rtx_on_ = true;
+  scheduleSentinelTimer(state_->getRTT() * CATCH_UP_RTT_INCREMENT);
+}
+
+void RTCLossDetectionAndRecovery::turnOffRTX() {
+  rtx_on_ = false;
+  clear();
+}
+
+void RTCLossDetectionAndRecovery::onTimeout(uint32_t seq) {
+  // always add timeouts to the RTX list to avoid to send the same packet as if
+  // it was not a rtx
+  addToRetransmissions(seq, seq + 1);
+  last_event_ = getNow();
+}
+
+void RTCLossDetectionAndRecovery::onDataPacketReceived(
+    const core::ContentObject &content_object) {
+  last_event_ = getNow();
+
+  uint32_t seq = content_object.getName().getSuffix();
+  if (deleteRtx(seq)) {
+    state_->onPacketRecovered(seq);
+  } else {
+    if (TRANSPORT_EXPECT_FALSE(!rtx_on_)) return;  // do not add if RTX is off
+    TRANSPORT_LOGD("received data. add from %u to %u ",
+                   state_->getHighestSeqReceivedInOrder() + 1, seq);
+    addToRetransmissions(state_->getHighestSeqReceivedInOrder() + 1, seq);
+  }
+}
+
+void RTCLossDetectionAndRecovery::onNackPacketReceived(
+    const core::ContentObject &nack) {
+  last_event_ = getNow();
+
+  uint32_t seq = nack.getName().getSuffix();
+
+  if (TRANSPORT_EXPECT_FALSE(!rtx_on_)) return;  // do not add if RTX is off
+
+  struct nack_packet_t *nack_pkt =
+      (struct nack_packet_t *)nack.getPayload()->data();
+  uint32_t production_seq = nack_pkt->getProductionSegement();
+
+  if (production_seq > seq) {
+    // this is a past nack, all data before productionSeq are lost. if
+    // productionSeq > state_->getHighestSeqReceivedInOrder() is impossible to
+    // recover any packet. If this is not the case we can try to recover the
+    // packets between state_->getHighestSeqReceivedInOrder() and productionSeq.
+    // e.g.: the client receives packets 8 10 11 9 where 9 is a nack with
+    // productionSeq = 14. 9 is lost but we can try to recover packets 12 13 and
+    // 14 that are not arrived yet
+    deleteRtx(seq);
+    TRANSPORT_LOGD("received past nack. add from %u to %u ",
+                   state_->getHighestSeqReceivedInOrder() + 1, production_seq);
+    addToRetransmissions(state_->getHighestSeqReceivedInOrder() + 1,
+                         production_seq);
+  } else {
+    // future nack. here there should be a gap between the last data received
+    // and this packet and is it possible to recover the packets between the
+    // last received data and the production seq. we should not use the seq
+    // number of the nack since we know that is too early to ask for this seq
+    // number
+    // e.g.: // e.g.: the client receives packets 10 11 12 20 where 20 is a nack
+    // with productionSeq = 18. this says that all the packets between 12 and 18
+    // may got lost and we should ask them
+    deleteRtx(seq);
+    TRANSPORT_LOGD("received futrue nack. add from %u to %u ",
+                   state_->getHighestSeqReceivedInOrder() + 1, production_seq);
+    addToRetransmissions(state_->getHighestSeqReceivedInOrder() + 1,
+                         production_seq);
+  }
+}
+
+void RTCLossDetectionAndRecovery::onProbePacketReceived(
+    const core::ContentObject &probe) {
+  // we don't log the reception of a probe packet for the sentinel timer because
+  // probes are not taken into account into the sync window. we use them as
+  // future nacks to detect possible packets lost
+  if (TRANSPORT_EXPECT_FALSE(!rtx_on_)) return;  // do not add if RTX is off
+  struct nack_packet_t *probe_pkt =
+      (struct nack_packet_t *)probe.getPayload()->data();
+  uint32_t production_seq = probe_pkt->getProductionSegement();
+  TRANSPORT_LOGD("received probe. add from %u to %u ",
+                 state_->getHighestSeqReceivedInOrder() + 1, production_seq);
+  addToRetransmissions(state_->getHighestSeqReceivedInOrder() + 1,
+                       production_seq);
+}
+
+void RTCLossDetectionAndRecovery::clear() {
+  rtx_state_.clear();
+  rtx_timers_.clear();
+  sentinel_timer_->cancel();
+  if (next_rtx_timer_ != MAX_TIMER_RTX) {
+    next_rtx_timer_ = MAX_TIMER_RTX;
+    timer_->cancel();
+  }
+}
+
+void RTCLossDetectionAndRecovery::addToRetransmissions(uint32_t start,
+                                                       uint32_t stop) {
+  // skip nacked packets
+  if (start <= state_->getLastSeqNacked()) {
+    start = state_->getLastSeqNacked() + 1;
+  }
+
+  // skip received or lost packets
+  if (start <= state_->getHighestSeqReceivedInOrder()) {
+    start = state_->getHighestSeqReceivedInOrder() + 1;
+  }
+
+  for (uint32_t seq = start; seq < stop; seq++) {
+    if (!isRtx(seq) &&  // is not already an rtx
+                        // is not received or lost
+        state_->isReceivedOrLost(seq) == PacketState::UNKNOWN) {
+      // add rtx
+      rtxState state;
+      state.first_send_ = state_->getInterestSentTime(seq);
+      if (state.first_send_ == 0)  // this interest was never sent before
+        state.first_send_ = getNow();
+      state.next_send_ = computeNextSend(seq, true);
+      state.rtx_count_ = 0;
+      TRANSPORT_LOGD("add %u to retransmissions. next rtx is %lu ", seq,
+                     (state.next_send_ - getNow()));
+      rtx_state_.insert(std::pair<uint32_t, rtxState>(seq, state));
+      rtx_timers_.insert(std::pair<uint64_t, uint32_t>(state.next_send_, seq));
+    }
+  }
+  scheduleNextRtx();
+}
+
+uint64_t RTCLossDetectionAndRecovery::computeNextSend(uint32_t seq,
+                                                      bool new_rtx) {
+  uint64_t now = getNow();
+  if (new_rtx) {
+    // for the new rtx we wait one estimated IAT after the loss detection. this
+    // is bacause, assuming that packets arrive with a constant IAT, we should
+    // get a new packet every IAT
+    double prod_rate = state_->getProducerRate();
+    uint32_t estimated_iat = SENTINEL_TIMER_INTERVAL;
+    uint32_t jitter = 0;
+
+    if (prod_rate != 0) {
+      double packet_size = state_->getAveragePacketSize();
+      estimated_iat = ceil(1000.0 / (prod_rate / packet_size));
+      jitter = ceil(state_->getJitter());
+    }
+
+    uint32_t wait = estimated_iat + jitter;
+    TRANSPORT_LOGD("first rtx for %u in %u ms, rtt = %lu ait = %u jttr = %u",
+                   seq, wait, state_->getRTT(), estimated_iat, jitter);
+
+    return now + wait;
+  } else {
+    // wait one RTT
+    // however if the IAT is larger than the RTT, wait one IAT
+    uint32_t wait = SENTINEL_TIMER_INTERVAL;
+
+    double prod_rate = state_->getProducerRate();
+    if (prod_rate == 0) {
+      return now + SENTINEL_TIMER_INTERVAL;
+    }
+
+    double packet_size = state_->getAveragePacketSize();
+    uint32_t estimated_iat = ceil(1000.0 / (prod_rate / packet_size));
+
+    uint64_t rtt = state_->getRTT();
+    if (rtt == 0) rtt = SENTINEL_TIMER_INTERVAL;
+    wait = rtt;
+
+    if (estimated_iat > rtt) wait = estimated_iat;
+
+    uint32_t jitter = ceil(state_->getJitter());
+    wait += jitter;
+
+    // it may happen that the channel is congested and we have some additional
+    // queuing delay to take into account
+    uint32_t queue = ceil(state_->getQueuing());
+    wait += queue;
+
+    TRANSPORT_LOGD(
+        "next rtx for %u in %u ms, rtt = %lu ait = %u jttr = %u queue = %u",
+        seq, wait, state_->getRTT(), estimated_iat, jitter, queue);
+
+    return now + wait;
+  }
+}
+
+void RTCLossDetectionAndRecovery::retransmit() {
+  if (rtx_timers_.size() == 0) return;
+
+  uint64_t now = getNow();
+
+  auto it = rtx_timers_.begin();
+  std::unordered_set<uint32_t> lost_pkt;
+  uint32_t sent_counter = 0;
+  while (it != rtx_timers_.end() && it->first <= now &&
+         sent_counter < MAX_INTERESTS_IN_BATCH) {
+    uint32_t seq = it->second;
+    auto rtx_it =
+        rtx_state_.find(seq);  // this should always return a valid iter
+    if (rtx_it->second.rtx_count_ >= RTC_MAX_RTX ||
+        (now - rtx_it->second.first_send_) >= RTC_MAX_AGE ||
+        seq < state_->getLastSeqNacked()) {
+      // max rtx reached or packet too old or packet nacked, this packet is lost
+      TRANSPORT_LOGD(
+          "packet %u lost because 1) max rtx: %u 2) max age: %u 3) naked: %u",
+          seq, (rtx_it->second.rtx_count_ >= RTC_MAX_RTX),
+          ((now - rtx_it->second.first_send_) >= RTC_MAX_AGE),
+          (seq < state_->getLastSeqNacked()));
+      lost_pkt.insert(seq);
+      it++;
+    } else {
+      // resend the packet
+      state_->onRetransmission(seq);
+      double prod_rate = state_->getProducerRate();
+      if (prod_rate != 0) rtx_it->second.rtx_count_++;
+      rtx_it->second.next_send_ = computeNextSend(seq, false);
+      it = rtx_timers_.erase(it);
+      rtx_timers_.insert(
+          std::pair<uint64_t, uint32_t>(rtx_it->second.next_send_, seq));
+      TRANSPORT_LOGD("send rtx for sequence %u, next send in %lu", seq,
+                     (rtx_it->second.next_send_ - now));
+      send_rtx_callback_(seq);
+      sent_counter++;
+    }
+  }
+
+  // remove packets if needed
+  for (auto lost_it = lost_pkt.begin(); lost_it != lost_pkt.end(); lost_it++) {
+    uint32_t seq = *lost_it;
+    state_->onPacketLost(seq);
+    deleteRtx(seq);
+  }
+}
+
+void RTCLossDetectionAndRecovery::scheduleNextRtx() {
+  if (rtx_timers_.size() == 0) {
+    // all the rtx were removed, reset timer
+    next_rtx_timer_ = MAX_TIMER_RTX;
+    return;
+  }
+
+  // check if timer is alreay set
+  if (next_rtx_timer_ != MAX_TIMER_RTX) {
+    // a new check for rtx is already scheduled
+    if (next_rtx_timer_ > rtx_timers_.begin()->first) {
+      // we need to re-schedule it
+      timer_->cancel();
+    } else {
+      // wait for the next timer
+      return;
+    }
+  }
+
+  // set a new timer
+  next_rtx_timer_ = rtx_timers_.begin()->first;
+  uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
+                     std::chrono::steady_clock::now().time_since_epoch())
+                     .count();
+  uint64_t wait = 1;
+  if (next_rtx_timer_ != MAX_TIMER_RTX && next_rtx_timer_ > now)
+    wait = next_rtx_timer_ - now;
+
+  std::weak_ptr<RTCLossDetectionAndRecovery> self(shared_from_this());
+  timer_->expires_from_now(std::chrono::milliseconds(wait));
+  timer_->async_wait([self](std::error_code ec) {
+    if (ec) return;
+    if (auto s = self.lock()) {
+      s->retransmit();
+      s->next_rtx_timer_ = MAX_TIMER_RTX;
+      s->scheduleNextRtx();
+    }
+  });
+}
+
+bool RTCLossDetectionAndRecovery::deleteRtx(uint32_t seq) {
+  auto it_rtx = rtx_state_.find(seq);
+  if (it_rtx == rtx_state_.end()) return false;  // rtx not found
+
+  uint64_t ts = it_rtx->second.next_send_;
+  auto it_timers = rtx_timers_.find(ts);
+  while (it_timers != rtx_timers_.end() && it_timers->first == ts) {
+    if (it_timers->second == seq) {
+      rtx_timers_.erase(it_timers);
+      break;
+    }
+    it_timers++;
+  }
+
+  bool lost = it_rtx->second.rtx_count_ > 0;
+  rtx_state_.erase(it_rtx);
+
+  return lost;
+}
+
+void RTCLossDetectionAndRecovery::scheduleSentinelTimer(
+    uint64_t expires_from_now) {
+  std::weak_ptr<RTCLossDetectionAndRecovery> self(shared_from_this());
+  sentinel_timer_->expires_from_now(
+      std::chrono::milliseconds(expires_from_now));
+  sentinel_timer_->async_wait([self](std::error_code ec) {
+    if (ec) return;
+    if (auto s = self.lock()) {
+      s->sentinelTimer();
+    }
+  });
+}
+
+void RTCLossDetectionAndRecovery::sentinelTimer() {
+  uint64_t now = getNow();
+
+  bool expired = false;
+  bool sent = false;
+  if ((now - last_event_) >= sentinel_timer_interval_) {
+    // at least a sentinel_timer_interval_ elapsed since last event
+    expired = true;
+    if (TRANSPORT_EXPECT_FALSE(!state_->isProducerActive())) {
+      // this happens at the beginning (or if the producer stops for some
+      // reason) we need to keep sending interest 0 until we get an answer
+      TRANSPORT_LOGD(
+          "sentinel timer: the producer is not active, send packet 0");
+      state_->onRetransmission(0);
+      send_rtx_callback_(0);
+    } else {
+      TRANSPORT_LOGD(
+          "sentinel timer: the producer is active, send the 10 oldest packets");
+      sent = true;
+      uint32_t rtx = 0;
+      auto it = state_->getPendingInterestsMapBegin();
+      auto end = state_->getPendingInterestsMapEnd();
+      while (it != end && rtx < MAX_RTX_WITH_SENTINEL) {
+        uint32_t seq = it->first;
+        TRANSPORT_LOGD("sentinel timer, add %u to the rtx list", seq);
+        addToRetransmissions(seq, seq + 1);
+        rtx++;
+        it++;
+      }
+    }
+  } else {
+    // sentinel timer did not expire because we registered at least one event
+  }
+
+  uint32_t next_timer;
+  double prod_rate = state_->getProducerRate();
+  if (TRANSPORT_EXPECT_FALSE(!state_->isProducerActive()) || prod_rate == 0) {
+    TRANSPORT_LOGD("next timer in %u", SENTINEL_TIMER_INTERVAL);
+    next_timer = SENTINEL_TIMER_INTERVAL;
+  } else {
+    double prod_rate = state_->getProducerRate();
+    double packet_size = state_->getAveragePacketSize();
+    uint32_t estimated_iat = ceil(1000.0 / (prod_rate / packet_size));
+    uint32_t jitter = ceil(state_->getJitter());
+
+    // try to reduce the number of timers if the estimated IAT is too small
+    next_timer = std::max((estimated_iat + jitter) * 20, (uint32_t)1);
+    TRANSPORT_LOGD("next sentinel in %u ms, rate: %f, iat: %u, jitter: %u",
+                   next_timer, ((prod_rate * 8.0) / 1000000.0), estimated_iat,
+                   jitter);
+
+    if (!expired) {
+      // discount the amout of time that is already passed
+      uint32_t discount = now - last_event_;
+      if (next_timer > discount) {
+        next_timer = next_timer - discount;
+      } else {
+        // in this case we trigger the timer in 1 ms
+        next_timer = 1;
+      }
+      TRANSPORT_LOGD("timer after discout: %u", next_timer);
+    } else if (sent) {
+      // wait at least one producer stats interval + owd to check if the
+      // production rate is reducing.
+      uint32_t min_wait = PRODUCER_STATS_INTERVAL + ceil(state_->getQueuing());
+      next_timer = std::max(next_timer, min_wait);
+      TRANSPORT_LOGD("wait for updates from prod, next timer: %u", next_timer);
+    }
+  }
+
+  scheduleSentinelTimer(next_timer);
+}
+
+}  // namespace rtc
+
+}  // namespace protocol
+
+}  // namespace transport