sockethandler.cpp.svn-base

絲路server源碼 Silk Road server source

/** \file SocketHandler.cpp
 **	\date  2004-02-13
 **	\author [e-mail]
**/
/*
Copyright (C) 2004-2007  Anders Hedstrom

This library is made available under the terms of the GNU GPL.

If you would like to use this library in a closed-source application,
a separate license agreement is available. For information about 
the closed-source license agreement for the C++ sockets library,
please visit http://www.alhem.net/Sockets/license.html and/or
email [e-mail].

This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; either version 2
of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
*/
#ifdef _WIN32
#ifdef _MSC_VER
#pragma warning(disable:4786)
#endif
#endif
#include <stdlib.h>
#include <errno.h>

#include "SocketHandler.h"
#include "UdpSocket.h"
#include "ResolvSocket.h"
#include "ResolvServer.h"
#include "TcpSocket.h"
#include "Mutex.h"
#include "Utility.h"
#include "SocketAddress.h"

#ifdef SOCKETS_NAMESPACE
namespace SOCKETS_NAMESPACE {
#endif


//#ifdef _DEBUG
//#define DEB(x) x; fflush(stderr);
//#else
#define DEB(x) 
//#endif


SocketHandler::SocketHandler(StdLog *p)
:m_stdlog(p)
,m_mutex(m_mutex)
,m_b_use_mutex(false)
,m_maxsock(0)
,m_preverror(-1)
,m_errcnt(0)
,m_tlast(0)
#ifdef ENABLE_SOCKS4
,m_socks4_host(0)
,m_socks4_port(0)
,m_bTryDirect(false)
#endif
#ifdef ENABLE_RESOLVER
,m_resolv_id(0)
,m_resolver(NULL)
#endif
#ifdef ENABLE_POOL
,m_b_enable_pool(false)
#endif
#ifdef ENABLE_TRIGGERS
,m_next_trigger_id(0)
#endif
#ifdef ENABLE_DETACH
,m_slave(false)
#endif
{
	FD_ZERO(&m_rfds);
	FD_ZERO(&m_wfds);
	FD_ZERO(&m_efds);
}


SocketHandler::SocketHandler(Mutex& mutex,StdLog *p)
:m_stdlog(p)
,m_mutex(mutex)
,m_b_use_mutex(true)
,m_maxsock(0)
,m_preverror(-1)
,m_errcnt(0)
,m_tlast(0)
#ifdef ENABLE_SOCKS4
,m_socks4_host(0)
,m_socks4_port(0)
,m_bTryDirect(false)
#endif
#ifdef ENABLE_RESOLVER
,m_resolv_id(0)
,m_resolver(NULL)
#endif
#ifdef ENABLE_POOL
,m_b_enable_pool(false)
#endif
#ifdef ENABLE_TRIGGERS
,m_next_trigger_id(0)
#endif
#ifdef ENABLE_DETACH
,m_slave(false)
#endif
{
	m_mutex.Lock();
	FD_ZERO(&m_rfds);
	FD_ZERO(&m_wfds);
	FD_ZERO(&m_efds);
}


SocketHandler::~SocketHandler()
{
#ifdef ENABLE_RESOLVER
	if (m_resolver)
	{
		m_resolver -> Quit();
	}
#endif
	{
		while (m_sockets.size())
		{
DEB(			fprintf(stderr, "Emptying sockets list in SocketHandler destructor, %d instances\n", (int)m_sockets.size());)
			socket_m::iterator it = m_sockets.begin();
			Socket *p = it -> second;
			if (p)
			{
DEB(				fprintf(stderr, "  fd %d\n", p -> GetSocket());)
				p -> Close();
DEB(				fprintf(stderr, "  fd closed %d\n", p -> GetSocket());)
//				p -> OnDelete(); // hey, I turn this back on. what's the worst that could happen??!!
				// MinionSocket breaks, calling MinderHandler methods in OnDelete -
				// MinderHandler is already gone when that happens...

				// only delete socket when controlled
				// ie master sockethandler can delete non-detached sockets
				// and a slave sockethandler can only delete a detach socket
				if (p -> DeleteByHandler()
#ifdef ENABLE_DETACH
					&& !(m_slave ^ p -> IsDetached()) 
#endif
					)
				{
					p -> SetErasedByHandler();
					delete p;
				}
				m_sockets.erase(it);
			}
			else
			{
				m_sockets.erase(it);
			}
DEB(			fprintf(stderr, "next\n");)
		}
DEB(		fprintf(stderr, "/Emptying sockets list in SocketHandler destructor, %d instances\n", (int)m_sockets.size());)
	}
#ifdef ENABLE_RESOLVER
	if (m_resolver)
	{
		delete m_resolver;
	}
#endif
	if (m_b_use_mutex)
	{
		m_mutex.Unlock();
	}
}


Mutex& SocketHandler::GetMutex() const
{
	return m_mutex; 
}


#ifdef ENABLE_DETACH
void SocketHandler::SetSlave(bool x)
{
	m_slave = x;
}


bool SocketHandler::IsSlave()
{
	return m_slave;
}
#endif


void SocketHandler::RegStdLog(StdLog *log)
{
	m_stdlog = log;
}


void SocketHandler::LogError(Socket *p,const std::string& user_text,int err,const std::string& sys_err,loglevel_t t)
{
	if (m_stdlog)
	{
		m_stdlog -> error(this, p, user_text, err, sys_err, t);
	}
}


void SocketHandler::Add(Socket *p)
{
	if (p -> GetSocket() == INVALID_SOCKET)
	{
		LogError(p, "Add", -1, "Invalid socket", LOG_LEVEL_WARNING);
		if (p -> CloseAndDelete())
		{
			m_delete.push_back(p);
		}
		return;
	}
	if (m_add.find(p -> GetSocket()) != m_add.end())
	{
		LogError(p, "Add", (int)p -> GetSocket(), "Attempt to add socket already in add queue", LOG_LEVEL_FATAL);
		m_delete.push_back(p);
		return;
	}
	m_add[p -> GetSocket()] = p;
}


void SocketHandler::Get(SOCKET s,bool& r,bool& w,bool& e)
{
	if (s >= 0)
	{
		r = FD_ISSET(s, &m_rfds) ? true : false;
		w = FD_ISSET(s, &m_wfds) ? true : false;
		e = FD_ISSET(s, &m_efds) ? true : false;
	}
}


void SocketHandler::Set(SOCKET s,bool bRead,bool bWrite,bool bException)
{
DEB(	fprintf(stderr, "Set(%d, %s, %s, %s)\n", s, bRead ? "true" : "false", bWrite ? "true" : "false", bException ? "true" : "false");)
	if (s >= 0)
	{
		if (bRead)
		{
			if (!FD_ISSET(s, &m_rfds))
			{
				FD_SET(s, &m_rfds);
			}
		}
		else
		{
			FD_CLR(s, &m_rfds);
		}
		if (bWrite)
		{
			if (!FD_ISSET(s, &m_wfds))
			{
				FD_SET(s, &m_wfds);
			}
		}
		else
		{
			FD_CLR(s, &m_wfds);
		}
		if (bException)
		{
			if (!FD_ISSET(s, &m_efds))
			{
				FD_SET(s, &m_efds);
			}
		}
		else
		{
			FD_CLR(s, &m_efds);
		}
	}
}


int SocketHandler::Select(long sec,long usec)
{
	struct timeval tv;
	tv.tv_sec = sec;
	tv.tv_usec = usec;
	return Select(&tv);
}


int SocketHandler::Select()
{
	if (!m_fds_callonconnect.empty() ||
#ifdef ENABLE_DETACH
		(!m_slave && !m_fds_detach.empty()) ||
#endif
		!m_fds_timeout.empty() ||
		!m_fds_retry.empty() ||
		!m_fds_close.empty() ||
		!m_fds_erase.empty())
	{
		return Select(0, 200000);
	}
	return Select(NULL);
}


int SocketHandler::Select(struct timeval *tsel)
{
	size_t ignore = 0;
	while (m_add.size() > ignore)
	{
		if (m_sockets.size() >= FD_SETSIZE)
		{
			LogError(NULL, "Select", (int)m_sockets.size(), "FD_SETSIZE reached", LOG_LEVEL_WARNING);
			break;
		}
		socket_m::iterator it = m_add.begin();
		SOCKET s = it -> first;
		Socket *p = it -> second;
DEB(		fprintf(stderr, "Trying to add fd %d,  m_add.size() %d,  ignore %d\n", (int)s, (int)m_add.size(), (int)ignore);)
		//
		if (m_sockets.find(p -> GetSocket()) != m_sockets.end())
		{
			LogError(p, "Add", (int)p -> GetSocket(), "Attempt to add socket already in controlled queue", LOG_LEVEL_FATAL);
			// %! it's a dup, don't add to delete queue, just ignore it
			m_delete.push_back(p);
			m_add.erase(it);
//			ignore++;
			continue;
		}
		if (!p -> CloseAndDelete())
		{
			StreamSocket *scp = dynamic_cast<StreamSocket *>(p);
			if (scp && scp -> Connecting()) // 'Open' called before adding socket
			{
				Set(s,false,true);
			}
			else
			{
				TcpSocket *tcp = dynamic_cast<TcpSocket *>(p);
				bool bWrite = tcp ? tcp -> GetOutputLength() != 0 : false;
				if (p -> IsDisableRead())
				{
					Set(s, false, bWrite);
				}
				else
				{
					Set(s, true, bWrite);
				}
			}
			m_maxsock = (s > m_maxsock) ? s : m_maxsock;
		}
		else
		{
			LogError(p, "Add", (int)p -> GetSocket(), "Trying to add socket with SetCloseAndDelete() true", LOG_LEVEL_WARNING);
		}
		// only add to m_fds (process fd_set events) if
		//  slave handler and detached/detaching socket
		//  master handler and non-detached socket
#ifdef ENABLE_DETACH
		if (!(m_slave ^ p -> IsDetach()))
#endif
		{
			m_fds.push_back(s);
		}
		m_sockets[s] = p;
		//
		m_add.erase(it);
	}
#ifdef MACOSX
	fd_set rfds;
	fd_set wfds;
	fd_set efds;
	FD_COPY(&m_rfds, &rfds);
	FD_COPY(&m_wfds, &wfds);
	FD_COPY(&m_efds, &efds);
#else
	fd_set rfds = m_rfds;
	fd_set wfds = m_wfds;
	fd_set efds = m_efds;
#endif
	int n;
	if (m_b_use_mutex)
	{
		m_mutex.Unlock();
		n = select( (int)(m_maxsock + 1),&rfds,&wfds,&efds,tsel);
		m_mutex.Lock();
	}
	else
	{
		n = select( (int)(m_maxsock + 1),&rfds,&wfds,&efds,tsel);
	}
	if (n == -1)
	{
		/*
			EBADF  An invalid file descriptor was given in one of the sets.
			EINTR  A non blocked signal was caught.
			EINVAL n is negative. Or struct timeval contains bad time values (<0).
			ENOMEM select was unable to allocate memory for internal tables.
		*/
		if (Errno != m_preverror || m_errcnt++ % 10000 == 0)
		{
			LogError(NULL, "select", Errno, StrError(Errno));
DEB(			fprintf(stderr, "m_maxsock: %d\n", m_maxsock);
			fprintf(stderr, "%s\n", Errno == EINVAL ? "EINVAL" :
				Errno == EINTR ? "EINTR" :
				Errno == EBADF ? "EBADF" :
				Errno == ENOMEM ? "ENOMEM" : "<another>");
			// test bad fd
			for (SOCKET i = 0; i <= m_maxsock; i++)
			{
				bool t = false;
				FD_ZERO(&rfds);
				FD_ZERO(&wfds);
				FD_ZERO(&efds);
				if (FD_ISSET(i, &m_rfds))
				{
					FD_SET(i, &rfds);
					t = true;
				}
				if (FD_ISSET(i, &m_wfds))
				{
					FD_SET(i, &wfds);
					t = true;
				}
				if (FD_ISSET(i, &m_efds))
				{
					FD_SET(i, &efds);
					t = true;
				}
				if (t && m_sockets.find(i) == m_sockets.end())
				{
					fprintf(stderr, "Bad fd in fd_set: %d\n", i);
				}
			}
) // DEB
			m_preverror = Errno;
		}
		/// \todo rebuild fd_set's from active sockets list (m_sockets) here
	}
	else
	if (!n)
	{
		m_preverror = -1;
	}
	else
	if (n > 0)
	{
		for (socket_v::iterator it2 = m_fds.begin(); it2 != m_fds.end() && n; it2++)
		{
			SOCKET i = *it2;
			if (FD_ISSET(i, &rfds))
			{
				socket_m::iterator itmp = m_sockets.find(i);
				if (itmp != m_sockets.end()) // found
				{
					Socket *p = itmp -> second;
					// new SSL negotiate method
#ifdef HAVE_OPENSSL
					if (p -> IsSSLNegotiate())
					{
						p -> SSLNegotiate();
					}