From 60b5d45c53238a70c75a925acf4de0ebc6132365 Mon Sep 17 00:00:00 2001 From: lotodore Date: Mon, 4 May 2009 21:11:29 +0000 Subject: [PATCH] Adopting changes from asio branch. Currently broken. --- pokerth_game.pro | 1 + pokerth_lib.pro | 2 + pokerth_server.pro | 17 +- src/core/common/timermanager.cpp | 111 +++++++++++ src/core/timermanager.h | 63 ++++++ src/net/clientthread.h | 2 +- src/net/common/clientstate.cpp | 17 +- src/net/common/clientthread.cpp | 31 ++- src/net/common/connectdata.cpp | 10 +- src/net/common/receiverhelper.cpp | 4 +- src/net/common/serveracceptthread.cpp | 267 ++++++++----------------- src/net/common/servercontext.cpp | 11 +- src/net/common/serverlobbythread.cpp | 275 +++++++++++++++----------- src/net/common/servermanager.cpp | 22 +-- src/net/common/sessiondata.cpp | 21 +- src/net/connectdata.h | 9 +- src/net/netcontext.h | 3 +- src/net/receiverhelper.h | 3 +- src/net/serveracceptthread.h | 36 ++-- src/net/servercontext.h | 12 +- src/net/serverlobbythread.h | 31 +-- src/net/servermanager.h | 2 + src/net/sessiondata.h | 10 +- 23 files changed, 539 insertions(+), 421 deletions(-) create mode 100644 src/core/common/timermanager.cpp create mode 100644 src/core/timermanager.h diff --git a/pokerth_game.pro b/pokerth_game.pro index 4b4b57cc..ba9b935f 100644 --- a/pokerth_game.pro +++ b/pokerth_game.pro @@ -334,6 +334,7 @@ win32 { -lmsimg32 \ -lshell32 \ -lkernel32 \ + -lmswsock \ -lws2_32 \ -ladvapi32 \ -lsdl \ diff --git a/pokerth_lib.pro b/pokerth_lib.pro index 11f98301..3d707d5d 100644 --- a/pokerth_lib.pro +++ b/pokerth_lib.pro @@ -56,6 +56,7 @@ HEADERS += \ src/core/crypthelper.h \ src/core/avatarmanager.h \ src/core/pokerthexception.h \ + src/core/timermanager.h \ src/engine/boardinterface.h \ src/engine/enginefactory.h \ src/engine/handinterface.h \ @@ -131,6 +132,7 @@ SOURCES += \ src/core/common/crypthelper.cpp \ src/core/common/avatarmanager.cpp \ src/core/common/pokerthexception.cpp \ + src/core/common/timermanager.cpp \ src/third_party/tinyxml/tinystr.cpp \ src/third_party/tinyxml/tinyxml.cpp \ src/third_party/tinyxml/tinyxmlerror.cpp \ diff --git a/pokerth_server.pro b/pokerth_server.pro index 522de319..84dc2c90 100644 --- a/pokerth_server.pro +++ b/pokerth_server.pro @@ -149,7 +149,22 @@ win32 { LIBS += -llibboost_program_options-mgw43-mt-1_38 } - LIBS += -lgdi32 -lcomdlg32 -loleaut32 -limm32 -lwinmm -lwinspool -lole32 -luuid -luser32 -lmsimg32 -lshell32 -lkernel32 -lws2_32 -ladvapi32 -lwldap32 + LIBS += -lgdi32 \ + -lcomdlg32 \ + -loleaut32 \ + -limm32 \ + -lwinmm \ + -lwinspool \ + -lole32 \ + -luuid \ + -luser32 \ + -lmsimg32 \ + -lshell32 \ + -lkernel32 \ + -lmswsock \ + -lws2_32 \ + -ladvapi32 \ + -lwldap32 } !win32 { DEPENDPATH += src/net/linux/ src/core/linux diff --git a/src/core/common/timermanager.cpp b/src/core/common/timermanager.cpp new file mode 100644 index 00000000..66c89b8b --- /dev/null +++ b/src/core/common/timermanager.cpp @@ -0,0 +1,111 @@ +/*************************************************************************** + * Copyright (C) 2009 by Lothar May * + * * + * 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. * + ***************************************************************************/ + +#include + + +using namespace std; + + +TimerManager::TimerManager() +: m_curTimerId(0) +{ +} + +unsigned +TimerManager::RegisterTimer(unsigned timeoutMsec, boost::function timerHandler, bool timerRepeat) +{ + // Register a new timer callback. + boost::recursive_mutex::scoped_lock lock(m_timerMutex); + // Use a unique id for each timer. + unsigned id = GetNextTimerId(); + // Sort all timers by tick on which the callback occurs. + unsigned absoluteTimer = static_cast(m_softwareTimer.elapsed().total_milliseconds()) + timeoutMsec; + m_timerMap.insert( + TimerMap::value_type(absoluteTimer, TimerData(id, timeoutMsec, timerHandler, timerRepeat))); + return id; +} + +bool +TimerManager::UnregisterTimer(unsigned timerId) +{ + // Remove a timer callback from the map. + boost::recursive_mutex::scoped_lock lock(m_timerMutex); + bool unregistered = false; + TimerMap::iterator i = m_timerMap.begin(); + TimerMap::iterator end = m_timerMap.end(); + while (i != end) + { + if (i->second.id == timerId) + { + m_timerMap.erase(i); + unregistered = true; + break; + } + ++i; + } + return unregistered; +} + +void +TimerManager::Process() +{ + boost::recursive_mutex::scoped_lock lock(m_timerMutex); + bool timerOccurred; + do + { + timerOccurred = false; + TimerMap::iterator i = m_timerMap.begin(); + if (i != m_timerMap.end()) + { + // Check whether the timer occured. + unsigned currentTicks = static_cast(m_softwareTimer.elapsed().total_milliseconds()); + unsigned timerTicks = i->first; + if (currentTicks >= timerTicks) + { + // Grab a copy of the timer. + TimerData timer = i->second; + // Remove the timer, re-add if it is repeating. + m_timerMap.erase(i); + if (timer.repeat) + { + // Re-add the repeating timer. + // Try to be precise. + unsigned absoluteTimer = currentTicks + timer.msec; + unsigned tickDiff = currentTicks - timerTicks; + if (absoluteTimer >= tickDiff) + absoluteTimer -= tickDiff; + m_timerMap.insert( + TimerMap::value_type(absoluteTimer, timer)); + } + + // The callback may register/unregister a timer (a recursive mutex is used). + timer.handler(); + timerOccurred = true; + } + } + } while (timerOccurred); +} + +unsigned +TimerManager::GetNextTimerId() +{ + return ++m_curTimerId; +} + diff --git a/src/core/timermanager.h b/src/core/timermanager.h new file mode 100644 index 00000000..778e9e06 --- /dev/null +++ b/src/core/timermanager.h @@ -0,0 +1,63 @@ +/*************************************************************************** + * Copyright (C) 2009 by Lothar May * + * * + * 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. * + ***************************************************************************/ +/* A manager for asynchronous software timer. */ + +#ifndef _TIMERMANAGER_H_ +#define _TIMERMANAGER_H_ + +#include +#include +#include +#include + +class TimerManager +{ +public: + TimerManager(); + + unsigned RegisterTimer(unsigned timeoutMsec, boost::function timerHandler, bool timerRepeat = false); + bool UnregisterTimer(unsigned timerId); + + void Process(); + +protected: + + struct TimerData + { + TimerData(unsigned timerId, unsigned timeoutMsec, boost::function timerHandler, bool timerRepeat) + : id(timerId), msec(timeoutMsec), handler(timerHandler), repeat(timerRepeat) {} + unsigned id; + unsigned msec; + boost::function handler; + bool repeat; + }; + + typedef std::multimap TimerMap; + + unsigned GetNextTimerId(); + +private: + mutable boost::recursive_mutex m_timerMutex; + TimerMap m_timerMap; + unsigned m_curTimerId; + + boost::timers::portable::microsec_timer m_softwareTimer; +}; + +#endif diff --git a/src/net/clientthread.h b/src/net/clientthread.h index 33ea2529..6b74eeac 100644 --- a/src/net/clientthread.h +++ b/src/net/clientthread.h @@ -121,7 +121,7 @@ protected: const ClientContext &GetContext() const; ClientContext &GetContext(); - void SetContextSocket(SOCKET s); + void CreateContextSession(); ClientState &GetState(); void SetState(ClientState &newState); diff --git a/src/net/common/clientstate.cpp b/src/net/common/clientstate.cpp index dc70db60..505b3e4e 100644 --- a/src/net/common/clientstate.cpp +++ b/src/net/common/clientstate.cpp @@ -86,22 +86,7 @@ ClientStateInit::Process(ClientThread &client) if (context.GetServerPort() < 1024) throw ClientException(__FILE__, __LINE__, ERR_SOCK_INVALID_PORT, 0); - client.SetContextSocket(socket(context.GetAddrFamily(), SOCK_STREAM, context.GetProtocol())); - if (!IS_VALID_SOCKET(context.GetSocket())) - throw ClientException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); - - unsigned long mode = 1; - if (IOCTLSOCKET(context.GetSocket(), FIONBIO, &mode) == SOCKET_ERROR) - throw ClientException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); - - // The following calls are optional - the return value is not checked. - int nodelay = 1; - setsockopt(context.GetSocket(), SOL_SOCKET, TCP_NODELAY, (char *)&nodelay, sizeof(nodelay)); - -#ifdef SO_NOSIGPIPE - int nosigpipe = 1; - setsockopt(context.GetSocket(), SOL_SOCKET, SO_NOSIGPIPE, (char *)&nosigpipe, sizeof(nosigpipe)); -#endif + client.CreateContextSession(); if (context.GetUseServerList()) client.SetState(ClientStateStartServerListDownload::Instance()); diff --git a/src/net/common/clientthread.cpp b/src/net/common/clientthread.cpp index 89b5a8cb..f7ef5ec3 100644 --- a/src/net/common/clientthread.cpp +++ b/src/net/common/clientthread.cpp @@ -17,6 +17,7 @@ * 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. * ***************************************************************************/ +#include #include #include #include @@ -40,6 +41,7 @@ #define TEMP_AVATAR_FILENAME "avatar.tmp" using namespace std; +using boost::asio::ip::tcp; class ClientSenderCallback : public SenderCallback, public SessionDataCallback @@ -719,14 +721,29 @@ ClientThread::GetContext() } void -ClientThread::SetContextSocket(SOCKET s) +ClientThread::CreateContextSession() { - GetContext().SetSessionData(boost::shared_ptr(new SessionData( - s, - SESSION_ID_GENERIC, - m_senderThread, - *m_senderCallback, - *m_ioService))); + bool validSocket = false; + // TODO ipv6 + // TODO sctp + try { + boost::shared_ptr newSock(new boost::asio::ip::tcp::socket(*m_ioService, tcp::v4())); + boost::asio::socket_base::non_blocking_io command(true); + newSock->io_control(command); + newSock->set_option(tcp::no_delay(true)); + newSock->set_option(boost::asio::socket_base::keep_alive(true)); + + GetContext().SetSessionData(boost::shared_ptr(new SessionData( + newSock, + SESSION_ID_GENERIC, + m_senderThread, + *m_senderCallback))); + validSocket = true; + } catch (...) + { + } + if (!validSocket) + throw ClientException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); } ClientState & diff --git a/src/net/common/connectdata.cpp b/src/net/common/connectdata.cpp index f0872a2d..f6b6d19b 100644 --- a/src/net/common/connectdata.cpp +++ b/src/net/common/connectdata.cpp @@ -23,7 +23,7 @@ using namespace std; ConnectData::ConnectData() -: m_sockfd(INVALID_SOCKET), m_peerAddrSize(sizeof(m_peerAddr)) +: m_sockfd(INVALID_SOCKET), m_peerAddrSize(sizeof(struct sockaddr_storage)) { memset(&m_peerAddr, 0, sizeof(m_peerAddr)); } @@ -34,11 +34,3 @@ ConnectData::~ConnectData() CLOSESOCKET(m_sockfd); } -SOCKET -ConnectData::ReleaseSocket() -{ - SOCKET tmpSock = m_sockfd; - m_sockfd = INVALID_SOCKET; - return tmpSock; -} - diff --git a/src/net/common/receiverhelper.cpp b/src/net/common/receiverhelper.cpp index 60e626b6..711cf514 100644 --- a/src/net/common/receiverhelper.cpp +++ b/src/net/common/receiverhelper.cpp @@ -77,7 +77,7 @@ ReceiverHelper::Recv(SOCKET sock, ReceiveBuffer &buf) else { buf.recvBufUsed += bytesRecvd; - InternalGetPackets(buf); + ScanPackets(buf); } } } @@ -92,7 +92,7 @@ ReceiverHelper::Recv(SOCKET sock, ReceiveBuffer &buf) } void -ReceiverHelper::InternalGetPackets(ReceiveBuffer &buf) +ReceiverHelper::ScanPackets(ReceiveBuffer &buf) { bool dataAvailable = true; do diff --git a/src/net/common/serveracceptthread.cpp b/src/net/common/serveracceptthread.cpp index 5e798eb8..5cb65100 100644 --- a/src/net/common/serveracceptthread.cpp +++ b/src/net/common/serveracceptthread.cpp @@ -17,28 +17,22 @@ * 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. * ***************************************************************************/ -#include #include -#include -#include -#include #include #include #include -#include #include -#include -#define ACCEPT_TIMEOUT_MSEC 50 #define NET_SERVER_LISTEN_BACKLOG 5 using namespace std; +using boost::asio::ip::tcp; -ServerAcceptThread::ServerAcceptThread(ServerCallback &serverCallback) -: m_serverCallback(serverCallback) +ServerAcceptThread::ServerAcceptThread(ServerCallback &serverCallback, boost::shared_ptr ioService) +: m_ioService(ioService), m_serverCallback(serverCallback) { - m_context.reset(new ServerContext); + m_acceptor.reset(new tcp::acceptor(*m_ioService)); } ServerAcceptThread::~ServerAcceptThread() @@ -46,24 +40,85 @@ ServerAcceptThread::~ServerAcceptThread() } void -ServerAcceptThread::Init(unsigned serverPort, bool ipv6, bool sctp, const string &pwd, const string &logDir, boost::shared_ptr lobbyThread) +ServerAcceptThread::Listen(unsigned serverPort, bool ipv6, bool sctp, const string &pwd, const string &logDir, boost::shared_ptr lobbyThread) { - if (IsRunning()) - { - assert(false); - return; - } - - ServerContext &context = GetContext(); - - context.SetProtocol(sctp ? SOCKET_IPPROTO_SCTP : 0); - // If a "dual stack" is available, the pokerth server always uses ipv6. - // In this case, ipv4 requests will be mapped to ipv6. - context.SetAddrFamily(socket_has_dual_stack() ? AF_INET6 : (ipv6 ? AF_INET6 : AF_INET)); - context.SetServerPort(serverPort); - m_lobbyThread = lobbyThread; - GetLobbyThread().Init(pwd, logDir); + + try + { + InternalListen(serverPort, ipv6, sctp); + } + catch (const PokerTHException &e) + { + LOG_ERROR(e.what()); + GetCallback().SignalNetServerError(e.GetErrorId(), e.GetOsErrorCode()); + } + catch (...) + { + // This is probably an asio exception. Assume that bind failed, + // which is the most frequent case. + LOG_ERROR("Cannot bind/listen on TCP port."); + GetCallback().SignalNetServerError(ERR_SOCK_BIND_FAILED, 0); + } +} + +void +ServerAcceptThread::InternalListen(unsigned serverPort, bool ipv6, bool sctp) +{ + if (serverPort < 1024) + throw ServerException(__FILE__, __LINE__, ERR_SOCK_INVALID_PORT, 0); + + // TODO consider sctp + // Prepare Listen. + if (ipv6) + m_endpoint.reset(new tcp::endpoint(tcp::v6(), serverPort)); + else + m_endpoint.reset(new tcp::endpoint(tcp::v4(), serverPort)); + + // TODO use non blocking I/O + //boost::asio::socket_base::non_blocking_io command(true); + //m_acceptor->io_control(command); + m_acceptor->open(m_endpoint->protocol()); + m_acceptor->set_option(tcp::acceptor::reuse_address(true)); + if (ipv6) // In IPv6 mode: Be compatible with IPv4. + m_acceptor->set_option(boost::asio::ip::v6_only(false)); + m_acceptor->bind(*m_endpoint); + m_acceptor->listen(NET_SERVER_LISTEN_BACKLOG); + + // Start first asynchronous Accept. + boost::shared_ptr newSocket(new tcp::socket(*m_ioService)); + m_acceptor->async_accept( + *newSocket, + boost::bind(&ServerAcceptThread::HandleAccept, this, newSocket, + boost::asio::placeholders::error) + ); +} + +void +ServerAcceptThread::HandleAccept(boost::shared_ptr acceptedSocket, + const boost::system::error_code& error) +{ + if (!error) + { + boost::asio::socket_base::non_blocking_io command(true); + acceptedSocket->io_control(command); + acceptedSocket->set_option(tcp::no_delay(true)); + acceptedSocket->set_option(boost::asio::socket_base::keep_alive(true)); + GetLobbyThread().AddConnection(acceptedSocket); + + boost::shared_ptr newSocket(new tcp::socket(*m_ioService)); + m_acceptor->async_accept( + *newSocket, + boost::bind(&ServerAcceptThread::HandleAccept, this, newSocket, + boost::asio::placeholders::error) + ); + } + else + { + // Accept failed. This is a fatal error. + LOG_ERROR("In boost::asio handler: Accept failed."); + GetCallback().SignalNetServerError(ERR_SOCK_ACCEPT_FAILED, 0); + } } ServerCallback & @@ -72,166 +127,6 @@ ServerAcceptThread::GetCallback() return m_serverCallback; } -void -ServerAcceptThread::Main() -{ - try - { - Listen(); - - while (!ShouldTerminate()) - { - // The main server thread is simple. It only accepts connections. - AcceptLoop(); - } - } catch (const PokerTHException &e) - { - GetCallback().SignalNetServerError(e.GetErrorId(), e.GetOsErrorCode()); - LOG_ERROR(e.what()); - } -} - -void -ServerAcceptThread::Listen() -{ - ServerContext &context = GetContext(); - - if (context.GetServerPort() < 1024) - throw ServerException(__FILE__, __LINE__, ERR_SOCK_INVALID_PORT, 0); - -#ifdef _WIN32 - context.SetSocket(WSASocket(context.GetAddrFamily(), SOCK_STREAM, context.GetProtocol(), 0, 0, WSA_FLAG_OVERLAPPED)); -#else - context.SetSocket(socket(context.GetAddrFamily(), SOCK_STREAM, context.GetProtocol())); -#endif - - if (!IS_VALID_SOCKET(context.GetSocket())) - throw ServerException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); - - unsigned long mode = 1; - if (IOCTLSOCKET(context.GetSocket(), FIONBIO, &mode) == SOCKET_ERROR) - throw ServerException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); - - // The following three calls are optional. If they fail, we don't care. - int reuse = 1; - setsockopt(context.GetSocket(), SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(reuse)); - int nodelay = 1; - setsockopt(context.GetSocket(), SOL_SOCKET, TCP_NODELAY, (char *)&nodelay, sizeof(nodelay)); - // Enable dual-stack socket on Windows Vista. - if (context.GetAddrFamily() == AF_INET6) - { - int ipv6only = 0; - setsockopt(context.GetSocket(), IPPROTO_IPV6, IPV6_V6ONLY, (char *)&ipv6only, sizeof(ipv6only)); - } - - context.GetServerSockaddr()->ss_family = context.GetAddrFamily(); - - const char *localAddr = (context.GetAddrFamily() == AF_INET6) ? "::0" : "0.0.0.0"; - if (!socket_string_to_addr( - localAddr, - context.GetAddrFamily(), - (struct sockaddr *)context.GetServerSockaddr(), - context.GetServerSockaddrSize())) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_SET_ADDR_FAILED, 0); - } - if (!socket_set_port( - context.GetServerPort(), - context.GetAddrFamily(), - (struct sockaddr *)context.GetServerSockaddr(), - context.GetServerSockaddrSize())) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_SET_PORT_FAILED, 0); - } - - if (!IS_VALID_BIND(bind( - context.GetSocket(), - (const struct sockaddr *)context.GetServerSockaddr(), - context.GetServerSockaddrSize()))) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_BIND_FAILED, SOCKET_ERRNO()); - } - - if (!IS_VALID_LISTEN(listen(context.GetSocket(), NET_SERVER_LISTEN_BACKLOG))) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_LISTEN_FAILED, SOCKET_ERRNO()); - } -} - -void -ServerAcceptThread::AcceptLoop() -{ - ServerContext &context = GetContext(); - - fd_set readSet; - struct timeval timeout; - - FD_ZERO(&readSet); - FD_SET(context.GetSocket(), &readSet); - - timeout.tv_sec = 0; - timeout.tv_usec = ACCEPT_TIMEOUT_MSEC * 1000; - int selectResult = select(context.GetSocket() + 1, &readSet, NULL, NULL, &timeout); - if (!IS_VALID_SELECT(selectResult)) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_SELECT_FAILED, SOCKET_ERRNO()); - } - if (selectResult > 0) // accept is possible - { - boost::shared_ptr tmpData(new ConnectData); - tmpData->SetSocket(accept(context.GetSocket(), NULL, NULL)); - - if (!IS_VALID_SOCKET(tmpData->GetSocket())) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_ACCEPT_FAILED, SOCKET_ERRNO()); - } - unsigned long mode = 1; - if (IOCTLSOCKET(tmpData->GetSocket(), FIONBIO, &mode) == SOCKET_ERROR) - { - throw ServerException(__FILE__, __LINE__, ERR_SOCK_CREATION_FAILED, SOCKET_ERRNO()); - } - - // Retrieve peer address. - socklen_t addrLen = (socklen_t)context.GetServerSockaddrSize(); - if (getpeername(tmpData->GetSocket(), tmpData->GetPeerAddr(), &addrLen) != 0) - { - // Something went wrong with the connection, just continue (socket will be closed). - LOG_ERROR("getpeername() failed: " << SOCKET_ERRNO()); - } - else - { - // Set the size of the peer address. - tmpData->SetPeerAddrSize(addrLen); - - // Optional calls - don't check return value. - // Enable keepalive - won't be of much use but better than nothing. - int keepalive = 1; - setsockopt(tmpData->GetSocket(), SOL_SOCKET, SO_KEEPALIVE, (char *)&keepalive, sizeof(keepalive)); - - #ifdef SO_NOSIGPIPE - int nosigpipe = 1; - setsockopt(tmpData->GetSocket(), SOL_SOCKET, SO_NOSIGPIPE, (char *)&nosigpipe, sizeof(nosigpipe)); - #endif - - GetLobbyThread().AddConnection(tmpData); - } - } -} - -const ServerContext & -ServerAcceptThread::GetContext() const -{ - assert(m_context.get()); - return *m_context; -} - -ServerContext & -ServerAcceptThread::GetContext() -{ - assert(m_context.get()); - return *m_context; -} - ServerLobbyThread & ServerAcceptThread::GetLobbyThread() { diff --git a/src/net/common/servercontext.cpp b/src/net/common/servercontext.cpp index f4e7746d..44b4e35a 100644 --- a/src/net/common/servercontext.cpp +++ b/src/net/common/servercontext.cpp @@ -20,26 +20,23 @@ #include ServerContext::ServerContext() -: m_sockfd(INVALID_SOCKET), m_protocol(0), m_addrFamily(AF_INET), m_serverPort(0) +: m_protocol(0), m_addrFamily(AF_INET), m_serverPort(0) { - bzero(&m_serverSockaddr, sizeof(m_serverSockaddr)); } ServerContext::~ServerContext() { - if (m_sockfd != INVALID_SOCKET) - CLOSESOCKET(m_sockfd); } SOCKET ServerContext::GetSocket() const { - return m_sockfd; + return m_sock; } void -ServerContext::SetSocket(SOCKET sockfd) +ServerContext::SetSocket(SOCKET sock) { - m_sockfd = sockfd; + m_sock = sock; } diff --git a/src/net/common/serverlobbythread.cpp b/src/net/common/serverlobbythread.cpp index 669ad4d9..a4b20755 100644 --- a/src/net/common/serverlobbythread.cpp +++ b/src/net/common/serverlobbythread.cpp @@ -56,6 +56,7 @@ #define SERVER_STATISTICS_STR_CUR_PLAYERS "CurPlayersLoggedIn" using namespace std; +using boost::asio::ip::tcp; class ServerSenderCallback : public SenderCallback, public SessionDataCallback @@ -80,12 +81,12 @@ private: }; -ServerLobbyThread::ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig, AvatarManager &avatarManager) -: m_curBanId(0), m_gui(gui), m_avatarManager(avatarManager), m_playerConfig(playerConfig), +ServerLobbyThread::ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig, AvatarManager &avatarManager, + boost::shared_ptr ioService) +: m_ioService(ioService), m_curBanId(0), m_gui(gui), m_avatarManager(avatarManager), m_playerConfig(playerConfig), m_curGameId(0), m_curUniquePlayerId(0), m_curSessionId(INVALID_SESSION + 1), m_statDataChanged(false), m_startTime(boost::posix_time::second_clock::local_time()) { - m_ioService.reset(new boost::asio::io_service()); m_senderCallback.reset(new ServerSenderCallback(*this)); m_sender.reset(new SenderThread(*m_senderCallback, m_ioService)); m_receiver.reset(new ReceiverHelper); @@ -114,10 +115,10 @@ ServerLobbyThread::Init(const string &pwd, const string &logDir) } void -ServerLobbyThread::AddConnection(boost::shared_ptr data) +ServerLobbyThread::AddConnection(boost::shared_ptr sock) { boost::mutex::scoped_lock lock(m_connectQueueMutex); - m_connectQueue.push_back(data); + m_connectQueue.push_back(sock); } void @@ -442,6 +443,18 @@ ServerLobbyThread::Main() { try { + m_timerManager.RegisterTimer( + SERVER_CHECK_SESSION_TIMEOUTS_INTERVAL_MSEC, + boost::bind(&ServerLobbyThread::TimerCheckSessionTimeouts, this), + true); + m_timerManager.RegisterTimer( + SERVER_CACHE_CLEANUP_INTERVAL_SEC * 1000, + boost::bind(&ServerLobbyThread::TimerCleanupAvatarCache, this), + true); + m_timerManager.RegisterTimer( + SERVER_SAVE_STATISTICS_INTERVAL_SEC * 1000, + boost::bind(&ServerLobbyThread::TimerSaveStatisticsFile, this), + true); m_sender->Start(); while (!ShouldTerminate()) @@ -450,22 +463,17 @@ ServerLobbyThread::Main() NewConnectionLoop(); // Process re-added sessions. NewSessionLoop(); - // Main loop. - ProcessLoop(); // Remove games. RemoveGameLoop(); // Kick players. RemovePlayerLoop(); // Resubscribe Lobby Messages if needed. ResubscribeLobbyMsgLoop(); - // Check session timeouts. - CheckSessionTimeoutsLoop(); // Update avatar limitation lock. UpdateAvatarClientTimerLoop(); - // Cleanup cache. - CleanupAvatarCache(); - // Save statistics if needed. - SaveStatisticsFile(); + // Process timers. + m_timerManager.Process(); + Thread::Msleep(10); } } catch (const PokerTHException &e) { @@ -486,70 +494,104 @@ ServerLobbyThread::Main() } void -ServerLobbyThread::ProcessLoop() +ServerLobbyThread::HandleRead(SessionId sessionId, const boost::system::error_code& error, size_t bytesRead) { - // Wait for data. - SessionWrapper session = m_sessionManager.Select(RECV_TIMEOUT_MSEC); - - if (session.sessionData.get()) + SessionWrapper session = m_sessionManager.GetSessionById(sessionId); + if (!session.sessionData) + session = m_gameSessionManager.GetSessionById(sessionId); + if (session.sessionData) { - boost::shared_ptr packet; - try + unsigned gameId = session.sessionData->GetGameId(); + if (!error) { - // Receive the next packet. - packet = GetReceiver().Recv(session.sessionData->GetSocket(), session.sessionData->GetReceiveBuffer()); - } catch (const NetException &) - { - // On error: Close this session. - CloseSession(session); - return; + + ReceiveBuffer &buf = session.sessionData->GetReceiveBuffer(); + buf.recvBufUsed += bytesRead; + GetReceiver().ScanPackets(buf); + + while (!buf.receivedPackets.empty()) + { + boost::shared_ptr packet = buf.receivedPackets.front(); + buf.receivedPackets.pop_front(); + if (game) + game->HandlePacket(session, packet); + else + HandlePacket(session, packet); + } + session.sessionData->GetAsioSocket()->async_read_some( + boost::asio::buffer(buf.recvBuf + buf.recvBufUsed, RECV_BUF_SIZE - buf.recvBufUsed), + boost::bind( + &ServerLobbyThread::HandleRead, + this, + sessionId, + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); } - if (!packet.get()) - LOG_VERBOSE("Select successful but no packet received for session #" << session.sessionData->GetId() << "."); else { - if (packet->IsClientActivity()) - session.sessionData->ResetActivityTimer(); + // On error: Close this session. + boost::shared_ptr game; + if (gameId) + { + GameMap::iterator pos = m_gameMap.find(joinGameData.gameId); - if (session.sessionData->GetState() == SessionData::Init) - { - if (packet->ToNetPacketInit()) - HandleNetPacketInit(session, *packet->ToNetPacketInit()); - else if (packet->ToNetPacketAvatarHeader()) - HandleNetPacketAvatarHeader(session, *packet->ToNetPacketAvatarHeader()); - else if (packet->ToNetPacketUnknownAvatar()) - HandleNetPacketUnknownAvatar(session, *packet->ToNetPacketUnknownAvatar()); - else - SessionError(session, ERR_SOCK_INVALID_STATE); - } - else if (session.sessionData->GetState() == SessionData::ReceivingAvatar) - { - if (packet->ToNetPacketAvatarFile()) - HandleNetPacketAvatarFile(session, *packet->ToNetPacketAvatarFile()); - else if (packet->ToNetPacketAvatarEnd()) - HandleNetPacketAvatarEnd(session, *packet->ToNetPacketAvatarEnd()); - else - SessionError(session, ERR_SOCK_INVALID_STATE); + if (pos != m_gameMap.end()) + game = pos->second; } + if (game) + game->ErrorRemoveSession(session); else - { - if (packet->ToNetPacketRetrievePlayerInfo()) - HandleNetPacketRetrievePlayerInfo(session, *packet->ToNetPacketRetrievePlayerInfo()); - else if (packet->ToNetPacketRetrieveAvatar()) - HandleNetPacketRetrieveAvatar(session, *packet->ToNetPacketRetrieveAvatar()); - else if (packet->ToNetPacketResetTimeout()) - {} - else if (packet->ToNetPacketUnsubscribeGameList()) - session.sessionData->ResetWantsLobbyMsg(); - else if (packet->ToNetPacketResubscribeGameList()) - InternalResubscribeMsg(session); - else if (packet->ToNetPacketCreateGame()) - HandleNetPacketCreateGame(session, *packet->ToNetPacketCreateGame()); - else if (packet->ToNetPacketJoinGame()) - HandleNetPacketJoinGame(session, *packet->ToNetPacketJoinGame()); - else - SessionError(session, ERR_SOCK_INVALID_STATE); - } + CloseSession(session); + } + } +} + +void +ServerLobbyThread::HandlePacket(SessionWrapper session, boost::shared_ptr packet) +{ + if (session.sessionData && packet) + { + if (packet->IsClientActivity()) + session.sessionData->ResetActivityTimer(); + + if (session.sessionData->GetState() == SessionData::Init) + { + if (packet->ToNetPacketInit()) + HandleNetPacketInit(session, *packet->ToNetPacketInit()); + else if (packet->ToNetPacketAvatarHeader()) + HandleNetPacketAvatarHeader(session, *packet->ToNetPacketAvatarHeader()); + else if (packet->ToNetPacketUnknownAvatar()) + HandleNetPacketUnknownAvatar(session, *packet->ToNetPacketUnknownAvatar()); + else + SessionError(session, ERR_SOCK_INVALID_STATE); + } + else if (session.sessionData->GetState() == SessionData::ReceivingAvatar) + { + if (packet->ToNetPacketAvatarFile()) + HandleNetPacketAvatarFile(session, *packet->ToNetPacketAvatarFile()); + else if (packet->ToNetPacketAvatarEnd()) + HandleNetPacketAvatarEnd(session, *packet->ToNetPacketAvatarEnd()); + else + SessionError(session, ERR_SOCK_INVALID_STATE); + } + else + { + if (packet->ToNetPacketRetrievePlayerInfo()) + HandleNetPacketRetrievePlayerInfo(session, *packet->ToNetPacketRetrievePlayerInfo()); + else if (packet->ToNetPacketRetrieveAvatar()) + HandleNetPacketRetrieveAvatar(session, *packet->ToNetPacketRetrieveAvatar()); + else if (packet->ToNetPacketResetTimeout()) + {} + else if (packet->ToNetPacketUnsubscribeGameList()) + session.sessionData->ResetWantsLobbyMsg(); + else if (packet->ToNetPacketResubscribeGameList()) + InternalResubscribeMsg(session); + else if (packet->ToNetPacketCreateGame()) + HandleNetPacketCreateGame(session, *packet->ToNetPacketCreateGame()); + else if (packet->ToNetPacketJoinGame()) + HandleNetPacketJoinGame(session, *packet->ToNetPacketJoinGame()); + else + SessionError(session, ERR_SOCK_INVALID_STATE); } } } @@ -923,7 +965,7 @@ void ServerLobbyThread::NewConnectionLoop() { // Handle one incoming connection at a time. - boost::shared_ptr tmpData; + boost::shared_ptr tmpData; { boost::mutex::scoped_lock lock(m_connectQueueMutex); if (!m_connectQueue.empty()) @@ -1020,18 +1062,6 @@ ServerLobbyThread::ResubscribeLobbyMsgLoop() } } -void -ServerLobbyThread::CheckSessionTimeoutsLoop() -{ - if (m_checkSessionTimeoutsTimer.elapsed().total_milliseconds() >= SERVER_CHECK_SESSION_TIMEOUTS_INTERVAL_MSEC) - { - m_sessionManager.ForEach(boost::bind(&ServerLobbyThread::InternalCheckSessionTimeouts, boost::ref(*this), _1)); - m_gameSessionManager.ForEach(boost::bind(&ServerLobbyThread::InternalCheckSessionTimeouts, boost::ref(*this), _1)); - m_checkSessionTimeoutsTimer.reset(); - m_checkSessionTimeoutsTimer.start(); - } -} - void ServerLobbyThread::UpdateAvatarClientTimerLoop() { @@ -1051,17 +1081,21 @@ ServerLobbyThread::UpdateAvatarClientTimerLoop() } void -ServerLobbyThread::CleanupAvatarCache() +ServerLobbyThread::TimerCheckSessionTimeouts() { - // Only act on timer and if there are no sessions. - if (m_cacheCleanupTimer.elapsed().total_seconds() >= SERVER_CACHE_CLEANUP_INTERVAL_SEC - && !m_sessionManager.HasSessions() && !m_gameSessionManager.HasSessions()) + m_sessionManager.ForEach(boost::bind(&ServerLobbyThread::InternalCheckSessionTimeouts, boost::ref(*this), _1)); + m_gameSessionManager.ForEach(boost::bind(&ServerLobbyThread::InternalCheckSessionTimeouts, boost::ref(*this), _1)); +} + +void +ServerLobbyThread::TimerCleanupAvatarCache() +{ + // Only act if there are no sessions. + if (!m_sessionManager.HasSessions() && !m_gameSessionManager.HasSessions()) { LOG_VERBOSE("Cleaning up avatar cache."); m_avatarManager.RemoveOldAvatarCacheEntries(); - m_cacheCleanupTimer.reset(); - m_cacheCleanupTimer.start(); } } @@ -1169,7 +1203,7 @@ ServerLobbyThread::TerminateGames() } void -ServerLobbyThread::HandleNewConnection(boost::shared_ptr connData) +ServerLobbyThread::HandleNewConnection(boost::shared_ptr sock) { // Create a random session id. // This id can be used to reconnect to the server if the connection was lost. @@ -1182,20 +1216,38 @@ ServerLobbyThread::HandleNewConnection(boost::shared_ptr connData) //} // Create a new session. - boost::shared_ptr sessionData(new SessionData(connData->ReleaseSocket(), m_curSessionId++, m_sender, *m_senderCallback, *m_ioService)); + boost::shared_ptr sessionData(new SessionData(sock, m_curSessionId++, m_sender, *m_senderCallback)); m_sessionManager.AddSession(sessionData); LOG_VERBOSE("Accepted connection - session #" << sessionData->GetId() << "."); + bool hasClientIp = false; if (m_sessionManager.GetRawSessionCount() <= SERVER_MAX_NUM_SESSIONS) { - char tmpAddress[MAX_ADDR_STRING_LEN]; - // Only consider address, set port to zero. - if (socket_set_port(0, connData->GetPeerAddr()->sa_family, connData->GetPeerAddr(), connData->GetPeerAddrSize()) - && socket_addr_to_string(connData->GetPeerAddr(), connData->GetPeerAddrSize(), connData->GetPeerAddr()->sa_family, tmpAddress, sizeof(tmpAddress))) + boost::system::error_code errCode; + tcp::endpoint clientEndpoint = sock->remote_endpoint(errCode); + if (!errCode) { - tmpAddress[sizeof(tmpAddress) - 1] = 0; // paranoia - sessionData->SetClientAddr(tmpAddress); + string ipAddress = clientEndpoint.address().to_string(errCode); + if (!errCode && !ipAddress.empty()) + { + sessionData->SetClientAddr(ipAddress); + hasClientIp = true; + sock->async_read_some( + boost::asio::buffer(sessionData->GetReceiveBuffer().recvBuf, RECV_BUF_SIZE), + boost::bind( + &ServerLobbyThread::HandleRead, + this, + sessionData->GetId(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); + } + } + if (!hasClientIp) + { + // We do not accept sessions if we cannot + // retrieve the client address. + SessionError(SessionWrapper(sessionData, boost::shared_ptr()), ERR_NET_INVALID_SESSION); } } else @@ -1386,30 +1438,23 @@ ServerLobbyThread::ReadStatisticsFile() } void -ServerLobbyThread::SaveStatisticsFile() +ServerLobbyThread::TimerSaveStatisticsFile() { - if (m_saveStatisticsTimer.elapsed().total_seconds() >= SERVER_SAVE_STATISTICS_INTERVAL_SEC) + LOG_VERBOSE("Saving statistics."); + boost::mutex::scoped_lock lock(m_statMutex); + if (m_statDataChanged) { - LOG_VERBOSE("Saving statistics."); + ofstream o(m_statisticsFileName.c_str(), ios_base::out | ios_base::trunc); + if (!o.fail()) { - boost::mutex::scoped_lock lock(m_statMutex); - if (m_statDataChanged) - { - ofstream o(m_statisticsFileName.c_str(), ios_base::out | ios_base::trunc); - if (!o.fail()) - { - o << SERVER_STATISTICS_STR_TOTAL_PLAYERS " " << m_statData.totalPlayersEverLoggedIn << endl; - o << SERVER_STATISTICS_STR_TOTAL_GAMES " " << m_statData.totalGamesEverCreated << endl; - o << SERVER_STATISTICS_STR_MAX_PLAYERS " " << m_statData.maxPlayersLoggedIn << endl; - o << SERVER_STATISTICS_STR_MAX_GAMES " " << m_statData.maxGamesOpen << endl; - o << SERVER_STATISTICS_STR_CUR_PLAYERS " " << m_statData.numberOfPlayersOnServer << endl; - o << SERVER_STATISTICS_STR_CUR_GAMES " " << m_statData.numberOfGamesOpen << endl; - m_statDataChanged = false; - } - } + o << SERVER_STATISTICS_STR_TOTAL_PLAYERS " " << m_statData.totalPlayersEverLoggedIn << endl; + o << SERVER_STATISTICS_STR_TOTAL_GAMES " " << m_statData.totalGamesEverCreated << endl; + o << SERVER_STATISTICS_STR_MAX_PLAYERS " " << m_statData.maxPlayersLoggedIn << endl; + o << SERVER_STATISTICS_STR_MAX_GAMES " " << m_statData.maxGamesOpen << endl; + o << SERVER_STATISTICS_STR_CUR_PLAYERS " " << m_statData.numberOfPlayersOnServer << endl; + o << SERVER_STATISTICS_STR_CUR_GAMES " " << m_statData.numberOfGamesOpen << endl; + m_statDataChanged = false; } - m_saveStatisticsTimer.reset(); - m_saveStatisticsTimer.start(); } } diff --git a/src/net/common/servermanager.cpp b/src/net/common/servermanager.cpp index 124df6e3..4e227d71 100644 --- a/src/net/common/servermanager.cpp +++ b/src/net/common/servermanager.cpp @@ -39,6 +39,7 @@ using namespace std; ServerManager::ServerManager(GuiInterface &gui, ConfigFile *config, AvatarManager &avatarManager) : m_gui(gui), m_playerConfig(config), m_avatarManager(avatarManager) { + m_ioService.reset(new boost::asio::io_service()); } ServerManager::~ServerManager() @@ -48,19 +49,19 @@ ServerManager::~ServerManager() void ServerManager::Init(unsigned serverPort, bool ipv6, ServerNetworkMode mode, const string &pwd, const string &logDir, boost::shared_ptr ircThread) { - m_lobbyThread.reset(new ServerLobbyThread(GetGui(), m_playerConfig, m_avatarManager)); + m_lobbyThread.reset(new ServerLobbyThread(GetGui(), m_playerConfig, m_avatarManager, m_ioService)); GetLobbyThread().Init(pwd, logDir); if (mode & NETWORK_MODE_TCP) { - boost::shared_ptr tcpAcceptThread(new ServerAcceptThread(GetGui())); - tcpAcceptThread->Init(serverPort, ipv6, false, pwd, logDir, m_lobbyThread); + boost::shared_ptr tcpAcceptThread(new ServerAcceptThread(GetGui(), m_ioService)); + tcpAcceptThread->Listen(serverPort, ipv6, false, pwd, logDir, m_lobbyThread); m_acceptThreadPool.push_back(tcpAcceptThread); } if (mode & NETWORK_MODE_SCTP) { - boost::shared_ptr sctpAcceptThread(new ServerAcceptThread(GetGui())); - sctpAcceptThread->Init(serverPort, ipv6, true, pwd, logDir, m_lobbyThread); + boost::shared_ptr sctpAcceptThread(new ServerAcceptThread(GetGui(), m_ioService)); + sctpAcceptThread->Listen(serverPort, ipv6, true, pwd, logDir, m_lobbyThread); m_acceptThreadPool.push_back(sctpAcceptThread); } m_ircThread = ircThread; @@ -269,7 +270,7 @@ ServerManager::RunAll() if (m_ircThread) m_ircThread->Run(); GetLobbyThread().Run(); - for_each(m_acceptThreadPool.begin(), m_acceptThreadPool.end(), boost::mem_fn(&ServerAcceptThread::Run)); +// for_each(m_acceptThreadPool.begin(), m_acceptThreadPool.end(), boost::mem_fn(&ServerAcceptThread::Run)); } void @@ -298,7 +299,7 @@ ServerManager::SignalTerminationAll() if (m_ircThread) m_ircThread->SignalTermination(); GetLobbyThread().SignalTermination(); - for_each(m_acceptThreadPool.begin(), m_acceptThreadPool.end(), boost::mem_fn(&ServerAcceptThread::SignalTermination)); +// for_each(m_acceptThreadPool.begin(), m_acceptThreadPool.end(), boost::mem_fn(&ServerAcceptThread::SignalTermination)); } bool @@ -307,16 +308,15 @@ ServerManager::JoinAll(bool wait) if (m_ircThread) m_ircThread->Join(wait ? NET_ADMIN_IRC_TERMINATE_TIMEOUT_MSEC : 0); bool lobbyThreadTerminated = GetLobbyThread().Join(wait ? NET_LOBBY_THREAD_TERMINATE_TIMEOUT_MSEC : 0); - bool allAcceptThreadsTerminated = true; - AcceptThreadList::iterator i = m_acceptThreadPool.begin(); +/* AcceptThreadList::iterator i = m_acceptThreadPool.begin(); AcceptThreadList::iterator end = m_acceptThreadPool.end(); while (i != end) { if (!(*i)->Join(wait ? NET_ACCEPT_THREAD_TERMINATE_TIMEOUT_MSEC : 0)) allAcceptThreadsTerminated = false; ++i; - } - return lobbyThreadTerminated || allAcceptThreadsTerminated; + }*/ + return lobbyThreadTerminated; } ServerLobbyThread & diff --git a/src/net/common/sessiondata.cpp b/src/net/common/sessiondata.cpp index 934d4f02..5d8b6b10 100644 --- a/src/net/common/sessiondata.cpp +++ b/src/net/common/sessiondata.cpp @@ -20,13 +20,12 @@ #include #include -SessionData::SessionData(SOCKET sockfd, SessionId id, boost::shared_ptr sender, SessionDataCallback &cb, boost::asio::io_service &ioService) -: m_id(id), m_state(SessionData::Init), m_readyFlag(false), +SessionData::SessionData(boost::shared_ptr sock, SessionId id, + boost::shared_ptr sender, SessionDataCallback &cb) +: m_socket(sock), m_id(id), m_gameId(0), m_state(SessionData::Init), m_readyFlag(false), m_wantsLobbyMsg(true), m_activityTimeoutNoticeSent(false), m_callback(cb), m_maxNumPlayers(0) { - m_socket.reset(new boost::asio::ip::tcp::socket( - ioService, boost::asio::ip::tcp::v6(), sockfd)); m_sender = sender; } @@ -42,6 +41,20 @@ SessionData::GetId() const return m_id; } +unsigned +SessionData::GetGameId() const +{ + boost::mutex::scoped_lock lock(m_dataMutex); + return m_gameId; +} + +void +SessionData::SetGameId(unsigned gameId) +{ + boost::mutex::scoped_lock lock(m_dataMutex); + m_gameId = gameId; +} + SessionData::State SessionData::GetState() const { diff --git a/src/net/connectdata.h b/src/net/connectdata.h index 0a42865d..f882e032 100644 --- a/src/net/connectdata.h +++ b/src/net/connectdata.h @@ -34,16 +34,11 @@ public: void SetSocket(SOCKET sockfd) {m_sockfd = sockfd;} - SOCKET ReleaseSocket(); - - struct sockaddr *GetPeerAddr() {return (sockaddr *)&m_peerAddr;} - int GetPeerAddrSize() const {return m_peerAddrSize;} - - void SetPeerAddrSize(int addrSize) {m_peerAddrSize = addrSize;} + SOCKET GetSocket(); private: - SOCKET m_sockfd; + SOCKET m_sockfd; struct sockaddr_storage m_peerAddr; int m_peerAddrSize; }; diff --git a/src/net/netcontext.h b/src/net/netcontext.h index 496c90c5..b55da783 100644 --- a/src/net/netcontext.h +++ b/src/net/netcontext.h @@ -21,8 +21,7 @@ #ifndef _NETCONTEXT_H_ #define _NETCONTEXT_H_ -#include -#include +#include class NetContext diff --git a/src/net/receiverhelper.h b/src/net/receiverhelper.h index e955316f..1aa70bfc 100644 --- a/src/net/receiverhelper.h +++ b/src/net/receiverhelper.h @@ -39,8 +39,7 @@ public: boost::shared_ptr Recv(SOCKET sock, ReceiveBuffer &buf); -protected: - void InternalGetPackets(ReceiveBuffer &buf); + void ScanPackets(ReceiveBuffer &buf); }; #endif diff --git a/src/net/serveracceptthread.h b/src/net/serveracceptthread.h index 5e5eacdc..003c953e 100644 --- a/src/net/serveracceptthread.h +++ b/src/net/serveracceptthread.h @@ -21,51 +21,39 @@ #ifndef _SERVERACCEPTTHREAD_H_ #define _SERVERACCEPTTHREAD_H_ +#include #include -#include - #include #include -#define NET_ACCEPT_THREAD_TERMINATE_TIMEOUT_MSEC 2000 - -class ServerContext; class ServerLobbyThread; -class ServerSenderCallback; -class SenderThread; -class ConfigFile; -class AvatarManager; -class IrcThread; -struct GameData; -class ServerAcceptThread : public Thread +class ServerAcceptThread { public: - ServerAcceptThread(ServerCallback &serverCallback); + ServerAcceptThread(ServerCallback &serverCallback, boost::shared_ptr ioService); virtual ~ServerAcceptThread(); // Set the parameters. - void Init(unsigned serverPort, bool ipv6, bool sctp, const std::string &pwd, const std::string &logDir, boost::shared_ptr lobbyThread); - - ServerCallback &GetCallback(); + void Listen(unsigned serverPort, bool ipv6, bool sctp, const std::string &pwd, const std::string &logDir, + boost::shared_ptr lobbyThread); protected: - // Main function of the thread. - virtual void Main(); + void InternalListen(unsigned serverPort, bool ipv6, bool sctp); + void HandleAccept(boost::shared_ptr acceptedSocket, + const boost::system::error_code& error); - void Listen(); - void AcceptLoop(); - - const ServerContext &GetContext() const; - ServerContext &GetContext(); + ServerCallback &GetCallback(); ServerLobbyThread &GetLobbyThread(); private: + boost::shared_ptr m_ioService; + boost::shared_ptr m_acceptor; + boost::shared_ptr m_endpoint; ServerCallback &m_serverCallback; - boost::shared_ptr m_context; boost::shared_ptr m_lobbyThread; }; diff --git a/src/net/servercontext.h b/src/net/servercontext.h index 61cbbc21..54071f88 100644 --- a/src/net/servercontext.h +++ b/src/net/servercontext.h @@ -32,7 +32,7 @@ public: virtual SOCKET GetSocket() const; - void SetSocket(SOCKET sockfd); + void SetSocket(SOCKET sock); int GetProtocol() const {return m_protocol;} @@ -46,20 +46,12 @@ public: {return m_serverPort;} void SetServerPort(unsigned serverPort) {m_serverPort = serverPort;} - const sockaddr_storage *GetServerSockaddr() const - {return &m_serverSockaddr;} - sockaddr_storage *GetServerSockaddr() - {return &m_serverSockaddr;} - - int GetServerSockaddrSize() const - {return m_addrFamily == AF_INET6 ? sizeof(sockaddr_in6) : sizeof(sockaddr_in);} private: - SOCKET m_sockfd; + SOCKET m_sock; int m_protocol; int m_addrFamily; unsigned m_serverPort; - sockaddr_storage m_serverSockaddr; }; #endif diff --git a/src/net/serverlobbythread.h b/src/net/serverlobbythread.h index 31340455..f23d7140 100644 --- a/src/net/serverlobbythread.h +++ b/src/net/serverlobbythread.h @@ -21,7 +21,8 @@ #ifndef _SERVERRECVTHREAD_H_ #define _SERVERRECVTHREAD_H_ -#include +#include +#include #include #include #include @@ -48,12 +49,13 @@ class Game; class ServerLobbyThread : public Thread { public: - ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig, AvatarManager &avatarManager); + ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig, AvatarManager &avatarManager, + boost::shared_ptr ioService); virtual ~ServerLobbyThread(); void Init(const std::string &pwd, const std::string &logDir); - void AddConnection(boost::shared_ptr data); + void AddConnection(boost::shared_ptr sock); void ReAddSession(SessionWrapper session, int reason); void MoveSessionToGame(ServerGameThread &game, SessionWrapper session); void RemoveSessionFromGame(SessionWrapper session); @@ -98,7 +100,7 @@ public: protected: - typedef std::deque > ConnectQueue; + typedef std::deque > ConnectQueue; typedef std::deque SessionQueue; typedef std::list SessionList; typedef std::list SessionIdList; @@ -112,7 +114,8 @@ protected: // Main function of the thread. virtual void Main(); - void ProcessLoop(); + void HandleRead(SessionId sessionId, const boost::system::error_code& error, size_t bytesRead); + void HandlePacket(SessionWrapper session, boost::shared_ptr packet); void HandleNetPacketInit(SessionWrapper session, const NetPacketInit &tmpPacket); void HandleNetPacketAvatarHeader(SessionWrapper session, const NetPacketAvatarHeader &tmpPacket); void HandleNetPacketUnknownAvatar(SessionWrapper session, const NetPacketUnknownAvatar &tmpPacket); @@ -129,9 +132,9 @@ protected: void RemoveGameLoop(); void RemovePlayerLoop(); void ResubscribeLobbyMsgLoop(); - void CheckSessionTimeoutsLoop(); void UpdateAvatarClientTimerLoop(); - void CleanupAvatarCache(); + void TimerCheckSessionTimeouts(); + void TimerCleanupAvatarCache(); void InternalAddGame(boost::shared_ptr game); void InternalRemoveGame(boost::shared_ptr game); @@ -140,7 +143,7 @@ protected: void TerminateGames(); - void HandleNewConnection(boost::shared_ptr connData); + void HandleNewConnection(boost::shared_ptr sock); void HandleReAddedSession(SessionWrapper session); void InternalCheckSessionTimeouts(SessionWrapper session); @@ -156,7 +159,7 @@ protected: void BroadcastStatisticsUpdate(const ServerStats &stats); void ReadStatisticsFile(); - void SaveStatisticsFile(); + void TimerSaveStatisticsFile(); ReceiverHelper &GetReceiver(); @@ -174,6 +177,10 @@ protected: private: + boost::shared_ptr m_ioService; + + TimerManager m_timerManager; + ConnectQueue m_connectQueue; mutable boost::mutex m_connectQueueMutex; @@ -224,13 +231,7 @@ private: bool m_statDataChanged; mutable boost::mutex m_statMutex; - boost::timers::portable::microsec_timer m_cacheCleanupTimer; - boost::timers::portable::microsec_timer m_saveStatisticsTimer; - boost::timers::portable::microsec_timer m_checkSessionTimeoutsTimer; - const boost::posix_time::ptime m_startTime; - - boost::shared_ptr m_ioService; }; #endif diff --git a/src/net/servermanager.h b/src/net/servermanager.h index 86e16f71..aeb5f86c 100644 --- a/src/net/servermanager.h +++ b/src/net/servermanager.h @@ -21,6 +21,7 @@ #ifndef _SERVERMANAGER_H_ #define _SERVERMANAGER_H_ +#include #include #include @@ -77,6 +78,7 @@ private: std::string m_ircNick; + boost::shared_ptr m_ioService; boost::shared_ptr m_lobbyThread; boost::shared_ptr m_ircThread; boost::timers::portable::microsec_timer m_ircRestartTimer; diff --git a/src/net/sessiondata.h b/src/net/sessiondata.h index ee792ea8..25f4e66b 100644 --- a/src/net/sessiondata.h +++ b/src/net/sessiondata.h @@ -23,13 +23,13 @@ typedef unsigned SessionId; +#include #include #include #include #include #include #include -#include #define INVALID_SESSION 0 #define SESSION_ID_INIT INVALID_SESSION @@ -42,10 +42,15 @@ class SessionData public: enum State { Init, ReceivingAvatar, Established, Game }; - SessionData(SOCKET sockfd, SessionId id, boost::shared_ptr sender, SessionDataCallback &cb, boost::asio::io_service &ioService); + SessionData(boost::shared_ptr sock, SessionId id, + boost::shared_ptr sender, SessionDataCallback &cb); ~SessionData(); SessionId GetId() const; + + unsigned GetGameId() const; + void SetGameId(unsigned gameId); + State GetState() const; void SetState(State state); @@ -77,6 +82,7 @@ public: private: boost::shared_ptr m_socket; const SessionId m_id; + unsigned m_gameId; State m_state; std::string m_clientAddr; ReceiveBuffer m_receiveBuffer;