segment-fetcher.hpp

Go to the documentation of this file.

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/*
 * Copyright (c) 2013-2021 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.hpp"

#include <queue>
#include <set>

namespace ndn {
namespace util {

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,
  };

  class Options
  {
  public:
    Options()
    {
    }

    void
    validate();

  public:
    time::milliseconds interestLifetime = 4_s; 
    time::milliseconds maxTimeout = 60_s; 
    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; 
    RttEstimator::Options rttOptions; 
    size_t flowControlWindow = 25000; 
  };

  static shared_ptr<SegmentFetcher>
  start(Face& face,
        const Interest& baseInterest,
        security::Validator& validator,
        const Options& 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<SegmentFetcher, ConstBufferPtr> onComplete;

  Signal<SegmentFetcher, uint32_t, std::string> onError;

  Signal<SegmentFetcher, Data> afterSegmentReceived;

  Signal<SegmentFetcher, Data> afterSegmentValidated;

  Signal<SegmentFetcher> afterSegmentNacked;

  Signal<SegmentFetcher> afterSegmentTimedOut;

  Signal<SegmentFetcher, ConstBufferPtr> onInOrderData;

  Signal<SegmentFetcher> onInOrderComplete;

private:
  enum class SegmentState {
    FirstInterest, 
    InRetxQueue,   
    Retransmitted, 
  };

  class PendingSegment
  {
  public:
    SegmentState state;
    time::steady_clock::TimePoint 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;
  RttEstimator m_rttEstimator;

  time::steady_clock::TimePoint 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 util
} // namespace ndn

#endif // NDN_CXX_UTIL_SEGMENT_FETCHER_HPP