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

Fixed cleanup issue - the previous version had kind of a resource leak.

Browse Source
This commit is contained in:
lotodore
2009-01-31 22:20:08 +00:00
parent 3f8d3c8d55
commit c998edf219
6 changed files with 66 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/net/common/clientcontext.cpp
+2 -2
View File
@@ -43,10 +43,10 @@ ClientContext::ClientContext()
m_hasSubscribedLobbyMsg(true)
{
bzero(&m_clientSockaddr, sizeof(m_clientSockaddr));
m_ioService.reset(new boost::asio::io_service());
m_senderCallback.reset(new ClientSenderCallback());
m_senderThread.reset(new SenderThread(*m_senderCallback));
m_senderThread.reset(new SenderThread(*m_senderCallback, m_ioService));
m_senderThread->Start();
m_ioService = dynamic_cast<SenderThread *>(m_senderThread.get())->GetIOService();
}
ClientContext::~ClientContext()
src/net/common/senderthread.cpp
+42 -17
View File
@@ -43,8 +43,8 @@ typedef std::list<boost::shared_ptr<NetPacket> > SendDataList;
class SendDataManager : public boost::enable_shared_from_this<SendDataManager>
{
public:
SendDataManager(boost::shared_ptr<SessionData> s)
: session(s), writeInProgress(false)
SendDataManager(boost::shared_ptr<boost::asio::ip::tcp::socket> s)
: socket(s), writeInProgress(false)
{
}
@@ -52,7 +52,7 @@ class SendDataManager : public boost::enable_shared_from_this<SendDataManager>
void AsyncSendNextPacket(bool handlerMode = false);
boost::shared_ptr<SessionData> session;
boost::shared_ptr<boost::asio::ip::tcp::socket> socket;
mutable boost::mutex dataMutex;
SendDataList list;
@@ -79,7 +79,7 @@ SendDataManager::AsyncSendNextPacket(bool handlerMode)
{
boost::shared_ptr<NetPacket> nextPacket = list.front();
boost::asio::async_write(
*session->GetAsioSocket(),
*socket,
boost::asio::buffer(nextPacket->GetRawData(),
nextPacket->GetLen()),
boost::bind(&SendDataManager::HandleWrite, shared_from_this(),
@@ -91,8 +91,8 @@ SendDataManager::AsyncSendNextPacket(bool handlerMode)
}
}
SenderThread::SenderThread(SenderCallback &cb)
: m_callback(cb)
SenderThread::SenderThread(SenderCallback &cb, boost::shared_ptr<boost::asio::io_service> ioService)
: m_callback(cb), m_ioService(ioService)
{
}
@@ -103,9 +103,7 @@ SenderThread::~SenderThread()
void
SenderThread::Start()
{
m_ioServiceBarrier.reset(new boost::barrier(2));
Run();
m_ioServiceBarrier->wait();
}
void
@@ -131,7 +129,7 @@ SenderThread::Send(boost::shared_ptr<SessionData> session, boost::shared_ptr<Net
boost::mutex::scoped_lock lock(m_sendQueueMapMutex);
SendQueueMap::iterator pos = m_sendQueueMap.find(session->GetId());
if (pos == m_sendQueueMap.end())
pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr<SendDataManager>(new SendDataManager(session)))).first;
pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr<SendDataManager>(new SendDataManager(session->GetAsioSocket())))).first;
tmpManager = pos->second;
}
{
@@ -143,7 +141,7 @@ SenderThread::Send(boost::shared_ptr<SessionData> session, boost::shared_ptr<Net
{
// Third: Update notification list.
boost::mutex::scoped_lock lock(m_changedSessionsMutex);
m_changedSessions.push_back(tmpManager->session->GetId());
m_changedSessions.push_back(session->GetId());
}
}
}
@@ -159,7 +157,7 @@ SenderThread::Send(boost::shared_ptr<SessionData> session, const NetPacketList &
boost::mutex::scoped_lock lock(m_sendQueueMapMutex);
SendQueueMap::iterator pos = m_sendQueueMap.find(session->GetId());
if (pos == m_sendQueueMap.end())
pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr<SendDataManager>(new SendDataManager(session)))).first;
pos = m_sendQueueMap.insert(SendQueueMap::value_type(session->GetId(), boost::shared_ptr<SendDataManager>(new SendDataManager(session->GetAsioSocket())))).first;
tmpManager = pos->second;
}
{
@@ -179,25 +177,52 @@ SenderThread::Send(boost::shared_ptr<SessionData> session, const NetPacketList &
{
// Third: Update notification list.
boost::mutex::scoped_lock lock(m_changedSessionsMutex);
m_changedSessions.push_back(tmpManager->session->GetId());
m_changedSessions.push_back(session->GetId());
}
}
}
boost::shared_ptr<boost::asio::io_service>
SenderThread::GetIOService()
void
SenderThread::SignalSessionTerminated(unsigned sessionId)
{
return m_ioService;
boost::mutex::scoped_lock lock(m_removedSessionsMutex);
m_removedSessions.push_back(sessionId);
}
void
SenderThread::Main()
{
m_ioService.reset(new boost::asio::io_service());
m_ioServiceBarrier->wait();
boost::asio::io_service::work ioWork(*m_ioService);
while (!ShouldTerminate())
{
// Close sessions if they were destructed.
{
boost::mutex::scoped_lock lock(m_removedSessionsMutex);
if (!m_removedSessions.empty())
{
SessionIdList newRemovedSessions;
SessionIdList::iterator i = m_removedSessions.begin();
SessionIdList::iterator end = m_removedSessions.end();
boost::mutex::scoped_lock lock(m_sendQueueMapMutex);
while (i != end)
{
SendQueueMap::iterator pos = m_sendQueueMap.find(*i);
if (pos != m_sendQueueMap.end())
{
// Remove session if no write is in progress, else wait.
if (!pos->second->writeInProgress)
m_sendQueueMap.erase(pos);
else
newRemovedSessions.push_back(*i);
}
++i;
}
m_removedSessions = newRemovedSessions;
}
}
// Iterate through all changed sessions, and send data if needed.
bool sessionValid;
do
{
src/net/common/serverlobbythread.cpp
+11 -3
View File
@@ -69,8 +69,9 @@ 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*/)
virtual void SignalSessionTerminated(unsigned session)
{
m_server.GetSender();
}
private:
@@ -83,8 +84,9 @@ ServerLobbyThread::ServerLobbyThread(GuiInterface &gui, ConfigFile *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_sender.reset(new SenderThread(*m_senderCallback, m_ioService));
m_receiver.reset(new ReceiverHelper);
}
@@ -323,6 +325,13 @@ ServerLobbyThread::GetStartTime() const
return m_startTime;
}
SenderInterface &
ServerLobbyThread::GetSender()
{
assert(m_sender);
return *m_sender;
}
u_int32_t
ServerLobbyThread::GetNextUniquePlayerId()
{
@@ -350,7 +359,6 @@ ServerLobbyThread::Main()
try
{
m_sender->Start();
m_ioService = dynamic_cast<SenderThread *>(m_sender.get())->GetIOService();
while (!ShouldTerminate())
{
src/net/senderinterface.h
+2
View File
@@ -36,6 +36,8 @@ public:
virtual void Send(boost::shared_ptr<SessionData> session, boost::shared_ptr<NetPacket> packet) = 0;
virtual void Send(boost::shared_ptr<SessionData> session, const NetPacketList &packetList) = 0;
virtual void SignalSessionTerminated(unsigned sessionId) = 0;
};
#endif
src/net/senderthread.h
+7 -6
View File
@@ -36,7 +36,7 @@ class SendDataManager;
class SenderThread : public Thread, public SenderInterface
{
public:
SenderThread(SenderCallback &cb);
SenderThread(SenderCallback &cb, boost::shared_ptr<boost::asio::io_service> ioService);
virtual ~SenderThread();
virtual void Start();
@@ -46,10 +46,10 @@ public:
virtual void Send(boost::shared_ptr<SessionData> session, boost::shared_ptr<NetPacket> packet);
virtual void Send(boost::shared_ptr<SessionData> session, const NetPacketList &packetList);
boost::shared_ptr<boost::asio::io_service> GetIOService();
virtual void SignalSessionTerminated(unsigned sessionId);
protected:
typedef std::list<unsigned> ChangedSessionList;
typedef std::list<unsigned> SessionIdList;
typedef std::map<SessionId, boost::shared_ptr<SendDataManager> > SendQueueMap;
@@ -61,13 +61,14 @@ private:
SendQueueMap m_sendQueueMap;
mutable boost::mutex m_sendQueueMapMutex;
ChangedSessionList m_changedSessions;
SessionIdList m_changedSessions;
mutable boost::mutex m_changedSessionsMutex;
SessionIdList m_removedSessions;
mutable boost::mutex m_removedSessionsMutex;
SenderCallback &m_callback;
boost::shared_ptr<boost::asio::io_service> m_ioService;
mutable boost::shared_ptr<boost::barrier> m_ioServiceBarrier;
};
#endif
src/net/serverlobbythread.h
+2
View File
@@ -87,6 +87,8 @@ public:
ServerStats GetStats() const;
boost::posix_time::ptime GetStartTime() const;
SenderInterface &GetSender();
protected:
typedef std::deque<boost::shared_ptr<ConnectData> > ConnectQueue;