Fixed one thread per session implementation. Kind of. Still seems to have a handle leak on Windows. However, it does not scale at all and needs to be reverted. It won't even handle 500 players with 500 games on a dual core high end machine.

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 \

src/net/clientcontext.h

@@ -24,9 +24,10 @@
 #include <net/netcontext.h>
 #include <net/receivebuffer.h>
 #include <net/sessiondata.h>
+#include <net/senderinterface.h>
 #include <boost/shared_ptr.hpp>
 
-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<SenderCallback> m_senderCallback;
+	boost::shared_ptr<ClientSenderCallback> m_senderCallback;
+	boost::shared_ptr<SenderInterface> m_senderThread;
 };
 
 #endif

src/net/common/clientcontext.cpp

@@ -18,9 +18,9 @@
  ***************************************************************************/
 
 #include <net/clientcontext.h>
-#include <net/sendercallback.h>
+#include <net/senderthread.h>
 
-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<SessionData>

src/net/common/senderthread.cpp

@@ -74,8 +74,9 @@ SenderThread::Send(boost::shared_ptr<SessionData> session, boost::shared_ptr<Net
 			}
 		}
 		{
-			boost::mutex::scoped_lock lock(m_sessionMutex);
+			boost::mutex::scoped_lock lock(m_sessionDataMutex);
-			m_session = session;
+			m_sessionSocket = session->GetSocket();
+			m_sessionId = session->GetId();
 		}
 	}
 }
@@ -97,8 +98,9 @@ SenderThread::Send(boost::shared_ptr<SessionData> session, const NetPacketList &
 			}
 		}
 		{
-			boost::mutex::scoped_lock lock(m_sessionMutex);
+			boost::mutex::scoped_lock lock(m_sessionDataMutex);
-			m_session = session;
+			m_sessionSocket = session->GetSocket();
+			m_sessionId = session->GetId();
 		}
 	}
 }
@@ -127,8 +129,8 @@ SenderThread::Main()
 			{
 				SOCKET tmpSocket;
 				{
-					boost::mutex::scoped_lock lock(m_sessionMutex);
+					boost::mutex::scoped_lock lock(m_sessionDataMutex);
-					tmpSocket = m_session->GetSocket();
+					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);
+									boost::mutex::scoped_lock lock(m_sessionDataMutex);
-									m_callback.SignalNetError(m_session->GetId(), ERR_SOCK_SELECT_FAILED, errCode);
+									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);
+							boost::mutex::scoped_lock lock(m_sessionDataMutex);
-							m_callback.SignalNetError(m_session->GetId(), ERR_SOCK_SEND_FAILED, errCode);
+							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);
 }
 

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<SenderInterface> 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<ConnectData> connData)
 	//}
 
 	// Create a new session.
-	boost::shared_ptr<SessionData> sessionData(new SessionData(connData->ReleaseSocket(), m_curSessionId++, *m_senderCallback));
+	boost::shared_ptr<SenderInterface> senderThread(new SenderThread(*m_senderCallback));
+	senderThread->Start();
+	boost::shared_ptr<SessionData> 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() << ".");
 

src/net/common/sessiondata.cpp

@@ -18,22 +18,18 @@
  ***************************************************************************/
 
 #include <net/sessiondata.h>
-#include <net/senderthread.h>
+#include <net/senderinterface.h>
 
-SessionData::SessionData(SOCKET sockfd, SessionId id, SenderCallback &cb)
+SessionData::SessionData(SOCKET sockfd, SessionId id, boost::shared_ptr<SenderInterface> 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 = sender;
-	m_sender->Start();
 }
 
 SessionData::~SessionData()
 {
-	m_sender->SignalStop();
+	m_callback.SignalSessionTerminated(m_id);
-	m_sender->WaitStop();
-	if (m_sockfd != INVALID_SOCKET)
-		CLOSESOCKET(m_sockfd);
 }
 
 SessionId

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 <net/sessiondatacallback.h>
+
+
+SessionDataCallback::~SessionDataCallback()
+{
+}
+

src/net/senderthread.h

@@ -23,13 +23,13 @@
 
 #include <core/thread.h>
 #include <net/senderinterface.h>
-#include <net/sessiondata.h>
 #include <net/netpacket.h>
 #include <net/sendercallback.h>
 
 #include <list>
 #include <boost/shared_ptr.hpp>
 
+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<SessionData> m_session;
+	mutable boost::mutex m_sessionDataMutex;
-	mutable boost::mutex m_sessionMutex;
+	SOCKET m_sessionSocket;
+	unsigned m_sessionId;
 	boost::shared_ptr<NetPacket> m_curPacket;
 	unsigned m_bytesSent;
 	SenderCallback &m_callback;

src/net/serverlobbythread.h

@@ -77,6 +77,7 @@ public:
 	void RemoveComputerPlayer(boost::shared_ptr<PlayerData> 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<unsigned, boost::shared_ptr<ServerGameThread> > GameMap;
 	typedef std::map<std::string, boost::timers::portable::microsec_timer> TimerClientAddressMap;
 	typedef std::list<unsigned> RemoveGameList;
+	typedef std::list<unsigned> RemoveSenderList;
+	typedef std::map<unsigned, boost::shared_ptr<SenderInterface> >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<ReceiverHelper> m_receiver;
 	boost::shared_ptr<ServerSenderCallback> m_senderCallback;

src/net/sessiondata.h

@@ -21,8 +21,11 @@
 #ifndef _SESSIONDATA_H_
 #define _SESSIONDATA_H_
 
+typedef unsigned SessionId;
+
 #include <net/socket_helper.h>
 #include <net/receivebuffer.h>
+#include <net/sessiondatacallback.h>
 #include <string>
 #include <boost/thread.hpp>
 #include <third_party/boost/timers.hpp>
@@ -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<SenderInterface> 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<SenderThread>	m_sender;
+	boost::shared_ptr<SenderInterface>	m_sender;
+	SessionDataCallback				&m_callback;
 
 	mutable boost::mutex			m_dataMutex;
 };

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