jiigen666/i2pd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix udp tunnels reload

Browse Source 
Signed-off-by: R4SAS <r4sas@i2pmail.org>
This commit is contained in:
R4SAS
2022-05-09 10:59:25 +03:00
parent 2cd3ebbdb3
commit 923eb9fdb3
4 changed files with 183 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
libi2pd_client/ClientContext.cpp
View File

@@ -623,6 +623,14 @@ namespace client
}
else
{
// TODO: update
if (ins.first->second->GetLocalDestination () != clientTunnel->GetLocalDestination ())
{
LogPrint (eLogInfo, "Clients: I2P UDP client tunnel destination updated");
ins.first->second->Stop ();
ins.first->second->SetLocalDestination (clientTunnel->GetLocalDestination ());
ins.first->second->Start ();
}
ins.first->second->isUpdated = true;
LogPrint(eLogError, "Clients: I2P Client forward for endpoint ", end, " already exists");
}
@@ -976,11 +984,11 @@ namespace client
}
}
/* // TODO: Write correct UDP tunnels stop
// TODO: Write correct UDP tunnels stop
for (auto it = m_ClientForwards.begin (); it != m_ClientForwards.end ();)
{
if(clean && !it->second->isUpdated) {
it->second = nullptr;
it->second->Stop ();
it = m_ClientForwards.erase(it);
} else {
it->second->isUpdated = false;
@@ -991,13 +999,13 @@ namespace client
for (auto it = m_ServerForwards.begin (); it != m_ServerForwards.end ();)
{
if(clean && !it->second->isUpdated) {
it->second = nullptr;
it->second->Stop ();
it = m_ServerForwards.erase(it);
} else {
it->second->isUpdated = false;
it++;
}
} */
}
}
}
}

122
libi2pd_client/I2PTunnel.cpp
View File

@@ -796,7 +796,8 @@ namespace client
}
}
void I2PUDPServerTunnel::ExpireStale(const uint64_t delta) {
void I2PUDPServerTunnel::ExpireStale(const uint64_t delta)
{
std::lock_guard<std::mutex> lock(m_SessionsMutex);
uint64_t now = i2p::util::GetMillisecondsSinceEpoch();
auto itr = m_Sessions.begin();
@@ -808,7 +809,8 @@ namespace client
}
}
void I2PUDPClientTunnel::ExpireStale(const uint64_t delta) {
void I2PUDPClientTunnel::ExpireStale(const uint64_t delta)
{
std::lock_guard<std::mutex> lock(m_SessionsMutex);
uint64_t now = i2p::util::GetMillisecondsSinceEpoch();
std::vector<uint16_t> removePorts;
@@ -864,7 +866,8 @@ namespace client
Receive();
}
void UDPSession::Receive() {
void UDPSession::Receive()
{
LogPrint(eLogDebug, "UDPSession: Receive");
IPSocket.async_receive_from(boost::asio::buffer(m_Buffer, I2P_UDP_MAX_MTU),
FromEndpoint, std::bind(&UDPSession::HandleReceived, this, std::placeholders::_1, std::placeholders::_2));
@@ -903,29 +906,28 @@ namespace client
I2PUDPServerTunnel::I2PUDPServerTunnel (const std::string & name, std::shared_ptr<i2p::client::ClientDestination> localDestination,
boost::asio::ip::address localAddress, boost::asio::ip::udp::endpoint forwardTo, uint16_t port, bool gzip) :
m_IsUniqueLocal(true),
m_Name(name),
m_LocalAddress(localAddress),
m_RemoteEndpoint(forwardTo)
m_IsUniqueLocal (true), m_Name (name), m_LocalAddress (localAddress), m_LocalDest (localDestination), m_RemoteEndpoint (forwardTo), m_Gzip (gzip)
{
m_LocalDest = localDestination;
m_LocalDest->Start();
auto dgram = m_LocalDest->CreateDatagramDestination(gzip);
dgram->SetReceiver(std::bind(&I2PUDPServerTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5));
dgram->SetRawReceiver(std::bind(&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
}
I2PUDPServerTunnel::~I2PUDPServerTunnel ()
{
auto dgram = m_LocalDest->GetDatagramDestination();
if (dgram) dgram->ResetReceiver();
LogPrint(eLogInfo, "UDPServer: Done");
Stop ();
}
void I2PUDPServerTunnel::Start ()
{
m_LocalDest->Start ();
auto dgram = m_LocalDest->CreateDatagramDestination (m_Gzip);
dgram->SetReceiver (std::bind (&I2PUDPServerTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5));
dgram->SetRawReceiver (std::bind (&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
}
void I2PUDPServerTunnel::Stop ()
{
auto dgram = m_LocalDest->GetDatagramDestination ();
if (dgram) dgram->ResetReceiver ();
}
std::vector<std::shared_ptr<DatagramSessionInfo> > I2PUDPServerTunnel::GetSessions ()
@@ -954,38 +956,67 @@ namespace client
boost::asio::ip::udp::endpoint localEndpoint,
std::shared_ptr<i2p::client::ClientDestination> localDestination,
uint16_t remotePort, bool gzip) :
m_Name(name),
m_RemoteDest(remoteDest),
m_LocalDest(localDestination),
m_LocalEndpoint(localEndpoint),
m_RemoteIdent(nullptr),
m_ResolveThread(nullptr),
m_LocalSocket(localDestination->GetService(), localEndpoint),
RemotePort(remotePort), m_LastPort (0),
m_cancel_resolve(false)
m_Name (name), m_RemoteDest (remoteDest), m_LocalDest (localDestination), m_LocalEndpoint (localEndpoint),
m_RemoteIdent (nullptr), m_ResolveThread (nullptr), m_LocalSocket (nullptr), RemotePort (remotePort),
m_LastPort (0), m_cancel_resolve (false), m_Gzip (gzip)
{
m_LocalSocket.set_option (boost::asio::socket_base::receive_buffer_size (I2P_UDP_MAX_MTU));
m_LocalSocket.set_option (boost::asio::socket_base::reuse_address (true));
}
auto dgram = m_LocalDest->CreateDatagramDestination(gzip);
I2PUDPClientTunnel::~I2PUDPClientTunnel ()
{
Stop ();
}
void I2PUDPClientTunnel::Start ()
{
// Reset flag in case of tunnel reload
if (m_cancel_resolve) m_cancel_resolve = false;
m_LocalSocket.reset (new boost::asio::ip::udp::socket (m_LocalDest->GetService (), m_LocalEndpoint));
m_LocalSocket->set_option (boost::asio::socket_base::receive_buffer_size (I2P_UDP_MAX_MTU));
m_LocalSocket->set_option (boost::asio::socket_base::reuse_address (true));
auto dgram = m_LocalDest->CreateDatagramDestination (m_Gzip);
dgram->SetReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2P, this,
std::placeholders::_1, std::placeholders::_2,
std::placeholders::_3, std::placeholders::_4,
std::placeholders::_5));
dgram->SetRawReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2PRaw, this,
std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
}
void I2PUDPClientTunnel::Start() {
m_LocalDest->Start ();
if (m_ResolveThread == nullptr)
m_ResolveThread = new std::thread (std::bind (&I2PUDPClientTunnel::TryResolving, this));
RecvFromLocal ();
}
void I2PUDPClientTunnel::Stop ()
{
auto dgram = m_LocalDest->GetDatagramDestination ();
if (dgram) dgram->ResetReceiver ();
m_cancel_resolve = true;
m_Sessions.clear();
if(m_LocalSocket && m_LocalSocket->is_open ())
m_LocalSocket->close ();
if(m_ResolveThread)
{
m_ResolveThread->join ();
delete m_ResolveThread;
m_ResolveThread = nullptr;
}
if (m_RemoteIdent)
{
delete m_RemoteIdent;
m_RemoteIdent = nullptr;
}
}
void I2PUDPClientTunnel::RecvFromLocal ()
{
m_LocalSocket.async_receive_from(boost::asio::buffer(m_RecvBuff, I2P_UDP_MAX_MTU),
m_LocalSocket->async_receive_from (boost::asio::buffer (m_RecvBuff, I2P_UDP_MAX_MTU),
m_RecvEndpoint, std::bind (&I2PUDPClientTunnel::HandleRecvFromLocal, this, std::placeholders::_1, std::placeholders::_2));
}
@@ -1030,9 +1061,9 @@ namespace client
while (numPackets < i2p::datagram::DATAGRAM_SEND_QUEUE_MAX_SIZE)
{
boost::system::error_code ec;
size_t moreBytes = m_LocalSocket.available(ec);
size_t moreBytes = m_LocalSocket->available (ec);
if (ec || !moreBytes) break;
transferred = m_LocalSocket.receive_from (boost::asio::buffer (m_RecvBuff, I2P_UDP_MAX_MTU), m_RecvEndpoint, 0, ec);
transferred = m_LocalSocket->receive_from (boost::asio::buffer (m_RecvBuff, I2P_UDP_MAX_MTU), m_RecvEndpoint, 0, ec);
remotePort = m_RecvEndpoint.port ();
// TODO: check remotePort
m_LocalDest->GetDatagramDestination ()->SendRawDatagram (session, m_RecvBuff, transferred, remotePort, RemotePort);
@@ -1055,7 +1086,8 @@ namespace client
return infos;
}
void I2PUDPClientTunnel::TryResolving() {
void I2PUDPClientTunnel::TryResolving ()
{
i2p::util::SetThreadName ("UDP Resolver");
LogPrint (eLogInfo, "UDP Tunnel: Trying to resolve ", m_RemoteDest);
@@ -1098,7 +1130,7 @@ namespace client
if (len > 0)
{
LogPrint (eLogDebug, "UDP Client: Got ", len, "B from ", m_RemoteIdent ? m_RemoteIdent->ToBase32 () : "");
m_LocalSocket.send_to(boost::asio::buffer(buf, len), itr->second->first);
m_LocalSocket->send_to (boost::asio::buffer (buf, len), itr->second->first);
// mark convo as active
itr->second->second = i2p::util::GetMillisecondsSinceEpoch ();
}
@@ -1106,25 +1138,5 @@ namespace client
else
LogPrint (eLogWarning, "UDP Client: Not tracking udp session using port ", (int) toPort);
}
I2PUDPClientTunnel::~I2PUDPClientTunnel()
{
auto dgram = m_LocalDest->GetDatagramDestination();
if (dgram) dgram->ResetReceiver();
m_cancel_resolve = true;
m_Sessions.clear();
if(m_LocalSocket.is_open())
m_LocalSocket.close();
if(m_ResolveThread)
{
m_ResolveThread->join();
delete m_ResolveThread;
m_ResolveThread = nullptr;
}
if (m_RemoteIdent) delete m_RemoteIdent;
}
}
}

15
libi2pd_client/I2PTunnel.h
View File

@@ -235,9 +235,11 @@ namespace client
boost::asio::ip::address localAddress,
boost::asio::ip::udp::endpoint forwardTo, uint16_t port, bool gzip);
~I2PUDPServerTunnel ();
/** expire stale udp conversations */
void ExpireStale (const uint64_t delta=I2P_UDP_SESSION_TIMEOUT);
void Start ();
void Stop ();
const char * GetName () const { return m_Name.c_str(); }
std::vector<std::shared_ptr<DatagramSessionInfo> > GetSessions ();
std::shared_ptr<ClientDestination> GetLocalDestination () const { return m_LocalDest; }
@@ -260,6 +262,7 @@ namespace client
std::vector<UDPSessionPtr> m_Sessions;
std::shared_ptr<i2p::client::ClientDestination> m_LocalDest;
UDPSessionPtr m_LastSession;
bool m_Gzip;
public:
@@ -274,13 +277,22 @@ namespace client
boost::asio::ip::udp::endpoint localEndpoint, std::shared_ptr<i2p::client::ClientDestination> localDestination,
uint16_t remotePort, bool gzip);
~I2PUDPClientTunnel ();
void Start ();
void Stop ();
const char * GetName () const { return m_Name.c_str(); }
std::vector<std::shared_ptr<DatagramSessionInfo> > GetSessions ();
bool IsLocalDestination (const i2p::data::IdentHash & destination) const { return destination == m_LocalDest->GetIdentHash(); }
std::shared_ptr<ClientDestination> GetLocalDestination () const { return m_LocalDest; }
inline void SetLocalDestination (std::shared_ptr<ClientDestination> dest)
{
if (m_LocalDest) m_LocalDest->Release ();
if (dest) dest->Acquire ();
m_LocalDest = dest;
}
void ExpireStale (const uint64_t delta=I2P_UDP_SESSION_TIMEOUT);
private:
@@ -302,11 +314,12 @@ namespace client
const boost::asio::ip::udp::endpoint m_LocalEndpoint;
i2p::data::IdentHash * m_RemoteIdent;
std::thread * m_ResolveThread;
boost::asio::ip::udp::socket m_LocalSocket;
std::unique_ptr<boost::asio::ip::udp::socket> m_LocalSocket;
boost::asio::ip::udp::endpoint m_RecvEndpoint;
uint8_t m_RecvBuff[I2P_UDP_MAX_MTU];
uint16_t RemotePort, m_LastPort;
bool m_cancel_resolve;
bool m_Gzip;
std::shared_ptr<UDPConvo> m_LastSession;
public: