Mirrors/i2pd
2
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2025-02-02 11:04:00 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

handle SSU v4 and v6 messages in one thread

Browse source
This commit is contained in:
orignal 2020-10-14 21:06:51 -04:00
parent 2648f1ba89
commit 417b5ed6cc
4 changed files with 16 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
libi2pd/SSU.cpp
View file

@ -19,27 +19,25 @@ namespace transport
{ {
SSUServer::SSUServer (const boost::asio::ip::address & addr, int port): SSUServer::SSUServer (const boost::asio::ip::address & addr, int port):
m_OnlyV6(true), m_IsRunning(false), m_OnlyV6(true), m_IsRunning(false), m_Thread (nullptr),
m_Thread (nullptr), m_ThreadV6 (nullptr), m_ReceiversThread (nullptr), m_ReceiversThread (nullptr), m_ReceiversThreadV6 (nullptr), m_Work (m_Service),
m_ReceiversThreadV6 (nullptr), m_Work (m_Service), m_WorkV6 (m_ServiceV6),
m_ReceiversWork (m_ReceiversService), m_ReceiversWorkV6 (m_ReceiversServiceV6), m_ReceiversWork (m_ReceiversService), m_ReceiversWorkV6 (m_ReceiversServiceV6),
m_EndpointV6 (addr, port), m_Socket (m_ReceiversService, m_Endpoint), m_EndpointV6 (addr, port), m_Socket (m_ReceiversService, m_Endpoint),
m_SocketV6 (m_ReceiversServiceV6), m_IntroducersUpdateTimer (m_Service), m_SocketV6 (m_ReceiversServiceV6), m_IntroducersUpdateTimer (m_Service),
m_PeerTestsCleanupTimer (m_Service), m_TerminationTimer (m_Service), m_PeerTestsCleanupTimer (m_Service), m_TerminationTimer (m_Service),
m_TerminationTimerV6 (m_ServiceV6) m_TerminationTimerV6 (m_Service)
{ {
OpenSocketV6 (); OpenSocketV6 ();
} }
SSUServer::SSUServer (int port): SSUServer::SSUServer (int port):
m_OnlyV6(false), m_IsRunning(false), m_OnlyV6(false), m_IsRunning(false), m_Thread (nullptr),
m_Thread (nullptr), m_ThreadV6 (nullptr), m_ReceiversThread (nullptr), m_ReceiversThread (nullptr), m_ReceiversThreadV6 (nullptr), m_Work (m_Service),
m_ReceiversThreadV6 (nullptr), m_Work (m_Service), m_WorkV6 (m_ServiceV6),
m_ReceiversWork (m_ReceiversService), m_ReceiversWorkV6 (m_ReceiversServiceV6), m_ReceiversWork (m_ReceiversService), m_ReceiversWorkV6 (m_ReceiversServiceV6),
m_Endpoint (boost::asio::ip::udp::v4 (), port), m_EndpointV6 (boost::asio::ip::udp::v6 (), port), m_Endpoint (boost::asio::ip::udp::v4 (), port), m_EndpointV6 (boost::asio::ip::udp::v6 (), port),
m_Socket (m_ReceiversService), m_SocketV6 (m_ReceiversServiceV6), m_Socket (m_ReceiversService), m_SocketV6 (m_ReceiversServiceV6),
m_IntroducersUpdateTimer (m_Service), m_PeerTestsCleanupTimer (m_Service), m_IntroducersUpdateTimer (m_Service), m_PeerTestsCleanupTimer (m_Service),
m_TerminationTimer (m_Service), m_TerminationTimerV6 (m_ServiceV6) m_TerminationTimer (m_Service), m_TerminationTimerV6 (m_Service)
{ {
OpenSocket (); OpenSocket ();
if (context.SupportsV6 ()) if (context.SupportsV6 ())
@ -98,7 +96,8 @@ namespace transport
if (context.SupportsV6 ()) if (context.SupportsV6 ())
{ {
m_ReceiversThreadV6 = new std::thread (std::bind (&SSUServer::RunReceiversV6, this)); m_ReceiversThreadV6 = new std::thread (std::bind (&SSUServer::RunReceiversV6, this));
m_ThreadV6 = new std::thread (std::bind (&SSUServer::RunV6, this)); if (!m_Thread)
m_Thread = new std::thread (std::bind (&SSUServer::Run, this));
m_ReceiversServiceV6.post (std::bind (&SSUServer::ReceiveV6, this)); m_ReceiversServiceV6.post (std::bind (&SSUServer::ReceiveV6, this));
ScheduleTerminationV6 (); ScheduleTerminationV6 ();
} }
@ -114,7 +113,6 @@ namespace transport
m_TerminationTimerV6.cancel (); m_TerminationTimerV6.cancel ();
m_Service.stop (); m_Service.stop ();
m_Socket.close (); m_Socket.close ();
m_ServiceV6.stop ();
m_SocketV6.close (); m_SocketV6.close ();
m_ReceiversService.stop (); m_ReceiversService.stop ();
m_ReceiversServiceV6.stop (); m_ReceiversServiceV6.stop ();
@ -136,12 +134,6 @@ namespace transport
delete m_ReceiversThreadV6; delete m_ReceiversThreadV6;
m_ReceiversThreadV6 = nullptr; m_ReceiversThreadV6 = nullptr;
} }
if (m_ThreadV6)
{
m_ThreadV6->join ();
delete m_ThreadV6;
m_ThreadV6 = nullptr;
}
} }
void SSUServer::Run () void SSUServer::Run ()
@ -159,21 +151,6 @@ namespace transport
} }
} }
void SSUServer::RunV6 ()
{
while (m_IsRunning)
{
try
{
m_ServiceV6.run ();
}
catch (std::exception& ex)
{
LogPrint (eLogError, "SSU: v6 server runtime exception: ", ex.what ());
}
}
}
void SSUServer::RunReceivers () void SSUServer::RunReceivers ()
{ {
while (m_IsRunning) while (m_IsRunning)
@ -358,7 +335,7 @@ namespace transport
} }
} }
m_ServiceV6.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets, &m_SessionsV6)); m_Service.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets, &m_SessionsV6));
ReceiveV6 (); ReceiveV6 ();
} }
else else
@ -456,8 +433,7 @@ namespace transport
else else
{ {
boost::asio::ip::udp::endpoint remoteEndpoint (addr, port); boost::asio::ip::udp::endpoint remoteEndpoint (addr, port);
auto& s = addr.is_v6 () ? m_ServiceV6 : m_Service; m_Service.post (std::bind (&SSUServer::CreateDirectSession, this, router, remoteEndpoint, peerTest));
s.post (std::bind (&SSUServer::CreateDirectSession, this, router, remoteEndpoint, peerTest));
} }
} }
} }
@ -841,7 +817,7 @@ namespace transport
auto session = it.second; auto session = it.second;
if (it.first != session->GetRemoteEndpoint ()) if (it.first != session->GetRemoteEndpoint ())
LogPrint (eLogWarning, "SSU: remote endpoint ", session->GetRemoteEndpoint (), " doesn't match key ", it.first); LogPrint (eLogWarning, "SSU: remote endpoint ", session->GetRemoteEndpoint (), " doesn't match key ", it.first);
m_ServiceV6.post ([session] m_Service.post ([session]
{ {
LogPrint (eLogWarning, "SSU: no activity with ", session->GetRemoteEndpoint (), " for ", session->GetTerminationTimeout (), " seconds"); LogPrint (eLogWarning, "SSU: no activity with ", session->GetRemoteEndpoint (), " for ", session->GetTerminationTimeout (), " seconds");
session->Failed (); session->Failed ();

8
libi2pd/SSU.h
View file

@ -64,7 +64,6 @@ namespace transport
void DeleteAllSessions (); void DeleteAllSessions ();
boost::asio::io_service& GetService () { return m_Service; }; boost::asio::io_service& GetService () { return m_Service; };
boost::asio::io_service& GetServiceV6 () { return m_ServiceV6; };
const boost::asio::ip::udp::endpoint& GetEndpoint () const { return m_Endpoint; }; const boost::asio::ip::udp::endpoint& GetEndpoint () const { return m_Endpoint; };
void Send (const uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& to); void Send (const uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& to);
void AddRelay (uint32_t tag, std::shared_ptr<SSUSession> relay); void AddRelay (uint32_t tag, std::shared_ptr<SSUSession> relay);
@ -82,7 +81,6 @@ namespace transport
void OpenSocket (); void OpenSocket ();
void OpenSocketV6 (); void OpenSocketV6 ();
void Run (); void Run ();
void RunV6 ();
void RunReceivers (); void RunReceivers ();
void RunReceiversV6 (); void RunReceiversV6 ();
void Receive (); void Receive ();
@ -122,9 +120,9 @@ namespace transport
bool m_OnlyV6; bool m_OnlyV6;
bool m_IsRunning; bool m_IsRunning;
std::thread * m_Thread, * m_ThreadV6, * m_ReceiversThread, * m_ReceiversThreadV6; std::thread * m_Thread, * m_ReceiversThread, * m_ReceiversThreadV6;
boost::asio::io_service m_Service, m_ServiceV6, m_ReceiversService, m_ReceiversServiceV6; boost::asio::io_service m_Service, m_ReceiversService, m_ReceiversServiceV6;
boost::asio::io_service::work m_Work, m_WorkV6, m_ReceiversWork, m_ReceiversWorkV6; boost::asio::io_service::work m_Work, m_ReceiversWork, m_ReceiversWorkV6;
boost::asio::ip::udp::endpoint m_Endpoint, m_EndpointV6; boost::asio::ip::udp::endpoint m_Endpoint, m_EndpointV6;
boost::asio::ip::udp::socket m_Socket, m_SocketV6; boost::asio::ip::udp::socket m_Socket, m_SocketV6;
boost::asio::deadline_timer m_IntroducersUpdateTimer, m_PeerTestsCleanupTimer, boost::asio::deadline_timer m_IntroducersUpdateTimer, m_PeerTestsCleanupTimer,

2
libi2pd/SSUSession.cpp
View file

@ -49,7 +49,7 @@ namespace transport
boost::asio::io_service& SSUSession::GetService () boost::asio::io_service& SSUSession::GetService ()
{ {
return IsV6 () ? m_Server.GetServiceV6 () : m_Server.GetService (); return m_Server.GetService ();
} }
void SSUSession::CreateAESandMacKey (const uint8_t * pubKey) void SSUSession::CreateAESandMacKey (const uint8_t * pubKey)

2
libi2pd/Transports.cpp
View file

@ -501,7 +501,7 @@ namespace transport
{ {
auto addr = router->GetSSUV6Address (); auto addr = router->GetSSUV6Address ();
if (addr) if (addr)
m_SSUServer->GetServiceV6 ().post ([this, router, addr] m_SSUServer->GetService ().post ([this, router, addr]
{ {
m_SSUServer->CreateDirectSession (router, { addr->host, (uint16_t)addr->port }, false); m_SSUServer->CreateDirectSession (router, { addr->host, (uint16_t)addr->port }, false);
}); });