Skip to content

Commit

Permalink
separate SSU sessions lists for V4 and V6
Browse files Browse the repository at this point in the history
  • Loading branch information
@orignal
orignal committed Nov 30, 2015
1 parent 2b8e662 commit c5308e3
Show file tree
Hide file tree
Showing 4 changed files with 46 additions and 38 deletions.
14 changes: 11 additions & 3 deletions HTTPServer.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -772,16 +772,24 @@ namespace util
s << "<br>SSU<br>";
for (auto it: ssuServer->GetSessions ())
{
// incoming connections don't have remote router
auto endpoint = it.second->GetRemoteEndpoint ();
if (it.second->IsOutgoing ()) s << "-->";
s << endpoint.address ().to_string () << ":" << endpoint.port ();
if (!it.second->IsOutgoing ()) s << "-->";
s << " [" << it.second->GetNumSentBytes () << ":" << it.second->GetNumReceivedBytes () << "]";
if (it.second->GetRelayTag ())
s << " [itag:" << it.second->GetRelayTag () << "]";
s << "<br>";
s << std::endl;
s << "<br>" << std::endl;
}
s << "<br>SSU6<br>";
for (auto it: ssuServer->GetSessionsV6 ())
{
auto endpoint = it.second->GetRemoteEndpoint ();
if (it.second->IsOutgoing ()) s << "-->";
s << endpoint.address ().to_string () << ":" << endpoint.port ();
if (!it.second->IsOutgoing ()) s << "-->";
s << " [" << it.second->GetNumSentBytes () << ":" << it.second->GetNumReceivedBytes () << "]";
s << "<br>" << std::endl;
}
}
}
Expand Down
57 changes: 28 additions & 29 deletions SSU.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,7 @@ namespace transport
moreBytes = m_Socket.available();
}

m_Service.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets));
m_Service.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets, m_Sessions));
Receive ();
}
else
Expand All @@ -201,7 +201,7 @@ namespace transport
moreBytes = m_SocketV6.available();
}

m_ServiceV6.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets));
m_ServiceV6.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets, m_SessionsV6));
ReceiveV6 ();
}
else
Expand All @@ -211,7 +211,8 @@ namespace transport
}
}

void SSUServer::HandleReceivedPackets (std::vector<SSUPacket *> packets)
void SSUServer::HandleReceivedPackets (std::vector<SSUPacket *> packets,
std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSUSession> >& sessions)
{
std::shared_ptr<SSUSession> session;
for (auto it1: packets)
Expand All @@ -222,17 +223,14 @@ namespace transport
if (!session || session->GetRemoteEndpoint () != packet->from) // we received packet for other session than previous
{
if (session) session->FlushData ();
auto it = m_Sessions.find (packet->from);
if (it != m_Sessions.end ())
auto it = sessions.find (packet->from);
if (it != sessions.end ())
session = it->second;
if (!session)
{
session = std::make_shared<SSUSession> (*this, packet->from);
session->WaitForConnect ();
{
std::unique_lock<std::mutex> l(m_SessionsMutex);
m_Sessions[packet->from] = session;
}
sessions[packet->from] = session;
LogPrint (eLogInfo, "New SSU session from ", packet->from.address ().to_string (), ":", packet->from.port (), " created");
}
}
Expand Down Expand Up @@ -265,8 +263,9 @@ namespace transport

std::shared_ptr<SSUSession> SSUServer::FindSession (const boost::asio::ip::udp::endpoint& e) const
{
auto it = m_Sessions.find (e);
if (it != m_Sessions.end ())
auto& sessions = e.address ().is_v6 () ? m_SessionsV6 : m_Sessions;
auto it = sessions.find (e);
if (it != sessions.end ())
return it->second;
else
return nullptr;
Expand Down Expand Up @@ -295,8 +294,9 @@ namespace transport

void SSUServer::CreateDirectSession (std::shared_ptr<const i2p::data::RouterInfo> router, boost::asio::ip::udp::endpoint remoteEndpoint, bool peerTest)
{
auto it = m_Sessions.find (remoteEndpoint);
if (it != m_Sessions.end ())
auto& sessions = remoteEndpoint.address ().is_v6 () ? m_SessionsV6 : m_Sessions;
auto it = sessions.find (remoteEndpoint);
if (it != sessions.end ())
{
auto session = it->second;
if (peerTest && session->GetState () == eSessionStateEstablished)
Expand All @@ -306,10 +306,7 @@ namespace transport
{
// otherwise create new session
auto session = std::make_shared<SSUSession> (*this, remoteEndpoint, router, peerTest);
{
std::unique_lock<std::mutex> l(m_SessionsMutex);
m_Sessions[remoteEndpoint] = session;
}
sessions[remoteEndpoint] = session;
// connect
LogPrint ("Creating new SSU session to [", i2p::data::GetIdentHashAbbreviation (router->GetIdentHash ()), "] ",
remoteEndpoint.address ().to_string (), ":", remoteEndpoint.port ());
Expand Down Expand Up @@ -360,15 +357,11 @@ namespace transport
introducer = &(address->introducers[0]); // TODO:
boost::asio::ip::udp::endpoint introducerEndpoint (introducer->iHost, introducer->iPort);
introducerSession = std::make_shared<SSUSession> (*this, introducerEndpoint, router);
std::unique_lock<std::mutex> l(m_SessionsMutex);
m_Sessions[introducerEndpoint] = introducerSession;
}
// create session
auto session = std::make_shared<SSUSession> (*this, remoteEndpoint, router, peerTest);
{
std::unique_lock<std::mutex> l(m_SessionsMutex);
m_Sessions[remoteEndpoint] = session;
}
m_Sessions[remoteEndpoint] = session;
// introduce
LogPrint ("Introduce new SSU session to [", i2p::data::GetIdentHashAbbreviation (router->GetIdentHash ()),
"] through introducer ", introducer->iHost, ":", introducer->iPort);
Expand All @@ -393,21 +386,27 @@ namespace transport
if (session)
{
session->Close ();
std::unique_lock<std::mutex> l(m_SessionsMutex);
m_Sessions.erase (session->GetRemoteEndpoint ());
auto& ep = session->GetRemoteEndpoint ();
if (ep.address ().is_v6 ())
m_SessionsV6.erase (ep);
else
m_Sessions.erase (ep);
}
}

void SSUServer::DeleteAllSessions ()
{
std::unique_lock<std::mutex> l(m_SessionsMutex);
for (auto it: m_Sessions)
it.second->Close ();
m_Sessions.clear ();

for (auto it: m_SessionsV6)
it.second->Close ();
m_SessionsV6.clear ();
}

template<typename Filter>
std::shared_ptr<SSUSession> SSUServer::GetRandomSession (Filter filter)
std::shared_ptr<SSUSession> SSUServer::GetRandomV4Session (Filter filter) // v4 only
{
std::vector<std::shared_ptr<SSUSession> > filteredSessions;
for (auto s :m_Sessions)
Expand All @@ -420,9 +419,9 @@ namespace transport
return nullptr;
}

std::shared_ptr<SSUSession> SSUServer::GetRandomEstablishedSession (std::shared_ptr<const SSUSession> excluded)
std::shared_ptr<SSUSession> SSUServer::GetRandomEstablishedV4Session (std::shared_ptr<const SSUSession> excluded) // v4 only
{
return GetRandomSession (
return GetRandomV4Session (
[excluded](std::shared_ptr<SSUSession> session)->bool
{
return session->GetState () == eSessionStateEstablished && !session->IsV6 () &&
Expand All @@ -437,7 +436,7 @@ namespace transport
std::set<SSUSession *> ret;
for (int i = 0; i < maxNumIntroducers; i++)
{
auto session = GetRandomSession (
auto session = GetRandomV4Session (
[&ret, ts](std::shared_ptr<SSUSession> session)->bool
{
return session->GetRelayTag () && !ret.count (session.get ()) &&
Expand Down
11 changes: 6 additions & 5 deletions SSU.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ namespace transport
void CreateSession (std::shared_ptr<const i2p::data::RouterInfo> router, bool peerTest = false);
std::shared_ptr<SSUSession> FindSession (std::shared_ptr<const i2p::data::RouterInfo> router) const;
std::shared_ptr<SSUSession> FindSession (const boost::asio::ip::udp::endpoint& e) const;
std::shared_ptr<SSUSession> GetRandomEstablishedSession (std::shared_ptr<const SSUSession> excluded);
std::shared_ptr<SSUSession> GetRandomEstablishedV4Session (std::shared_ptr<const SSUSession> excluded);
void DeleteSession (std::shared_ptr<SSUSession> session);
void DeleteAllSessions ();

Expand All @@ -69,12 +69,13 @@ namespace transport
void ReceiveV6 ();
void HandleReceivedFrom (const boost::system::error_code& ecode, std::size_t bytes_transferred, SSUPacket * packet);
void HandleReceivedFromV6 (const boost::system::error_code& ecode, std::size_t bytes_transferred, SSUPacket * packet);
void HandleReceivedPackets (std::vector<SSUPacket *> packets);
void HandleReceivedPackets (std::vector<SSUPacket *> packets,
std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSUSession> >& sessions);

void CreateSessionThroughIntroducer (std::shared_ptr<const i2p::data::RouterInfo> router, bool peerTest = false);
void CreateDirectSession (std::shared_ptr<const i2p::data::RouterInfo> router, boost::asio::ip::udp::endpoint remoteEndpoint, bool peerTest);
template<typename Filter>
std::shared_ptr<SSUSession> GetRandomSession (Filter filter);
std::shared_ptr<SSUSession> GetRandomV4Session (Filter filter);

std::set<SSUSession *> FindIntroducers (int maxNumIntroducers);
void ScheduleIntroducersUpdateTimer ();
Expand All @@ -100,14 +101,14 @@ namespace transport
boost::asio::ip::udp::socket m_Socket, m_SocketV6;
boost::asio::deadline_timer m_IntroducersUpdateTimer, m_PeerTestsCleanupTimer;
std::list<boost::asio::ip::udp::endpoint> m_Introducers; // introducers we are connected to
mutable std::mutex m_SessionsMutex;
std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSUSession> > m_Sessions;
std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSUSession> > m_Sessions, m_SessionsV6;
std::map<uint32_t, boost::asio::ip::udp::endpoint> m_Relays; // we are introducer
std::map<uint32_t, PeerTest> m_PeerTests; // nonce -> creation time in milliseconds

public:
// for HTTP only
const decltype(m_Sessions)& GetSessions () const { return m_Sessions; };
const decltype(m_SessionsV6)& GetSessionsV6 () const { return m_SessionsV6; };
};
}
}
Expand Down
2 changes: 1 addition & 1 deletion SSUSession.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -931,7 +931,7 @@ namespace transport
else
{
LogPrint (eLogDebug, "SSU peer test from Alice. We are Bob");
auto session = m_Server.GetRandomEstablishedSession (shared_from_this ()); // Charlie
auto session = m_Server.GetRandomEstablishedV4Session (shared_from_this ()); // Charlie, TODO: implement v6 support
if (session)
{
m_Server.NewPeerTest (nonce, ePeerTestParticipantBob, shared_from_this ());
Expand Down

0 comments on commit c5308e3

Please sign in to comment.