cmd_control.c

自己开发基于P2P通讯的网络服务器

/*********************************************************************
 * 模块名称:cmd_control
 * 说明:命令处理模块
 * 其它说明: 
 * 作者: 刘青山 
 * 时间 : 2004-09-26 7:13:30 
*********************************************************************/
#include "natsvr.h"
#include "memory.h"
#include "netcomm.h"
#include "cmd_control.h"
#include "sthread.h"
#include "assure.h"
#include "log.h"

#ifndef max
#define max(a,b) (((a) > (b)) ? (a) : (b))
#endif
int		g_iWorkThreads		= 20 ;
static	WORD	m_wAckSeq	= 1 ;
static	WORD	m_wSeq		= 1 ;

static  int		g_iIsCSCEnable = 1;
static pthread_mutex_t		m_mtGlobalAnswer ;
LIST_HEAD(g_quAnswer) ;

void LockGlobalAnswer()
{
	pthread_mutex_lock(&m_mtGlobalAnswer);
}

extern void UnlockGlobalAnswer()
{
	pthread_mutex_unlock(&m_mtGlobalAnswer);
}

static LPANS_PACKET CreateAnsPacket( LPREQ_PACKET pReq,LPP2P_USER pUser,WORD wSeq )
{
	LPANS_PACKET	lpAns;

	lpAns = (LPANS_PACKET)malloc(sizeof(ANS_PACKET));
	if (!lpAns)
	{
		LogOut(KLOG_ERROR, "function(CreateAnsPacket) can't alloc memory");
		return NULL;
	}
	memset(lpAns, 0, sizeof(ANS_PACKET));
	memcpy(lpAns->buf, pReq->buf ,P2P_HDR_LEN);
	lpAns->pHeader				= (LPP2P_HDR)lpAns->buf ;
	lpAns->pHeader->wDatalen	= P2P_HDR_LEN ;
	lpAns->pHeader->wSeq		= wSeq;
	lpAns->pHeader->wAckseq		= m_wAckSeq++ ;
	lpAns->pData				= lpAns->buf + P2P_HDR_LEN;
	lpAns->byTry				= 0;

	INIT_LIST_HEAD(&lpAns->listUser);
	INIT_LIST_HEAD(&lpAns->listGlobal);
	if (!pUser)
	{
		lpAns->pUser = pReq->pUser;
		pthread_mutex_lock(&pReq->pUser->mtUser);
		list_add_tail(&lpAns->listUser, &pReq->pUser->listAns);
		pthread_mutex_unlock(&pReq->pUser->mtUser);
	}
	else
	{
		lpAns->pUser = pUser;
		pthread_mutex_lock(&pReq->pUser->mtUser);
		list_add_tail(&lpAns->listUser, &pUser->listAns);
		pthread_mutex_unlock(&pReq->pUser->mtUser);
	}
	lpAns->bSended = false ;

	LockGlobalAnswer();
	list_add_tail(&lpAns->listGlobal, &g_quAnswer);
	UnlockGlobalAnswer();

	return lpAns;
}

static LPANS_PACKET CreateAckPacket( LPREQ_PACKET pReq)
{
	LPANS_PACKET  lpAns ;

	lpAns = (LPANS_PACKET)malloc(sizeof(ANS_PACKET));
	if (!lpAns)
	{
		LogOut(KLOG_ERROR, "function(CreateAckPacket) can't alloc memory");
		return NULL;
	}
	memset(lpAns, 0, sizeof(ANS_PACKET));
	memcpy( lpAns->buf,pReq->buf ,P2P_HDR_LEN);
	lpAns->pHeader				= (LPP2P_HDR)lpAns->buf ;
	lpAns->pHeader->wDatalen	= P2P_HDR_LEN ;
	lpAns->pHeader->dwSrc		= 0;
	lpAns->pHeader->dwDst		= pReq->pHeader->dwSrc ;
//	lpAns->pHeader->wAckseq		= 0 ;	// 不要对方应答，为0
	lpAns->pHeader->wCmd		= CMD_ACK;
	lpAns->pUser				= pReq->pUser ;
	lpAns->byTry				= 0;
	
	INIT_LIST_HEAD(&lpAns->listUser);
	INIT_LIST_HEAD(&lpAns->listGlobal);
	return lpAns;
}

static void WriteData( LPANS_PACKET	pAns, void *pSrc, WORD wLen )
{
	if (pAns->pHeader->wDatalen + wLen > P2P_DATA_LEN)
	{
		LogOut(KLOG_INFO, "Data len[%d + %d] have exceed buf length", pAns->pHeader->wDatalen, wLen );
		MemDump(pAns, sizeof(ANS_PACKET));
		return;
	}
	memcpy(pAns->buf + pAns->pHeader->wDatalen, pSrc, wLen);
	pAns->pHeader->wDatalen += wLen ;

}

///////////////////////////////////////////////////////////////////////
//
// 函数名       : OnRegUser
// 功能描述     : 在NAT服务器上注册用户
// 参数         : LPREQ_PACKET pReq
// 返回值       : static void 
//
///////////////////////////////////////////////////////////////////////
static void OnRegUser(LPREQ_PACKET pReq)
{
	LPP2P_USER		lpUser ;
	LPP2P_HDR		lpHeader;
	LPANS_PACKET	lpAns ;

	lpUser = (LPP2P_USER)GetHashUserByID(pReq->pHeader->dwSrc) ;
	if (lpUser)	// 已经注册过
	{
		if (lpUser != pReq->pUser)
		{
			LogOut(KLOG_INFO, "the uin:%u had registered,drop him and register him",
				pReq->pHeader->dwSrc);
			ClearUser(NOT_REF_COUNT, lpUser);

			PutHashUserByID(pReq->pUser, pReq->pHeader->dwSrc) ;
		}
		else
			return;
	}
	else
	{
		lpUser = pReq->pUser;
		PutHashUserByID(lpUser, pReq->pHeader->dwSrc) ;
	}

	lpAns = CreateAnsPacket(pReq, NULL, pReq->pHeader->wSeq);
	if (!lpAns)
	{
		LogOut(KLOG_ERROR, "function(OnRegUser) can't alloc memory");
		return ;
	}
	lpHeader		= lpAns->pHeader ;
	lpHeader->dwDst = pReq->pHeader->dwSrc ;
	lpHeader->dwSrc = 0 ;
	WriteData(lpAns, (void*)&ANS_REG_SUCCESS, 2); 
	WriteData(lpAns, (void*)&pReq->pUser->dwNatIP, 4); 
	WriteData(lpAns, (void*)&pReq->pUser->wNatPort, 2); 
	WriteData(lpAns, (void*)&g_iIsCSCEnable, 4);

	SendPacket(lpAns);
}

// 在NAT服务器上查询用户的NAT信息
// 请求包:(请求人数)(2)+(KKID)(4)+....
// 返回包：(命令号)(2)+(成功人数)(2)+(KKID+Local IP+Local Port +Nat IP +Nat Port)(16)+....
static void OnUserInfo(LPREQ_PACKET pReq)
{
	LPP2P_USER		lpUser ;
	LPP2P_HDR		lpHeader;
	LPANS_PACKET	lpAns ;
	WORD			wUsers,wUsersRet;
	BYTE			*pData ;
	int				i;

	pData = pReq->pData ;
	memcpy(&wUsers, pData, 2);
	lpAns = CreateAnsPacket(pReq,NULL,pReq->pHeader->wSeq);

	if (!lpAns)
	{
		LogOut(KLOG_ERROR, "function(OnUserInfo) can't alloc memory");
		return ;
	}
	lpHeader		= lpAns->pHeader ;
	lpHeader->dwDst = pReq->pHeader->dwSrc ;
	lpHeader->dwSrc = 0 ;
	if (wUsers > 60)	// 超过了最大用户数量
	{
		WriteData(lpAns, (void*)&ANS_EXCEED_MAX_USER, 2);
		SendPacket(lpAns);
		return ;
	}
	WriteData(lpAns, (void*)&ANS_OK, 2);	
	WriteData(lpAns, (void*)&wUsersRet, 2); 
	
	wUsersRet	= 0 ;
	pData		+= 2 ;
	for( i=0 ;i<wUsers;i++ )
	{
		DWORD	dwKKID;
		memcpy(&dwKKID,pData,4 );
			
		lpUser = GetHashUserByID(dwKKID) ;
		if ( lpUser )
		{
			WriteData(lpAns, (void*)&lpUser->dwID, 4); 
			WriteData(lpAns, (void*)&lpUser->dwLocalIP, 4); 
			WriteData(lpAns, (void*)&lpUser->wLocalPort, 2); 
			WriteData(lpAns, (void*)&lpUser->dwNatIP , 4); 
			WriteData(lpAns, (void*)&lpUser->wNatPort, 2); 

			wUsersRet++ ;
		}
		pData+=4 ;
	}
	// 重新写入返回用户数据量
	memcpy( lpAns->buf+P2P_HDR_LEN+2,&wUsersRet,2);
	SendPacket( lpAns );
	
}
// 申请对多个用户打洞
// 请求包:(请求人数)(2)+(KKID)(4)+....
// 返回包:(命令号)(2)+(成功人数)(2)+(KKID)(4)+....
// 客户端请求包:(KKID+Local IP+Local Port +Nat IP +Nat Port)(16)
static void OnHole(LPREQ_PACKET pReq)
{
	LPP2P_HDR		lpHeader;
	LPANS_PACKET	lpAns ;
	WORD			wUsers,wUsersRet;
	BYTE			*pData ;
	int				i;
	LPP2P_USER		lpDstUser;

	pData = pReq->pData ;
	memcpy( &wUsers,pData,2 );
	lpAns = CreateAnsPacket(pReq, NULL, pReq->pHeader->wSeq);

	if (!lpAns)
	{
		LogOut(KLOG_ERROR, "function(OnHole) can't alloc memory");
		return ;
	}
	lpHeader = lpAns->pHeader ;
	lpHeader->dwDst = pReq->pHeader->dwSrc ;
	lpHeader->dwSrc = 0 ;
	lpHeader->wCmd  = CMD_ACK_HOLE ;

	if (wUsers > 60)
	{
		WriteData(lpAns, (void*)&ANS_NOT_REG, 2);
		SendPacket(lpAns);
		return ;
	}

	wUsersRet = 0 ;
	WriteData(lpAns, (void*)&ANS_OK, 2);		
	WriteData(lpAns, (void*)&wUsersRet, 2); 

	pData +=2 ;
	for( i=0 ;i<wUsers;i++ )
	{
		DWORD	dwKKID;
		memcpy(&dwKKID,pData,4 );

		lpDstUser = GetHashUserByID(dwKKID);
		if (lpDstUser)
		{
			LPANS_PACKET lpDstAns = CreateAnsPacket(pReq, lpDstUser, m_wSeq++);

			lpDstAns->pHeader->dwDst = dwKKID ;
			lpDstAns->pHeader->dwSrc = 0 ;
			WriteData(lpDstAns, &pReq->pUser->dwID, 4); 
			WriteData(lpDstAns, &pReq->pUser->dwLocalIP, 4); 
			WriteData(lpDstAns, &pReq->pUser->wLocalPort, 2); 
			WriteData(lpDstAns, &pReq->pUser->dwNatIP , 4); 
			WriteData(lpDstAns, &pReq->pUser->wNatPort, 2); 
			SendPacket(lpDstAns);
			
			WriteData(lpAns,&lpDstUser->dwID, 4); 
			WriteData(lpAns,&lpDstUser->dwLocalIP, 4); 
			WriteData(lpAns,&lpDstUser->wLocalPort, 2); 
			WriteData(lpAns,&lpDstUser->dwNatIP, 4); 
			WriteData(lpAns,&lpDstUser->wNatPort, 2); 
			wUsersRet++ ;
		}
		pData+=4 ;
	}
	// 重新写入返回用户数据量
	memcpy(lpAns->buf + P2P_HDR_LEN + 2, &wUsersRet, 2);
	SendPacket(lpAns);
}
// 申请对多个用户进行CRC中转
// 请求包:(请求对象ID)(4)+(请求人数)(2)
// 返回包:(命令号)(2)+(成功人数)(2)+(KKID)(4)+....
// 客户端请求包:(KKID+Local IP+Local Port +Nat IP +Nat Port)(16)
static void OnCRC(LPREQ_PACKET pReq)
{
	LPP2P_HDR		lpHeader;
	LPP2P_USER		lpDstUser, lpCRCUser;
	LPANS_PACKET	lpSrcAns, lpDesAns;
	DWORD			dwDesUser, dwSrcUser;
	WORD			wUsers,wUsersRet;
	BYTE			*pData;
	int				i;
	
	struct list_head *pos, *head;
	LIST_HEAD(CRCUsers_List);

	LogOut(KLOG_INFO, "Create CRC_REQ Answer Packet!");
	lpSrcAns = CreateAnsPacket(pReq, NULL, pReq->pHeader->wSeq);
	if (!lpSrcAns )
	{
		LogOut(KLOG_ERROR, "Function(OnCRC) can't alloc memory");
		return;
	}

	//获取CRC双方和所需用户数
	dwSrcUser = pReq->pHeader->dwSrc;
	pData = pReq->pData;
	memcpy(&dwDesUser, pData, 4);
	pData += 4;
	memcpy(&wUsers, pData, 2);
	
	lpDstUser = GetHashUserByID(dwDesUser);
	// 目标对象尚未登陆或一次请求用户过多
	if (wUsers > 12 || !lpDstUser) 
	{
		if (wUsers > 12)
			LogOut(KLOG_WARN, "%d Requested Too Many CRC Users - %d.", dwSrcUser, wUsers);
		else
			LogOut(KLOG_WARN, "%d Requested DesUser %d not exist!", dwSrcUser, dwDesUser);
					
		WriteData(lpSrcAns, (void*)&ANS_FAIL, 2);
		SendPacket(lpSrcAns);
		LogOut(KLOG_WARN, "====Send ANS_FAIL!====");
		return;
	}

	lpDesAns = CreateAnsPacket(pReq, lpDstUser, m_wSeq++);
	if (!lpDesAns )
	{
		LogOut(KLOG_ERROR, "Function(OnCRC) can't alloc memory");
		return;
	}
	//Set Src Header
	lpHeader = lpSrcAns->pHeader;
	lpHeader->dwDst = dwSrcUser;
	lpHeader->dwSrc = 0;
	lpHeader->wCmd  = CMD_ACK_CRC;

	//Set Dst Header
	lpHeader = lpDesAns->pHeader;
	lpHeader->dwDst = dwDesUser;
	lpHeader->dwSrc = 0;
	lpHeader->wCmd  = CMD_ACK_CRC;
	LogOut(KLOG_INFO, "Write CRC_REQ Answer Packet!");
	wUsersRet = 0;
	WriteData(lpSrcAns, (void*)&ANS_OK, 2);
	WriteData(lpSrcAns, (void*)&dwSrcUser, 4);
	WriteData(lpSrcAns, (void*)&dwDesUser, 4);
	WriteData(lpSrcAns, (void*)&wUsers, 2);
	WriteData(lpSrcAns, (void*)&wUsersRet, 2); 

	LogOut(KLOG_INFO, "Finding CRC Users!");
	head = &g_KeepAliveList;
	pos  = head->next;
	int	iChecked = 0;

	pthread_mutex_lock(&g_mtKeepAlive);
	for( i=0; i < wUsers; i++)
	{
		lpCRCUser == NULL;
		// 已经完成了全部的查找必须退出
		if (pos == head)
			break;

		while (pos != head) 
		{
			lpCRCUser = LIST_ENTRY(pos, P2P_USER, listItem);
			if (lpCRCUser->dwID != dwDesUser 
				&& lpCRCUser->dwID != dwSrcUser
			//	&& lpCRCUser->dwNatIP == lpCRCUser->dwLocalIP
				&& lpCRCUser->wRefCRC <= 3) 
			{
				LogOut(KLOG_INFO, "For %d---%d Choose a CRC User %d",
					dwSrcUser, dwDesUser, lpCRCUser->dwID);
				pos = pos->next;
				iChecked = 0;
				break;
			}
			
			iChecked++;
			pos = pos->next;
			LogOut(KLOG_INFO, "OnCRC Switch to another User %d.", iChecked);
		}

		if (lpCRCUser)