实现可靠的UDP传输类

#include "StdAfx.h" #include "UDT.h" #include <afxcoll.h> CUDTSend * CUDTSend::m_pObj = NULL; CUDTRecv * CUDTRecv::m_pObj = NULL; CUDTSend::CUDTSend(void) { //序号 m_Squence = 0; //记录序号和内存位置对应关系的 m_pdwPos = (DWORD *)malloc(65536 * sizeof(DWORD)); memset(m_pdwPos,0,65536 * sizeof(DWORD)); //线程句柄 m_hThread = NULL; //控制线程状态 m_bIsRun = TRUE; //线程休息时间 m_dwSleepTime = 500; //最大发送次数 m_dwMaxCount = 30; //启动检查线程 UINT nId = 0; m_hThread = (HANDLE)::_beginthreadex(NULL,0,CheckThread,this,0,&nId); } CUDTSend::~CUDTSend(void) { DWORD dwCode = 0; //停止线程 m_bIsRun = FALSE; Sleep(100); //强制退出线程 if(m_hThread != NULL) { GetExitCodeThread(m_hThread,&dwCode); if(dwCode == STILL_ACTIVE) { TerminateThread(m_hThread,dwCode); } } m_hThread = NULL; //释放所有的发送缓冲区 FreeAll(); //释放索引结构 if(m_pdwPos != NULL) { free(m_pdwPos); m_pdwPos = NULL; } } //发送数据 BOOL CUDTSend::UDTSend(SOCKET hSock,CHAR * pchBuf,int nBufLen,char * pchIP,USHORT uPort) { SOCKADDR_IN To; //设置目标地址 memset(&To,0,sizeof(SOCKADDR_IN)); To.sin_family = AF_INET; To.sin_addr.S_un.S_addr = inet_addr(pchIP); To.sin_port = htons(uPort); return UDTSend(hSock,pchBuf,nBufLen,(PSOCKADDR)&To); } BOOL CUDTSend::UDTSend(SOCKET hSock,CHAR * pchBuf,int nBufLen,PSOCKADDR To) { UDT_SND_ITEM * pItem = NULL; UDT_HEAD head; UINT nOffset = 0,nTotal = 0; BOOL bSuc = TRUE; DWORD dwStart = 0; if(pchBuf == NULL || nBufLen == 0)return FALSE; //清空头数据 memset(&head,0,sizeof(head)); //计算总包数 nTotal = nBufLen / DATA_SNDSIZE; if(nBufLen % DATA_SNDSIZE != 0)nTotal++; //不能发送大于1024个单元的包 if(nTotal > 1024)return FALSE; //设置序号区间,并获得起始序号 head.dwSeqSpan = GetSeqSpan(nTotal,&dwStart); //设置包类型 = 数据包 head.uPacketType = 1; for(UINT i = 0; i < nTotal; i++) { //设置包序号 head.dwPacketId = dwStart++; //获取本次数据长度 head.dwPacketLen = (i==(nTotal-1))?(nBufLen % DATA_SNDSIZE):DATA_SNDSIZE; //设置效验和 head.nCheck = head.dwPacketId + head.uPacketType + head.dwSeqSpan + head.dwPacketLen + head.dwAckSeq; //分配发送单元 pItem = AllocItem(); if(pItem == NULL) { bSuc = FALSE; break; } //设置UDP句柄 pItem->hSock = hSock; //拷贝目的地址 memcpy(&(pItem->To),To,sizeof(SOCKADDR)); //拷贝包头 memcpy(pItem->chBuf,&head,sizeof(head)); //拷贝数据 memcpy(pItem->chBuf + sizeof(head),pchBuf + nOffset,head.dwPacketLen); //设置发送单元的长度标识 = 包头长度 + 本次数据的长度 pItem->uBufLen = (USHORT)(sizeof(head) + head.dwPacketLen); //调整缓冲区位置 nOffset = nOffset + head.dwPacketLen; //将数据插入发送缓冲队列中 //失败则直接退出 if(!PutItem(pItem)) { bSuc = FALSE; break; } //发送数据 SendTo(pItem); } return bSuc; } //确认一个发送单元 BOOL CUDTSend::AckItem(UDT_HEAD * pHead) { UDT_SND_ITEM * pItem = NULL; //如果不是确认包，则不处理 if(pHead->uPacketType != 2)return FALSE; //获的要确认的包序号 USHORT uSeq = (USHORT)((pHead->dwAckSeq) % 65536); //开始同步 m_cs_pos.Lock(); //获取该序列号所对应的发送单元 pItem = (UDT_SND_ITEM *)m_pdwPos[uSeq]; m_pdwPos[uSeq] = 0; m_cs_pos.Unlock(); //如果不为NULL，则释放 if(pItem!=NULL) { this->FreeItem(pItem); } return TRUE; } //获得序列号区间(0~15表示起始序号,16~31表示结束序号) DWORD CUDTSend::GetSeqSpan(int nCnt,DWORD * dwStart) { USHORT uStart = 0,uStop = 0; DWORD dwSpan = 0; m_cs_seq.Lock(); //记录序列号的起始值 *dwStart = m_Squence; uStart = (USHORT)(m_Squence % 65536); for(int i = 0;i < nCnt; i++) m_Squence++; uStop = (USHORT)(m_Squence % 65536); m_cs_seq.Unlock(); uStop--; dwSpan = (dwSpan + uStop) << 16; dwSpan = (dwSpan | uStart); return dwSpan; } //发送一个数据单元 int CUDTSend::SendTo(UDT_SND_ITEM * pItem) { //设置最后发送时间 pItem->dwLastTime = GetTickCount(); //设置已经发送次数 pItem->uLimitCnt++; //发送数据 return sendto(pItem->hSock,pItem->chBuf,pItem->uBufLen,0,(PSOCKADDR)&(pItem->To),sizeof(pItem->To)); } //将数据插入到链表中 BOOL CUDTSend::PutItem(UDT_SND_ITEM * pItem) { USHORT uSeq = 0; //设置最后发送时间 pItem->dwLastTime = 0; //设置已经发送次数 pItem->uLimitCnt = 0; //获取UDT包头 UDT_HEAD * pHead = (UDT_HEAD *)(pItem->chBuf); //获取包序列号 uSeq = (USHORT)((pHead->dwPacketId) % 65536); //如果对应结构已经被删除，则 if(m_pdwPos == NULL)return FALSE; //开始同步 m_cs_pos.Lock(); //获取该序列号所对应的位置 UDT_SND_ITEM * pItemO = (UDT_SND_ITEM *)m_pdwPos[uSeq]; //将数据插入发送队列中 m_pdwPos[uSeq] = (DWORD)pItem; m_cs_pos.Unlock(); //删除原始位置的发送单元 if(pItemO != NULL) { FreeItem(pItemO); } return TRUE; } //分配一个发送单元 UDT_SND_ITEM * CUDTSend::AllocItem() { UDT_SND_ITEM * pItem = NULL; pItem = new UDT_SND_ITEM(); if(pItem != NULL) { memset(pItem,0,sizeof(UDT_SND_ITEM)); } return pItem; } //释放一个发送单元 void CUDTSend::FreeItem(UDT_SND_ITEM * pItem) { //释放内存 delete pItem; pItem = NULL; } //释放所有的发送单元 void CUDTSend::FreeAll() { UDT_SND_ITEM * pItem = NULL; for(int i=0;i<65536;i++) { m_cs_pos.Lock(); //获得该位置的单元数据 pItem = (UDT_SND_ITEM *)m_pdwPos[i]; if(pItem != NULL) { FreeItem(pItem); pItem = NULL; } m_cs_pos.Unlock(); } } //定时检测线程 UINT CUDTSend::CheckThread(LPVOID lParam) { UDT_SND_ITEM * pItem = NULL; CDWordArray sendArray; CDWordArray delArray; CUDTSend * pObj = (CUDTSend *) lParam; while(pObj->m_bIsRun) { //停止一段时间 Sleep(pObj->m_dwSleepTime); for(int i=0;i<65536;i++) { if(!pObj->m_bIsRun) return 1; pObj->m_cs_pos.Lock(); //获取该序列号所对应的发送单元 pItem = (UDT_SND_ITEM *)(pObj->m_pdwPos[i]); //如果该位置不存在数据，则返回 if(pItem != NULL) { //1.检查发送次数是否超过，超过则释放 //2.没超过则继续发送 //3.检查是否有确认过的包，有则释放 if(pItem->uLimitCnt >= pObj->m_dwMaxCount || pItem->dwLastTime == 0xFFFFFFFF) { //清空位置 pObj->m_pdwPos[i] = 0; //添加到待删除队列中 delArray.Add((DWORD)pItem); } else { //发送 sendArray.Add((DWORD)pItem); } }//end if(pItem != NULL0 pObj->m_cs_pos.Unlock(); }//end for //删除所有待删除的内存 for(int i = 0; i< delArray.GetCount(); i++) { UDT_SND_ITEM * pItem = (UDT_SND_ITEM *)delArray.GetAt(i); //删除内存 pObj->FreeItem(pItem); } delArray.RemoveAll(); //发送所有未确认的包 for(int i = 0; i< sendArray.GetCount(); i++) { UDT_SND_ITEM * pItem = (UDT_SND_ITEM *)sendArray.GetAt(i); if(pItem != NULL) { //连续发送16个包,让线程停止一会儿 //Sleep的定时精度<=16ms if( (i + 1) % 16 == 0) Sleep(15); pObj->SendTo(pItem); } } sendArray.RemoveAll(); }//end while return 1; } //构造函数 CUDTRecv::CUDTRecv(void) { //节点头指针 m_PeerHead = NULL; //线程句柄 HANDLE m_hThread = NULL; //控制线程状态 m_bIsRun = TRUE; //线程休息时间(1000) m_dwSleepTime = 1000; //节点最大接收超时时间(60000) m_dwPeerTimeout = 60000; //每个包最大接收超时时间(15000) m_dwItemsTimeout = 15000; //启动检查线程 UINT nId = 0; m_hThread = (HANDLE)::_beginthreadex(NULL,0,CheckThread,this,0,&nId); } CUDTRecv::~CUDTRecv(void) { DWORD dwCode = 0; //停止线程 m_bIsRun = FALSE; Sleep(100); //强制退出线程 if(m_hThread != NULL) { GetExitCodeThread(m_hThread,&dwCode); if(dwCode == STILL_ACTIVE) { TerminateThread(m_hThread,dwCode); } } m_hThread = NULL; //释放所有的接收缓冲区 Free