diff --git a/pokerth_lib.pro b/pokerth_lib.pro index 3b9e32bb..f65b32fc 100644 --- a/pokerth_lib.pro +++ b/pokerth_lib.pro @@ -76,6 +76,7 @@ HEADERS += \ src/net/servergamethread.h \ src/net/servergamestate.h \ src/net/serverlobbythread.h \ + src/net/sessiondatacallback.h \ src/net/socket_helper.h \ src/net/socket_msg.h \ src/net/socket_startup.h \ @@ -167,6 +168,7 @@ SOURCES += \ src/net/common/serverlobbythread.cpp \ src/net/common/servercallback.cpp \ src/net/common/sessiondata.cpp \ + src/net/common/sessiondatacallback.cpp \ src/net/common/sessionmanager.cpp \ src/net/common/socket_startup_cmn.cpp \ src/net/common/socket_helper_cmn.cpp \ diff --git a/src/net/clientcontext.h b/src/net/clientcontext.h index 80757294..966ba707 100644 --- a/src/net/clientcontext.h +++ b/src/net/clientcontext.h @@ -24,9 +24,10 @@ #include #include #include +#include #include -class SenderCallback; +class ClientSenderCallback; class ClientContext : public NetContext { @@ -108,7 +109,8 @@ private: std::string m_cacheDir; bool m_hasSubscribedLobbyMsg; ReceiveBuffer m_receiveBuffer; - boost::shared_ptr m_senderCallback; + boost::shared_ptr m_senderCallback; + boost::shared_ptr m_senderThread; }; #endif diff --git a/src/net/common/clientcontext.cpp b/src/net/common/clientcontext.cpp index cf983935..27817536 100644 --- a/src/net/common/clientcontext.cpp +++ b/src/net/common/clientcontext.cpp @@ -18,9 +18,9 @@ ***************************************************************************/ #include -#include +#include -class ClientSenderCallback : public SenderCallback +class ClientSenderCallback : public SenderCallback, public SessionDataCallback { public: ClientSenderCallback() {} @@ -30,6 +30,10 @@ public: { } + virtual void SignalSessionTerminated(unsigned /*session*/) + { + } + private: }; @@ -40,10 +44,14 @@ ClientContext::ClientContext() { bzero(&m_clientSockaddr, sizeof(m_clientSockaddr)); m_senderCallback.reset(new ClientSenderCallback()); + m_senderThread.reset(new SenderThread(*m_senderCallback)); + m_senderThread->Start(); } ClientContext::~ClientContext() { + m_senderThread->SignalStop(); + m_senderThread->WaitStop(); } SOCKET @@ -56,7 +64,7 @@ ClientContext::GetSocket() const void ClientContext::SetSocket(SOCKET sockfd) { - m_sessionData.reset(new SessionData(sockfd, SESSION_ID_GENERIC, *m_senderCallback)); + m_sessionData.reset(new SessionData(sockfd, SESSION_ID_GENERIC, m_senderThread, *m_senderCallback)); } boost::shared_ptr diff --git a/src/net/common/senderthread.cpp b/src/net/common/senderthread.cpp index 737f8e92..63fcfeb8 100644 --- a/src/net/common/senderthread.cpp +++ b/src/net/common/senderthread.cpp @@ -74,8 +74,9 @@ SenderThread::Send(boost::shared_ptr session, boost::shared_ptrGetSocket(); + m_sessionId = session->GetId(); } } } @@ -97,8 +98,9 @@ SenderThread::Send(boost::shared_ptr session, const NetPacketList & } } { - boost::mutex::scoped_lock lock(m_sessionMutex); - m_session = session; + boost::mutex::scoped_lock lock(m_sessionDataMutex); + m_sessionSocket = session->GetSocket(); + m_sessionId = session->GetId(); } } } @@ -127,8 +129,8 @@ SenderThread::Main() { SOCKET tmpSocket; { - boost::mutex::scoped_lock lock(m_sessionMutex); - tmpSocket = m_session->GetSocket(); + boost::mutex::scoped_lock lock(m_sessionDataMutex); + tmpSocket = m_sessionSocket; } // send next chunk of data @@ -159,8 +161,8 @@ SenderThread::Main() // Ignore invalid or not connected sockets. if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK) { - boost::mutex::scoped_lock lock(m_sessionMutex); - m_callback.SignalNetError(m_session->GetId(), ERR_SOCK_SELECT_FAILED, errCode); + boost::mutex::scoped_lock lock(m_sessionDataMutex); + m_callback.SignalNetError(m_sessionId, ERR_SOCK_SELECT_FAILED, errCode); } } Msleep(SEND_TIMEOUT_MSEC); @@ -172,8 +174,8 @@ SenderThread::Main() // Ignore invalid or not connected sockets. if (errCode != SOCKET_ERR_NOTCONN && errCode != SOCKET_ERR_NOTSOCK) { - boost::mutex::scoped_lock lock(m_sessionMutex); - m_callback.SignalNetError(m_session->GetId(), ERR_SOCK_SEND_FAILED, errCode); + boost::mutex::scoped_lock lock(m_sessionDataMutex); + m_callback.SignalNetError(m_sessionId, ERR_SOCK_SEND_FAILED, errCode); } Msleep(SEND_TIMEOUT_MSEC); } @@ -198,5 +200,8 @@ SenderThread::Main() else Msleep(SEND_TIMEOUT_MSEC); } + boost::mutex::scoped_lock lock(m_sessionDataMutex); + if (m_sessionSocket != INVALID_SOCKET) + CLOSESOCKET(m_sessionSocket); } diff --git a/src/net/common/serverlobbythread.cpp b/src/net/common/serverlobbythread.cpp index 519c1bdf..1e6de971 100644 --- a/src/net/common/serverlobbythread.cpp +++ b/src/net/common/serverlobbythread.cpp @@ -57,7 +57,7 @@ using namespace std; -class ServerSenderCallback : public SenderCallback +class ServerSenderCallback : public SenderCallback, public SessionDataCallback { public: ServerSenderCallback(ServerLobbyThread &server) : m_server(server) {} @@ -69,6 +69,10 @@ public: // A serious send error should trigger a read error or a read // returning 0 afterwards, and we will handle this error. } + virtual void SignalSessionTerminated(unsigned session) + { + m_server.RemoveSender(session); + } private: ServerLobbyThread &m_server; @@ -300,6 +304,13 @@ ServerLobbyThread::RemoveGame(unsigned id) m_removeGameList.push_back(id); } +void +ServerLobbyThread::RemoveSender(unsigned session) +{ + boost::mutex::scoped_lock lock(m_removeSenderListMutex); + m_removeSenderList.push_back(session); +} + AvatarManager & ServerLobbyThread::GetAvatarManager() { @@ -357,6 +368,8 @@ ServerLobbyThread::Main() RemoveGameLoop(); // Kick players. RemovePlayerLoop(); + // Remove sender threads. + RemoveSenderLoop(); // Resubscribe Lobby Messages if needed. ResubscribeLobbyMsgLoop(); // Check session timeouts. @@ -862,6 +875,30 @@ ServerLobbyThread::RemovePlayerLoop() } } +void +ServerLobbyThread::RemoveSenderLoop() +{ + boost::mutex::scoped_lock lock(m_removeSenderListMutex); + + RemoveSenderList::iterator i = m_removeSenderList.begin(); + RemoveSenderList::iterator end = m_removeSenderList.end(); + + // Synchronously remove Sender Threads whose Sessions have been closed. + while (i != end) + { + SenderMap::iterator pos = m_senderMap.find(*i); + if (pos != m_senderMap.end()) + { + boost::shared_ptr tmpSender = pos->second; + tmpSender->SignalStop(); + tmpSender->WaitStop(); + m_senderMap.erase(pos); + } + ++i; + } + m_removeSenderList.clear(); +} + void ServerLobbyThread::ResubscribeLobbyMsgLoop() { @@ -1047,8 +1084,11 @@ ServerLobbyThread::HandleNewConnection(boost::shared_ptr connData) //} // Create a new session. - boost::shared_ptr sessionData(new SessionData(connData->ReleaseSocket(), m_curSessionId++, *m_senderCallback)); + boost::shared_ptr senderThread(new SenderThread(*m_senderCallback)); + senderThread->Start(); + boost::shared_ptr sessionData(new SessionData(connData->ReleaseSocket(), m_curSessionId++, senderThread, *m_senderCallback)); m_sessionManager.AddSession(sessionData); + m_senderMap[sessionData->GetId()] = senderThread; LOG_VERBOSE("Accepted connection - session #" << sessionData->GetId() << "."); diff --git a/src/net/common/sessiondata.cpp b/src/net/common/sessiondata.cpp index dd633a3a..ddb05c63 100644 --- a/src/net/common/sessiondata.cpp +++ b/src/net/common/sessiondata.cpp @@ -18,22 +18,18 @@ ***************************************************************************/ #include -#include +#include -SessionData::SessionData(SOCKET sockfd, SessionId id, SenderCallback &cb) +SessionData::SessionData(SOCKET sockfd, SessionId id, boost::shared_ptr sender, SessionDataCallback &cb) : m_sockfd(sockfd), m_id(id), m_state(SessionData::Init), m_readyFlag(false), - m_wantsLobbyMsg(true), m_activityTimeoutNoticeSent(false) + m_wantsLobbyMsg(true), m_activityTimeoutNoticeSent(false), m_callback(cb) { - m_sender.reset(new SenderThread(cb)); - m_sender->Start(); + m_sender = sender; } SessionData::~SessionData() { - m_sender->SignalStop(); - m_sender->WaitStop(); - if (m_sockfd != INVALID_SOCKET) - CLOSESOCKET(m_sockfd); + m_callback.SignalSessionTerminated(m_id); } SessionId diff --git a/src/net/common/sessiondatacallback.cpp b/src/net/common/sessiondatacallback.cpp new file mode 100644 index 00000000..ae8086dc --- /dev/null +++ b/src/net/common/sessiondatacallback.cpp @@ -0,0 +1,26 @@ +/*************************************************************************** + * 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 + + +SessionDataCallback::~SessionDataCallback() +{ +} + diff --git a/src/net/senderthread.h b/src/net/senderthread.h index 3fd3464d..0405b084 100644 --- a/src/net/senderthread.h +++ b/src/net/senderthread.h @@ -23,13 +23,13 @@ #include #include -#include #include #include #include #include +class SessionData; #define SENDER_THREAD_TERMINATE_TIMEOUT THREAD_WAIT_INFINITE class SenderThread : public Thread, public SenderInterface @@ -56,8 +56,9 @@ private: SendDataList m_sendQueue; mutable boost::mutex m_sendQueueMutex; - boost::shared_ptr m_session; - mutable boost::mutex m_sessionMutex; + mutable boost::mutex m_sessionDataMutex; + SOCKET m_sessionSocket; + unsigned m_sessionId; boost::shared_ptr m_curPacket; unsigned m_bytesSent; SenderCallback &m_callback; diff --git a/src/net/serverlobbythread.h b/src/net/serverlobbythread.h index 22b8a882..e9ca775f 100644 --- a/src/net/serverlobbythread.h +++ b/src/net/serverlobbythread.h @@ -77,6 +77,7 @@ public: void RemoveComputerPlayer(boost::shared_ptr player); void RemoveGame(unsigned id); + void RemoveSender(unsigned session); u_int32_t GetNextUniquePlayerId(); u_int32_t GetNextGameId(); @@ -97,6 +98,8 @@ protected: typedef std::map > GameMap; typedef std::map TimerClientAddressMap; typedef std::list RemoveGameList; + typedef std::list RemoveSenderList; + typedef std::map >SenderMap; // Main function of the thread. virtual void Main(); @@ -117,6 +120,7 @@ protected: void NewSessionLoop(); void RemoveGameLoop(); void RemovePlayerLoop(); + void RemoveSenderLoop(); void ResubscribeLobbyMsgLoop(); void CheckSessionTimeoutsLoop(); void UpdateAvatarClientTimerLoop(); @@ -179,6 +183,9 @@ private: RemovePlayerList m_removePlayerList; mutable boost::mutex m_removePlayerListMutex; + RemoveSenderList m_removeSenderList; + mutable boost::mutex m_removeSenderListMutex; + PlayerDataMap m_computerPlayers; mutable boost::mutex m_computerPlayersMutex; @@ -186,6 +193,7 @@ private: mutable boost::mutex m_resubscribeListMutex; GameMap m_gameMap; + SenderMap m_senderMap; boost::shared_ptr m_receiver; boost::shared_ptr m_senderCallback; diff --git a/src/net/sessiondata.h b/src/net/sessiondata.h index 88a2c54a..63702924 100644 --- a/src/net/sessiondata.h +++ b/src/net/sessiondata.h @@ -21,8 +21,11 @@ #ifndef _SESSIONDATA_H_ #define _SESSIONDATA_H_ +typedef unsigned SessionId; + #include #include +#include #include #include #include @@ -31,17 +34,14 @@ #define SESSION_ID_INIT INVALID_SESSION #define SESSION_ID_GENERIC 0xFFFFFFFF -typedef unsigned SessionId; -class SenderThread; class SenderInterface; -class SenderCallback; class SessionData { public: enum State { Init, ReceivingAvatar, Established, Game }; - SessionData(SOCKET sockfd, SessionId id, SenderCallback &cb); + SessionData(SOCKET sockfd, SessionId id, boost::shared_ptr sender, SessionDataCallback &cb); ~SessionData(); SessionId GetId() const; @@ -80,7 +80,8 @@ private: boost::timers::portable::microsec_timer m_activityTimer; bool m_activityTimeoutNoticeSent; boost::timers::portable::microsec_timer m_autoDisconnectTimer; - boost::shared_ptr m_sender; + boost::shared_ptr m_sender; + SessionDataCallback &m_callback; mutable boost::mutex m_dataMutex; }; diff --git a/src/net/sessiondatacallback.h b/src/net/sessiondatacallback.h new file mode 100644 index 00000000..8f4429b3 --- /dev/null +++ b/src/net/sessiondatacallback.h @@ -0,0 +1,32 @@ +/*************************************************************************** + * 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. * + ***************************************************************************/ +/* Callback interface for session data. */ + +#ifndef _SESSIONDATACALLBACK_H_ +#define _SESSIONDATACALLBACK_H_ + +class SessionDataCallback +{ +public: + virtual ~SessionDataCallback(); + + virtual void SignalSessionTerminated(unsigned session) = 0; +}; + +#endif