PurpleI2P/i2pd
1
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2025-01-08 22:57:52 +00:00
Code Issues Releases Activity

fixed race condition

Browse Source
This commit is contained in:
orignal 2014-10-04 16:27:21 -04:00
parent 9a617297f2
commit 9750fb73b5
2 changed files with 40 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
NetDb.cpp
View File

@ -102,7 +102,7 @@ namespace data
{ {
try try
{ {
I2NPMessage * msg = m_Queue.GetNextWithTimeout (10000); // 10 sec I2NPMessage * msg = m_Queue.GetNextWithTimeout (30000); // 30 sec
if (msg) if (msg)
{ {
while (msg) while (msg)

59
Streaming.cpp
View File

@ -37,20 +37,25 @@ namespace stream
} }
Stream::~Stream () Stream::~Stream ()
{ {
Close ();
m_ReceiveTimer.cancel ();
m_ResendTimer.cancel ();
while (!m_ReceiveQueue.empty ()) while (!m_ReceiveQueue.empty ())
{ {
auto packet = m_ReceiveQueue.front (); auto packet = m_ReceiveQueue.front ();
m_ReceiveQueue.pop (); m_ReceiveQueue.pop ();
delete packet; delete packet;
} }
for (auto it: m_SavedPackets) m_ReceiveTimer.cancel ();
delete it;
for (auto it: m_SentPackets) for (auto it: m_SentPackets)
delete it; delete it;
m_SentPackets.clear ();
m_ResendTimer.cancel ();
for (auto it: m_SavedPackets)
delete it;
m_SavedPackets.clear ();
Close ();
} }
void Stream::HandleNextPacket (Packet * packet) void Stream::HandleNextPacket (Packet * packet)
@ -387,11 +392,14 @@ namespace stream
if (packet) if (packet)
{ {
SendPackets (std::vector<Packet *> { packet }); SendPackets (std::vector<Packet *> { packet });
bool isEmpty = m_SentPackets.empty (); if (m_IsOpen)
m_SentPackets.insert (packet); {
if (isEmpty) bool isEmpty = m_SentPackets.empty ();
ScheduleResend (); m_SentPackets.insert (packet);
return true; if (isEmpty)
ScheduleResend ();
}
return true;
} }
else else
return false; return false;
@ -565,8 +573,12 @@ namespace stream
StreamingDestination::~StreamingDestination () StreamingDestination::~StreamingDestination ()
{ {
for (auto it: m_Streams) {
delete it.second; std::unique_lock<std::mutex> l(m_StreamsMutex);
for (auto it: m_Streams)
delete it.second;
m_Streams.clear ();
}
if (m_Pool) if (m_Pool)
i2p::tunnel::tunnels.DeleteTunnelPool (m_Pool); i2p::tunnel::tunnels.DeleteTunnelPool (m_Pool);
delete m_LeaseSet; delete m_LeaseSet;
@ -619,10 +631,14 @@ namespace stream
void StreamingDestination::DeleteStream (Stream * stream) void StreamingDestination::DeleteStream (Stream * stream)
{ {
if (stream) if (stream)
{ {
std::unique_lock<std::mutex> l(m_StreamsMutex); std::unique_lock<std::mutex> l(m_StreamsMutex);
m_Streams.erase (stream->GetRecvStreamID ()); auto it = m_Streams.find (stream->GetRecvStreamID ());
delete stream; if (it != m_Streams.end ())
{
m_Streams.erase (it);
delete stream;
}
} }
} }
@ -738,9 +754,12 @@ namespace stream
auto it = m_Destinations.find (destination->GetIdentHash ()); auto it = m_Destinations.find (destination->GetIdentHash ());
if (it != m_Destinations.end ()) if (it != m_Destinations.end ())
{ {
delete it->second; auto d = it->second;
std::unique_lock<std::mutex> l(m_DestinationsMutex); {
m_Destinations.erase (it); std::unique_lock<std::mutex> l(m_DestinationsMutex);
m_Destinations.erase (it);
}
delete d;
} }
} }