用UDP实现可靠文件传输

AnnO

用UDP实现可靠文件传输 大家都清楚，如果用TCP传输文件的话，是很简单的，根本都不用操心会丢包，除非是网络坏了，就得重来。用UDP的话，因为UDP是不可靠的，所以用它传输文件，要保证不丢包，就得我们自己写额外的代码来保障了。本文就说说如果保证可靠传输。 要实现无差错的传输数据，我们可以采用重发请求（ARQ）协议，它又可分为连续ARQ协议、选择重发ARQ协议、滑动窗口协议。本文重点介绍滑动窗口协议，其它的两种有兴趣的可参考相关的网络通信之类的书。 采用滑动窗口协议，限制已发送出去但未被确认的数据帧的数目。循环重复使用已收到的那些数据帧的序号。具体实现是在发送端和接收端分别设定发送窗口和接收窗口。 （1）发送窗口 发送窗口用来对发送端进行流量控制。发送窗口的大小Wt代表在还没有收到对方确认的条件下，发送端最多可以发送的数据帧的个数。具体意思请参考下图： （2）接收窗口 接收窗口用来控制接收数据帧。只有当接收到的数据帧的发送序号落在接收窗口内，才允许将该数据帧收下，否则一律丢弃。接收窗口的大小用Wr来表示，在连续ARQ协议中，Wr = 1。接收窗口的意义可参考下图： 在接收窗口和发送窗口间存在着这样的关系：接收窗口发生旋转后，发送窗口才可能向前旋转，接收窗口保持不动时，发送窗口是不会旋转的。这种收发窗口按如此规律顺时钟方向不断旋转的协议就犯法为滑动窗口协议。 好了，在上面对滑动窗口协议有大致了解后，我们还是进入正题吧：） 发送端的发送线程： int ret; int nPacketCount = 0; DWORD dwRet; SendBuf sendbuf; DWORD dwRead; DWORD dwReadSize; SendBuf* pushbuf; //计算一共要读的文件次数，若文件已读完，但客户端没有接收完， //则要发送的内容不再从文件里读取，而从m_bufqueue里提取 nPacketCount = m_dwFileSize / sizeof(sendbuf.buf); //若不能整除，则应加1 if(m_dwFileSize % sizeof(sendbuf.buf) != 0) ++nPacketCount; SetEvent(m_hEvent); CHtime htime; //若已发送大小小于文件大小并且发送窗口前沿等于后沿，则继续发送 //否则退出循环 if(m_dwSend < m_dwFileSize) // 文件没有传输完时才继续传输 { while(1) { dwRet = WaitForSingleObject(m_hEvent, 1000); if(dwRet == WAIT_FAILED) { return false; } else if(dwRet == WAIT_TIMEOUT) { //重发 ::EnterCriticalSection(&m_csQueue); // 进入m_bufqueue的排斥区 ret = m_hsocket.hsendto((char*)m_bufqueue.front(), sizeof(sendbuf)); :eaveCriticalSection(&m_csQueue); // 退出m_bufqueue的排斥区 if(ret == SOCKET_ERROR) { cout << "重发失败，继续重发" << endl; continue; } ResetEvent(m_hEvent); continue; } //若发送窗口大小 < 预定大小 && 已读文件次数(nReadIndex) < 需要读文件的次数(nReadCount)，则继续读取发送 //否则，要发送的内容从m_bufqueue里提取 if(m_dwSend < m_dwFileSize) { dwReadSize = m_dwFileSize - m_dwSend; dwReadSize = dwReadSize < MAXBUF_SIZE ? dwReadSize : MAXBUF_SIZE; memset(sendbuf.buf, 0, sizeof(sendbuf.buf)); if(!ReadFile(m_hFile, sendbuf.buf, dwReadSize, &dwRead, NULL)) { //AfxMessageBox("读取文件失败,请确认文件存在或有读取权限."); cout << "读取文件失败,请确认文件存在或有读取权限." << endl; return false; } m_dwSend += dwRead; sendbuf.index = m_nSendIndexHead; m_nSendIndexHead = (m_nSendIndexHead + 1) % Sliding_Window_Size; // 发送窗口前沿向前移一格 sendbuf.dwLen = dwRead; //保存发送过的数据，以便重发 ::EnterCriticalSection(&m_csQueue); // 进入m_bufqueue的排斥区 pushbuf = GetBufFromLookaside(); memcpy(pushbuf, &sendbuf, sizeof(sendbuf)); m_bufqueue.push(pushbuf); if(m_dwSend >= m_dwFileSize) // 文件已读完，在队列中加一File_End标志，以便判断是否需要继续发送 { pushbuf = GetBufFromLookaside(); pushbuf->index = File_End; pushbuf->dwLen = File_End; memset(pushbuf->buf, 0, sizeof(pushbuf->buf)); m_bufqueue.push(pushbuf); } :eaveCriticalSection(&m_csQueue); // 退出m_bufqueue的排斥区 } ::EnterCriticalSection(&m_csQueue); // 进入m_bufqueue的排斥区 if(m_bufqueue.front()->index == File_End) // 所有数据包已发送完毕,退出循环 { :eaveCriticalSection(&m_csQueue); // 退出m_bufqueue的排斥区 break; } else if(m_bufqueue.size() <= Send_Window_Size) // 发送窗口小于指定值，继续发送 { ret = m_hsocket.hsendto((char*)m_bufqueue.front(), sizeof(sendbuf)); if(ret == SOCKET_ERROR) { :eaveCriticalSection(&m_csQueue); // 退出m_bufqueue的排斥区 cout << "发送失败，重发" << endl; continue; } //延时，防止丢包 Sleep(50); } else // 发送窗口大于指定值，等持接收线程接收确认消息 { ResetEvent(m_hEvent); } :eaveCriticalSection(&m_csQueue); // 退出m_bufqueue的排斥区 } } 发送端的接收线程： int ret; RecvBuf recvbuf; while(m_hFile != NULL) { ret = m_hsocket.hrecvfrom((char*)&recvbuf, sizeof(recvbuf)); if(ret == SOCKET_ERROR) { //AfxMessageBox("接收确认消息出错"); ::EnterCriticalSection(&m_csQueue); if(m_bufqueue.front()->index == File_End) // 文件传输完毕 { :eaveCriticalSection(&m_csQueue); break; } :eaveCriticalSection(&m_csQueue); cout << "接收确认消息出错: " << GetLastError() << endl; return false; } if(recvbuf.flag == Flag_Ack && recvbuf.index == m_nSendIndexTail) { m_nSendIndexTail = (m_nSendIndexTail + 1) % Sliding_Window_Size; //该结点已得到确认，将其加入旁视列表，以备再用 ::EnterCriticalSection(&m_csQueue); m_bufLookaside.push(m_bufqueue.front()); m_bufqueue.pop(); :eaveCriticalSection(&m_csQueue); SetEvent(m_hEvent); } } 接收端的接收线程： int ret; DWORD dwWritten; SendBuf recvbuf; RecvBuf sendbuf; int nerror = 0; // 设置文件指针位置，指向上次已发送的大小 SetFilePointer(m_hFile, 0, NULL, FILE_END); //若已接收文件大小小于需要接收的大小，则继续 while(m_dwSend < m_dwFileSize) { //接收 memset(&recvbuf, 0, sizeof(recvbuf)); ret = m_hsocket.hrecvfrom((char*)&recvbuf, sizeof(recvbuf)); if(ret == SOCKET_ERROR) { return false; } //不是希望接收的，丢弃，继续接收 if(recvbuf.index != (m_nRecvIndex) % Sliding_Window_Size) { nerror++; cout << recvbuf.index << "error?" << m_nRecvIndex << endl; continue; } if(!WriteFile(m_hFile, recvbuf.buf, recvbuf.dwLen, &dwWritten, NULL)) { //AfxMessageBox("写入文件失败"); cout << "写入文件失败" << endl; return false; } //已接收文件大小 m_dwSend += dwWritten; //发送确认消息 sendbuf.flag = Flag_Ack; sendbuf.index = m_nRecvIndex; ret = m_hsocket.hsendto((char*)&sendbuf, sizeof(sendbuf)); if(ret == SOCKET_ERROR) { return false; } //接收窗口前移一格 m_nRecvIndex = (m_nRecvIndex + 1) % Sliding_Window_Size; }