segment-fetcher.hpp

Go to the documentation of this file.

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/*
 * Copyright (c) 2013-2024 Regents of the University of California.
 *
 * This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
 *
 * ndn-cxx 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 3 of the License, or (at your option) any later version.
 *
 * ndn-cxx 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 copies of the GNU General Public License and GNU Lesser
 * General Public License along with ndn-cxx, e.g., in COPYING.md file.  If not, see
 * <http://www.gnu.org/licenses/>.
 *
 * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
 */
 
#ifndef NDN_CXX_UTIL_SEGMENT_FETCHER_HPP
#define NDN_CXX_UTIL_SEGMENT_FETCHER_HPP
 
#include "ndn-cxx/face.hpp"
#include "ndn-cxx/security/validator.hpp"
#include "ndn-cxx/util/rtt-estimator.hpp"
#include "ndn-cxx/util/scheduler.hpp"
#include "ndn-cxx/util/signal/signal.hpp"
 
#include <queue>
#include <set>
 
namespace ndn {
 
struct SegmentFetcherOptions
{
  time::milliseconds interestLifetime = 4_s;
  time::milliseconds maxTimeout = 60_s;
  bool probeLatestVersion = true;
  bool inOrder = false;
  bool useConstantInterestTimeout = false;
  bool useConstantCwnd = false;
  bool disableCwa = false;
  bool resetCwndToInit = false;
  bool ignoreCongMarks = false;
  double initCwnd = 1.0;
  double initSsthresh = std::numeric_limits<double>::max();
  double aiStep = 1.0;
  double mdCoef = 0.5;
  util::RttEstimator::Options rttOptions;
  size_t flowControlWindow = 25000;
 
  void
  validate();
};
 
class SegmentFetcher : noncopyable
{
public:
  enum ErrorCode {
    INTEREST_TIMEOUT = 1,
    DATA_HAS_NO_SEGMENT = 2,
    SEGMENT_VALIDATION_FAIL = 3,
    NACK_ERROR = 4,
    FINALBLOCKID_NOT_SEGMENT = 5,
  };
 
  using Options = SegmentFetcherOptions;
 
  static shared_ptr<SegmentFetcher>
  start(Face& face, const Interest& baseInterest, security::Validator& validator,
        const Options& options = {});
 
  void
  stop();
 
private:
  class PendingSegment;
 
  SegmentFetcher(Face& face, security::Validator& validator, const Options& options);
 
  static bool
  shouldStop(const weak_ptr<SegmentFetcher>& weakSelf);
 
  void
  fetchFirstSegment(const Interest& baseInterest, bool isRetransmission);
 
  void
  fetchSegmentsInWindow(const Interest& origInterest);
 
  void
  sendInterest(uint64_t segNum, const Interest& interest, bool isRetransmission);
 
  void
  afterSegmentReceivedCb(const Interest& origInterest, const Data& data,
                         const weak_ptr<SegmentFetcher>& weakSelf);
 
  void
  afterValidationSuccess(const Data& data, const Interest& origInterest,
                         std::map<uint64_t, PendingSegment>::iterator pendingSegmentIt,
                         const weak_ptr<SegmentFetcher>& weakSelf);
 
  void
  afterValidationFailure(const Data& data,
                         const security::ValidationError& error,
                         const weak_ptr<SegmentFetcher>& weakSelf);
 
  void
  afterNackReceivedCb(const Interest& origInterest, const lp::Nack& nack,
                      const weak_ptr<SegmentFetcher>& weakSelf);
 
  void
  afterTimeoutCb(const Interest& origInterest,
                 const weak_ptr<SegmentFetcher>& weakSelf);
 
  void
  afterNackOrTimeout(const Interest& origInterest);
 
  void
  finalizeFetch();
 
  void
  windowIncrease();
 
  void
  windowDecrease();
 
  void
  signalError(uint32_t code, const std::string& msg);
 
  void
  updateRetransmittedSegment(uint64_t segmentNum,
                             const PendingInterestHandle& pendingInterest,
                             scheduler::EventId timeoutEvent);
 
  void
  cancelExcessInFlightSegments();
 
  bool
  checkAllSegmentsReceived();
 
  time::milliseconds
  getEstimatedRto();
 
public:
  signal::Signal<SegmentFetcher, ConstBufferPtr> onComplete;
 
  signal::Signal<SegmentFetcher, uint32_t, std::string> onError;
 
  signal::Signal<SegmentFetcher, Data> afterSegmentReceived;
 
  signal::Signal<SegmentFetcher, Data> afterSegmentValidated;
 
  signal::Signal<SegmentFetcher> afterSegmentNacked;
 
  signal::Signal<SegmentFetcher> afterSegmentTimedOut;
 
  signal::Signal<SegmentFetcher, ConstBufferPtr> onInOrderData;
 
  signal::Signal<SegmentFetcher> onInOrderComplete;
 
private:
  enum class SegmentState {
    FirstInterest, 
    InRetxQueue,   
    Retransmitted, 
  };
 
  class PendingSegment
  {
  public:
    SegmentState state;
    time::steady_clock::time_point sendTime;
    ScopedPendingInterestHandle hdl;
    scheduler::ScopedEventId timeoutEvent;
  };
 
NDN_CXX_PUBLIC_WITH_TESTS_ELSE_PRIVATE:
  static constexpr double MIN_SSTHRESH = 2.0;
 
  shared_ptr<SegmentFetcher> m_this;
 
  Options m_options;
  Face& m_face;
  Scheduler m_scheduler;
  security::Validator& m_validator;
  util::RttEstimator m_rttEstimator;
 
  time::steady_clock::time_point m_timeLastSegmentReceived;
  std::queue<uint64_t> m_retxQueue;
  Name m_versionedDataName;
  uint64_t m_nextSegmentNum = 0;
  double m_cwnd;
  double m_ssthresh;
  int64_t m_nSegmentsInFlight = 0;
  int64_t m_nSegments = 0;
  uint64_t m_highInterest = 0;
  uint64_t m_highData = 0;
  uint64_t m_recPoint = 0;
  int64_t m_nReceived = 0;
  int64_t m_nBytesReceived = 0;
  uint64_t m_nextSegmentInOrder = 0;
 
  std::map<uint64_t, Buffer> m_segmentBuffer;
  std::map<uint64_t, PendingSegment> m_pendingSegments;
  std::set<uint64_t> m_receivedSegments;
};
 
} // namespace ndn
 
#endif // NDN_CXX_UTIL_SEGMENT_FETCHER_HPP