udptrsmtfile.cpp

进行UDP的有效传输

// UDPTrsmtFile.cpp: implementation of the CUDPTrsmtFile class.

#include "stdafx.h"
//#include "HICQ.h"
#include "UDPTrsmtFile.h"
#include "..\HICQ_Server\htime.h"

CUDPTrsmtFile::CUDPTrsmtFile(CHsocket& hsocket) 
	: m_hsocket(hsocket)
{
	CUDPTrsmtFile();
}

CUDPTrsmtFile::CUDPTrsmtFile(SOCKET socket)
	: m_hsocket(socket)
{
	CUDPTrsmtFile();
}

CUDPTrsmtFile::CUDPTrsmtFile()
{
	m_nSendIndexHead = 0;
	m_nSendIndexTail = 0;
	m_nAckCount  = 0;
	m_nRecvIndex = 0;
	m_dwFileSize = 0;
	m_dwSend = 0;
	m_hFile = NULL;
	InitializeCriticalSection(&m_csIndex);
	InitializeCriticalSection(&m_csQueue);
}

CUDPTrsmtFile::~CUDPTrsmtFile()
{
	while(!m_bufLookaside.empty())// clear
	{
		delete m_bufLookaside.front();
		m_bufLookaside.pop();
	}

	DeleteCriticalSection(&m_csIndex);
}

void CUDPTrsmtFile::CreateSocket(unsigned int nPort)
{
	m_hsocket.hsocket(nPort, SOCK_DGRAM, m_hsocket.getlocaladdr());
}

void CUDPTrsmtFile::SetSendSockAddr(const char* pszIP, unsigned int nPort)
{
	m_hsocket.makesendsockaddr(pszIP, nPort);
}

void CUDPTrsmtFile::SetRunType(bool bServer)
{
	m_bServer = bServer;
}

bool CUDPTrsmtFile::GetRunType() const
{
	return m_bServer;
}

LPSendBuf CUDPTrsmtFile::GetBufFromLookaside()
{
	LPSendBuf tmpbuf = NULL;
	if(m_bufLookaside.empty())
	{
		tmpbuf = new SendBuf;
		return tmpbuf;
	}

	tmpbuf = m_bufLookaside.front();
	m_bufLookaside.pop();
	return tmpbuf;
}

void CUDPTrsmtFile::SetSocket(CHsocket& hsocket)
{
	m_hsocket.setsocket(hsocket);
}

CHsocket& CUDPTrsmtFile::GetSocket()
{
	return m_hsocket;
}

void CUDPTrsmtFile::SetFile(HANDLE hFile)
{
	m_hFile = hFile;
}

void CUDPTrsmtFile::SetFile(LPCTSTR lpszFileName)
{
	m_hFile = CreateFile(
		lpszFileName, 
		m_bServer ? GENERIC_READ : GENERIC_WRITE,
		NULL, 
		NULL, 
		/*m_bServer ? OPEN_EXISTING : */OPEN_ALWAYS,
		NULL,
		NULL);

	if(m_hFile == INVALID_HANDLE_VALUE)
	{
		//AfxMessageBox("打开文件出错");
		cout << "打开文件出错" << endl;
		return;
	}

	// the file always exists, then get the file size
	if(GetLastError() == ERROR_ALREADY_EXISTS)
	{
		//若该类作为服务器端运行，即发送文件，则m_dwFileSize为总共要发送的大小
		//否则，即接收文件，则m_dwSend为上一次已接收的大小
		if(m_bServer)
			m_dwFileSize = GetFileSize(m_hFile, NULL);
		else
			m_dwSend = GetFileSize(m_hFile, NULL);
	}

	cout << m_dwFileSize << endl;
}

void CUDPTrsmtFile::SetFileSize(DWORD dwFileSize)
{
	m_dwFileSize = dwFileSize;
}

// begin transmitting file
void CUDPTrsmtFile::BeginTransmitting()
{
	if(m_bServer)
		m_hEvent = CreateEvent(NULL, TRUE, FALSE, NULL); //　创建一个事件，无信号状态，等待接收打洞成功消息

	m_hSendThread = CreateThread(NULL, NULL, SendThread, (LPVOID)this, NULL, NULL);
	m_hRecvThread = CreateThread(NULL, NULL, RecvThread, (LPVOID)this, NULL, NULL);
}

void CUDPTrsmtFile::StopTransmitting()
{
	TerminateThread(m_hSendThread, 0);
	CloseHandle(m_hSendThread);

	TerminateThread(m_hRecvThread, 0);
	CloseHandle(m_hRecvThread);

	CloseHandle(m_hEvent);
}

DWORD CALLBACK CUDPTrsmtFile::SendThread(LPVOID lParam)
{
	CUDPTrsmtFile* pThis = (CUDPTrsmtFile*)lParam;

	if(pThis->GetRunType())
	{
		if(!pThis->ServerSend())
			return 1;
	}
	else
	{
		if(!pThis->ClientSend())
			return 1;
	}

	return 0;
}

DWORD CALLBACK CUDPTrsmtFile::RecvThread(LPVOID lParam)
{
	CUDPTrsmtFile* pThis = (CUDPTrsmtFile*)lParam;

	if(pThis->GetRunType())
	{
		if(!pThis->ServerRecv())
			return 1;
	}
	else
	{
		if(!pThis->ClientRecv())
			return 1;
	}

	return 0;
}

bool CUDPTrsmtFile::ServerSend()
{
	if(!s_SendHoleMsg())
		return false;

	// 设置文件指针位置，指向上次已发送的大小
	SetFilePointer(m_hFile, m_dwSend, NULL, FILE_BEGIN);

	int		ret;
	int		nPacketCount = 0;
	DWORD	dwRet;
	SendBuf sendbuf;
	DWORD	dwRead;
	DWORD	dwReadSize;

	SendBuf* pushbuf;

	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));
				::LeaveCriticalSection(&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);
				}
				::LeaveCriticalSection(&m_csQueue);			// 退出m_bufqueue的排斥区
			}

			::EnterCriticalSection(&m_csQueue);				// 进入m_bufqueue的排斥区
			if(m_bufqueue.front()->index == File_End)		// 所有数据包已发送完毕,退出循环
			{
				::LeaveCriticalSection(&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)
				{
					::LeaveCriticalSection(&m_csQueue);		// 退出m_bufqueue的排斥区
					cout << "发送失败，重发" << endl;
					continue;
				}

				//延时，防止丢包
				Sleep(50);		
			}
			else											// 发送窗口大于指定值，等持接收线程接收确认消息
			{
				ResetEvent(m_hEvent);
			}
			::LeaveCriticalSection(&m_csQueue);				// 退出m_bufqueue的排斥区
		}
	}
	cout << "共分成 " << nPacketCount << " 个数据包" << endl;
	cout << "共花时: " << htime.difference() << "s" << endl;

	CloseHandle(m_hEvent);

	CloseHandle(m_hFile);
	m_hFile = NULL;

	m_hsocket.hclosesocket();
	 
	while(!m_bufLookaside.empty())	// clear
	{
		delete m_bufLookaside.front();
		m_bufLookaside.pop();
	}
	
	//AfxMessageBox("文件传输完成");
	cout << "文件传输完成" << endl;

	return true;
}

bool CUDPTrsmtFile::ServerRecv()
{
	if(!s_RecvHoleSuccessMsg())
		return false;

	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)	// 文件传输完毕
			{
				::LeaveCriticalSection(&m_csQueue);
				break;
			}
			::LeaveCriticalSection(&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();
			::LeaveCriticalSection(&m_csQueue);

			SetEvent(m_hEvent);
		}
	}

	return true;
}
bool CUDPTrsmtFile::ClientSend()
{
	return true;
}

bool CUDPTrsmtFile::ClientRecv()
{
	if(!c_RecvHole_SendHoleSuccessMsg())
		return false;

	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;
	}

	cout << "client end recv." << endl;
	cout << "nerror: " << nerror << endl;

	CloseHandle(m_hFile);
	m_hFile = NULL;

	m_hsocket.hclosesocket();

	return true;
}

bool CUDPTrsmtFile::s_SendHoleMsg(void)
{
	int		ret;
	int		count = 0;
	DWORD	dwRet;
	char	sendbuf[32];

	//发送打洞消息,最多发三次，若三次都失败，则不能通信

	while(1)
	{
		if(count == 3)
		{
			//AfxMessageBox("连接超时");
			cout << "连接超时" << endl;
			return false;
		}

		strcpy(sendbuf, "hole");
		ret = m_hsocket.hsendto(sendbuf, strlen(sendbuf));
		if(ret == SOCKET_ERROR)
		{
			cout << "发送打洞消息失败" << endl;
			return false;
		}

		
		dwRet = WaitForSingleObject(m_hEvent, 2 * 1000);	// 等待接收线程接收打洞成功消息
		if(dwRet == WAIT_FAILED)
		{
			return false;
		}
		else if(dwRet == WAIT_TIMEOUT)
		{			
			cout << " 重发打洞消息" << endl;
			count++;
			continue;	// 重发打洞消息
		}

		break;
	}
	
	//发送文件总大小
	RecvBuf filesize;
	filesize.flag = Flag_FileSize;
	filesize.dwFileSize = m_dwFileSize;
	ret = m_hsocket.hsendto((char*)&filesize, sizeof(filesize));
	if(ret == SOCKET_ERROR)
	{
		cout << "发送文件总大小失败" << endl;
		return false;
	}	
	
	ResetEvent(m_hEvent);							//设置m_hEvent为无信号状态，等待接收线程接收已发送文件大小
	
	dwRet = WaitForSingleObject(m_hEvent, 30 * 1000);
	if(dwRet == WAIT_FAILED)
	{
		cout << "WaitForSingleObject Error" << endl;
		return false;
	}
	else if(dwRet == WAIT_TIMEOUT)
	{
		cout << "接收已发送文件大小消息超时" << endl;
		return false;
	}
	
	SetEvent(m_hEvent);		//设置m_hEvent为有信号状态，以便发送文件

	return true;
}

bool CUDPTrsmtFile::s_RecvHoleSuccessMsg(void)
{
	int ret;

	char szbuf[64];
	RecvBuf recvbuf;
	
	ret = m_hsocket.hrecvfrom(szbuf, sizeof(szbuf));		//接收打洞成功消息
	if(ret == SOCKET_ERROR)
	{
		//AfxMessageBox("接收打洞成功消息出错");
		cout << "接收打洞成功消息出错: " << GetLastError() << endl;
		return false;
	}
	
	SetEvent(m_hEvent);		//接收成功，设置m_hEvent为有信号状态，以便发送线程退出发送打洞消息循环
	
	ret = m_hsocket.hrecvfrom((char*)&recvbuf, sizeof(recvbuf));		//接收已传输送文件大小
	if(ret == SOCKET_ERROR)
	{
		//AfxMessageBox("接收已发送文件大小消息出错");
		cout << "接收已发送文件大小消息出错" << endl;
		return false;
	}

	m_dwSend = recvbuf.dwFileSize;

	Sleep(10);	// 不明白这里为什么要等待10毫秒
	
	SetEvent(m_hEvent);		//接收成功，设置m_hEvent为有信号状态，发送线程开始发送文件

	return true;
}

bool CUDPTrsmtFile::c_RecvHole_SendHoleSuccessMsg(void)
{
	int	 ret;
	char recvbuf[64];

	while(1)
	{
		ret = m_hsocket.hrecvfrom(recvbuf, sizeof(recvbuf));
		if(ret == SOCKET_ERROR)
		{
			//AfxMessageBox("接收打洞消息失败");
			cout << "接收打洞消息失败: " << GetLastError() << endl;
			return false;
		}

		recvbuf[ret] = 0;

		//收到对方发来的打洞消息，结束循环。同时新建套接字，向发对方发送打洞成功消息，否则继续等待接收
		if(strcmp(recvbuf, "hole") == 0)
		{
			//发送打洞成功消息
			char szsend[] = "holesuccess";
			m_hsocket.makesendsockaddr(m_hsocket.getsockaddr(), m_hsocket.getsockport());
			ret = m_hsocket.hsendto(szsend, sizeof(szsend));
			if(ret == SOCKET_ERROR)
			{
				//AfxMessageBox("发送打洞成功消息失败");
				cout << "发送打洞成功消息失败" << endl;
				return false;
			}
			
			break;
		}
		else
		{
			cout << "recv again" << endl;
			continue;
		}
	}

	Sleep(10);	// 等待对方接收

	//接收文件总大小
	RecvBuf sendbuf;
	ret = m_hsocket.hrecvfrom((char*)&sendbuf, sizeof(sendbuf));
	if(ret == SOCKET_ERROR)
	{
		cout << "接收文件总大小失败" << endl;
		return false;
	}

	m_dwFileSize = sendbuf.dwFileSize;

	//发送已传输文件大小
	sendbuf.dwFileSize = m_dwSend;
	sendbuf.flag = Flag_ReSend;
	ret = m_hsocket.hsendto((char*)&sendbuf, sizeof(sendbuf));
	if(ret == SOCKET_ERROR)
	{
		//AfxMessageBox("发送已传送文件大小失败");
		cout << "发送已传送文件大小失败" << endl;
		return false;
	}

	return true;
}