udt.cc

udt的一个源代码

   last_feedback_time_ = new double [size_];
   count_ = new int [size_];
   next_ = new int [size_];
   prior_ = new int [size_];

   // -1 means there is no data in the node
   for (int i = 0; i < size; ++ i)
   {
      data1_[i] = -1;
      data2_[i] = -1;
   }

   length_ = 0;
   head_ = -1;
   tail_ = -1;
}

RcvLossList::~RcvLossList()
{
   delete [] data1_;
   delete [] data2_;
   delete [] last_feedback_time_;
   delete [] count_;
   delete [] next_;
   delete [] prior_;
}

void RcvLossList::insert(const int& seqno1, const int& seqno2)
{
   // Data to be inserted must be larger than all those in the list
   // guaranteed by the UDT receiver

   if (0 == length_)
   {
      // insert data into an empty list
      head_ = 0;
      tail_ = 0;
      data1_[head_] = seqno1;
      if (seqno2 != seqno1)
         data2_[head_] = seqno2;

      last_feedback_time_[head_] = Scheduler::instance().clock();
      count_[head_] = 2;

      next_[head_] = -1;
      prior_[head_] = -1;
      length_ += getLength(seqno1, seqno2);

      return;
   }

   // otherwise searching for the position where the node should be

   int offset = seqno1 - data1_[head_];

   if (offset < -seq_no_th_)
      offset += max_seq_no_;

   int loc = (head_ + offset) % size_;

   if ((-1 != data2_[tail_]) && (incSeqNo(data2_[tail_]) == seqno1))
   {
      // coalesce with prior node, e.g., [2, 5], [6, 7] becomes [2, 7]
      loc = tail_;
      data2_[loc] = seqno2;
   }
   else
   {
      // create new node
      data1_[loc] = seqno1;

      if (seqno2 != seqno1)
         data2_[loc] = seqno2;

      next_[tail_] = loc;
      prior_[loc] = tail_;
      next_[loc] = -1;
      tail_ = loc;
   }

   // Initilize time stamp
   last_feedback_time_[loc] = Scheduler::instance().clock();
   count_[loc] = 2;

   length_ += getLength(seqno1, seqno2);
}

bool RcvLossList::remove(const int& seqno)
{
   if (0 == length_)
      return false; 

   // locate the position of "seqno" in the list
   int offset = seqno - data1_[head_];

   if (offset < -seq_no_th_)
      offset += max_seq_no_;

   if (offset < 0)
      return false;

   int loc = (head_ + offset) % size_;

   if (seqno == data1_[loc])
   {
      // This is a seq. no. that starts the loss sequence

      if (-1 == data2_[loc])
      {
         // there is only 1 loss in the sequence, delete it from the node
         if (head_ == loc)
         {
            head_ = next_[head_];
            if (-1 != head_)
               prior_[head_] = -1;
         }
         else
         {
            next_[prior_[loc]] = next_[loc];
            if (-1 != next_[loc])
               prior_[next_[loc]] = prior_[loc];
            else
               tail_ = prior_[loc];
         }

         data1_[loc] = -1;
      }
      else
      {
         // there are more than 1 loss in the sequence
         // move the node to the next and update the starter as the next loss inSeqNo(seqno)

         // find next node
         int i = (loc + 1) % size_;

         // remove the "seqno" and change the starter as next seq. no.
         data1_[i] = incSeqNo(data1_[loc]);

         // process the sequence end
         if (greaterthan(data2_[loc], incSeqNo(data1_[loc])))
            data2_[i] = data2_[loc];

         // replicate the time stamp and report counter
         last_feedback_time_[i] = last_feedback_time_[loc];
         count_[i] = count_[loc];

         // remove the current node
         data1_[loc] = -1;
         data2_[loc] = -1;
 
         // update list pointer
         next_[i] = next_[loc];
         prior_[i] = prior_[loc];

         if (head_ == loc)
            head_ = i;
         else
            next_[prior_[i]] = i;

         if (tail_ == loc)
            tail_ = i;
         else
            prior_[next_[i]] = i;
      }

      length_ --;

      return true;
   }

   // There is no loss sequence in the current position
   // the "seqno" may be contained in a previous node

   // searching previous node
   int i = (loc - 1 + size_) % size_;
   while (-1 == data1_[i])
      i = (i - 1 + size_) % size_;

   // not contained in this node, return
   if ((-1 == data2_[i]) || greaterthan(seqno, data2_[i]))
       return false;

   if (seqno == data2_[i])
   {
      // it is the sequence end

      if (seqno == incSeqNo(data1_[i]))
         data2_[i] = -1;
      else
         data2_[i] = decSeqNo(seqno);
   }
   else
   {
      // split the sequence

      // construct the second sequence from incSeqNo(seqno) to the original sequence end
      // located at "loc + 1"
      loc = (loc + 1) % size_;

      data1_[loc] = incSeqNo(seqno);
      if (greaterthan(data2_[i], incSeqNo(seqno)))
         data2_[loc] = data2_[i];

      // the first (original) sequence is between the original sequence start to decSeqNo(seqno)
      if (seqno == incSeqNo(data1_[i]))
         data2_[i] = -1;
      else
         data2_[i] = decSeqNo(seqno);

      // replicate the time stamp and report counter
      last_feedback_time_[loc] = last_feedback_time_[i];
      count_[loc] = count_[i];

      // update the list pointer
      next_[loc] = next_[i];
      next_[i] = loc;
      prior_[loc] = i;

      if (tail_ == i)
         tail_ = loc;
      else
         prior_[next_[loc]] = loc;
   }

   length_ --;

   return true;
}

int RcvLossList::getLossLength() const
{
   return length_;
}

int RcvLossList::getFirstLostSeq() const
{
   if (0 == length_)
      return -1;

   return data1_[head_];
}

void RcvLossList::getLossArray(int* array, int* len, const int& limit, const double& threshold)
{
   double currtime = Scheduler::instance().clock();

   int i  = head_;

   len = 0;

   while ((*len < limit - 1) && (-1 != i))
   {
      if (currtime - last_feedback_time_[i] > count_[i] * threshold)
      {
         array[*len] = data1_[i];
         if (-1 != data2_[i])
         {
            // there are more than 1 loss in the sequence
            array[*len] |= 0x80000000;
            ++ *len;
            array[*len] = data2_[i];
         }

         ++ *len;

         // update the timestamp
         last_feedback_time_[i] = Scheduler::instance().clock();
         // update how many times this loss has been fed back, the "k" in UDT paper
         ++ count_[i];
      }

      i = next_[i];
   }
}





////////////////////////////////////////////////////////////////////////////
//
AckWindow::AckWindow():
size_(1024),
head_(0),
tail_(0)
{
   ack_seqno_ = new int[size_];
   ack_ = new int[size_];
   ts_ = new double[size_];

   ack_seqno_[0] = -1;
}

AckWindow::~AckWindow()
{
   delete [] ack_seqno_;
   delete [] ack_;
   delete [] ts_;
}

void AckWindow::store(const int& seq, const int& ack)
{
   head_ = (head_ + 1) % size_;
   ack_seqno_[head_] = seq;
   ack_[head_] = ack;
   *(ts_ + head_) = Scheduler::instance().clock();

   // overwrite the oldest ACK since it is not likely to be acknowledged
   if (head_ == tail_)
      tail_ = (tail_ + 1) % size_;
}

double AckWindow::acknowledge(const int& seq, int& ack)
{
   if (head_ >= tail_)
   {
      // Head has not exceeded the physical boundary of the window
      for (int i = tail_; i <= head_; i ++)
         // looking for indentical ACK Seq. No.
         if (seq == ack_seqno_[i])
         {
            // return the Data ACK it carried
            ack = ack_[i];

            // calculate RTT
            double rtt = Scheduler::instance().clock() - ts_[i];
            if (i == head_)
               tail_ = head_ = 0;
            else
               tail_ = (i + 1) % size_;

            return rtt;
         }

      // Bad input, the ACK node has been overwritten
      return -1;
   }

   // head has exceeded the physical window boundary, so it is behind to tail
   for (int i = tail_; i <= head_ + size_; i ++)
      // looking for indentical ACK seq. no.
      if (seq == ack_seqno_[i % size_])
      {
         // return Data ACK
         i %= size_;
         ack = ack_[i];

         // calculate RTT
         double currtime = Scheduler::instance().clock();
         double rtt = currtime - ts_[i];
         if (i == head_)
            tail_ = head_ = 0;
         else
            tail_ = (i + 1) % size_;

         return rtt;
      }

   // bad input, the ACK node has been overwritten
   return -1;
}

//
TimeWindow::TimeWindow():
size_(16)
{
   pkt_window_ = new double[size_];
   rtt_window_ = new double[size_];
   pct_window_ = new double[size_];
   pdt_window_ = new double[size_];
   probe_window_ = new double[size_];

   pkt_window_ptr_ = 0;
   rtt_window_ptr_ = 0;
   probe_window_ptr_ = 0;

   first_round_ = true;

   for (int i = 0; i < size_; ++ i)
   {
      pkt_window_[i] = 1.0;
      rtt_window_[i] = pct_window_[i] = pdt_window_[i] = 0.0;
      probe_window_[i] = 1000.0;
   }

   last_arr_time_ = Scheduler::instance().clock();
}

TimeWindow::~TimeWindow()
{
   delete [] pkt_window_;
   delete [] rtt_window_;
   delete [] pct_window_;
   delete [] pdt_window_;
}

int TimeWindow::getbandwidth() const
{
   double temp;
   for (int i = 0; i < ((size_ >> 1) + 1); ++ i)
      for (int j = i; j < size_; ++ j)
         if (probe_window_[i] > probe_window_[j])
         {
            temp = probe_window_[i];
            probe_window_[i] = probe_window_[j];
            probe_window_[j] = temp;
         }

   if (0 == probe_window_[size_ >> 1])
      return 0;

   return int(ceil(1.0 / probe_window_[size_ >> 1]));
}

int TimeWindow::getpktspeed() const
{
   if ((first_round_) && (pkt_window_ptr_ > 0))
   {
      if ((pkt_window_ptr_ > 1) && (pkt_window_[pkt_window_ptr_ - 1] < 2 * pkt_window_[pkt_window_ptr_ - 2]))
         return (int)ceil(1.0 / pkt_window_[pkt_window_ptr_ - 1]);

      return 0;
   }

   double temp;
   for (int i = 0; i < ((size_ >> 1) + 1); ++ i)
      for (int j = i; j < size_; ++ j)
         if (pkt_window_[i] > pkt_window_[j])
         {
            temp = pkt_window_[i];
            pkt_window_[i] = pkt_window_[j];
            pkt_window_[j] = temp;
         }

   double median = pkt_window_[size_ >> 1];
   int count = 0;
   double sum = 0.0;

   for (int i = 0; i < size_; ++ i)
      if ((pkt_window_[i] < (median * 2)) && (pkt_window_[i] > (median / 2)))
      {
         ++ count;
         sum += pkt_window_[i];
      }

   if (count > (size_ >> 1))
      return (int)ceil(1.0 / (sum / count));
   else
      return 0;
}

bool TimeWindow::getdelaytrend() const
{
   double pct = 0.0;
   double pdt = 0.0;

   for (int i = 0; i < size_; ++i)
      if (i != rtt_window_ptr_)
      {
         pct += pct_window_[i];
         pdt += pdt_window_[i];
      }

   pct /= size_ - 1.0;
   if (0.0 != pdt)
      pdt = (rtt_window_[(rtt_window_ptr_ - 1 + size_) % size_] - rtt_window_[rtt_window_ptr_]) / pdt;

   return ((pct > 0.66) && (pdt > 0.45)) || ((pct > 0.54) && (pdt > 0.55));
}

void TimeWindow::pktarrival()
{
   curr_arr_time_ = Scheduler::instance().clock();

   pkt_window_[pkt_window_ptr_] = curr_arr_time_ - last_arr_time_;

   pkt_window_ptr_ = (pkt_window_ptr_ + 1) % size_;
 
   if (0 == pkt_window_ptr_)
      first_round_ = false;

   last_arr_time_ = curr_arr_time_;
}

void TimeWindow::probe1arrival()
{
   probe_time_ = Scheduler::instance().clock();
}

void TimeWindow::probe2arrival()
{
   probe_window_[probe_window_ptr_] = Scheduler::instance().clock() - probe_time_;;

   probe_window_ptr_ = (probe_window_ptr_ + 1) % size_;

   last_arr_time_ = Scheduler::instance().clock();
}

void TimeWindow::ack2arrival(const double& rtt)
{
   rtt_window_[rtt_window_ptr_] = rtt;
   pct_window_[rtt_window_ptr_] = (rtt > rtt_window_[(rtt_window_ptr_ - 1 + size_) % size_]) ? 1 : 0;
   pdt_window_[rtt_window_ptr_] = fabs(rtt - rtt_window_[(rtt_window_ptr_ - 1 + size_) % size_]);

   rtt_window_ptr_ = (rtt_window_ptr_ + 1) % size_;
}