Browse Source

use memory pool for SSU2 received packets arrays

gha
orignal 2 months ago
parent
commit
62b811c2c1
  1. 36
      libi2pd/SSU2.cpp
  2. 18
      libi2pd/SSU2.h
  3. 10
      libi2pd/util.h

36
libi2pd/SSU2.cpp

@ -368,9 +368,9 @@ namespace transport
size_t moreBytes = socket.available (ec); size_t moreBytes = socket.available (ec);
if (!ec && moreBytes) if (!ec && moreBytes)
{ {
std::vector<Packet *> packets; auto packets = m_PacketsArrayPool.AcquireMt ();
packets.push_back (packet); packets->AddPacket (packet);
while (moreBytes && packets.size () < SSU2_MAX_NUM_PACKETS_PER_BATCH) while (moreBytes && packets->numPackets < SSU2_MAX_NUM_PACKETS_PER_BATCH)
{ {
packet = m_PacketsPool.AcquireMt (); packet = m_PacketsPool.AcquireMt ();
packet->len = socket.receive_from (boost::asio::buffer (packet->buf, SSU2_MAX_PACKET_SIZE), packet->from, 0, ec); packet->len = socket.receive_from (boost::asio::buffer (packet->buf, SSU2_MAX_PACKET_SIZE), packet->from, 0, ec);
@ -378,7 +378,13 @@ namespace transport
{ {
i2p::transport::transports.UpdateReceivedBytes (packet->len); i2p::transport::transports.UpdateReceivedBytes (packet->len);
if (packet->len >= SSU2_MIN_RECEIVED_PACKET_SIZE) if (packet->len >= SSU2_MIN_RECEIVED_PACKET_SIZE)
packets.push_back (packet); {
if (!packets->AddPacket (packet))
{
LogPrint (eLogError, "SSU2: Received packets array is full");
m_PacketsPool.ReleaseMt (packet);
}
}
else // drop too short packets else // drop too short packets
m_PacketsPool.ReleaseMt (packet); m_PacketsPool.ReleaseMt (packet);
moreBytes = socket.available(ec); moreBytes = socket.available(ec);
@ -391,10 +397,7 @@ namespace transport
break; break;
} }
} }
GetService ().post ([packets = std::move (packets), this]() mutable GetService ().post (std::bind (&SSU2Server::HandleReceivedPackets, this, packets));
{
HandleReceivedPackets (std::move (packets));
});
} }
else else
GetService ().post (std::bind (&SSU2Server::HandleReceivedPacket, this, packet)); GetService ().post (std::bind (&SSU2Server::HandleReceivedPacket, this, packet));
@ -438,15 +441,23 @@ namespace transport
} }
} }
void SSU2Server::HandleReceivedPackets (std::vector<Packet *>&& packets) void SSU2Server::HandleReceivedPackets (Packets * packets)
{ {
if (!packets) return;
if (m_IsThroughProxy) if (m_IsThroughProxy)
for (auto& packet: packets) for (size_t i = 0; i < packets->numPackets; i++)
{
auto& packet = (*packets)[i];
ProcessNextPacketFromProxy (packet->buf, packet->len); ProcessNextPacketFromProxy (packet->buf, packet->len);
}
else else
for (auto& packet: packets) for (size_t i = 0; i < packets->numPackets; i++)
{
auto& packet = (*packets)[i];
ProcessNextPacket (packet->buf, packet->len, packet->from); ProcessNextPacket (packet->buf, packet->len, packet->from);
m_PacketsPool.ReleaseMt (std::move (packets)); }
m_PacketsPool.ReleaseMt (packets->data (), packets->numPackets);
m_PacketsArrayPool.ReleaseMt (packets);
if (m_LastSession && m_LastSession->GetState () != eSSU2SessionStateTerminated) if (m_LastSession && m_LastSession->GetState () != eSSU2SessionStateTerminated)
m_LastSession->FlushData (); m_LastSession->FlushData ();
} }
@ -1098,6 +1109,7 @@ namespace transport
} }
m_PacketsPool.CleanUpMt (); m_PacketsPool.CleanUpMt ();
m_PacketsArrayPool.CleanUpMt ();
m_SentPacketsPool.CleanUp (); m_SentPacketsPool.CleanUp ();
m_IncompleteMessagesPool.CleanUp (); m_IncompleteMessagesPool.CleanUp ();
m_FragmentsPool.CleanUp (); m_FragmentsPool.CleanUp ();

18
libi2pd/SSU2.h

@ -12,6 +12,7 @@
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
#include <array>
#include <mutex> #include <mutex>
#include <random> #include <random>
#include "util.h" #include "util.h"
@ -51,6 +52,20 @@ namespace transport
boost::asio::ip::udp::endpoint from; boost::asio::ip::udp::endpoint from;
}; };
struct Packets: public std::array<Packet *, SSU2_MAX_NUM_PACKETS_PER_BATCH>
{
size_t numPackets = 0;
bool AddPacket (Packet *p)
{
if (p && numPackets < size ())
{
data()[numPackets] = p; numPackets++;
return true;
}
return false;
}
};
class ReceiveService: public i2p::util::RunnableService class ReceiveService: public i2p::util::RunnableService
{ {
public: public:
@ -127,7 +142,7 @@ namespace transport
void HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred, void HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred,
Packet * packet, boost::asio::ip::udp::socket& socket); Packet * packet, boost::asio::ip::udp::socket& socket);
void HandleReceivedPacket (Packet * packet); void HandleReceivedPacket (Packet * packet);
void HandleReceivedPackets (std::vector<Packet *>&& packets); void HandleReceivedPackets (Packets * packets);
void ProcessNextPacket (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint); void ProcessNextPacket (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint);
void ScheduleTermination (); void ScheduleTermination ();
@ -172,6 +187,7 @@ namespace transport
std::unordered_map<uint32_t, std::weak_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session std::unordered_map<uint32_t, std::weak_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session
std::list<std::pair<i2p::data::IdentHash, uint32_t> > m_Introducers, m_IntroducersV6; // introducers we are connected to std::list<std::pair<i2p::data::IdentHash, uint32_t> > m_Introducers, m_IntroducersV6; // introducers we are connected to
i2p::util::MemoryPoolMt<Packet> m_PacketsPool; i2p::util::MemoryPoolMt<Packet> m_PacketsPool;
i2p::util::MemoryPoolMt<Packets> m_PacketsArrayPool;
i2p::util::MemoryPool<SSU2SentPacket> m_SentPacketsPool; i2p::util::MemoryPool<SSU2SentPacket> m_SentPacketsPool;
i2p::util::MemoryPool<SSU2IncompleteMessage> m_IncompleteMessagesPool; i2p::util::MemoryPool<SSU2IncompleteMessage> m_IncompleteMessagesPool;
i2p::util::MemoryPool<SSU2IncompleteMessage::Fragment> m_FragmentsPool; i2p::util::MemoryPool<SSU2IncompleteMessage::Fragment> m_FragmentsPool;

10
libi2pd/util.h

@ -131,8 +131,16 @@ namespace util
this->Release (t); this->Release (t);
} }
void ReleaseMt (T * * arr, size_t num)
{
if (!arr || !num) return;
std::lock_guard<std::mutex> l(m_Mutex);
for (size_t i = 0; i < num; i++)
this->Release (arr[i]);
}
template<template<typename, typename...>class C, typename... R> template<template<typename, typename...>class C, typename... R>
void ReleaseMt(C<T *, R...>&& c) void ReleaseMt(const C<T *, R...>& c)
{ {
std::lock_guard<std::mutex> l(m_Mutex); std::lock_guard<std::mutex> l(m_Mutex);
for (auto& it: c) for (auto& it: c)

Loading…
Cancel
Save