socketserver.cc

五行MMORPG引擎系统V1.0

// SocketServer.cc: interface for the SocketLayer class.
/*/////////////////////////////////////////////////////////////////////////////
	Socket层，封装了服务器基本功能
	李亦
	2006.06.21

/*//////////////////////////////////////////////////////////////////////////////

//#include "Platforms.h"

//#ifdef _LINUX
//#include "unistd.h"
//#endif

#ifdef TGE_RPGSERVER

#include "server/net/IOTCPModule.h"
#include "server/net/IOUDPModule.h"
//#include "server/net/auth.h"
//#include "server/net/Nullauth.h"
//#include "server/encrypt/Encryption.h"
//#include "server/encrypt/XOREncryptionModule.h"
//#include "server/encrypt/EncryptDESHashModule.h"
#include "server/SocketServer.h"


#define  THREAD_LIFETIME			5000
#define	SOCKETPOOL_NUM				1
#define	SOCKETPOOL_MAX				16
#define	SOCKETPOOL_MIN				2

#define	LISTENPOOL_NUM				1
#define	LISTENPOOL_MAX				8
#define	LISTENPOOL_MIN				1

#define  DISCARD_LIFETIME			120000
#define	DISCARDPOOL_NUM			1
#define	DISCARDPOOL_MAX			4
#define	DISCARDPOOL_MIN			1


#define	MAX_LISTEN_SOCKETS		64
#define	MAX_CLIENT_SOCKETS		2048
#define  CREATESERVER_TRYDELAY	100
#define  SERVER_IDLE					20


///////////////////////////////////////////////////////////////////////////////
//CS SDK
namespace CS
{



//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

IMPLEMENT_SOCKETPOOL(SocketServer,Send)
{
	m_pServer->ProcessSendSocket(pSocket);
}

IMPLEMENT_SOCKETPOOL(SocketServer,Recv)
{
	m_pServer->ProcessRecvSocket(pSocket);
}

IMPLEMENT_SOCKETPOOL(SocketServer,Listen)
{
	m_pServer->ProcessAcceptSocket(pSocket);
}

IMPLEMENT_SOCKETPOOL(SocketServer,Discard)
{
	m_pServer->ProcessDiscardSocket(pSocket);
}




//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

SocketServer::SocketServer()
:m_sendPool(SOCKETPOOL_NUM,		SOCKETPOOL_MIN,	SOCKETPOOL_MAX,	THREAD_LIFETIME)
,m_recvPool(SOCKETPOOL_NUM,		SOCKETPOOL_MIN,	SOCKETPOOL_MAX,	THREAD_LIFETIME)
,m_listenPool(LISTENPOOL_NUM,		LISTENPOOL_MIN,	LISTENPOOL_MAX,	THREAD_LIFETIME)
,m_discardPool(DISCARDPOOL_NUM,	DISCARDPOOL_MIN,	DISCARDPOOL_MAX,	DISCARD_LIFETIME)
{
	m_sendPool.m_pServer		= this;
	m_recvPool.m_pServer		= this;
	m_listenPool.m_pServer	= this;
	m_discardPool.m_pServer	= this;

	//m_uIdleTimeOut			= 5000;
	m_bStopServer		= FALSE;
}

SocketServer::~SocketServer()
{
}


IMPLEMENT_CONOBJECT(SocketServer);

ConsoleMethod(SocketServer,create,bool,2,2,"obj.create()")
{
	CSTR	szAddr;
	CSTR	szNetMod;
	CSTR	szEncrypt;
	CSTR	szAuth;

	S32	nListenNum = Con::getIntVariable("Pref::CS::ListenAmount",1);
	if(nListenNum <= 0)
	{
		AssertFatal(0,"需要指定Listen端口...");
		return false;
	}
	//if(nListenNum == 1)
	//{
	//	szAddr		= Con::getVariable("Pref::CS::Addr");
	//	szNetMod		= Con::getVariable("Pref::CS::IO");
	//	szEncrypt	= Con::getVariable("Pref::CS::Encrypt");
	//	szAuth		= Con::getVariable("Pref::CS::Auth");

	//	//Con::getVariable("$Pref::CS::port");
	//	return object->CreateServer(szAddr,szNetMod, szEncrypt, szAuth) == E_IOOK;
	//}

	//bool bRet(false);
	S32 n;
	for(n = 0; n < nListenNum; n++)
	{
		szAddr		= Con::getVariable(avar("Pref::CS::Addr%d",n));
		szNetMod		= Con::getVariable(avar("Pref::CS::IO%d",n));
		szEncrypt	= Con::getVariable(avar("Pref::CS::Encrypt%d",n));
		szAuth		= Con::getVariable(avar("Pref::CS::Auth%d",n));

		//Con::getVariable("$Pref::CS::port");
		if( object->CreateServer(szAddr,szNetMod, szEncrypt, szAuth) != E_IOOK)
			return false;
		Con::printf(" @建立Listen -> %s",szAddr);
	}
	return nListenNum > 0;
}

ConsoleMethod(SocketServer,shutdown,void,2,2,"obj.shutdown()")
{
	object->ShutdownServer();
}

ConsoleMethod(SocketServer,stop,void,2,2,"obj.stop()")
{
	object->StopServer();
}

ConsoleMethod(SocketServer,run,void,2,2,"obj.run()")
{
	object->RunServer();
}


int SocketServer::CreateServer(CSTR szAddr, CSTR szNetMod, CSTR szEncrypt, CSTR szAuth)
{
	if(m_arListenSockets.size() >= MAX_LISTEN_SOCKETS)
		return E_IOFAILED;

	//m_uIdleTimeOut	= nIdleTimeOut;
	//m_nCommCount	= 0;
	//m_nConnCount	= 0;

	if(	!szNetMod 		|| szNetMod[0] == 0 		)
		szNetMod = "NetTCP";

	if(!szAddr || szAddr[0] == 0)
		szAddr = "localhost:2323";

	if(	!szEncrypt 		|| szEncrypt[0] == 0 	)
		 szEncrypt = "XOR";

	if(	!szAuth 		|| szAuth[0] == 0 	)
		 szAuth = "NULL";


	CAuthSocket *pSockNew;
	//do 
	//{
		pSockNew = ListenAuthSocket(NULL,0,NULL,
							szAddr,
							szNetMod,
							szEncrypt,
							szAuth);
	
		if(pSockNew != INVALID_AUTHPTR) 
		{

			m_uIdleTimeOut = Con::getIntVariable("Pref::CS::ServerIdle",CS_NET_IDLE);
			m_arListenSockets.push_back(pSockNew);
			return E_IOOK;
		}

	//	Platform::sleep(CREATESERVER_TRYDELAY);

	//} while(pSockNew == INVALID_AUTHPTR);

	return E_IOFAILED;
}


void SocketServer::ShutdownServer()
{
	int n;

	//关闭所有Socket池

	// discard all connections
	m_mutexClients.lock(true);

	for(n=0; n < m_arClientSockets.size(); n++)
	{
		m_discardPool.ProcessSockets(m_arClientSockets[n]);
		//m_arClientSockets[n]->Close();
		//delete m_arClientSockets[n];
	}
	m_arClientSockets.clear();
	m_mutexClients.unlock();



	// Terminate all bound sockets
	m_mutexListens.lock(true);

	for(n=0; n < m_arListenSockets.size(); n++) 
	{
		m_discardPool.ProcessSockets(m_arListenSockets[n]);
		//m_arListenSockets[n]->Close();
		//delete m_arListenSockets[n];
	}
	m_arListenSockets.clear();
	m_mutexListens.unlock();

	Con::printf("@网络守护线程已关闭... ");

}


BOOL SocketServer::RunServer()
{
	//m_sendPool.s
	return Parent::start();
}


void SocketServer::run(S32 arg)
{
	//轮询检测Accept
	arg;

	S32	n,
			nRet;
	CAuthSocket* pSocket;

	Con::printf("@网络守护线程启动... ");

	while(!m_bStopServer) 
	{
		//////////////////////////////////////////////////
		////检测listen，是否有新连接进来，
		/// 有的话，派发到listenPool处理
		for(n=0; n < m_arListenSockets.size(); n++) 
		{
			pSocket = m_arListenSockets[n];
			if(pSocket->CheckAccept(TRUE))
			{	
				pSocket->SetACCEPTING();
				m_listenPool.ProcessSockets(pSocket);
			}
		}


		//////////////////////////////////////////////////
		//Client连接处理
		for(n = m_arClientSockets.size()-1; n >= 0 ; n--) 
		{
			pSocket = m_arClientSockets[n];
			//AssertWarn(0,"检测连接是否已经关闭");
			//检测连接是否已经关闭
			if(!pSocket->IsAlive())
			{
				//delete pSocket; 丢到垃圾堆，等候处理
				m_mutexClients.lock(true);
				m_arClientSockets.erase(n);
				m_mutexClients.unlock();
				m_discardPool.ProcessSockets(pSocket);
				continue;
			}

			//未建立连接，不处列队发送、接收处理
			if(!pSocket->IsESTABLISHED())
				continue;

			////检测recv，是否有数据可读
			/// 有的话，派发到recvPool处理
			if(pSocket->CheckRecv(TRUE))
			{	
				pSocket->SetRECVING();
				m_recvPool.ProcessSockets(pSocket);
			}

			////检测Send，是否有数据可发送
			/// 有的话，派发到sendPool处理
			if(pSocket->CheckSend(TRUE))
			{	
				pSocket->SetSENDING();
				m_sendPool.ProcessSockets(pSocket);
			}
		}


		Platform::sleep(m_uIdleTimeOut);
	}//while

	// Terminate nCmd handlers
	ShutdownServer();
}


void SocketServer::StopServer()
{
	m_bStopServer = TRUE;
}


void SocketServer::ProcessDiscardSocket(CAuthSocket*  pSocket)
{
	if(pSocket == NULL)
		return;
	//进来这里的，须为 not alive
	if(pSocket->IsAlive())
	{
		AssertWarn(0,"进来垃圾堆的socket还是alive，有没搞错");
		return;
	}

	//等待工作完毕
	for(;;)
	{
		Platform::sleep(m_uIdleTimeOut);
		if( pSocket->IsACCEPTING() )
			continue;
		if( pSocket->IsSENDING() )
			continue;
		if( pSocket->IsRECVING() )
			continue;
		break;
	}

	pSocket->Close();

	Platform::sleep(m_uIdleTimeOut);
	//删除socket对象，自动调用Close();
	delete pSocket;
}


void SocketServer::ProcessAcceptSocket(CAuthSocket*  pSocket)
{
	if(pSocket == NULL)
		return;

	CAuthSocket *pSockNew;

	//pSocket->SetACCEPTING();

	for(; pSocket->CheckAccept(FALSE) ; )
	{
		pSockNew	= pSocket->Accept();

		if(pSockNew == NULL)
			break;
		

		if(m_arClientSockets.size() >= MAX_CLIENT_SOCKETS) 
		{	
			Con::printf("@@@@@@@@@ 服务端超载 @@@@@@@");
			pSockNew->Close();
			delete pSockNew;	
			break;
		}
		Con::printf("接受新连接...");
			
		m_mutexClients.lock(true);
		m_arClientSockets.push_back(pSockNew);
		m_mutexClients.unlock();
		pSockNew->SetESTABLISHED();

		Platform::sleep(SERVER_IDLE);
	}//for

	pSocket->SetACCEPTING(FALSE);
}


void SocketServer::ProcessRecvSocket(CAuthSocket*  pSocket)
{
	if(pSocket == NULL)
		return;
	
	BitStream* pStream;
	BOOL bRet;

	//pSocket->SetRECVING();
	for(; pSocket->CheckRecv(FALSE) ; )
	{
		if(!pSocket->BeginRecv())
			break;

		//要跳出，因为断线后返回E_IODISCONNECTED
		if(!pSocket->EndRecv())
			break;
	}//for

	//由SocketServer统一处理
	//if(nRet == E_IODISCONNECTED)
	//	pSocket->SetTOBEDIE();

	pSocket->SetRECVING(FALSE);

}


void SocketServer::ProcessSendSocket(CAuthSocket*  pSocket)
{
	if(pSocket == NULL)
		return;

	//pSocket->SetSENDING();
	//发送所有数据缓冲
	S32 nRet;

	do
	{
		nRet = pSocket->SendOutPacket();
	}while(nRet == E_IOOK);

	pSocket->SetSENDING(FALSE);
}


};//namespace CS



#endif //#ifdef TGE_RPGSERVER