织梦CMS - 轻松建站从此开始！
高级搜索|网站地图|TAG标签RSS订阅[设为首页] [加入收藏]

罗索

搜索

热门标签:

当前位置: 主页>基础技术>高性能服务器>

UDP分包重组算法

落鹤生 发布于 2010-09-02 15:23 点击:次 
落鹤生：由于UDP传输的特性：乱序，后发先至等，而在进行UDP编程的时候因为一个包太大，不适合用来传输，所以经常会用到UDP服务器端发送分包，以及客户端分包重组的问题。这里就是一个不错的例子源码。（BTW：Windows下用IOCP来发也可能会有同样的问题，所以本文同样适
TAG:

UDP分包重组算法（BTW：Windows下用IOCP来发也可能会有同样的问题，所以本文同样适用于TCP - IOCP下的分包及重组）

Packet.h

  1. #include "InetAddr.h" //对socket地址操作封装的类，比如char*IP转成ULONG 
  2. #include <vector> 
  3. using namespace std; 
  4.  
  5. //先定义包头
  6. typedef struct _HeadExt 
  7. { 
  8.     //每帧所有分片公用信息 
  9.     char   flag[5]; //可以任意字符或数字或其他方法，用来标志我们发送的数据 
  10.     UINT   m_nSeqNumber; //本帧序数 
  11.     USHORT m_nTotalFragment;//本帧数据可分成的总片数 
  12.     USHORT m_nTotalSize;  //本帧数据总长度 
  13.     //每片各自信息 
  14.     USHORT m_nFragmentIndex; //每帧分片序数,0 1 2 ... ,\ 
  15.     USHORT m_usPayloadSize; //本片数据长度 
  16.     USHORT m_usFragOffset;//本片数据相对总数据的偏移量 
  17.     USHORT m_bLastFragment;//是否最后一帧 
  18. //上 述数据有些重复，可以精简，有时间再修改 
  19.  }HeadExt; 
  20.  
  21.  
  22.  
  23. //从socket接收到的数据 
  24. class  PacketIn 
  25. { 
  26. public : 
  27.     PacketIn(char* lpszBuffer=NULL, UINT usBufferSize=0, UINT nDataSize=0); 
  28.     virtual ~PacketIn(); 
  29.     HeadExt head; 
  30.     BYTE*    m_lpszBuffer; 
  31.     ULONG   ip; 
  32.     UINT    port; 
  33.     CVLInetAddr addr; 
  34.     BOOL Normalize(); 
  35.     UINT   m_nDataLen; 
  36.     UINT   m_usBufferSize; 
  37.  
  38. }; 
  39.  
  40. //接收到数据后开始重组使用的一个中间缓冲区，多个PacketIn合成一个Packet 
  41. //发送时从一个Packet分割出多片，不过本程序里直接发送，没有封成Packet 
  42. class Packet  
  43. {
  44. public: 
  45.     enum { TIMEOUT_LOCKFRAGMENTS = 1000 };
  46.     Packet( );
  47.     ~Packet();
  48.     void reset();
  49.     void Set(const CVLInetAddr& iaFrom, UINT nSeqNumber );
  50.     BOOL InsertFragment(PacketIn* const pFragment);
  51.     bool m_bUsed; 
  52.     int recvedpacks; 
  53.     int recvedbytes; 
  54.     int seqnum; 
  55.     BYTE* m_pBuffer;
  56.  
  57. //测试用程序，直接访问了类的变量，没有封装成函数。上述变量正常应该写成 SetXXX ,GetXXX    
  58. private: 
  59.     BOOL IsFragComplete() const;
  60.     ULONG     m_ip;
  61.     USHORT   m_port; 
  62.     UINT        m_nSeqNumber; 
  63.     vector<int> SeqNumberList; 
  64. }; 

Packet.cpp

  1. #include "Packet.h" 
  2.  
  3. PacketIn::PacketIn(char* lpszBuffer/*=NULL*/, UINT usBufferSize/*=0*/,
  4.  UINT nDataSize/*=0*/) 
  5. { 
  6.  
  7.     m_lpszBuffer   = (BYTE*)lpszBuffer; 
  8.     m_usBufferSize = usBufferSize; //数据最大长度，其实没什么用，已经设置了这个值最大为64000 
  9.     m_nDataLen = nDataSize; 
  10. } 
  11. PacketIn::~PacketIn() 
  12. { 
  13. } 
  14. BOOL PacketIn::Normalize()  //判断收到的数据是否有效 
  15. { 
  16.     const USHORT usHeaderSize = sizeof(HeadExt); 
  17.     if ( !m_lpszBuffer || m_usBufferSize < usHeaderSize )//没什么用 
  18.     { 
  19.         return FALSE; 
  20.     } 
  21.  
  22.     HeadExt* pHeader = (HeadExt*)( m_lpszBuffer  ); 
  23.     if ( pHeader->m_usPayloadSize != m_nDataLen - usHeaderSize ) 
  24.     { 
  25.         return FALSE; 
  26.     } 
  27.  
  28.     head = *pHeader; 
  29.     if ( pHeader->m_usPayloadSize > 0 ) 
  30.     { 
  31. memmove( m_lpszBuffer, m_lpszBuffer + usHeaderSize, pHeader->m_usPayloadSize ); 
  32.     } 
  33.     
  34.     return TRUE; 
  35. } 
  36.  
  37. ///////////// 
  38. //这里是重组数据的临时缓冲，只看这里必然迷惑，先看socket收数据那里，再回来查看 
  39. void Packet::Set( const CVLInetAddr& iaFrom, UINT nSeqNumber ) 
  40. { 
  41.     m_iaFrom =  iaFrom; 
  42.     m_nSeqNumber = nSeqNumber; 
  43.     if(m_pBuffer) 
  44.         delete []m_pBuffer; 
  45. } 
  46.  
  47. void Packet::reset() 
  48. { 
  49.     m_bUsed = false; 
  50.     recvedpacks = 0; 
  51.     seqnum = 0; 
  52.     recvedbytes = 0; 
  53.     m_pBuffer = NULL; 
  54.     m_pBuffer = new BYTE[64000]; 
  55.     SeqNumberList.clear(); 
  56. } 
  57. Packet::Packet() 
  58. { 
  59.     //calculate the ttl 
  60.     //m_nTTL = RUDP_REASSEMBLE_TIMEOUT*CRudpPeer::PR_SLOWHZ; 
  61.     reset(); 
  62. } 
  63.  
  64. Packet::~Packet() 
  65. { 
  66.     if(m_pBuffer) 
  67.         delete []m_pBuffer; 
  68. } 
  69.  
  70. BOOL Packet::InsertFragment(PacketIn* const pFragment) 
  71. { 
  72.     int nSize = SeqNumberList.size(); 
  73.     for(int i = 0; i< nSize ;i ++) 
  74.     { 
  75.         if(nSize ==SeqNumberList[i] )//收到重复数据包 
  76.         { 
  77.             return FALSE; 
  78.         } 
  79.     } 
  80.     SeqNumberList.push_back(pFragment->head.m_nFragmentIndex); 
  81.  
  82.     memcpy( m_pBuffer + pFragment->head.m_usFragOffset, 
  83. pFragment->m_lpszBuffer, pFragment->head.m_usPayloadSize); 
  84.     recvedbytes += pFragment->head.m_usPayloadSize; 
  85.     recvedpacks++; 
  86.     m_bUsed = true; 
  87.  
  88.     CString str; 
  89. str.Format("收到数据：m_nSeqNumber%d ,m_nTotalFragment %d,m_nFragmentIndex %d, 
  90. m_usFragOffset %d,m_nTotalSize %d,recvedbytes %d\r\n", 
  91. pFragment->head.m_nSeqNumber,pFragment->head.m_nTotalFragment, 
  92. pFragment->head.m_nFragmentIndex,pFragment->head.m_usFragOffset, 
  93. pFragment->head.m_nTotalSize,pFragment->head.m_usPayloadSize); 
  94.         OutputDebugString(str); 
  95.  
  96.     if(recvedbytes == pFragment->head.m_nTotalSize ) 
  97.         return TRUE; 
  98.     
  99.     return FALSE; 
  100. } 

发送时的操作：

  1. #iclude "packet.h" 
  2.  
  3. const int RTP_SPLIT_PACKSIZE = 1300;
  4. //udp一般最大包为1500,一般这里都设置为1400，当然1300也没错 
  5. LONGLONG index = rand();
  6. //帧序数，从一个随机数开始，我机器上生成的是41 
  7. //分包发送,逻辑相当简单，按最常规的方法分割发送 
  8. void Send(BYTE* pBuff,UINT nLen,ULONG ip,USHORT port) 
  9. { 
  10.     HeadExt head; 
  11.     strcpy(head.flag ,"aaaa"); 
  12.       
  13.     head.m_nSeqNumber = index++; //帧序数 
  14.  
  15.     head.m_nTotalFragment = nLen/RTP_SPLIT_PACKSIZE; 
  16.     head.m_nTotalFragment  += nLen%RTP_SPLIT_PACKSIZE?1:0;
  17. //整除的话就是0，否则多出一个尾数据 
  18.     head.m_nFragmentIndex = 0;//第一个分段从0开始 
  19.     head.m_nTotalSize = nLen; 
  20.     head.m_bLastFragment = 0; 
  21.     head.m_usFragOffset = 0; 
  22.  
  23.     char tem[RTP_SPLIT_PACKSIZE+sizeof(HeadExt)]; 
  24.  
  25.     if(head.m_nTotalFragment == 1) 
  26.     { 
  27.         memcpy(tem,&head,sizeof(HeadExt)); 
  28.         memcpy(tem+sizeof(HeadExt),pBuff,nLen); 
  29.         head.m_usPayloadSize = nLen; 
  30.  
  31. //用UDP发送，常规做法，先对UDP做一个封装的类，这里调用类对象。本代码只说明原理，类似伪代友 
  32.        sendto(tem,nLen+sizeof(HeadExt),ip,port); 
  33.  
  34.         CString str; 
  35.         str.Format("发送数据：m_nSeqNumber%d ,m_nTotalFragment %d,m_nFragmentIndex %d,
  36. m_usFragOffset %d,m_nTotalSize %d,m_nDataLen %d\r\n", 
  37. head.m_nSeqNumber,head.m_nTotalFragment,head.m_nFragmentIndex,
  38. head.m_usFragOffset,head.m_nTotalSize,head.m_usPayloadSize); 
  39.         OutputDebugString(str); 
  40.  
  41.         return; 
  42.     } 
  43.  
  44.     int i = 0; 
  45.     for( i = 0; i < head.m_nTotalFragment-1; i++)//开始分割,最后一个单独处理 
  46.     { 
  47.         head.m_bLastFragment = 0; 
  48.         head.m_nFragmentIndex = i; 
  49.         head.m_usFragOffset = i*RTP_SPLIT_PACKSIZE; 
  50.         head.m_usPayloadSize = RTP_SPLIT_PACKSIZE; 
  51.         memcpy(tem,&head,sizeof(HeadExt)); 
  52.         memcpy(tem+sizeof(HeadExt),pBuff+i*RTP_SPLIT_PACKSIZE,RTP_SPLIT_PACKSIZE); 
  53.        sendto(tem,nLen+sizeof(HeadExt),ip,port); 
  54.  
  55.         CString str; 
  56.         str.Format("发送数据：m_nSeqNumber%d ,m_nTotalFragment %d,m_nFragmentIndex %d,
  57. m_usFragOffset %d,m_nTotalSize %d,m_nDataLen %d\r\n", 
  58. head.m_nSeqNumber,head.m_nTotalFragment,head.m_nFragmentIndex,
  59. head.m_usFragOffset,head.m_nTotalSize,head.m_usPayloadSize); 
  60.         OutputDebugString(str); 
  61.  
  62.     } 
  63.  
  64.     head.m_bLastFragment = 1; 
  65.     head.m_nFragmentIndex = i; 
  66.     head.m_usFragOffset = i*RTP_SPLIT_PACKSIZE; 
  67.     head.m_usPayloadSize = nLen - (i*RTP_SPLIT_PACKSIZE); 
  68.  
  69.     memcpy(tem,&head,sizeof(HeadExt)); 
  70.     memcpy(tem+sizeof(HeadExt),pBuff+i*RTP_SPLIT_PACKSIZE,nLen - (i*RTP_SPLIT_PACKSIZE)); 
  71.  
  72.     sendto(tem,nLen+sizeof(HeadExt),ip,port); 
  73.  
  74.     CString str; 
  75. str.Format("发送数据：m_nSeqNumber%d ,m_nTotalFragment %d,m_nFragmentIndex %d,
  76. m_usFragOffset %d,m_nTotalSize %d,m_nDataLen %d\r\n",head.m_nSeqNumber,
  77. head.m_nTotalFragment,head.m_nFragmentIndex,head.m_usFragOffset,
  78. head.m_nTotalSize,head.m_usPayloadSize); 
  79.  
  80. OutputDebugString(str); 
  81. } 

接收数据的回调

//代码很简单，收到数据后，形成PacketIn，然后找一个链表，插进去
//因为收到的数据的顺序是不定的，所以先创建一个链表，每帧首数据到来时，分配一个节点，其他分片到来的时候，都加到//这个节点里,等数据都到齐了，这个节点就是一个完整的包，回调出去。

  1. void RecvCallback(BYTE* pBuff,UINT nLen,ULONG ip,USHORT port) 
  2. { 
  3.     if(nLen < sizeof(HeadExt)) 
  4.         return;
  5.     HeadExt* pHead = (HeadExt*)pBuff; 
  6.     CString str = pHead->flag; 
  7.     if(str != "aaaa") 
  8.     {
  9.         return; 
  10.     }
  11.     if(pHead->m_nTotalFragment == 1)//只有一帧，不分片 
  12.     { 
  13.         //回调，上层处理 
  14.         if(m_pfDispatch)
  15. m_pfDispatch(pBuff+sizeof(HeadExt),nLen-sizeof(HeadExt),ip,port,m_lpParam);
  16.         return; 
  17.     } 
  18.  
  19.     PacketIn data( (char*)pBuff, 64000, nLen ); 
  20.     if ( !data.Normalize() ) 
  21.         return;
  22.     if ( data.head.m_nTotalFragment>1 ) 
  23.     {
  24.         Packet* pTemp = GetPartialPacket( iaRemote, data.head.m_nSeqNumber ); 
  25.         if ( pTemp ) 
  26.         { 
  27.             if ( pTemp ->InsertFragment( &data ) ) 
  28.             { 
  29.                 m_pfDispatch(pTemp ->m_pBuffer, 
  30.                     pTemp ->recvedbytes,ip,port,m_lpParam); 
  31.             } 
  32.         }//end of if 
  33.     } 
  34. } 

//上面用到的一些变量可以在一个.h里面定义，比如：
Packet TempPacket[16];//分配一个数组，每个节点是一个重组后临时缓冲
Packet* GetPartialPacket(const CVLInetAddr& iaFrom, UINT nSeqNumber);//从数组里取出一个缓冲节点

  1. //根据这几个关键字查找，不过只用 
  2. //到了nSeqNumber,要是3人以上的视频聊天，ip和port是必要的 
  3. Packet* GetPartialPacket(const ULONG ip,USHORT port, UINT nSeqNumber) 
  4. { 
  5.     Packet* tmp = NULL; 
  6.     int i=0; 
  7.     while(tmp==NULL && i<16) 
  8.     { 
  9.         //该包所属的帧已有其它数据包到达 
  10.         if(TempPacket[i].seqnum==nSeqNumber && TempPacket[i].recvedpacks>0) 
  11.         { 
  12.             tmp = &TempPacket[i]; 
  13.             break; 
  14.         } 
  15.         i++; 
  16.     } 
  17.     if(tmp == NULL)        //新的帧 
  18.     { 
  19.         //查找空闲元素 
  20.         for(i=0;i<16;i++) 
  21.         { 
  22.             if(!TempPacket[i].m_bUsed) 
  23.                 break; 
  24.         } 
  25.  
  26.         if(i>=16) 
  27.         { 
  28.             //没有空闲的元素，丢掉一个最早 
  29.             tmp = &TempPacket[0]; 
  30.             int j = 0; 
  31.             for(i=1;i<16;i++) 
  32.             { 
  33.                 if(TempPacket[i].seqnum < tmp->seqnum) 
  34.                 { 
  35.                     tmp = &TempPacket[i]; 
  36.                     j = i; 
  37.                 } 
  38.             } 
  39.             //找到最早的一帧 
  40.             if(tmp->m_pBuffer) 
  41.             { 
  42.                 delete []tmp->m_pBuffer; 
  43.                 tmp->reset(); 
  44.             } 
  45.         } 
  46.         else 
  47.             tmp = &TempPacket[i]; 
  48.     } 
  49.  
  50.     InsertFragment 
  51.     tmp->m_bUsed = true; 
  52.     tmp->seqnum = nSeqNumber; 
  53.  
  54.     return tmp; 
  55. } 


整个示例最重要的是两个函数：

GetPartialPacket：取到一个临时缓冲节点

InsertFragment：把收到的每个数据段插入临时缓冲组成一个完整帧

当然发送方也要按一定规则分割发送。

 

(秩名)
本站文章除注明转载外，均为本站原创或编译欢迎任何形式的转载，但请务必注明出处，尊重他人劳动，同学习共成长。转载请注明：文章转载自：罗索实验室 [http://www.rosoo.net/a/201009/10078.html]
本文出处：网络　作者：秩名
顶一下
(13)
92.9%
踩一下
(1)
7.1%
------分隔线----------------------------
相关文章
发表评论
请自觉遵守互联网相关的政策法规，严禁发布色情、暴力、反动的言论。
评价:
表情:
用户名: 验证码:点击我更换图片
最新评论 进入详细评论页>>
栏目列表
将本文分享到微信
织梦二维码生成器
推荐内容
热点内容