Click here to Skip to main content
15,894,460 members
Search within:


Articles / Desktop Programming / MFC
 

Multithreaded Non-blocking Socket Server and Client, Based on Synchronized Socket

Rate me:
Please Sign up or sign in to vote.
3.75/5 (4 votes)
11 Mar 2002 98K   3.9K   33  
Non-blocking socket class using synchronized socket.
 
// Synchronized socket server deamon that support multiple clients
// Author: Kevin Hua
// Create: 2002-2-25 14:00
// Revision: 1
// Contact: Kevin-Hua@Woncore.com
//////////////////////////////////////////////////////////////////////
#if !defined(AFX_SYNCSOCKETSERVER_H__6C21FFA2_485A_11D5_9692_0050BA8CD8A0__INCLUDED_)
#define AFX_SYNCSOCKETSERVER_H__6C21FFA2_485A_11D5_9692_0050BA8CD8A0__INCLUDED_

#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000

#pragma warning(disable:4786)

#include <winsock2.h>
#include <process.h>
#include <list>

using namespace std;


#ifdef _DEBUG
#define new DEBUG_NEW
#endif


#pragma warning(disable : 4786)


template<class TSocketWorker>
class CSyncSocketServer
{
private:
	typedef list<TSocketWorker* > NEW_CONNECTION_LIST;

	struct START_WORKER_THREAD_PARAM
	{
		void* pServer;
		void* pWorker;
	};

	struct REFUSE_THREAD_PARAM
	{
		void* pServer;
		SOCKET s;
	};

public:
	CSyncSocketServer()
	{
		m_bRunning = FALSE;
		m_nServerPort = 0;
		m_nClientsMax = 100;
		m_nTimeout = -1;
		m_hEventShutdown = CreateEvent(NULL, FALSE, FALSE, NULL);
		InitializeCriticalSection(&m_cs);

		WORD wVersionRequested;
		WSADATA wsaData;
		int result;
		 
		wVersionRequested = MAKEWORD(2, 2);
		 
		result = WSAStartup(wVersionRequested, &wsaData);
		if(result != 0)
		{
		    m_bWSAStartup = FALSE;
		}
		 
		if (LOBYTE( wsaData.wVersion) != 2 ||
		        HIBYTE(wsaData.wVersion) != 2) 
		{
		    WSACleanup();
		    m_bWSAStartup = FALSE; 
		}		
	}

	virtual	~CSyncSocketServer()
	{
		if(m_bRunning)
			Shutdown();

		CloseHandle(m_hEventShutdown);
		DeleteCriticalSection(&m_cs);

		if(m_bWSAStartup)
			WSACleanup();
	}


	BOOL IsRunning()
	{
		return m_bRunning;
	}

	unsigned int GetClientsCount()
	{
		EnterCriticalSection(&m_cs);
		int nCount = m_listNewConnections.size();
		LeaveCriticalSection(&m_cs);

		return nCount;
	}

	void SetClientsMax(unsigned int nMax)
	{
		m_nClientsMax = nMax;
	}

	void SetCannotConnectMsg(const char* pMsg)
	{
		m_strCannotConnectMsg = pMsg;
	}

	unsigned long GetTimeout()
	{
		return m_nTimeout;
	}

	unsigned long SetTimeout(unsigned long nNewVal)
	{
		if(nNewVal == 0)
			nNewVal = -1;

		unsigned long nOldVal = m_nTimeout;
		m_nTimeout = nNewVal;

		EnterCriticalSection(&m_cs);
		NEW_CONNECTION_LIST::iterator it;
		for(it = m_listNewConnections.begin(); it != m_listNewConnections.end(); ++it)
		{
			TSocketWorker* pWorker = *it;
			if(pWorker)
				pWorker->SetTimeout(m_nTimeout);
		}
		LeaveCriticalSection(&m_cs);
		
		return nOldVal;
	}

	
public:						
	BOOL Run(int nPort)
	{
		if(!m_bWSAStartup)
			return FALSE;
		
		if(m_bRunning)
		{
			if(nPort != m_nServerPort)
				return FALSE;
			else
				return TRUE;
		}
		
		
		m_nServerPort = nPort;

		m_hThreadH = (HANDLE)_beginthread(HelperThread, 0, this);
		if(-1 == (int)m_hThreadH)
		{
			return FALSE;
		}

		m_hThreadLaunchedEvent	= CreateEvent(NULL, FALSE, TRUE, NULL);

		// Launch Accept Thread
		ResetEvent(m_hThreadLaunchedEvent);
		m_hThreadA = (HANDLE)_beginthread(AcceptThread, 0, this);
		if(-1 == (int)m_hThreadA)
		{
			CloseHandle(m_hThreadLaunchedEvent);			
			return FALSE;
		}

		if(WaitForSingleObject(m_hThreadLaunchedEvent, 30000) == WAIT_TIMEOUT)
		{
			TRACE("CSyncSocketServer::Run()   Terminate AcceptThread\n");			
			TerminateThread(m_hThreadA, 0);
			CloseHandle(m_hThreadLaunchedEvent);
			return FALSE;
		}

		CloseHandle(m_hThreadLaunchedEvent);
		
		return m_bRunning = TRUE;
	}

	BOOL Shutdown()
	{
		if(!m_bRunning)
			return FALSE;
		
		BOOL bResult = TRUE;

		m_bShutingdown = TRUE;
		
		closesocket(m_socketAccept);
		m_socketAccept = INVALID_SOCKET;
		
		SetEvent(m_hEventShutdown);
		
		if(WaitForSingleObject(m_hThreadA, 10000) == WAIT_TIMEOUT)
		{
			TRACE("CSyncSocketServer::Shutdown()   Terminate RecvThread\n");
			TerminateThread(m_hThreadA, 0);
			CloseHandle(m_hThreadA); 		
		}


		Sleep(3000);
		
		EnterCriticalSection(&m_cs);

		// Close all connections to clients, and you 
		// should not call DeleteSocketWorker() here
		// because it'll causes memory leaks when app quits
		NEW_CONNECTION_LIST::iterator it;
		for(it = m_listNewConnections.begin(); it != m_listNewConnections.end(); ++it)
		{
			TSocketWorker* pWorker = *it;
			if(!pWorker)
				continue;

			if(pWorker->IsConnected())
				pWorker->Disconnect();
		}
		
		for(it = m_listNewConnections.begin(); it != m_listNewConnections.end(); ++it)
		{
			TSocketWorker* pWorker = *it;
			if(!pWorker)
				continue;

			try
			{
				pWorker->Stop();
				delete pWorker;
			}
			catch(...)
			{
				TRACE("CSyncSocketServer::DeleteSocketWorker() exception.\n");
			}
		}
		
		m_listNewConnections.clear();
	
		LeaveCriticalSection(&m_cs);

		Reset();

		return TRUE;
	}
	
private:
	BOOL StartSocketWorker(TSocketWorker* pWorker)
	{
		if(!pWorker)
			return FALSE;

		START_WORKER_THREAD_PARAM* pParam = new START_WORKER_THREAD_PARAM();
		pParam->pServer = (void*)this;
		pParam->pWorker = (void*)pWorker;
		
		if(-1 == _beginthread(StartWorkerThread, 0, pParam))
		{
			return FALSE;
		}

		return TRUE;
	}
	

	static void __cdecl StartWorkerThread(void* pParam)
	{
		if(pParam)
		{
			START_WORKER_THREAD_PARAM* p = (START_WORKER_THREAD_PARAM*)pParam;
			CSyncSocketServer *pServer = (CSyncSocketServer*)p->pServer;
			TSocketWorker* pWorker = (TSocketWorker*)p->pWorker;

			try
			{
				if(pWorker->Start(pServer->m_strCannotConnectMsg.c_str(), pServer->m_strCannotConnectMsg.size()))
				{
					EnterCriticalSection(&pServer->m_cs);
					pServer->m_listNewConnections.push_back(pWorker);
					LeaveCriticalSection(&pServer->m_cs);
				}
				else
				{
					pWorker->Stop();
					delete pWorker;
				}
				
			}
			catch(...)
			{
				TRACE("CSyncSocketServer::StartWorkerThread() exception.\n");
				pServer->DeleteSocketWorker(pWorker);
			}

			delete p;
		}
	}

	void DeleteSocketWorker(TSocketWorker *pWorker)
	{
		if(!pWorker)
			return;
		
		if(-1 == _beginthread(DeleteWorkerThread, 0, pWorker))
		{
			try
			{
				pWorker->Stop();
				delete pWorker;
			}
			catch(...)
			{
				TRACE("CSyncSocketServer::DeleteSocketWorker() exception.\n");
			}
		}
	}

	static void __cdecl DeleteWorkerThread(void* pParam)
	{
		TSocketWorker* pWorker = (TSocketWorker*)pParam;
		if(pWorker)
		{
			try
			{
				pWorker->Stop();
				delete pWorker;
			}
			catch(...)
			{
				TRACE("CSyncSocketServer::DeleteSocketWorker() exception.\n");
			}
		}
	}

	static void __cdecl HelperThread(void* pParam)
	{// delete disconnected workers
		CSyncSocketServer *pServer = (CSyncSocketServer*)pParam;
		while(WAIT_OBJECT_0 != WaitForSingleObject(pServer->m_hEventShutdown, 1000))
		{
			NEW_CONNECTION_LIST listRunning, listClosed;
			NEW_CONNECTION_LIST::iterator it;

			listRunning.clear();
			listClosed.clear();
			
			EnterCriticalSection(&pServer->m_cs);
			for(it = pServer->m_listNewConnections.begin(); it != pServer->m_listNewConnections.end(); ++it)
			{
				TSocketWorker* pWorker = *it;
				if(pWorker && pWorker->IsConnected())
					listRunning.push_back(pWorker);
				else if(pWorker)
					listClosed.push_back(pWorker);
			}

			if(listRunning.size() != pServer->m_listNewConnections.size())
				pServer->m_listNewConnections = listRunning;


			while(listClosed.size())
			{
				TSocketWorker* pWorker = *listClosed.begin();
				pServer->DeleteSocketWorker(pWorker);
				listClosed.pop_front();
			}
			
			LeaveCriticalSection(&pServer->m_cs);
		}
	}

	static void __cdecl	AcceptThread(void* pParam)   // accecpt new connection and return the socket
	{
		CSyncSocketServer *pServer = (CSyncSocketServer*)pParam;
		SOCKET s; // Main Listen Socket
		sockaddr_in saLocal;
		sockaddr ClientAddr;
		INT addrlen = sizeof(ClientAddr);
		sockaddr_in sain;
		char szClientAddr[50];
		int result;
		
		saLocal.sin_family		= AF_INET;
		saLocal.sin_port		= htons(pServer->m_nServerPort);
		saLocal.sin_addr.s_addr = INADDR_ANY;
				
		s = socket(AF_INET, SOCK_STREAM, 0);
		if(s == INVALID_SOCKET)
		{
			return;
		}
		
		//
		//	B I N D
		//
		result = bind(s, (struct sockaddr *)&saLocal, sizeof(saLocal));
		if(result == SOCKET_ERROR)
		{
			shutdown(s, SD_SEND);
			closesocket(s);
			return;
		}	

		//
		//	L I S T E N
		//
		result = listen(s, SOMAXCONN);
		if(result == SOCKET_ERROR)
		{
			shutdown(s, SD_SEND);
			closesocket(s);
			return;
		}	
		
		pServer->m_socketAccept = s;

		SetEvent(pServer->m_hThreadLaunchedEvent);

		for(;;)
		{

				SOCKET ClientSocket = accept(s, &ClientAddr, &addrlen);

				if(INVALID_SOCKET == ClientSocket )
				{ 
					int nErrCode = WSAGetLastError();
					if(nErrCode == WSAEMFILE || nErrCode == WSAENOBUFS)
						continue;
					
					if(pServer->m_bShutingdown)
						break; 
				}
				else
				{
					memcpy(&sain, &ClientAddr, addrlen);
					sprintf(szClientAddr, "%d.%d.%d.%d", 
						sain.sin_addr.S_un.S_un_b.s_b1, 
						sain.sin_addr.S_un.S_un_b.s_b2, 
						sain.sin_addr.S_un.S_un_b.s_b3, 
						sain.sin_addr.S_un.S_un_b.s_b4);

					pServer->AddClient(ClientSocket, szClientAddr, sain.sin_port);
				}
		}

		shutdown(s, SD_SEND);
		closesocket(s);
		return;
	}	

	static void __cdecl RefuseThread(void* pParam)
	{
		REFUSE_THREAD_PARAM* p = (REFUSE_THREAD_PARAM*)pParam;
		if(p)
		{
			CSyncSocketServer* pServer = (CSyncSocketServer*)p->pServer;
			send(p->s, pServer->m_strCannotConnectMsg.c_str(), pServer->m_strCannotConnectMsg.size(), 0);
			shutdown(p->s, SD_SEND);
			closesocket(p->s);
			delete p;
		}
	}
	
	BOOL AddClient(SOCKET s, const char* szClientAddress, int port)
	{
		if(!CanJoin())
		{
			REFUSE_THREAD_PARAM* p = new REFUSE_THREAD_PARAM();
			
			p->pServer = this;
			p->s = s;
			if(-1 == _beginthread(RefuseThread, 0, p))
			{
				send(s, m_strCannotConnectMsg.c_str(), m_strCannotConnectMsg.size(), 0);
				shutdown(s, SD_SEND);
				closesocket(s);		
				delete p;
			}
			return FALSE;
		}

		TSocketWorker* pWorker = new TSocketWorker(s, szClientAddress, port, TRUE);
		if(pWorker)
		{
			pWorker->SetTimeout(m_nTimeout);
			if(!StartSocketWorker(pWorker))
				DeleteSocketWorker(pWorker);
		}

		
		return TRUE;
	}

	void Reset() // Release resource allocated and reset all date members to initial states 
	{
		m_bRunning = FALSE;
		m_hThreadA = NULL;
		m_hThreadH = NULL;
	}

	
	BOOL CanJoin()
	{
		if(INVALID_SOCKET == m_socketAccept)
			return FALSE;
		
		EnterCriticalSection(&m_cs);

		BOOL bRet = m_listNewConnections.size() < m_nClientsMax;

		LeaveCriticalSection(&m_cs);

		return bRet;
	}

	
private:					
	HANDLE					m_hThreadA; 
	HANDLE					m_hThreadH;						
	HANDLE					m_hThreadLaunchedEvent;
	HANDLE					m_hEventShutdown;						
						
	CRITICAL_SECTION		m_cs;
	
	unsigned short			m_nServerPort;

	BOOL					m_bRunning;
	BOOL					m_bWSAStartup;
protected:
	
	NEW_CONNECTION_LIST		m_listNewConnections;
	
private:
	BOOL					m_bShutingdown;
	SOCKET					m_socketAccept;

	unsigned int			m_nClientsMax;
	string					m_strCannotConnectMsg;

	unsigned long			m_nTimeout;
};


#endif // !defined(AFX_SYNCSOCKETSERVER_H__6C21FFA2_485A_11D5_9692_0050BA8CD8A0__INCLUDED_)

By viewing downloads associated with this article you agree to the Terms of Service and the article's licence.

If a file you wish to view isn't highlighted, and is a text file (not binary), please let us know and we'll add colourisation support for it.

License

This article has no explicit license attached to it but may contain usage terms in the article text or the download files themselves. If in doubt please contact the author via the discussion board below.

A list of licenses authors might use can be found here


Written By
Kevin Hua
Web Developer
China China
This member has not yet provided a Biography. Assume it's interesting and varied, and probably something to do with programming.

Comments and Discussions

Permalink
Advertise
Privacy
Cookies
Terms of Use
Layout: fixed | fluid

Posted 11 Mar 2002

Article Copyright 2002 by Kevin Hua
Everything else Copyright © CodeProject, 1999-2024

Web01 2.8:2024-04-02:1