integrating the existing state of TON Storage / TON Payments / CPS Fift development branches

rldp2/Ack.cpp

@@ -0,0 +1,29 @@
+#include "Ack.h"
+
+namespace ton {
+namespace rldp2 {
+
+bool Ack::on_got_packet(td::uint32 seqno) {
+  if (seqno > max_seqno) {
+    td::uint32 diff = seqno - max_seqno;
+    if (diff >= 32) {
+      received_mask = 0;
+    } else {
+      received_mask <<= diff;
+    }
+    max_seqno = seqno;
+  }
+  td::uint32 offset = max_seqno - seqno;
+  if (offset < 32) {
+    td::uint32 mask = 1 << offset;
+    if ((received_mask & mask) == 0) {
+      received_count++;
+      received_mask |= mask;
+      return true;
+    }
+  }
+
+  return false;
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/Ack.h

@@ -0,0 +1,18 @@
+#pragma once
+
+#include "td/utils/int_types.h"
+
+namespace ton {
+namespace rldp2 {
+// Helper for receiver
+// Also this information is sent to the sender as an acknowlegement.
+struct Ack {
+  td::uint32 max_seqno{0};
+  td::uint32 received_mask{0};
+  td::uint32 received_count{0};
+
+  // returns true if we know that packet is new and hasn't been received yet
+  bool on_got_packet(td::uint32 seqno);
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/Bbr.cpp

@@ -0,0 +1,62 @@
+#include "Bbr.h"
+
+#include "BdwStats.h"
+#include "RttStats.h"
+
+#include "td/utils/Random.h"
+
+namespace ton {
+namespace rldp2 {
+void Bbr::step(const RttStats &rtt_stats, const BdwStats &bdw_stats, td::uint64 in_flight, td::Timestamp now) {
+  rtt_min_ = rtt_stats.windowed_min_rtt;
+  bdw_max_ = bdw_stats.windowed_max_bdw;
+  if (bdw_max_ > bdw_peak_ * 1.25) {
+    bdw_peak_ = bdw_max_;
+    bdw_peak_at_round = rtt_stats.rtt_round;
+    //LOG(ERROR) << "NEW PEAK " << bdw_peak_ * 768;
+  }
+
+  if (state_ == State::Start && bdw_peak_at_round + 3 < rtt_stats.rtt_round) {
+    //LOG(ERROR) << "START -> DRAIN";
+    state_ = State::Drain;
+  }
+
+  if (state_ == State::Drain && (double)in_flight < bdw_max_ * rtt_min_) {
+    //LOG(ERROR) << "DRAIN -> BPROBE BDW";
+    state_ = State::ProbeBdw;
+    probe_bdw_cycle_ = td::Random::fast(1, 5);
+    probe_bdw_cycle_at_ = now;
+  }
+
+  if (state_ == State::ProbeBdw && td::Timestamp::in(rtt_stats.windowed_min_rtt, probe_bdw_cycle_at_).is_in_past(now)) {
+    probe_bdw_cycle_at_ = now;
+    probe_bdw_cycle_ = (probe_bdw_cycle_ + 1) % 6;
+    //LOG(ERROR) << "NEW PROBE BDW CYCLE";
+  }
+
+  //TODO: ProbeRtt state. Don't want to implenent now without proper tests
+}
+
+double Bbr::get_rate() const {
+  if (state_ == State::Start) {
+    return bdw_max_ * 2.8;
+  }
+  if (state_ == State::Drain) {
+    return bdw_max_ / 2.8;
+  }
+  if (state_ == State::ProbeBdw) {
+    constexpr double probe_bdw_gain[6] = {0.75, 1, 1, 1, 1, 1.25};
+    return probe_bdw_gain[probe_bdw_cycle_] * bdw_max_;
+  }
+  UNREACHABLE();
+}
+
+td::uint32 Bbr::get_window_size() const {
+  if (state_ == State::Start || state_ == State::Drain) {
+    return td::max(td::uint32(bdw_max_ * rtt_min_ * 2.8 + 1), 10u);
+  }
+  return td::max(td::uint32(bdw_max_ * rtt_min_ * 2 + 1), 10u);
+}
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/Bbr.h

@@ -0,0 +1,29 @@
+#pragma once
+#include "td/utils/int_types.h"
+#include "td/utils/Time.h"
+
+namespace ton {
+namespace rldp2 {
+struct RttStats;
+struct BdwStats;
+
+struct Bbr {
+ public:
+  void step(const RttStats &rtt_stats, const BdwStats &bdw_stats, td::uint64 in_flight, td::Timestamp now);
+  double get_rate() const;
+
+  td::uint32 get_window_size() const;
+
+ private:
+  double bdw_peak_{-1};
+  td::uint32 bdw_peak_at_round{0};
+  td::uint32 probe_bdw_cycle_{0};
+  td::Timestamp probe_bdw_cycle_at_;
+  double rtt_min_{0};
+  double bdw_max_{0};
+  enum class State { Start, Drain, ProbeRtt, ProbeBdw } state_ = State::Start;
+};
+
+}  // namespace rldp2
+}  // namespace ton
+

rldp2/BdwStats.cpp

@@ -0,0 +1,50 @@
+#include "BdwStats.h"
+
+namespace ton {
+namespace rldp2 {
+
+BdwStats::PacketInfo BdwStats::on_packet_send(td::Timestamp first_sent_at) const {
+  PacketInfo packet;
+  packet.delivered_now = delivered_now;
+  packet.first_sent_at = first_sent_at;
+  packet.delivered_count = delivered_count;
+  packet.is_paused = static_cast<bool>(paused_at_);
+  return packet;
+}
+
+void BdwStats::on_packet_ack(const PacketInfo &info, td::Timestamp sent_at, td::Timestamp now) {
+  if (paused_at_.is_in_past(info.delivered_now)) {
+    paused_at_ = {};
+  }
+  auto sent_passed = sent_at.at() - info.first_sent_at.at();
+  auto ack_passed = now.at() - info.delivered_now.at();
+  auto passed = td::max(sent_passed, ack_passed);
+  if (passed < 0.01) {
+    LOG(ERROR) << "Invalid passed " << passed;
+  }
+  auto delivered = delivered_count - info.delivered_count;
+  on_rate_sample((double)delivered / passed, now, info.is_paused);
+}
+
+void BdwStats::on_update(td::Timestamp now, td::uint64 delivered_count_diff) {
+  this->delivered_now = now;
+  this->delivered_count += delivered_count_diff;
+}
+
+void BdwStats::on_pause(td::Timestamp now) {
+  paused_at_ = now;
+}
+
+void BdwStats::on_rate_sample(double rate, td::Timestamp now, bool is_paused) {
+  // ignore decrease of rate if is_paused == true
+  if (is_paused && rate < windowed_max_bdw) {
+    return;
+  }
+  windowed_max_bdw_stat.add_event(rate, now.at());
+  auto windowed_max_bdw_sample = windowed_max_bdw_stat.get_stat(now.at()).get_stat();
+  if (windowed_max_bdw_sample) {
+    windowed_max_bdw = windowed_max_bdw_sample.value();
+  }
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/BdwStats.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include "td/utils/Time.h"
+#include "td/utils/TimedStat.h"
+
+namespace ton {
+namespace rldp2 {
+struct BdwStats {
+  struct State {};
+
+  struct PacketInfo {
+    td::Timestamp first_sent_at;
+
+    td::Timestamp delivered_now;
+    td::uint64 delivered_count{0};
+    bool is_paused{false};
+  };
+
+  PacketInfo on_packet_send(td::Timestamp first_sent_at) const;
+  void on_packet_ack(const PacketInfo &info, td::Timestamp sent_at, td::Timestamp now);
+
+  void on_update(td::Timestamp now, td::uint64 delivered_count_diff);
+
+  void on_pause(td::Timestamp now);
+  double windowed_max_bdw{0};
+
+ private:
+  td::Timestamp delivered_now;
+  td::uint64 delivered_count{0};
+  td::TimedStat<td::MaxStat<double>> windowed_max_bdw_stat{5, 0};
+  td::Timestamp paused_at_;
+
+  void on_rate_sample(double rate, td::Timestamp now, bool is_paused);
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/CMakeLists.txt

@@ -0,0 +1,58 @@
+cmake_minimum_required(VERSION 3.0.2 FATAL_ERROR)
+
+if (NOT OPENSSL_FOUND)
+  find_package(OpenSSL REQUIRED)
+endif()
+
+if (NOT GSL_FOUND)
+  find_package(GSL)
+endif()
+
+set(RLDP_SOURCE
+  Ack.cpp
+  Bbr.cpp
+  BdwStats.cpp
+  FecHelper.cpp
+  InboundTransfer.cpp
+  LossSender.cpp
+  LossStats.cpp
+  OutboundTransfer.cpp
+  Pacer.cpp
+  rldp.cpp
+  RldpReceiver.cpp
+  RldpSender.cpp
+  RldpConnection.cpp
+  RttStats.cpp
+  SenderPackets.cpp
+
+  Ack.h
+  Bbr.h
+  BdwStats.h
+  FecHelper.h
+  InboundTransfer.h
+  LossSender.h
+  LossStats.h
+  OutboundTransfer.h
+  Pacer.h
+  rldp.h
+  rldp.hpp
+  RldpReceiver.h
+  RldpSender.h
+  RldpConnection.h
+  RttStats.h
+  SenderPackets.h
+)
+
+add_library(rldp2 STATIC ${RLDP_SOURCE})
+
+target_include_directories(rldp PUBLIC 
+  $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}> 
+  $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>/.. 
+  ${OPENSSL_INCLUDE_DIR}
+)
+if (GSL_FOUND)
+  target_link_libraries(rldp2 PRIVATE GSL::gsl)
+  target_compile_definitions(rldp2 PRIVATE -DTON_HAVE_GSL=1)
+endif()
+target_link_libraries(rldp2 PUBLIC tdutils tdactor fec adnl tl_api)
+

rldp2/FecHelper.cpp

@@ -0,0 +1,24 @@
+#include "FecHelper.h"
+
+#include "td/utils/check.h"
+
+namespace ton {
+namespace rldp2 {
+td::uint32 FecHelper::get_fec_symbols_count() const {
+  constexpr td::uint32 x = 5;
+  constexpr td::uint32 y = 5;
+  // smallest (symbols_count + x + y * i) >  received_symbols_count
+  if (symbols_count + x > received_symbols_count) {
+    return symbols_count + x;
+  }
+  td::uint32 i = (received_symbols_count - (symbols_count + x)) / y + 1;
+  return symbols_count + x + i * y;
+}
+
+td::uint32 FecHelper::get_left_fec_symbols_count() const {
+  auto fec_symbols_count = get_fec_symbols_count();
+  CHECK(fec_symbols_count > received_symbols_count);
+  return fec_symbols_count - received_symbols_count;
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/FecHelper.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include "td/utils/int_types.h"
+
+namespace ton {
+namespace rldp2 {
+
+struct FecHelper {
+  td::uint32 symbols_count{0};
+  td::uint32 received_symbols_count{0};
+
+  td::uint32 get_fec_symbols_count() const;
+  td::uint32 get_left_fec_symbols_count() const;
+};
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/InboundTransfer.cpp

@@ -0,0 +1,58 @@
+#include "InboundTransfer.h"
+
+#include "common/errorcode.h"
+
+namespace ton {
+namespace rldp2 {
+size_t InboundTransfer::total_size() const {
+  return data_.size();
+}
+
+std::map<td::uint32, InboundTransfer::Part> &InboundTransfer::parts() {
+  return parts_;
+}
+
+bool InboundTransfer::is_part_completed(td::uint32 part_i) {
+  return parts_.count(part_i) == 0 && part_i < next_part_;
+}
+
+td::Result<InboundTransfer::Part *> InboundTransfer::get_part(td::uint32 part_i, const ton::fec::FecType &fec_type) {
+  auto it = parts_.find(part_i);
+  if (it != parts_.end()) {
+    return &it->second;
+  }
+  //TODO: pass offset off and process even newer parts.
+  //LOG_CHECK(next_part_ >= part_i) << next_part_ << " >= " << part_i;
+  if (next_part_ == part_i && parts_.size() < 20) {
+    auto offset = offset_;
+    offset_ += fec_type.size();
+    if (offset_ > total_size()) {
+      return td::Status::Error(ErrorCode::protoviolation,
+                               PSTRING() << "too big part: offset=" << offset_ << " total_size=" << total_size()
+                                         << " total_size=" << fec_type.size() << " part=" << part_i);
+    }
+
+    auto decoder = fec_type.create_decoder().move_as_ok();
+    auto it = parts_.emplace(part_i, Part{std::move(decoder), RldpReceiver(RldpSender::Config()), offset});
+    next_part_++;
+    return &it.first->second;
+  }
+  return nullptr;
+}
+
+void InboundTransfer::finish_part(td::uint32 part_i, td::Slice data) {
+  auto it = parts_.find(part_i);
+  CHECK(it != parts_.end());
+  data_.as_slice().substr(it->second.offset).copy_from(data);
+  parts_.erase(it);
+}
+
+td::optional<td::Result<td::BufferSlice>> InboundTransfer::try_finish() {
+  if (parts_.empty() && offset_ == data_.size()) {
+    return std::move(data_);
+  }
+  return {};
+}
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/InboundTransfer.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include "td/utils/optional.h"
+
+#include "fec/fec.h"
+
+#include "RldpReceiver.h"
+
+#include <map>
+
+namespace ton {
+namespace rldp2 {
+struct InboundTransfer {
+  struct Part {
+    std::unique_ptr<td::fec::Decoder> decoder;
+    RldpReceiver receiver;
+    size_t offset;
+  };
+
+  explicit InboundTransfer(size_t total_size) : data_(total_size) {
+  }
+
+  size_t total_size() const;
+  std::map<td::uint32, Part> &parts();
+  bool is_part_completed(td::uint32 part_i);
+  td::Result<Part *> get_part(td::uint32 part_i, const ton::fec::FecType &fec_type);
+  void finish_part(td::uint32 part_i, td::Slice data);
+  td::optional<td::Result<td::BufferSlice>> try_finish();
+
+ private:
+  std::map<td::uint32, Part> parts_;
+  td::uint32 next_part_{0};
+  size_t offset_{0};
+  td::BufferSlice data_;
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/LossSender.cpp

@@ -0,0 +1,135 @@
+#include "LossSender.h"
+
+#include "td/utils/logging.h"
+
+#if TON_HAVE_GSL
+#include <gsl/gsl_cdf.h>
+#endif
+
+#include <cmath>
+
+namespace ton {
+namespace rldp2 {
+namespace {
+// works for 1e-x, where x in {1..10}
+double ndtri_fast(double p) {
+  if (p < 2e-10) {
+    return 6.361340902404;
+  }
+  if (p < 2e-9) {
+    return 5.997807015008;
+  }
+  if (p < 2e-8) {
+    return 5.612001244175;
+  }
+  if (p < 2e-7) {
+    return 5.199337582193;
+  }
+  if (p < 2e-6) {
+    return 4.753424308823;
+  }
+  if (p < 2e-5) {
+    return 4.264890793923;
+  }
+  if (p < 2e-4) {
+    return 3.719016485456;
+  }
+  if (p < 2e-3) {
+    return 3.090232306168;
+  }
+  if (p < 2e-2) {
+    return 2.326347874041;
+  }
+  return 1.281551565545;
+}
+}  // namespace
+
+LossSender::LossSender(double loss, double p) : loss_(loss), p_(p) {
+  v_.resize(2);
+  v_[0] = 1;
+  res_.push_back(0);
+  S_ = ndtri_fast(p_);
+  sigma_ = p * (1 - p);
+  //LOG(ERROR) << S_ << " " << ndtri(1 - p_);
+  //CHECK(fabs(S_ - ndtri(1 - p_)) < 1e-6);
+}
+
+int LossSender::send_n(int n) {
+  if (n < 50) {
+    return send_n_exact(n);
+  }
+  return send_n_approx_nbd(n);
+}
+
+int LossSender::send_n_exact(int n) {
+  while ((int)res_.size() <= n) {
+    step();
+  }
+  return res_[n];
+}
+
+int LossSender::send_n_approx_norm(int n) {
+  double a = (1 - loss_) * (1 - loss_);
+  double b = loss_ * (loss_ - 1) * (2 * n + S_ * S_);
+  double c = loss_ * loss_ * n * n + S_ * S_ * n * loss_ * (loss_ - 1);
+  double x = ((-b + sqrt(b * b - 4 * a * c)) / (2 * a));
+  return (int)(x + n + 1);
+}
+
+int LossSender::send_n_approx_nbd(int n) {
+#if TON_HAVE_GSL
+  auto mean = n * loss_ / (1 - loss_);
+  auto var = sqrt(mean / (1 - loss_));
+  auto min_k = static_cast<int>(mean + var);
+  auto max_k = min_k + static_cast<int>(var + 1) * 15;
+  while (min_k + 1 < max_k) {
+    int k = (min_k + max_k) / 2;
+    if (gsl_cdf_negative_binomial_P(k, 1 - loss_, n) > 1 - p_) {
+      max_k = k;
+    } else {
+      min_k = k;
+    }
+  }
+  return max_k + n;
+#endif
+  return send_n_approx_norm(n);
+}
+
+int LossSender::send_n_approx_pd(int n) {
+#if TON_HAVE_GSL
+  for (int k = 0;; k++) {
+    if (gsl_cdf_poisson_P(k, (n + k) * loss_) > 1 - p_) {
+      return k + n;
+    }
+  }
+#endif
+  return send_n_approx_norm(n);
+}
+bool LossSender::has_good_approx() {
+#if TON_HAVE_GSL
+  return true;
+#else
+  return false;
+#endif
+}
+
+void LossSender::step() {
+  n_++;
+  v_.push_back(0);
+  v_[n_] = v_[n_ - 1];
+  for (int j = n_; j >= 0; j--) {
+    v_[j + 1] += v_[j] * loss_;
+    v_[j] *= (1 - loss_);
+  }
+
+  while (res_i_ < n_ && v_[res_i_] < 1 - p_) {
+    res_i_++;
+  }
+  auto left_ = n_ - res_i_;
+  if ((int)res_.size() == left_) {
+    res_.push_back(n_);
+  }
+}
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/LossSender.h

@@ -0,0 +1,31 @@
+#pragma once
+#include <vector>
+
+namespace ton {
+namespace rldp2 {
+struct LossSender {
+  LossSender(double loss, double p);
+  int send_n(int n);
+
+  int send_n_exact(int n);
+  int send_n_approx_norm(int n);
+  int send_n_approx_nbd(int n);
+  int send_n_approx_pd(int n);
+
+  bool has_good_approx();
+
+ private:
+  double loss_;
+  double p_;
+  double S_;
+  double sigma_;
+  int n_{0};
+  std::vector<double> v_;
+  std::vector<int> res_;
+  int res_i_{0};
+
+  void step();
+};
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/LossStats.cpp

@@ -0,0 +1,23 @@
+#include "LossStats.h"
+#include "td/utils/misc.h"
+
+#include <cmath>
+
+namespace ton {
+namespace rldp2 {
+void LossStats::on_update(td::uint32 ack, td::uint32 lost) {
+  ack_ += ack;
+  lost_ += lost;
+
+  if (ack_ + lost_ > 1000) {
+    auto new_loss = td::clamp((double)lost_ / (ack_ + lost_), 0.001, 0.2);
+    if (fabs(new_loss - loss) > 5e-3) {
+      prob = LossSender(new_loss, 1e-9);
+    }
+    loss = new_loss;
+    ack_ = 0;
+    lost_ = 0;
+  }
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/LossStats.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include "LossSender.h"
+
+#include "td/utils/int_types.h"
+
+namespace ton {
+namespace rldp2 {
+struct LossStats {
+  void on_update(td::uint32 ack, td::uint32 lost);
+  double loss = 0.1;
+  LossSender prob{0.1, 1e-9};
+
+ private:
+  td::uint32 ack_{0};
+  td::uint32 lost_{0};
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/OutboundTransfer.cpp

@@ -0,0 +1,40 @@
+#include "OutboundTransfer.h"
+
+namespace ton {
+namespace rldp2 {
+size_t OutboundTransfer::total_size() const {
+  return data_.size();
+}
+std::map<td::uint32, OutboundTransfer::Part> &OutboundTransfer::parts(const RldpSender::Config &config) {
+  while (parts_.size() < 20) {
+    auto offset = next_part_ * part_size();
+    if (offset >= data_.size()) {
+      break;
+    }
+    td::BufferSlice D = data_.from_slice(data_.as_slice().substr(offset).truncate(part_size()));
+    ton::fec::FecType fec_type = td::fec::RaptorQEncoder::Parameters{D.size(), symbol_size(), 0};
+    auto encoder = fec_type.create_encoder(std::move(D)).move_as_ok();
+    auto symbols_count = fec_type.symbols_count();
+    parts_.emplace(next_part_, Part{std::move(encoder), RldpSender(config, symbols_count), std::move(fec_type)});
+    next_part_++;
+  }
+  return parts_;
+}
+
+void OutboundTransfer::drop_part(td::uint32 part_i) {
+  parts_.erase(part_i);
+}
+
+OutboundTransfer::Part *OutboundTransfer::get_part(td::uint32 part_i) {
+  auto it = parts_.find(part_i);
+  if (it == parts_.end()) {
+    return nullptr;
+  }
+  return &it->second;
+}
+
+bool OutboundTransfer::is_done() const {
+  return next_part_ * part_size() >= data_.size() && parts_.empty();
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/OutboundTransfer.h

@@ -0,0 +1,40 @@
+#pragma once
+
+#include "RldpSender.h"
+#include "fec/fec.h"
+
+#include <map>
+
+namespace ton {
+namespace rldp2 {
+struct OutboundTransfer {
+ public:
+  struct Part {
+    std::unique_ptr<td::fec::Encoder> encoder;
+    RldpSender sender;
+    ton::fec::FecType fec_type;
+  };
+
+  OutboundTransfer(td::BufferSlice data) : data_(std::move(data)) {
+  }
+
+  size_t total_size() const;
+  std::map<td::uint32, Part> &parts(const RldpSender::Config &config);
+  void drop_part(td::uint32 part_i);
+  Part *get_part(td::uint32 part_i);
+  bool is_done() const;
+
+ private:
+  td::BufferSlice data_;
+  std::map<td::uint32, Part> parts_;
+  td::uint32 next_part_{0};
+
+  static size_t part_size() {
+    return 2000000;
+  }
+  static size_t symbol_size() {
+    return 768;
+  }
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/Pacer.cpp

@@ -0,0 +1,45 @@
+#include "Pacer.h"
+namespace ton {
+namespace rldp2 {
+Pacer::Pacer(Options options)
+    : speed_(options.initial_speed)
+    , capacity_(options.initial_capacity)
+    , max_capacity_(options.max_capacity)
+    , time_granularity_(options.time_granularity) {
+}
+
+td::Timestamp Pacer::wakeup_at() const {
+  return wakeup_at_;
+}
+
+void Pacer::set_speed(double speed) {
+  if (speed < 1) {
+    speed = 1;
+  }
+  speed_ = speed;
+}
+
+td::optional<td::Timestamp> Pacer::send(double size, td::Timestamp now) {
+  update_capacity(now);
+
+  if (size < capacity_) {
+    capacity_ -= size;
+    return {};
+  }
+
+  size -= capacity_;
+  capacity_ = 0;
+  wakeup_at_ = td::Timestamp::in(size / speed_, now);
+  capacity_at_ = wakeup_at_;
+  return wakeup_at_;
+}
+
+void Pacer::update_capacity(td::Timestamp now) {
+  if (capacity_at_ && capacity_at_.is_in_past(now)) {
+    capacity_ += (now.at() - capacity_at_.at()) * speed_;
+    capacity_ = td::min(capacity_, td::max(max_capacity_, speed_ * time_granularity_));
+  }
+  capacity_at_ = now;
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/Pacer.h

@@ -0,0 +1,39 @@
+#pragma once
+#include "td/utils/optional.h"
+#include "td/utils/Time.h"
+
+namespace ton {
+namespace rldp2 {
+// NB: Should be careful with max_capacity < time_granularity * speed
+// We may send packet of any size.
+// After that we will be put to sleep till wakeup_at().
+// When we are awake we send packet of any size again.
+// Logic is - we don't have to wait to send a packet - it is poinless.
+// But we have to wait for some time after packet is sent
+class Pacer {
+ public:
+  struct Options {
+    Options() {
+    }
+    double initial_capacity{20};
+    double initial_speed{10};
+    double max_capacity{40};
+    double time_granularity{0.001};
+  };
+  Pacer(Options options = {});
+  td::Timestamp wakeup_at() const;
+  void set_speed(double speed);
+  td::optional<td::Timestamp> send(double size, td::Timestamp now);
+
+ private:
+  double speed_;
+  double capacity_;
+  double max_capacity_;
+  double time_granularity_;
+  td::Timestamp capacity_at_;
+  td::Timestamp wakeup_at_;
+
+  void update_capacity(td::Timestamp now);
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/RldpConnection.cpp

@@ -0,0 +1,369 @@
+#include "RldpConnection.h"
+
+#include "td/utils/overloaded.h"
+#include "td/utils/Random.h"
+#include "td/utils/tl_helpers.h"
+
+#include "tl-utils/tl-utils.hpp"
+#include "auto/tl/ton_api.h"
+#include "auto/tl/ton_api.hpp"
+
+#include "common/errorcode.h"
+
+#include "td/actor//actor.h"
+
+namespace ton {
+namespace rldp2 {
+void RldpConnection::add_limit(td::Timestamp timeout, Limit limit) {
+  CHECK(timeout);
+  auto p = limits_set_.insert(limit);
+  LOG_CHECK(p.second) << limit.transfer_id.to_hex();
+  limits_heap_.insert(timeout.at(), const_cast<Limit *>(&*p.first));
+}
+
+td::Timestamp RldpConnection::next_limit_expires_at() {
+  if (limits_heap_.empty()) {
+    return td::Timestamp::never();
+  }
+  return td::Timestamp::at(limits_heap_.top_key());
+}
+
+void RldpConnection::drop_limits(TransferId id) {
+  Limit limit;
+  limit.transfer_id = id;
+  auto it = limits_set_.find(limit);
+  if (it == limits_set_.end()) {
+    return;
+  }
+  limits_heap_.erase(const_cast<td::HeapNode *>(static_cast<const td::HeapNode *>(&*it)));
+  limits_set_.erase(it);
+}
+
+void RldpConnection::on_inbound_completed(TransferId transfer_id, td::Timestamp now) {
+  inbound_transfers_.erase(transfer_id);
+  completed_set_.insert(transfer_id);
+  completed_queue_.push(CompletedId{transfer_id, now.in(20)});
+  while (completed_queue_.size() > 128 && completed_queue_.front().timeout.is_in_past(now)) {
+    completed_set_.erase(completed_queue_.pop().transfer_id);
+  }
+}
+
+td::Timestamp RldpConnection::loop_limits(td::Timestamp now) {
+  while (!limits_heap_.empty() && td::Timestamp::at(limits_heap_.top_key()).is_in_past(now)) {
+    auto *limit = static_cast<Limit *>(limits_heap_.pop());
+    auto error = td::Status::Error(ErrorCode::timeout, "timeout");
+    if (limit->is_inbound) {
+      on_inbound_completed(limit->transfer_id, now);
+      to_receive_.emplace_back(limit->transfer_id, std::move(error));
+    } else {
+      auto it = outbound_transfers_.find(limit->transfer_id);
+      if (it != outbound_transfers_.end()) {
+        for (auto &part : it->second.parts(RldpSender::Config{})) {
+          in_flight_count_ -= part.second.sender.get_inflight_symbols_count();
+        }
+        outbound_transfers_.erase(it);
+        to_on_sent_.emplace_back(limit->transfer_id, std::move(error));
+      } else {
+        LOG(ERROR) << "Timeout on unknown transfer " << limit->transfer_id.to_hex();
+      }
+    }
+    limits_set_.erase(*limit);
+  }
+
+  return next_limit_expires_at();
+}
+
+void RldpConnection::set_receive_limits(TransferId transfer_id, td::Timestamp timeout, td::uint64 max_size) {
+  CHECK(timeout);
+  Limit limit;
+  limit.transfer_id = transfer_id;
+  limit.max_size = max_size;
+  limit.is_inbound = true;
+  add_limit(timeout, limit);
+}
+
+RldpConnection::RldpConnection() {
+  bdw_stats_.on_update(td::Timestamp::now(), 0);
+
+  rtt_stats_.windowed_min_rtt = 0.5;
+  bdw_stats_.windowed_max_bdw = 10;
+}
+
+void RldpConnection::send(TransferId transfer_id, td::BufferSlice data, td::Timestamp timeout) {
+  if (transfer_id.is_zero()) {
+    td::Random::secure_bytes(transfer_id.as_slice());
+  } else {
+    if (outbound_transfers_.find(transfer_id) != outbound_transfers_.end()) {
+      LOG(WARNING) << "Skip resend of " << transfer_id.to_hex();
+      return;
+    }
+  }
+
+  if (timeout) {
+    Limit limit;
+    limit.transfer_id = transfer_id;
+    limit.max_size = 0;
+    limit.is_inbound = false;
+    add_limit(timeout, limit);
+  }
+  outbound_transfers_.emplace(transfer_id, OutboundTransfer{std::move(data)});
+}
+
+void RldpConnection::receive_raw(td::BufferSlice packet) {
+  auto F = ton::fetch_tl_object<ton::ton_api::rldp2_MessagePart>(std::move(packet), true);
+  if (F.is_error()) {
+    return;
+  }
+  downcast_call(*F.move_as_ok(), [&](auto &obj) { this->receive_raw_obj(obj); });
+}
+
+void RldpConnection::loop_bbr(td::Timestamp now) {
+  bbr_.step(rtt_stats_, bdw_stats_, in_flight_count_, td::Timestamp::now());
+  //LOG(ERROR) << td::format::as_time(rtt_stats_.windowed_min_rtt) << " "
+  //<< td::format::as_size((td::int64)bdw_stats_.windowed_max_bdw * 768) << " " << rtt_stats_.rtt_round;
+  double speed = bbr_.get_rate();
+  td::uint32 congestion_window = bbr_.get_window_size();
+
+  static td::Timestamp next;
+  //FIXME: remove this UNSAFE debug output
+  if (next.is_in_past(now)) {
+    next = td::Timestamp::in(1, now);
+    if (td::actor::core::ActorExecuteContext::get()->actor().get_actor_info_ptr()->get_name() == "Alice") {
+      LOG(ERROR) << "speed=" << td::format::as_size((td::int64)speed * 768) << " "
+                 << "cgw=" << td::format::as_size((td::int64)congestion_window * 768) << " "
+                 << "loss=" << loss_stats_.loss * 100 << "%";
+    }
+  }
+
+  pacer_.set_speed(speed);
+  congestion_window_ = congestion_window;
+}
+
+td::Timestamp RldpConnection::run(ConnectionCallback &callback) {
+  auto now = td::Timestamp::now();
+  loop_bbr(now);
+
+  td::Timestamp alarm_timestamp;
+  td::VectorQueue<std::pair<const TransferId, OutboundTransfer> *> queue;
+  for (auto &outbound : outbound_transfers_) {
+    queue.push(&outbound);
+  }
+  while (!queue.empty()) {
+    auto outbound = queue.pop();
+    auto o_timeout = step(outbound->first, outbound->second, now);
+    if (o_timeout) {
+      alarm_timestamp.relax(o_timeout.unwrap());
+    } else {
+      queue.push(outbound);
+    }
+  }
+
+  if (in_flight_count_ > congestion_window_) {
+    bdw_stats_.on_pause(now);
+  }
+
+  for (auto &inbound : inbound_transfers_) {
+    alarm_timestamp.relax(run(inbound.first, inbound.second));
+  }
+
+  alarm_timestamp.relax(loop_limits(td::Timestamp::now()));
+
+  for (auto &data : to_receive_) {
+    callback.receive(data.first, std::move(data.second));
+  }
+  for (auto &raw : to_send_raw_) {
+    callback.send_raw(std::move(raw));
+  }
+  to_send_raw_.clear();
+  to_receive_.clear();
+  for (auto &res : to_on_sent_) {
+    callback.on_sent(res.first, std::move(res.second));
+  }
+  to_on_sent_.clear();
+
+  return alarm_timestamp;
+}
+
+td::Timestamp RldpConnection::run(const TransferId &transfer_id, InboundTransfer &inbound) {
+  td::Timestamp wakeup_at;
+  bool has_actions = true;
+  while (has_actions) {
+    has_actions = false;
+    for (auto &it : inbound.parts()) {
+      auto &inbound = it.second;
+      inbound.receiver.next_action(td::Timestamp::now())
+          .visit(td::overloaded([&](const RldpReceiver::ActionWait &wait) { wakeup_at.relax(wait.wait_till); },
+                                [&](const RldpReceiver::ActionSendAck &send) {
+                                  send_packet(ton::create_serialize_tl_object<ton::ton_api::rldp2_confirm>(
+                                      transfer_id, it.first, send.ack.max_seqno, send.ack.received_mask,
+                                      send.ack.received_count));
+                                  inbound.receiver.on_ack_sent(td::Timestamp::now());
+                                  has_actions = true;
+                                }));
+    }
+  }
+  return wakeup_at;
+}
+
+td::optional<td::Timestamp> RldpConnection::step(const TransferId &transfer_id, OutboundTransfer &outbound,
+                                                 td::Timestamp now) {
+  bool only_probe = in_flight_count_ > congestion_window_;
+
+  td::Timestamp wakeup_at;
+  if (!pacer_.wakeup_at().is_in_past(now)) {
+    wakeup_at = pacer_.wakeup_at();
+    only_probe = true;
+  }
+
+  for (auto &it : outbound.parts(RldpSender::Config{})) {
+    auto &part = it.second;
+
+    Guard guard(in_flight_count_, part.sender);
+    auto action = part.sender.next_action(now, only_probe);
+
+    bool was_send = false;
+    action.visit(td::overloaded(
+        [&](const RldpSender::ActionSend &send) {
+          auto seqno = send.seqno - 1;
+          if (part.encoder->get_info().ready_symbol_count <= seqno) {
+            part.encoder->prepare_more_symbols();
+          }
+          auto symbol = part.encoder->gen_symbol(seqno).data;
+          send_packet(ton::create_serialize_tl_object<ton::ton_api::rldp2_messagePart>(
+              transfer_id, part.fec_type.tl(), it.first, outbound.total_size(), seqno, std::move(symbol)));
+          if (!send.is_probe) {
+            pacer_.send(1, now);
+          }
+          part.sender.on_send(send.seqno, now, send.is_probe, rtt_stats_, bdw_stats_);
+          if (send.is_probe) {
+            //LOG(ERROR) << "PROBE " << it.first << " " << send.seqno;
+          }
+          //LOG(ERROR) << "SEND";
+          was_send = true;
+        },
+        [&](const RldpSender::ActionWait &wait) {
+          //LOG(ERROR) << "WAIT";
+          wakeup_at.relax(wait.wait_till);
+        }));
+    if (was_send) {
+      return {};
+    }
+  }
+
+  return wakeup_at;
+}
+
+void RldpConnection::receive_raw_obj(ton::ton_api::rldp2_messagePart &part) {
+  if (completed_set_.count(part.transfer_id_) > 0) {
+    send_packet(ton::create_serialize_tl_object<ton::ton_api::rldp2_complete>(part.transfer_id_, part.part_));
+    return;
+  }
+
+  auto r_total_size = td::narrow_cast_safe<td::size_t>(part.total_size_);
+  if (r_total_size.is_error()) {
+    return;
+  }
+  auto r_fec_type = ton::fec::FecType::create(std::move(part.fec_type_));
+  if (r_fec_type.is_error()) {
+    return;
+  }
+
+  auto total_size = r_total_size.move_as_ok();
+
+  auto transfer_id = part.transfer_id_;
+
+  // check total_size limits
+  td::uint64 max_size = default_mtu();
+  Limit key;
+  key.transfer_id = transfer_id;
+  auto limit_it = limits_set_.find(key);
+  bool has_limit = limit_it != limits_set_.end();
+  if (has_limit && limit_it->max_size != 0) {
+    max_size = limit_it->max_size;
+  }
+  if (total_size > max_size) {
+    LOG(INFO) << "Drop too big rldp query " << part.total_size_ << " > " << max_size;
+    return;
+  }
+
+  auto it = inbound_transfers_.find(transfer_id);
+  if (it == inbound_transfers_.end()) {
+    if (!has_limit) {
+      // set timeout even for small inbound queries
+      // TODO: other party stil may ddos us with small transfers
+      set_receive_limits(transfer_id, td::Timestamp::in(10), max_size);
+    }
+    it = inbound_transfers_.emplace(transfer_id, InboundTransfer{total_size}).first;
+  }
+
+  auto &inbound = it->second;
+  auto o_res = [&]() -> td::optional<td::Result<td::BufferSlice>> {
+    TRY_RESULT(in_part, inbound.get_part(part.part_, r_fec_type.move_as_ok()));
+    if (!in_part) {
+      if (inbound.is_part_completed(part.part_)) {
+        send_packet(ton::create_serialize_tl_object<ton::ton_api::rldp2_complete>(transfer_id, part.part_));
+      }
+      return {};
+    }
+    if (in_part->receiver.on_received(part.seqno_, td::Timestamp::now())) {
+      TRY_STATUS_PREFIX(in_part->decoder->add_symbol({static_cast<td::uint32>(part.seqno_), std::move(part.data_)}),
+                        td::Status::Error(ErrorCode::protoviolation, "invalid symbol"));
+      if (in_part->decoder->may_try_decode()) {
+        auto r_data = in_part->decoder->try_decode(false);
+        if (r_data.is_ok()) {
+          inbound.finish_part(part.part_, r_data.move_as_ok().data);
+        }
+      }
+    }
+    return inbound.try_finish();
+  }();
+
+  if (o_res) {
+    drop_limits(transfer_id);
+    on_inbound_completed(transfer_id, td::Timestamp::now());
+    to_receive_.emplace_back(transfer_id, o_res.unwrap());
+  }
+}
+
+void RldpConnection::receive_raw_obj(ton::ton_api::rldp2_complete &complete) {
+  auto transfer_id = complete.transfer_id_;
+  auto it = outbound_transfers_.find(transfer_id);
+  if (it == outbound_transfers_.end()) {
+    return;
+  }
+
+  auto *part = it->second.get_part(complete.part_);
+  if (part) {
+    in_flight_count_ -= part->sender.get_inflight_symbols_count();
+    it->second.drop_part(complete.part_);
+  }
+
+  if (it->second.is_done()) {
+    drop_limits(it->first);
+    to_on_sent_.emplace_back(it->first, td::Unit());
+    outbound_transfers_.erase(it);
+  }
+}
+
+void RldpConnection::receive_raw_obj(ton::ton_api::rldp2_confirm &confirm) {
+  auto transfer_id = confirm.transfer_id_;
+  auto it = outbound_transfers_.find(transfer_id);
+  if (it == outbound_transfers_.end()) {
+    return;
+  }
+  auto *part = it->second.get_part(confirm.part_);
+  if (!part) {
+    return;
+  }
+  Guard guard(in_flight_count_, part->sender);
+  Ack ack;
+  ack.max_seqno = confirm.max_seqno_;
+  ack.received_count = confirm.received_count_;
+  ack.received_mask = confirm.received_mask_;
+  auto update = part->sender.on_ack(ack, 0, td::Timestamp::now(), rtt_stats_, bdw_stats_, loss_stats_);
+  // update.new_received event
+  // update.o_loss_at event
+}
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/RldpConnection.h

@@ -0,0 +1,119 @@
+#pragma once
+
+#include "Bbr.h"
+#include "InboundTransfer.h"
+#include "LossStats.h"
+#include "OutboundTransfer.h"
+#include "Pacer.h"
+#include "RttStats.h"
+
+#include "common/bitstring.h"
+
+#include "td/utils/buffer.h"
+#include "td/utils/Heap.h"
+#include "td/utils/VectorQueue.h"
+
+#include <set>
+
+namespace ton {
+namespace rldp2 {
+using TransferId = td::Bits256;
+class ConnectionCallback {
+ public:
+  virtual ~ConnectionCallback() {
+  }
+  virtual void send_raw(td::BufferSlice small_datagram) = 0;
+  virtual void receive(TransferId transfer_id, td::Result<td::BufferSlice> r_data) = 0;
+  virtual void on_sent(TransferId transfer_id, td::Result<td::Unit> state) = 0;
+};
+
+class RldpConnection {
+ public:
+  RldpConnection();
+  RldpConnection(RldpConnection &&other) = delete;
+  RldpConnection &operator=(RldpConnection &&other) = delete;
+  void send(TransferId tranfer_id, td::BufferSlice data, td::Timestamp timeout = td::Timestamp::never());
+  void set_receive_limits(TransferId transfer_id, td::Timestamp timeout, td::uint64 max_size);
+
+  void receive_raw(td::BufferSlice packet);
+
+  td::Timestamp run(ConnectionCallback &callback);
+
+  void set_default_mtu(td::uint64 mtu) {
+    default_mtu_ = mtu;
+  }
+  td::uint64 default_mtu() const {
+    return default_mtu_;
+  }
+
+ private:
+  td::uint64 default_mtu_ = 7680;
+
+  std::map<TransferId, OutboundTransfer> outbound_transfers_;
+  td::uint32 in_flight_count_{0};
+  std::map<TransferId, InboundTransfer> inbound_transfers_;
+
+  struct Limit : public td::HeapNode {
+    TransferId transfer_id;
+    td::uint64 max_size;
+    bool is_inbound;
+    bool operator<(const Limit &other) const {
+      return transfer_id < other.transfer_id;
+    }
+  };
+  td::KHeap<double> limits_heap_;
+  std::set<Limit> limits_set_;
+
+  struct CompletedId {
+    TransferId transfer_id;
+    td::Timestamp timeout;
+  };
+  td::VectorQueue<CompletedId> completed_queue_;
+  std::set<TransferId> completed_set_;
+
+  void add_limit(td::Timestamp timeout, Limit limit);
+  td::Timestamp next_limit_expires_at();
+  void drop_limits(TransferId id);
+  void on_inbound_completed(TransferId transfer_id, td::Timestamp now);
+  td::Timestamp loop_limits(td::Timestamp now);
+
+  void loop_bbr(td::Timestamp now);
+
+  RttStats rtt_stats_;
+  BdwStats bdw_stats_;
+  LossStats loss_stats_;
+  Bbr bbr_;
+  Pacer pacer_;
+  td::uint32 congestion_window_{0};
+
+  std::vector<td::BufferSlice> to_send_raw_;
+  std::vector<std::pair<TransferId, td::Result<td::BufferSlice>>> to_receive_;
+  std::vector<std::pair<TransferId, td::Result<td::Unit>>> to_on_sent_;
+
+  void send_packet(td::BufferSlice packet) {
+    to_send_raw_.push_back(std::move(packet));
+  };
+
+  td::Timestamp run(const TransferId &transfer_id, InboundTransfer &inbound);
+  struct Guard {
+    td::uint32 &in_flight_count;
+    const RldpSender &sender;
+    td::uint32 before_in_flight{sender.get_inflight_symbols_count()};
+
+    Guard(td::uint32 &in_flight_count, const RldpSender &sender) : in_flight_count(in_flight_count), sender(sender){};
+    ~Guard() {
+      in_flight_count -= before_in_flight;
+      in_flight_count += sender.get_inflight_symbols_count();
+    }
+  };
+
+  td::optional<td::Timestamp> step(const TransferId &transfer_id, OutboundTransfer &outbound, td::Timestamp now);
+
+  void receive_raw_obj(ton::ton_api::rldp2_messagePart &part);
+
+  void receive_raw_obj(ton::ton_api::rldp2_complete &part);
+
+  void receive_raw_obj(ton::ton_api::rldp2_confirm &part);
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/RldpReceiver.cpp

@@ -0,0 +1,34 @@
+#include "RldpReceiver.h"
+
+namespace ton {
+namespace rldp2 {
+td::Variant<RldpReceiver::ActionSendAck, RldpReceiver::ActionWait> RldpReceiver::next_action(td::Timestamp now) {
+  if (send_ack_at_ && (send_ack_at_.is_in_past(now))) {
+    return ActionSendAck{ack};
+  }
+  return ActionWait{send_ack_at_};
+}
+
+void RldpReceiver::on_ack_sent(td::Timestamp now) {
+  if (cnt_ != 0) {
+    //LOG(ERROR) << "RESEND ACK " << cnt_;
+  }
+  cnt_++;
+  if (cnt_ > 7) {
+    send_ack_at_ = {};
+  } else {
+    send_ack_at_.relax(td::Timestamp::at(now.at() + config_.ack_delay * (1 << cnt_)));
+  }
+}
+
+bool RldpReceiver::on_received(td::uint32 seqno, td::Timestamp now) {
+  if (!ack.on_got_packet(seqno)) {
+    return false;
+  }
+  cnt_ = 0;
+  send_ack_at_.relax(td::Timestamp::at(now.at() + config_.ack_delay));
+  return true;
+}
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/RldpReceiver.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include "Ack.h"
+#include "RldpSender.h"
+
+namespace ton {
+namespace rldp2 {
+class RldpReceiver {
+ public:
+  RldpReceiver() = default;
+  RldpReceiver(RldpSender::Config config) : config_(config) {
+  }
+
+  struct ActionSendAck {
+    Ack ack;
+  };
+
+  struct ActionWait {
+    td::Timestamp wait_till;
+  };
+
+  td::Variant<ActionSendAck, ActionWait> next_action(td::Timestamp now);
+
+  bool on_received(td::uint32 seqno, td::Timestamp now);
+
+  void on_ack_sent(td::Timestamp now);
+
+ private:
+  Ack ack;
+  td::Timestamp send_ack_at_;
+  td::uint32 cnt_{0};
+
+  RldpSender::Config config_;
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/RldpSender.cpp

@@ -0,0 +1,95 @@
+#include "RldpSender.h"
+
+#include "RttStats.h"
+#include "LossStats.h"
+#include "BdwStats.h"
+
+#include "td/utils/misc.h"
+
+namespace ton {
+namespace rldp2 {
+td::Variant<RldpSender::ActionWait, RldpSender::ActionSend> RldpSender::next_action(td::Timestamp now,
+                                                                                    bool only_probe) {
+  if (!only_probe && extra_symbols_ > get_inflight_symbols_count()) {
+    //LOG(ERROR) << fec_helper_.symbols_count << " " << fec_helper_.get_extra_symbols_count();
+    return ActionSend{packets_.next_seqno(), false};
+  }
+
+  return next_probe(now);
+}
+
+td::Variant<RldpSender::ActionWait, RldpSender::ActionSend> RldpSender::next_probe(td::Timestamp now) {
+  if (probe_timeout_.is_in_past(now)) {
+    return ActionSend{packets_.next_seqno(), true};
+  }
+  return ActionWait{probe_timeout_};
+}
+
+SenderPackets::Update RldpSender::on_ack(const Ack &ack, double ack_delay, td::Timestamp now, RttStats &rtt_stats,
+                                         BdwStats &bdw_stats, LossStats &loss_stats) {
+  //LOG(ERROR) << "ON ACK " << ack.max_seqno << " " << ack.received_mask << " " << ack.received_count;
+  auto update = packets_.on_ack(ack);
+  if (!update.was_max_updated) {
+    CHECK(!update.new_received);
+    return update;
+  }
+
+  // update rtt
+  ack_delay = td::clamp(ack_delay, 0.0, config_.max_ack_delay);
+  auto rtt_sample = now.at() - packets_.max_packet().sent_at.at();
+  rtt_stats.on_rtt_sample(rtt_sample, ack_delay, now);
+
+  bdw_stats.on_update(now, update.new_received);
+  bdw_stats.on_packet_ack(packets_.max_packet().bdw_packet_info, packets_.max_packet().sent_at, now);
+
+  // drop ready packets
+  SenderPackets::Limits limits;
+  limits.sent_at = td::Timestamp::at(now.at() - get_loss_delay(rtt_stats));
+  limits.seqno = sub_or_zero(packets_.max_packet().seqno, get_loss_seqno_delay());
+  update.drop_update = packets_.drop_packets(limits);
+
+  loss_stats.on_update(update.drop_update.new_ack, update.drop_update.new_lost);
+
+  fec_helper_.received_symbols_count = packets_.received_count();
+  extra_symbols_ = loss_stats.prob.send_n(fec_helper_.get_left_fec_symbols_count());
+  return update;
+}
+
+void RldpSender::on_send(td::uint32 seqno, td::Timestamp now, bool is_probe, const RttStats &rtt_stats,
+                         const BdwStats &bdw_stats) {
+  SenderPackets::Packet packet;
+  packet.is_in_flight = true;
+  packet.sent_at = now;
+  packet.seqno = seqno;
+  packet.size = 0;
+  packet.bdw_packet_info = bdw_stats.on_packet_send(packets_.first_sent_at(now));
+  packets_.send(packet);
+
+  probe_timeout_ = td::Timestamp::at(now.at() + get_probe_delay(rtt_stats));
+
+  if (is_probe) {
+    //LOG(ERROR) << get_probe_delay(rtt_stats) << " " << rtt_stats.last_rtt << " " << packets_.in_flight_count() << " "
+    //<< packets_.received_count();
+    probe_k_ = std::min(probe_k_ * 2, 10u);
+  } else {
+    probe_k_ = 1;
+  }
+}
+
+double RldpSender::get_loss_delay(const RttStats &rtt_stats) {
+  auto rtt = std::max(rtt_stats.last_rtt, rtt_stats.smoothed_rtt);
+  if (rtt < 0) {
+    rtt = config_.initial_rtt;
+  }
+  return rtt * 8 / 7;
+}
+
+double RldpSender::get_probe_delay(const RttStats &rtt_stats) {
+  if (rtt_stats.last_rtt < 0) {
+    return config_.initial_rtt * 2;
+  } else {
+    return (rtt_stats.smoothed_rtt + rtt_stats.rtt_var * 4 + config_.max_ack_delay) * probe_k_;
+  }
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/RldpSender.h

@@ -0,0 +1,81 @@
+#pragma once
+
+#include "td/utils/Time.h"
+#include "td/utils/Variant.h"
+
+#include "FecHelper.h"
+#include "SenderPackets.h"
+
+namespace ton {
+namespace rldp2 {
+struct Ack;
+struct BdwStats;
+struct RttStats;
+struct LossStats;
+
+inline td::uint32 sub_or_zero(td::uint32 a, td::uint32 b) {
+  if (a < b) {
+    return 0;
+  }
+  return a - b;
+}
+
+class RldpSender {
+ public:
+  struct Config {
+    static constexpr double DEFAULT_MAX_ACK_DELAY = 0.01;
+    static constexpr td::uint32 DEFAULT_PACKET_TRESHOLD = 3;
+    static constexpr double DEFAULT_INITIAL_RTT = 0.5;
+
+    double max_ack_delay{DEFAULT_MAX_ACK_DELAY};
+    double ack_delay{DEFAULT_MAX_ACK_DELAY};
+    td::uint32 packet_treshold{DEFAULT_PACKET_TRESHOLD};
+    double initial_rtt{DEFAULT_INITIAL_RTT};
+  };
+
+  RldpSender() = default;
+  RldpSender(Config config, td::uint32 symbols_count) : config_(config) {
+    fec_helper_.symbols_count = symbols_count;
+    extra_symbols_ = fec_helper_.get_left_fec_symbols_count();
+  }
+
+  struct ActionWait {
+    td::Timestamp wait_till;
+  };
+
+  struct ActionSend {
+    td::uint32 seqno;
+    bool is_probe;
+  };
+
+  td::Variant<ActionWait, ActionSend> next_action(td::Timestamp now, bool only_probe = false);
+  td::Variant<ActionWait, ActionSend> next_probe(td::Timestamp now);
+
+  td::uint32 get_inflight_symbols_count() const {
+    return packets_.in_flight_count();
+  }
+
+  SenderPackets::Update on_ack(const Ack &ack, double ack_delay, td::Timestamp now, RttStats &rtt_stats,
+                               BdwStats &bdw_stats, LossStats &loss_stats);
+
+  void on_send(td::uint32 seqno, td::Timestamp now, bool is_probe, const RttStats &rtt_stats,
+               const BdwStats &bdw_state);
+
+ private:
+  Config config_;
+  SenderPackets packets_;
+  FecHelper fec_helper_;
+  td::Timestamp probe_timeout_;
+  td::uint32 probe_k_{1};
+  td::uint32 extra_symbols_{0};
+
+  double get_loss_delay(const RttStats &rtt_stats);
+
+  double get_probe_delay(const RttStats &rtt_stats);
+
+  td::uint32 get_loss_seqno_delay() {
+    return config_.packet_treshold;
+  }
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/RttStats.cpp

@@ -0,0 +1,51 @@
+#include "RttStats.h"
+#include <cmath>
+
+namespace ton {
+namespace rldp2 {
+void RttStats::on_rtt_sample(double rtt_sample, double ack_delay, td::Timestamp now) {
+  if (rtt_sample < 0.001 || rtt_sample > 10) {
+    LOG(WARNING) << "Suspicious rtt sample " << rtt_sample;
+    return;
+  }
+  if (ack_delay < -1e-9 || ack_delay > 10) {
+    LOG(WARNING) << "Suspicious ack_delay " << ack_delay;
+    return;
+  }
+  rtt_sample = td::max(0.01, rtt_sample);
+
+  last_rtt = rtt_sample;
+
+  windowed_min_rtt_stat.add_event(rtt_sample, now.at());
+  auto windowed_min_rtt_sample = windowed_min_rtt_stat.get_stat(now.at()).get_stat();
+  if (windowed_min_rtt_sample) {
+    windowed_min_rtt = windowed_min_rtt_sample.value();
+  }
+
+  if (smoothed_rtt < 0) {
+    // ignore ack_delay just because
+    min_rtt = last_rtt;
+    smoothed_rtt = last_rtt;
+    rtt_var = last_rtt / 2;
+  } else {
+    if (rtt_sample < min_rtt) {
+      min_rtt = rtt_sample;
+    }
+
+    double adjusted_rtt = rtt_sample;
+    if (adjusted_rtt - ack_delay > min_rtt) {
+      adjusted_rtt -= ack_delay;
+    }
+
+    smoothed_rtt += (adjusted_rtt - smoothed_rtt) / 8;
+    double var = fabs(smoothed_rtt - adjusted_rtt);
+    rtt_var += (var - rtt_var) / 4;
+  }
+
+  if (td::Timestamp::in(smoothed_rtt, rtt_round_at).is_in_past(now)) {
+    rtt_round_at = now;
+    rtt_round++;
+  }
+}
+}  // namespace rldp2
+}  // namespace ton

rldp2/RttStats.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "td/utils/Time.h"
+#include "td/utils/TimedStat.h"
+
+namespace ton {
+namespace rldp2 {
+struct RttStats {
+  void on_rtt_sample(double rtt_sample, double ack_delay, td::Timestamp now);
+
+  double min_rtt = -1;
+  double windowed_min_rtt = -1;
+  double last_rtt = -1;
+  double smoothed_rtt = -1;
+  double rtt_var = -1;
+  td::uint32 rtt_round{0};
+
+ private:
+  td::Timestamp rtt_round_at;
+  td::TimedStat<td::MinStat<double>> windowed_min_rtt_stat{5, 0};
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/SenderPackets.cpp

@@ -0,0 +1,125 @@
+#include "SenderPackets.h"
+
+#include "td/utils/bits.h"
+
+namespace ton {
+namespace rldp2 {
+td::uint32 SenderPackets::next_seqno() const {
+  return last_seqno_ + 1;
+}
+
+SenderPackets::DropUpdate SenderPackets::drop_packets(const Limits &limits) {
+  while (!packets.empty()) {
+    auto &packet = packets.front();
+    if (!limits.should_drop(packet)) {
+      break;
+    }
+    mark_ack_or_lost(packet);
+    packets.pop();
+  }
+  DropUpdate update;
+  update.new_ack = total_ack_ - last_total_ack_;
+  update.new_lost = total_lost_ - last_total_lost_;
+  last_total_ack_ = total_ack_;
+  last_total_lost_ = total_lost_;
+  update.o_loss_at = std::move(last_loss_);
+  return update;
+}
+
+SenderPackets::Update SenderPackets::on_ack(Ack ack) {
+  ack.max_seqno = td::min(ack.max_seqno, last_seqno_);
+  ack.received_count = td::min(ack.received_count, ack.max_seqno);
+
+  // TODO: seqno of rldp and seqno of a packet must be completly separate seqnos
+  Update update;
+  if (received_count_ < ack.received_count) {
+    update.new_received = ack.received_count - received_count_;
+    left_ack_ += update.new_received;
+    left_ack_ = td::min(left_ack_, in_flight_count_);
+    received_count_ = ack.received_count;
+  }
+
+  if (max_packet_.seqno > ack.max_seqno) {
+    return update;
+  }
+
+  auto packet = get_packet(ack.max_seqno);
+  if (!packet) {
+    return update;
+  }
+
+  if (max_packet_.seqno < ack.max_seqno) {
+    update.was_max_updated = true;
+    max_packet_ = *packet;
+  }
+
+  for (td::uint32 i : td::BitsRange(ack.received_mask)) {
+    if (ack.max_seqno < i) {
+      break;
+    }
+    auto seqno = ack.max_seqno - i;
+    auto packet = get_packet(seqno);
+    if (!packet) {
+      break;
+    }
+    mark_ack(*packet);
+  }
+
+  return update;
+}
+void SenderPackets::mark_ack_or_lost(Packet &packet) {
+  if (left_ack_) {
+    mark_ack(packet);
+  } else {
+    mark_lost(packet);
+  }
+}
+
+void SenderPackets::mark_lost(Packet &packet) {
+  if (!packet.is_in_flight) {
+    return;
+  }
+  total_lost_++;
+  in_flight_count_--;
+  packet.is_in_flight = false;
+  last_loss_ = packet.sent_at;
+}
+
+void SenderPackets::mark_ack(Packet &packet) {
+  if (!packet.is_in_flight) {
+    return;
+  }
+  if (left_ack_ > 0) {
+    left_ack_--;
+  }
+  total_ack_++;
+  in_flight_count_--;
+  packet.is_in_flight = false;
+}
+
+SenderPackets::Packet *SenderPackets::get_packet(td::uint32 seqno) {
+  if (packets.empty()) {
+    return nullptr;
+  }
+  auto front_seqno = packets.front().seqno;
+  if (front_seqno > seqno) {
+    return nullptr;
+  }
+  td::uint32 index = seqno - front_seqno;
+  if (index >= packets.size()) {
+    return nullptr;
+  }
+  auto packet = packets.data() + index;
+  CHECK(packet->seqno == seqno);
+  return packet;
+}
+
+void SenderPackets::send(Packet packet) {
+  CHECK(next_seqno() == packet.seqno);
+  packets.push(packet);
+  last_seqno_++;
+  in_flight_count_ += packet.is_in_flight;
+}
+
+}  // namespace rldp2
+}  // namespace ton

rldp2/SenderPackets.h

@@ -0,0 +1,89 @@
+#pragma once
+
+#include "td/utils/VectorQueue.h"
+#include "Ack.h"
+#include "BdwStats.h"
+
+namespace ton {
+namespace rldp2 {
+class SenderPackets {
+ public:
+  struct Packet {
+    bool is_in_flight{false};
+    td::Timestamp sent_at;
+    td::uint32 seqno{0};
+    td::uint32 size{0};
+
+    BdwStats::PacketInfo bdw_packet_info;
+  };
+
+  struct Limits {
+    td::Timestamp sent_at;
+    td::uint32 seqno{0};
+    bool should_drop(const Packet &packet) const {
+      return !packet.is_in_flight || packet.sent_at < sent_at || packet.seqno < seqno;
+    }
+  };
+
+  struct DropUpdate {
+    td::uint32 new_ack{0};  // ~= new_received
+    td::uint32 new_lost{0};
+    td::optional<td::Timestamp> o_loss_at;
+  };
+
+  struct Update {
+    bool was_max_updated{false};
+    td::uint32 new_received{0};
+
+    DropUpdate drop_update;
+  };
+
+  td::VectorQueue<Packet> packets;
+
+  void send(Packet packet);
+
+  td::uint32 next_seqno() const;
+  DropUpdate drop_packets(const Limits &limits);
+
+  Update on_ack(Ack ack);
+
+  td::uint32 in_flight_count() const {
+    return in_flight_count_;
+  }
+  td::uint32 received_count() const {
+    return received_count_;
+  }
+  const Packet &max_packet() const {
+    return max_packet_;
+  }
+  td::Timestamp first_sent_at(td::Timestamp now) const {
+    if (!packets.empty()) {
+      now.relax(packets.front().sent_at);
+    }
+    return now;
+  }
+
+ private:
+  td::uint32 in_flight_count_{0};  // sum(packet.is_in_flight for packet in packets)
+  td::uint32 received_count_{0};
+  td::uint32 last_seqno_{0};
+  Packet max_packet_;
+
+  td::uint32 total_ack_{0};
+  td::uint32 total_lost_{0};
+  td::uint32 last_total_ack_{0};
+  td::uint32 last_total_lost_{0};
+
+  td::optional<td::Timestamp> last_loss_;
+  td::uint32 left_ack_{0};
+
+  void mark_ack_or_lost(Packet &packet);
+
+  void mark_lost(Packet &packet);
+
+  void mark_ack(Packet &packet);
+
+  Packet *get_packet(td::uint32 seqno);
+};
+}  // namespace rldp2
+}  // namespace ton

rldp2/rldp-in.hpp

@@ -0,0 +1,113 @@
+/*
+    This file is part of TON Blockchain Library.
+
+    TON Blockchain Library is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation, either version 2 of the License, or
+    (at your option) any later version.
+
+    TON Blockchain Library is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
+
+    Copyright 2017-2019 Telegram Systems LLP
+*/
+#pragma once
+
+#include "rldp.hpp"
+
+#include "tl-utils/tl-utils.hpp"
+#include "adnl/adnl-query.h"
+#include "adnl/adnl-peer-table.h"
+
+#include "td/utils/List.h"
+
+#include <map>
+#include <set>
+
+namespace ton {
+
+namespace rldp2 {
+
+class RldpLru : public td::ListNode {
+ public:
+  TransferId transfer_id() {
+    return transfer_id_;
+  }
+
+  RldpLru(TransferId transfer_id) : transfer_id_(transfer_id) {
+  }
+  RldpLru() {
+  }
+
+  static RldpLru *from_list_node(td::ListNode *node) {
+    return static_cast<RldpLru *>(node);
+  }
+
+ private:
+  TransferId transfer_id_;
+};
+
+class RldpConnectionActor;
+class RldpIn : public RldpImpl {
+ public:
+  static constexpr td::uint64 mtu() {
+    return (1ull << 37);
+  }
+  static constexpr td::uint32 lru_size() {
+    return 128;
+  }
+  void on_sent(TransferId transfer_id, td::Result<td::Unit> state);
+
+  void send_message(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::BufferSlice data) override;
+  void send_message_ex(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::Timestamp timeout,
+                       td::BufferSlice data) override;
+
+  void send_query(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, std::string name,
+                  td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data) override {
+    send_query_ex(src, dst, name, std::move(promise), timeout, std::move(data), default_mtu());
+  }
+  void send_query_ex(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, std::string name,
+                     td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data,
+                     td::uint64 max_answer_size) override;
+  void answer_query(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::Timestamp timeout,
+                    adnl::AdnlQueryId query_id, TransferId transfer_id, td::BufferSlice data);
+
+  void receive_message_part(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, td::BufferSlice data);
+
+  void process_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                       ton_api::rldp_message &message);
+  void process_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                       ton_api::rldp_query &message);
+  void process_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                       ton_api::rldp_answer &message);
+  void receive_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                       td::Result<td::BufferSlice> data);
+
+  void add_id(adnl::AdnlNodeIdShort local_id) override;
+
+  RldpIn(td::actor::ActorId<adnl::AdnlPeerTable> adnl) : adnl_(adnl) {
+  }
+
+ private:
+  std::unique_ptr<adnl::Adnl::Callback> make_adnl_callback();
+
+  td::actor::ActorId<adnl::AdnlPeerTable> adnl_;
+
+  std::map<std::pair<adnl::AdnlNodeIdShort, adnl::AdnlNodeIdShort>, td::actor::ActorOwn<RldpConnectionActor>>
+      connections_;
+
+  std::map<TransferId, td::Promise<td::BufferSlice>> queries_;
+
+  std::set<adnl::AdnlNodeIdShort> local_ids_;
+
+  td::actor::ActorId<RldpConnectionActor> create_connection(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst);
+};
+
+}  // namespace rldp2
+
+}  // namespace ton

rldp2/rldp.cpp

@@ -0,0 +1,246 @@
+/*
+    This file is part of TON Blockchain Library.
+
+    TON Blockchain Library is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation, either version 2 of the License, or
+    (at your option) any later version.
+
+    TON Blockchain Library is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
+
+    Copyright 2017-2019 Telegram Systems LLP
+*/
+#include "rldp-in.hpp"
+#include "auto/tl/ton_api.h"
+#include "auto/tl/ton_api.hpp"
+#include "td/utils/Random.h"
+#include "fec/fec.h"
+#include "RldpConnection.h"
+
+namespace ton {
+
+namespace rldp2 {
+
+class RldpConnectionActor : public td::actor::Actor, private ConnectionCallback {
+ public:
+  RldpConnectionActor(td::actor::ActorId<RldpIn> rldp, adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst,
+                      td::actor::ActorId<adnl::Adnl> adnl)
+      : rldp_(std::move(rldp)), src_(src), dst_(dst), adnl_(std::move(adnl)){};
+
+  void send(TransferId transfer_id, td::BufferSlice query, td::Timestamp timeout = td::Timestamp::never()) {
+    connection_.send(transfer_id, std::move(query), timeout);
+    yield();
+  }
+  void set_receive_limits(TransferId transfer_id, td::Timestamp timeout, td::uint64 max_size) {
+    connection_.set_receive_limits(transfer_id, timeout, max_size);
+  }
+  void receive_raw(td::BufferSlice data) {
+    connection_.receive_raw(std::move(data));
+    yield();
+  }
+
+ private:
+  td::actor::ActorId<RldpIn> rldp_;
+  adnl::AdnlNodeIdShort src_;
+  adnl::AdnlNodeIdShort dst_;
+  td::actor::ActorId<adnl::Adnl> adnl_;
+  RldpConnection connection_;
+
+  void loop() override {
+    alarm_timestamp() = connection_.run(*this);
+  }
+
+  void send_raw(td::BufferSlice data) override {
+    send_closure(adnl_, &adnl::Adnl::send_message, src_, dst_, std::move(data));
+  }
+  void receive(TransferId transfer_id, td::Result<td::BufferSlice> data) override {
+    send_closure(rldp_, &RldpIn::receive_message, dst_, src_, transfer_id, std::move(data));
+  }
+  void on_sent(TransferId transfer_id, td::Result<td::Unit> state) override {
+    send_closure(rldp_, &RldpIn::on_sent, transfer_id, std::move(state));
+  }
+};
+
+namespace {
+TransferId get_random_transfer_id() {
+  TransferId transfer_id;
+  td::Random::secure_bytes(transfer_id.as_slice());
+  return transfer_id;
+}
+TransferId get_responce_transfer_id(TransferId transfer_id) {
+  return transfer_id ^ TransferId::ones();
+}
+}  // namespace
+
+void RldpIn::send_message(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::BufferSlice data) {
+  return send_message_ex(src, dst, td::Timestamp::in(10.0), std::move(data));
+}
+
+void RldpIn::send_message_ex(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::Timestamp timeout,
+                             td::BufferSlice data) {
+  td::Bits256 id;
+  td::Random::secure_bytes(id.as_slice());
+
+  auto B = serialize_tl_object(create_tl_object<ton_api::rldp_message>(id, std::move(data)), true);
+
+  auto transfer_id = get_random_transfer_id();
+  send_closure(create_connection(src, dst), &RldpConnectionActor::send, transfer_id, std::move(B), timeout);
+}
+
+void RldpIn::send_query_ex(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, std::string name,
+                           td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data,
+                           td::uint64 max_answer_size) {
+  auto query_id = adnl::AdnlQuery::random_query_id();
+
+  auto date = static_cast<td::uint32>(timeout.at_unix()) + 1;
+  auto B = serialize_tl_object(create_tl_object<ton_api::rldp_query>(query_id, max_answer_size, date, std::move(data)),
+                               true);
+
+  auto connection = create_connection(src, dst);
+  auto transfer_id = get_random_transfer_id();
+  auto response_transfer_id = get_responce_transfer_id(transfer_id);
+  send_closure(connection, &RldpConnectionActor::set_receive_limits, response_transfer_id, timeout, max_answer_size);
+  send_closure(connection, &RldpConnectionActor::send, transfer_id, std::move(B), timeout);
+
+  queries_.emplace(response_transfer_id, std::move(promise));
+}
+
+void RldpIn::answer_query(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::Timestamp timeout,
+                          adnl::AdnlQueryId query_id, TransferId transfer_id, td::BufferSlice data) {
+  auto B = serialize_tl_object(create_tl_object<ton_api::rldp_answer>(query_id, std::move(data)), true);
+
+  send_closure(create_connection(src, dst), &RldpConnectionActor::send, transfer_id, std::move(B), timeout);
+}
+
+void RldpIn::receive_message_part(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, td::BufferSlice data) {
+  send_closure(create_connection(local_id, source), &RldpConnectionActor::receive_raw, std::move(data));
+}
+
+td::actor::ActorId<RldpConnectionActor> RldpIn::create_connection(adnl::AdnlNodeIdShort src,
+                                                                  adnl::AdnlNodeIdShort dst) {
+  auto it = connections_.find(std::make_pair(src, dst));
+  if (it != connections_.end()) {
+    return it->second.get();
+  }
+  auto connection = td::actor::create_actor<RldpConnectionActor>("RldpConnection", actor_id(this), src, dst, adnl_);
+  auto res = connection.get();
+  connections_[std::make_pair(src, dst)] = std::move(connection);
+  return res;
+}
+
+void RldpIn::receive_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                             td::Result<td::BufferSlice> r_data) {
+  if (r_data.is_error()) {
+    auto it = queries_.find(transfer_id);
+    if (it != queries_.end()) {
+      it->second.set_error(r_data.move_as_error());
+      queries_.erase(it);
+    } else {
+      VLOG(RLDP_INFO) << "received error to unknown transfer_id " << transfer_id << " " << r_data.error();
+    }
+    return;
+  }
+
+  auto data = r_data.move_as_ok();
+  //LOG(ERROR) << "RECEIVE MESSAGE " << data.size();
+  auto F = fetch_tl_object<ton_api::rldp_Message>(std::move(data), true);
+  if (F.is_error()) {
+    VLOG(RLDP_INFO) << "failed to parse rldp packet [" << source << "->" << local_id << "]: " << F.move_as_error();
+    return;
+  }
+
+  ton_api::downcast_call(*F.move_as_ok().get(),
+                         [&](auto &obj) { this->process_message(source, local_id, transfer_id, obj); });
+}
+
+void RldpIn::process_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                             ton_api::rldp_message &message) {
+  td::actor::send_closure(adnl_, &adnl::AdnlPeerTable::deliver, source, local_id, std::move(message.data_));
+}
+
+void RldpIn::process_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                             ton_api::rldp_query &message) {
+  auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), source, local_id,
+                                       timeout = td::Timestamp::at_unix(message.timeout_), query_id = message.query_id_,
+                                       max_answer_size = static_cast<td::uint64>(message.max_answer_size_),
+                                       transfer_id](td::Result<td::BufferSlice> R) {
+    if (R.is_ok()) {
+      auto data = R.move_as_ok();
+      if (data.size() > max_answer_size) {
+        VLOG(RLDP_NOTICE) << "rldp query failed: answer too big";
+      } else {
+        td::actor::send_closure(SelfId, &RldpIn::answer_query, local_id, source, timeout, query_id,
+                                transfer_id ^ TransferId::ones(), std::move(data));
+      }
+    } else {
+      VLOG(RLDP_NOTICE) << "rldp query failed: " << R.move_as_error();
+    }
+  });
+  VLOG(RLDP_DEBUG) << "delivering rldp query";
+  td::actor::send_closure(adnl_, &adnl::AdnlPeerTable::deliver_query, source, local_id, std::move(message.data_),
+                          std::move(P));
+}
+
+void RldpIn::process_message(adnl::AdnlNodeIdShort source, adnl::AdnlNodeIdShort local_id, TransferId transfer_id,
+                             ton_api::rldp_answer &message) {
+  auto it = queries_.find(transfer_id);
+  if (it != queries_.end()) {
+    it->second.set_value(std::move(message.data_));
+    queries_.erase(it);
+  } else {
+    VLOG(RLDP_INFO) << "received answer to unknown query " << message.query_id_;
+  }
+}
+
+void RldpIn::on_sent(TransferId transfer_id, td::Result<td::Unit> state) {
+  //TODO: completed transfer
+}
+
+void RldpIn::add_id(adnl::AdnlNodeIdShort local_id) {
+  if (local_ids_.count(local_id) == 1) {
+    return;
+  }
+
+  std::vector<std::string> X{adnl::Adnl::int_to_bytestring(ton_api::rldp2_messagePart::ID),
+                             adnl::Adnl::int_to_bytestring(ton_api::rldp2_confirm::ID),
+                             adnl::Adnl::int_to_bytestring(ton_api::rldp2_complete::ID)};
+  for (auto &x : X) {
+    td::actor::send_closure(adnl_, &adnl::Adnl::subscribe, local_id, x, make_adnl_callback());
+  }
+
+  local_ids_.insert(local_id);
+}
+
+std::unique_ptr<adnl::Adnl::Callback> RldpIn::make_adnl_callback() {
+  class Callback : public adnl::Adnl::Callback {
+   private:
+    td::actor::ActorId<RldpIn> id_;
+
+   public:
+    Callback(td::actor::ActorId<RldpIn> id) : id_(id) {
+    }
+    void receive_message(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::BufferSlice data) override {
+      td::actor::send_closure(id_, &RldpIn::receive_message_part, src, dst, std::move(data));
+    }
+    void receive_query(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::BufferSlice data,
+                       td::Promise<td::BufferSlice> promise) override {
+      promise.set_error(td::Status::Error(ErrorCode::notready, "rldp does not support queries"));
+    }
+  };
+
+  return std::make_unique<Callback>(actor_id(this));
+}
+
+td::actor::ActorOwn<Rldp> Rldp::create(td::actor::ActorId<adnl::Adnl> adnl) {
+  return td::actor::create_actor<RldpIn>("rldp", td::actor::actor_dynamic_cast<adnl::AdnlPeerTable>(adnl));
+}
+
+}  // namespace rldp2
+
+}  // namespace ton

rldp2/rldp.h

@@ -0,0 +1,45 @@
+/*
+    This file is part of TON Blockchain Library.
+
+    TON Blockchain Library is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation, either version 2 of the License, or
+    (at your option) any later version.
+
+    TON Blockchain Library is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
+
+    Copyright 2017-2020 Telegram Systems LLP
+*/
+#pragma once
+
+#include "adnl/adnl.h"
+
+namespace ton {
+
+namespace rldp2 {
+
+class Rldp : public adnl::AdnlSenderInterface {
+ public:
+  virtual ~Rldp() = default;
+
+  static constexpr td::uint64 default_mtu() {
+    return adnl::Adnl::get_mtu();
+  }
+
+  virtual void add_id(adnl::AdnlNodeIdShort local_id) = 0;
+
+  virtual void send_message_ex(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::Timestamp timeout,
+                               td::BufferSlice data) = 0;
+
+  static td::actor::ActorOwn<Rldp> create(td::actor::ActorId<adnl::Adnl> adnl);
+};
+
+}  // namespace rldp2
+
+}  // namespace ton

rldp2/rldp.hpp

@@ -0,0 +1,48 @@
+/*
+    This file is part of TON Blockchain Library.
+
+    TON Blockchain Library is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation, either version 2 of the License, or
+    (at your option) any later version.
+
+    TON Blockchain Library is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+
+    You should have received a copy of the GNU Lesser General Public License
+    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
+
+    Copyright 2017-2019 Telegram Systems LLP
+*/
+#pragma once
+
+#include "rldp.h"
+
+#include "tl-utils/tl-utils.hpp"
+#include "adnl/adnl-query.h"
+
+#include <map>
+
+namespace ton {
+
+namespace rldp2 {
+
+constexpr int VERBOSITY_NAME(RLDP_WARNING) = verbosity_WARNING;
+constexpr int VERBOSITY_NAME(RLDP_NOTICE) = verbosity_INFO;
+constexpr int VERBOSITY_NAME(RLDP_INFO) = verbosity_DEBUG;
+constexpr int VERBOSITY_NAME(RLDP_DEBUG) = verbosity_DEBUG;
+constexpr int VERBOSITY_NAME(RLDP_EXTRA_DEBUG) = verbosity_DEBUG + 1;
+
+using TransferId = td::Bits256;
+
+class RldpImpl : public Rldp {
+ public:
+  //virtual void transfer_completed(TransferId transfer_id) = 0;
+  //virtual void in_transfer_completed(TransferId transfer_id) = 0;
+};
+
+}  // namespace rldp2
+
+}  // namespace ton