cc.h

udt的一个源代码

      if ((currtime.tv_sec - m_LastCCTime.tv_sec) * 1000000 + (currtime.tv_usec - m_LastCCTime.tv_usec) < m_iRTT)
         return;

      expected = m_dCWndSize * 1000.0 / m_iBaseRTT;
      actual = m_iPktSent / ((currtime.tv_sec - m_LastCCTime.tv_sec) * 1000.0 + (currtime.tv_usec - m_LastCCTime.tv_usec) / 1000.0);
      diff = expected - actual;

      if (m_bSlowStart)
      {
         if (diff < gamma)
            m_bSlowStart = false;

         if (m_iSSRound & 1)
            m_dCWndSize *= 2;

         m_iSSRound ++;
      }
      else
      {
         if (diff < alpha)
            m_dCWndSize += 1.0;
         else if (diff > beta)
            m_dCWndSize -= 1.0;
      }

      gettimeofday(&m_LastCCTime, 0);
      m_iPktSent = 0;
      if (m_iBaseRTT > m_iRTT)
         m_iBaseRTT = m_iRTT;
   }

   virtual void onPktSent(const CPacket* pkt)
   {
      m_pAckWindow->store(pkt->m_iSeqNo, pkt->m_iSeqNo);
      m_iPktSent ++;
   }

   virtual void onTimeout()
   {
   }

protected:
   int m_iSSRound;

   int m_iRTT;
   int m_iBaseRTT;
   timeval m_LastCCTime;

   int m_iPktSent;

   static const int alpha = 30; //kbps
   static const int beta = 60;  //kbps
   static const int gamma = 30; //kbps

   CACKWindow* m_pAckWindow;
};


/*****************************************************************************
FAST TCP
Reference:
1. C. Jin, D. X. Wei and S. H. Low, "FAST TCP: motivation, architecture, 
   algorithms, performance", IEEE Infocom, March 2004
2. C. Jin, D. X. Wei and S. H. Low, FAST TCP for High-Speed Long-Distance 
   Networks, Internet Draft, draft-jwl-tcp-fast-01.txt,
   http://netlab.caltech.edu/pub/papers/draft-jwl-tcp-fast-01.txt

Note:
   Precision of RTT measurement may make great difference in the throughput
*****************************************************************************/

class CFAST: public CVegas
{
public:
   CFAST()
   {
      m_dOldWin = m_dCWndSize;
      m_iNumACK = 100000;
   }

   virtual void onACK(const int& ack)
   {
      if (ack == m_iLastACK)
      {
         if (3 == ++ m_iDupACKCount)
         {
            m_dCWndSize *= 0.875;
            return;
         }
      }
      else
      {
         if (m_iDupACKCount >= 3)
         {
//            m_dCWndSize = m_issthresh;
//            return;
         }

         m_iLastACK = ack;
         m_iDupACKCount = 1;
      }

      if (0 == (++ m_iACKCount % m_iNumACK))
         m_dCWndSize += m_iIncDec;

      int rtt = m_pAckWindow->acknowledge(ack, const_cast<int&>(ack));
      if (rtt > 0)
         m_iRTT = (m_iRTT * 7 + rtt) >> 3;

      timeval currtime;
      gettimeofday(&currtime, 0);

      if ((currtime.tv_sec - m_LastCCTime.tv_sec) * 1000000 + (currtime.tv_usec - m_LastCCTime.tv_usec) < 2 * m_iRTT)
         return;

      m_dNewWin = 0.5 * (m_dOldWin + (double(m_iBaseRTT) / m_iRTT) * m_dCWndSize + alpha);
      if (m_dNewWin > 2.0 * m_dCWndSize)
        m_dNewWin = 2.0 * m_dCWndSize;

      m_iNumACK = int(ceil(fabs(m_dCWndSize / (m_dNewWin - m_dCWndSize)) / 2.0));
      if (m_dNewWin > m_dCWndSize)
         m_iIncDec = 1;
      else
         m_iIncDec = -1;

      m_dOldWin = m_dCWndSize;

      gettimeofday(&m_LastCCTime, 0);
      m_iPktSent = 0;
      if (m_iBaseRTT > m_iRTT)
         m_iBaseRTT = m_iRTT;
   }

private:
   static const int alpha = 200;

   double m_dOldWin;
   double m_dNewWin;

   int m_iNumACK;
   int m_iIncDec;
   int m_iACKCount;
};


/*****************************************************************************
Reliable UDP Blast

Note:
The class demostrates the simplest control mechanism. The sending rate can
be set at any time by using setRate().
*****************************************************************************/

class CUDPBlast: public CCC
{
public:
   CUDPBlast()
   {
      m_dPktSndPeriod = 1000000; 
      m_dCWndSize = 83333.0; 
   }

public:
   void setRate(int mbps)
   {
      m_dPktSndPeriod = (m_iSMSS * 8.0) / mbps;
   }

protected:
  static const int m_iSMSS = 1500;
};


/*****************************************************************************
Group Transport Protocol
Reference:
Ryan X. Wu, and Andrew Chien, "GTP: Group Transport Protocol for 
Lambda-Grids", in Proceedings of the 4th IEEE/ACM International Symposium on 
Cluster Computing and the Grid (CCGrid), April 2004

Note:
This is a demotration showing how to use UDT/CCC to implement group-based
control mechanisms, such GTP and CM.
*****************************************************************************/

struct gtpcomp;

class CGTP: public CCC
{
friend struct gtpcomp;

public:
   virtual void init()
   {
      m_dRequestRate = 1;

      m_llLastRecvPkt = 0;
      gettimeofday(&m_LastGCTime, 0);

      m_GTPSet.insert(this);
      rateAlloc();
   }

   virtual void close()
   {
      m_GTPSet.erase(this);
      rateAlloc();
   }

   virtual void onPktReceived()
   {
      timeval currtime;
      gettimeofday(&currtime, 0);

      int interval = (currtime.tv_sec - m_LastGCTime.tv_sec) * 1000000 + currtime.tv_usec - m_LastGCTime.tv_usec;

      if (interval < 2 * m_iRTT)
         return;

      const UDT::TRACEINFO* info = getPerfInfo();
      
      double realrate, lossrate = 0;
      realrate = (info->pktRecvTotal - m_llLastRecvPkt) * 1500 * 8.0 / interval;
      if (info->pktRecvTotal != m_llLastRecvPkt)
         lossrate = double(info->pktRcvLossTotal - m_iLastRcvLoss) / (info->pktRecvTotal - m_llLastRecvPkt);

      if (0 == lossrate)
         m_dRequestRate *= 1.02;
      else if (lossrate * 0.5 < 0.125)
         m_dRequestRate *= (1 - lossrate * 0.5); 
      else
         m_dRequestRate *= 0.875;

      if (m_dRequestRate > m_dTargetRate)
         m_dRequestRate = m_dTargetRate;

      requestRate(int(m_dRequestRate));

      m_llLastRecvPkt = info->pktRecvTotal;
      m_iLastRcvLoss = info->pktRcvLossTotal;
      m_LastGCTime = currtime;
      m_iRTT = int(info->msRTT * 1000);
   }

   virtual void processCustomPkt(CPacket* pkt)
   {
      if (m_iGTPPktType != pkt->getExtendedType())
         return;

      m_dPktSndPeriod = (1500 * 8.0) / *(int *)(pkt->m_pcData);
   }

public:
   void setBandwidth(const double& mbps)
   {
      m_dBandwidth = mbps;
   }

private:
   void rateAlloc();

   void requestRate(int mbps)
   {
      CPacket pkt;
      pkt.pack(0x111, const_cast<void*>((void*)&m_iGTPPktType), &mbps, sizeof(int));

      sendCustomMsg(pkt);
   }

private:
   double m_dTargetRate;
   double m_dBandwidth;

   double m_dRequestRate;

   timeval m_LastGCTime;
   int64_t m_llLastRecvPkt;
   int m_iLastRcvLoss;
   int m_iRTT;

private:
   static set<CGTP*> m_GTPSet;
   static const int m_iGTPPktType = 0xFFF;
};

set<CGTP*> CGTP::m_GTPSet;
struct gtpcomp
{
  bool operator()(const CGTP* g1, const CGTP* g2) const
  {
    return g1->m_dBandwidth < g2->m_dBandwidth;
  }
};

void CGTP::rateAlloc()
{
   if (0 == m_GTPSet.size())
      return;

   vector<CGTP*> GTPVec;
   copy(m_GTPSet.begin(), m_GTPSet.end(), GTPVec.begin());
   sort(GTPVec.begin(), GTPVec.end(), gtpcomp());

   int N = GTPVec.size();
   int n = 0;
   vector<CGTP*>::iterator i = GTPVec.begin();
   double availbw = (*(i + N - 1))->m_dBandwidth;
   double fairshare = availbw / N;

   while ((n < N) && ((*i)->m_dBandwidth < fairshare))
   {
      (*i)->m_dTargetRate = (*i)->m_dBandwidth;
      availbw -= (*i)->m_dTargetRate;
      fairshare = availbw / (N - n);

      ++ n;
      ++ i;
   }

   for (; i != GTPVec.end(); ++ i)
      (*i)->m_dTargetRate = fairshare;
}


/*****************************************************************************
Protocol using reliable control channel

Note:
The feedback method using sendCustomMsg() as shown in CGTP sends data
using unreliable channel. If some protocol nees reliable channel to 
transfer control message, a seperate TCP connection can be sarted.
The CReliableChannel class below can be used to derive such protocols.
*****************************************************************************/

class CReliableChannel: public CCC
{
public:
   int startTCPServer(sockaddr* addr)
   {
      if (-1 == (m_TCPSocket = socket(AF_INET, SOCK_STREAM, 0)))
         return -1;

      if (-1 == (bind(m_TCPSocket, addr, sizeof(sockaddr_in))))
         return -1;

      if (-1 == (listen(m_TCPSocket, 10)))
         return -1;

      if (-1 == (m_TCPSocket = accept(m_TCPSocket, NULL, NULL)))
         return -1;

      #ifndef WIN32
         pthread_create(&m_TCPThread, NULL, TCPProcessing, this);
      #endif

      return 0;
   }

   int startTCPClient(sockaddr* addr)
   {
      if (-1 == (m_TCPSocket = socket(AF_INET, SOCK_STREAM, 0)))
         return -1;

      if (-1 == (connect(m_TCPSocket, addr, sizeof(sockaddr_in))))
         return -1;

      #ifndef WIN32
         pthread_create(&m_TCPThread, NULL, TCPProcessing, this);
      #endif

      return 0;
   }

   int sendReliableMsg(const char* data, const int& size)
   {
      return send(m_TCPSocket, data, size, 0);
   }


protected:
   virtual void processRealiableMsg()
   {
      char data[1500];

      while (true)
      {
         recv(m_TCPSocket, data, 1500, 0);
         //process data
      }
   }

protected:
   int m_TCPSocket;

   pthread_t m_TCPThread;

private:
   static void* TCPProcessing(void* self)
   {
      ((CReliableChannel*)self)->processRealiableMsg();
      return NULL;
   }
};