sco/pokerth
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Server accept thread is no longer a thread.

Browse Source
This commit is contained in:
lotodore
2009-05-10 14:57:51 +00:00
parent ac4c2e9aef
commit d258bc25d0
7 changed files with 122 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/net/common/serveracceptthread.cpp → src/net/common/serveracceptmanager.cpp
+10 -10
View File
@@ -17,7 +17,7 @@
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
***************************************************************************/
#include <net/serveracceptthread.h>
#include <net/serveracceptmanager.h>
#include <net/serverlobbythread.h>
#include <net/serverexception.h>
#include <net/socket_msg.h>
@@ -29,18 +29,18 @@
using namespace std;
using boost::asio::ip::tcp;
ServerAcceptThread::ServerAcceptThread(ServerCallback &serverCallback, boost::shared_ptr<boost::asio::io_service> ioService)
ServerAcceptManager::ServerAcceptManager(ServerCallback &serverCallback, boost::shared_ptr<boost::asio::io_service> ioService)
: m_ioService(ioService), m_serverCallback(serverCallback)
{
m_acceptor.reset(new tcp::acceptor(*m_ioService));
}
ServerAcceptThread::~ServerAcceptThread()
ServerAcceptManager::~ServerAcceptManager()
{
}
void
ServerAcceptThread::Listen(unsigned serverPort, bool ipv6, bool sctp, const string &pwd, const string &logDir, boost::shared_ptr<ServerLobbyThread> lobbyThread)
ServerAcceptManager::Listen(unsigned serverPort, bool ipv6, bool sctp, const string &pwd, const string &logDir, boost::shared_ptr<ServerLobbyThread> lobbyThread)
{
m_lobbyThread = lobbyThread;
@@ -63,7 +63,7 @@ ServerAcceptThread::Listen(unsigned serverPort, bool ipv6, bool sctp, const stri
}
void
ServerAcceptThread::InternalListen(unsigned serverPort, bool ipv6, bool sctp)
ServerAcceptManager::InternalListen(unsigned serverPort, bool ipv6, bool sctp)
{
if (serverPort < 1024)
throw ServerException(__FILE__, __LINE__, ERR_SOCK_INVALID_PORT, 0);
@@ -89,13 +89,13 @@ ServerAcceptThread::InternalListen(unsigned serverPort, bool ipv6, bool sctp)
boost::shared_ptr<tcp::socket> newSocket(new tcp::socket(*m_ioService));
m_acceptor->async_accept(
*newSocket,
boost::bind(&ServerAcceptThread::HandleAccept, this, newSocket,
boost::bind(&ServerAcceptManager::HandleAccept, this, newSocket,
boost::asio::placeholders::error)
);
}
void
ServerAcceptThread::HandleAccept(boost::shared_ptr<boost::asio::ip::tcp::socket> acceptedSocket,
ServerAcceptManager::HandleAccept(boost::shared_ptr<boost::asio::ip::tcp::socket> acceptedSocket,
const boost::system::error_code& error)
{
if (!error)
@@ -109,7 +109,7 @@ ServerAcceptThread::HandleAccept(boost::shared_ptr<boost::asio::ip::tcp::socket>
boost::shared_ptr<tcp::socket> newSocket(new tcp::socket(*m_ioService));
m_acceptor->async_accept(
*newSocket,
boost::bind(&ServerAcceptThread::HandleAccept, this, newSocket,
boost::bind(&ServerAcceptManager::HandleAccept, this, newSocket,
boost::asio::placeholders::error)
);
}
@@ -122,13 +122,13 @@ ServerAcceptThread::HandleAccept(boost::shared_ptr<boost::asio::ip::tcp::socket>
}
ServerCallback &
ServerAcceptThread::GetCallback()
ServerAcceptManager::GetCallback()
{
return m_serverCallback;
}
ServerLobbyThread &
ServerAcceptThread::GetLobbyThread()
ServerAcceptManager::GetLobbyThread()
{
assert(m_lobbyThread.get());
return *m_lobbyThread;
src/net/common/serverlobbythread.cpp
+91 -107
View File
@@ -36,8 +36,13 @@
#include <boost/bind.hpp>
#define SERVER_MAX_NUM_SESSIONS 512 // Maximum number of idle users in lobby.
#define SERVER_CACHE_CLEANUP_INTERVAL_SEC 86400 // 1 day
#define SERVER_SAVE_STATISTICS_INTERVAL_SEC 60
#define SERVER_CHECK_SESSION_TIMEOUTS_INTERVAL_MSEC 500
#define SERVER_REMOVE_GAME_INTERVAL_MSEC 100
#define SERVER_REMOVE_PLAYER_INTERVAL_MSEC 100
#define SERVER_INIT_AVATAR_CLIENT_LOCK_SEC 30 // Forbid a client to send an additional avatar.
#define SERVER_INIT_SESSION_TIMEOUT_SEC 20
@@ -45,7 +50,6 @@
#define SERVER_SESSION_ACTIVITY_TIMEOUT_SEC 1800 // 30 min, MUST be > SERVER_TIMEOUT_WARNING_REMAINING_SEC
#define SERVER_SESSION_FORCED_TIMEOUT_SEC 86400 // 1 day, should be quite large.
#define SERVER_CHECK_SESSION_TIMEOUTS_INTERVAL_MSEC 500
#define SERVER_STATISTICS_FILE_NAME "server_statistics.log"
#define SERVER_STATISTICS_STR_TOTAL_PLAYERS "TotalNumPlayersLoggedIn"
@@ -94,7 +98,6 @@ ServerLobbyThread::ServerLobbyThread(GuiInterface &gui, ConfigFile *playerConfig
ServerLobbyThread::~ServerLobbyThread()
{
CleanupConnectQueue();
}
void
@@ -117,8 +120,57 @@ ServerLobbyThread::Init(const string &pwd, const string &logDir)
void
ServerLobbyThread::AddConnection(boost::shared_ptr<tcp::socket> sock)
{
boost::mutex::scoped_lock lock(m_connectQueueMutex);
m_connectQueue.push_back(sock);
// Create a random session id.
// This id can be used to reconnect to the server if the connection was lost.
//unsigned sessionId;
// TODO: use randomized method.
//if(!RAND_bytes((unsigned char *)&sessionId, sizeof(sessionId)))
//{
// RAND_pseudo_bytes((unsigned char *)&sessionId, sizeof(sessionId));
//}
// Create a new session.
boost::shared_ptr<SessionData> 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)
{
boost::system::error_code errCode;
tcp::endpoint clientEndpoint = sock->remote_endpoint(errCode);
if (!errCode)
{
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<PlayerData>()), ERR_NET_INVALID_SESSION);
}
}
else
{
// Server is full.
// Gracefully close this session.
SessionError(SessionWrapper(sessionData, boost::shared_ptr<PlayerData>()), ERR_NET_SERVER_FULL);
}
}
void
@@ -443,30 +495,16 @@ 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);
// Register all timers.
RegisterTimers();
// Start send thread.
m_sender->Start();
while (!ShouldTerminate())
{
// Process new connections.
NewConnectionLoop();
// Process re-added sessions.
NewSessionLoop();
// Remove games.
RemoveGameLoop();
// Kick players.
RemovePlayerLoop();
// Resubscribe Lobby Messages if needed.
ResubscribeLobbyMsgLoop();
// Update avatar limitation lock.
@@ -489,8 +527,36 @@ ServerLobbyThread::Main()
// Stop sender thread.
m_sender->SignalStop();
m_sender->WaitStop();
}
CleanupConnectQueue();
void
ServerLobbyThread::RegisterTimers()
{
// Remove closed games.
m_timerManager.RegisterTimer(
SERVER_REMOVE_GAME_INTERVAL_MSEC,
boost::bind(&ServerLobbyThread::TimerRemoveGame, this),
true);
// Remove inactive/kicked players.
m_timerManager.RegisterTimer(
SERVER_REMOVE_PLAYER_INTERVAL_MSEC,
boost::bind(&ServerLobbyThread::TimerRemovePlayer, this),
true);
// Check the timeout of sessions which have not been initialised.
m_timerManager.RegisterTimer(
SERVER_CHECK_SESSION_TIMEOUTS_INTERVAL_MSEC,
boost::bind(&ServerLobbyThread::TimerCheckSessionTimeouts, this),
true);
// Cleanup the avatar cache. Note: Only works if there are no users on the server.
m_timerManager.RegisterTimer(
SERVER_CACHE_CLEANUP_INTERVAL_SEC * 1000,
boost::bind(&ServerLobbyThread::TimerCleanupAvatarCache, this),
true);
// Update the statistics file.
m_timerManager.RegisterTimer(
SERVER_SAVE_STATISTICS_INTERVAL_SEC * 1000,
boost::bind(&ServerLobbyThread::TimerSaveStatisticsFile, this),
true);
}
void
@@ -961,23 +1027,6 @@ ServerLobbyThread::RequestPlayerAvatar(SessionWrapper session)
session.sessionData->GetSender().Send(session.sessionData, retrieveAvatar);
}
void
ServerLobbyThread::NewConnectionLoop()
{
// Handle one incoming connection at a time.
boost::shared_ptr<tcp::socket> tmpData;
{
boost::mutex::scoped_lock lock(m_connectQueueMutex);
if (!m_connectQueue.empty())
{
tmpData = m_connectQueue.front();
m_connectQueue.pop_front();
}
}
if (tmpData.get())
HandleNewConnection(tmpData);
}
void
ServerLobbyThread::NewSessionLoop()
{
@@ -996,7 +1045,7 @@ ServerLobbyThread::NewSessionLoop()
}
void
ServerLobbyThread::RemoveGameLoop()
ServerLobbyThread::TimerRemoveGame()
{
boost::mutex::scoped_lock lock(m_removeGameListMutex);
@@ -1021,7 +1070,7 @@ ServerLobbyThread::RemoveGameLoop()
}
void
ServerLobbyThread::RemovePlayerLoop()
ServerLobbyThread::TimerRemovePlayer()
{
boost::mutex::scoped_lock lock(m_removePlayerListMutex);
@@ -1202,62 +1251,6 @@ ServerLobbyThread::TerminateGames()
m_gameMap.clear();
}
void
ServerLobbyThread::HandleNewConnection(boost::shared_ptr<tcp::socket> sock)
{
// Create a random session id.
// This id can be used to reconnect to the server if the connection was lost.
//unsigned sessionId;
// TODO: use randomized method.
//if(!RAND_bytes((unsigned char *)&sessionId, sizeof(sessionId)))
//{
// RAND_pseudo_bytes((unsigned char *)&sessionId, sizeof(sessionId));
//}
// Create a new session.
boost::shared_ptr<SessionData> 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)
{
boost::system::error_code errCode;
tcp::endpoint clientEndpoint = sock->remote_endpoint(errCode);
if (!errCode)
{
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<PlayerData>()), ERR_NET_INVALID_SESSION);
}
}
else
{
// Server is full.
// Gracefully close this session.
SessionError(SessionWrapper(sessionData, boost::shared_ptr<PlayerData>()), ERR_NET_SERVER_FULL);
}
}
void
ServerLobbyThread::HandleReAddedSession(SessionWrapper session)
{
@@ -1317,15 +1310,6 @@ ServerLobbyThread::InternalCheckSessionTimeouts(SessionWrapper session)
}
}
void
ServerLobbyThread::CleanupConnectQueue()
{
boost::mutex::scoped_lock lock(m_connectQueueMutex);
// Sockets will be closed automatically.
m_connectQueue.clear();
}
void
ServerLobbyThread::SessionError(SessionWrapper session, int errorCode)
{
src/net/common/servermanager.cpp
+6 -16
View File
@@ -23,7 +23,7 @@
#include <net/ircthread.h>
#include <net/connectdata.h>
#include <net/serverlobbythread.h>
#include <net/serveracceptthread.h>
#include <net/serveracceptmanager.h>
#include <net/serverexception.h>
#include <net/socket_msg.h>
#include <net/socket_startup.h>
@@ -54,15 +54,15 @@ ServerManager::Init(unsigned serverPort, bool ipv6, ServerNetworkMode mode, cons
if (mode & NETWORK_MODE_TCP)
{
boost::shared_ptr<ServerAcceptThread> tcpAcceptThread(new ServerAcceptThread(GetGui(), m_ioService));
boost::shared_ptr<ServerAcceptManager> tcpAcceptThread(new ServerAcceptManager(GetGui(), m_ioService));
tcpAcceptThread->Listen(serverPort, ipv6, false, pwd, logDir, m_lobbyThread);
m_acceptThreadPool.push_back(tcpAcceptThread);
m_acceptManagerPool.push_back(tcpAcceptThread);
}
if (mode & NETWORK_MODE_SCTP)
{
boost::shared_ptr<ServerAcceptThread> sctpAcceptThread(new ServerAcceptThread(GetGui(), m_ioService));
boost::shared_ptr<ServerAcceptManager> sctpAcceptThread(new ServerAcceptManager(GetGui(), m_ioService));
sctpAcceptThread->Listen(serverPort, ipv6, true, pwd, logDir, m_lobbyThread);
m_acceptThreadPool.push_back(sctpAcceptThread);
m_acceptManagerPool.push_back(sctpAcceptThread);
}
m_ircThread = ircThread;
}
@@ -299,7 +299,6 @@ ServerManager::SignalTerminationAll()
if (m_ircThread)
m_ircThread->SignalTermination();
GetLobbyThread().SignalTermination();
// for_each(m_acceptThreadPool.begin(), m_acceptThreadPool.end(), boost::mem_fn(&ServerAcceptThread::SignalTermination));
}
bool
@@ -307,16 +306,7 @@ 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);
/* 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;
return GetLobbyThread().Join(wait ? NET_LOBBY_THREAD_TERMINATE_TIMEOUT_MSEC : 0);
}
ServerLobbyThread &
src/net/serveracceptthread.h → src/net/serveracceptmanager.h
+7 -7
View File
@@ -1,5 +1,5 @@
/***************************************************************************
* Copyright (C) 2007 by Lothar May *
* Copyright (C) 2007-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 *
@@ -16,10 +16,10 @@
* Free Software Foundation, Inc., *
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
***************************************************************************/
/* Network server thread to accept connections. */
/* Network server manager to accept connections. */
#ifndef _SERVERACCEPTTHREAD_H_
#define _SERVERACCEPTTHREAD_H_
#ifndef _SERVERACCEPTMANAGER_H_
#define _SERVERACCEPTMANAGER_H_
#include <boost/asio.hpp>
#include <string>
@@ -29,11 +29,11 @@
class ServerLobbyThread;
class ServerAcceptThread
class ServerAcceptManager
{
public:
ServerAcceptThread(ServerCallback &serverCallback, boost::shared_ptr<boost::asio::io_service> ioService);
virtual ~ServerAcceptThread();
ServerAcceptManager(ServerCallback &serverCallback, boost::shared_ptr<boost::asio::io_service> ioService);
virtual ~ServerAcceptManager();
// Set the parameters.
void Listen(unsigned serverPort, bool ipv6, bool sctp, const std::string &pwd, const std::string &logDir,
src/net/serverlobbythread.h
+3 -8
View File
@@ -112,6 +112,7 @@ protected:
// Main function of the thread.
virtual void Main();
void RegisterTimers();
void HandleRead(SessionId sessionId, const boost::system::error_code& error, size_t bytesRead);
void HandlePacket(SessionWrapper session, boost::shared_ptr<NetPacket> packet);
@@ -126,10 +127,9 @@ protected:
void HandleNetPacketJoinGame(SessionWrapper session, const NetPacketJoinGame &tmpPacket);
void EstablishSession(SessionWrapper session);
void RequestPlayerAvatar(SessionWrapper session);
void NewConnectionLoop();
void NewSessionLoop();
void RemoveGameLoop();
void RemovePlayerLoop();
void TimerRemoveGame();
void TimerRemovePlayer();
void ResubscribeLobbyMsgLoop();
void UpdateAvatarClientTimerLoop();
void TimerCheckSessionTimeouts();
@@ -142,12 +142,10 @@ protected:
void TerminateGames();
void HandleNewConnection(boost::shared_ptr<boost::asio::ip::tcp::socket> sock);
void HandleReAddedSession(SessionWrapper session);
void InternalCheckSessionTimeouts(SessionWrapper session);
void CleanupConnectQueue();
void CleanupSessionMap();
void CloseSession(SessionWrapper session);
@@ -180,9 +178,6 @@ private:
TimerManager m_timerManager;
ConnectQueue m_connectQueue;
mutable boost::mutex m_connectQueueMutex;
SessionQueue m_sessionQueue;
mutable boost::mutex m_sessionQueueMutex;
src/net/servermanager.h
+3 -3
View File
@@ -30,7 +30,7 @@
#include <gui/guiinterface.h>
class ServerLobbyThread;
class ServerAcceptThread;
class ServerAcceptManager;
class SenderThread;
class ConfigFile;
class AvatarManager;
@@ -67,7 +67,7 @@ public:
virtual void SignalIrcServerError(int errorCode);
protected:
typedef std::list<boost::shared_ptr<ServerAcceptThread> > AcceptThreadList;
typedef std::list<boost::shared_ptr<ServerAcceptManager> > AcceptManagerList;
ServerLobbyThread &GetLobbyThread();
@@ -82,7 +82,7 @@ private:
boost::shared_ptr<ServerLobbyThread> m_lobbyThread;
boost::shared_ptr<IrcThread> m_ircThread;
boost::timers::portable::microsec_timer m_ircRestartTimer;
AcceptThreadList m_acceptThreadPool;
AcceptManagerList m_acceptManagerPool;
};
#endif