diff --git a/libi2pd/SSU2.cpp b/libi2pd/SSU2.cpp index cfe5a04c..841bbd92 100644 --- a/libi2pd/SSU2.cpp +++ b/libi2pd/SSU2.cpp @@ -225,7 +225,10 @@ namespace transport } } if (port) + { OpenSocket (port); + Receive (); + } else LogPrint (eLogError, "SSU2: Can't start server because port not specified "); break; @@ -238,7 +241,7 @@ namespace transport { StopIOService (); } - + void SSU2Server::OpenSocket (int port) { try @@ -255,6 +258,36 @@ namespace transport ThrowFatal ("Unable to start SSU2 transport at port ", port, ": ", ex.what ()); } } + + void SSU2Server::Receive () + { + Packet * packet = m_PacketsPool.AcquireMt (); + m_Socket.async_receive_from (boost::asio::buffer (packet->buf, SSU2_MTU), packet->from, + std::bind (&SSU2Server::HandleReceivedFrom, this, std::placeholders::_1, std::placeholders::_2, packet)); + } + + void SSU2Server::HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred, Packet * packet) + { + if (!ecode) + { + packet->len = bytes_transferred; + ProcessNextPacket (packet->buf, packet->len, packet->from); + m_PacketsPool.ReleaseMt (packet); + Receive (); + } + else + { + m_PacketsPool.ReleaseMt (packet); + if (ecode != boost::asio::error::operation_aborted) + { + LogPrint (eLogError, "SSU2: Receive error: code ", ecode.value(), ": ", ecode.message ()); + auto port = m_Socket.local_endpoint ().port (); + m_Socket.close (); + OpenSocket (port); + Receive (); + } + } + } void SSU2Server::AddSession (uint64_t connID, std::shared_ptr session) { diff --git a/libi2pd/SSU2.h b/libi2pd/SSU2.h index 528fd9c0..b78e7811 100644 --- a/libi2pd/SSU2.h +++ b/libi2pd/SSU2.h @@ -24,6 +24,7 @@ namespace transport const int SSU2_TERMINATION_TIMEOUT = 330; // 5.5 minutes const size_t SSU2_SOCKET_RECEIVE_BUFFER_SIZE = 0x1FFFF; // 128K const size_t SSU2_SOCKET_SEND_BUFFER_SIZE = 0x1FFFF; // 128K + const size_t SSU2_MTU = 1488; enum SSU2MessageType { @@ -78,6 +79,13 @@ namespace transport class SSU2Server: private i2p::util::RunnableServiceWithWork { + struct Packet + { + uint8_t buf[SSU2_MTU]; + size_t len; + boost::asio::ip::udp::endpoint from; + }; + public: SSU2Server (); @@ -96,6 +104,8 @@ namespace transport private: void OpenSocket (int port); + void Receive (); + void HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred, Packet * packet); void ProcessNextPacket (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint); private: @@ -103,6 +113,7 @@ namespace transport boost::asio::ip::udp::socket m_Socket; std::unordered_map > m_Sessions; std::map > m_PendingOutgoingSessions; + i2p::util::MemoryPoolMt m_PacketsPool; }; } }