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

memory pool for sent packets

Browse source
This commit is contained in:
orignal 2022-08-04 18:13:44 -04:00
parent df073bb306
commit 3e40852999
4 changed files with 19 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
libi2pd/SSU2.cpp
View file

@ -742,6 +742,7 @@ namespace transport
} }
m_PacketsPool.CleanUpMt (); m_PacketsPool.CleanUpMt ();
m_SentPacketsPool.CleanUp ();
ScheduleTermination (); ScheduleTermination ();
} }
} }

3
libi2pd/SSU2.h
View file

@ -87,6 +87,8 @@ namespace transport
void RescheduleIntroducersUpdateTimer (); void RescheduleIntroducersUpdateTimer ();
void RescheduleIntroducersUpdateTimerV6 (); void RescheduleIntroducersUpdateTimerV6 ();
i2p::util::MemoryPool<SSU2SentPacket>& GetSentPacketsPool () { return m_SentPacketsPool; };
private: private:
@ -124,6 +126,7 @@ namespace transport
std::map<uint32_t, std::shared_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session std::map<uint32_t, std::shared_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session
std::list<i2p::data::IdentHash> m_Introducers, m_IntroducersV6; // introducers we are connected to std::list<i2p::data::IdentHash> m_Introducers, m_IntroducersV6; // introducers we are connected to
i2p::util::MemoryPoolMt<Packet> m_PacketsPool; i2p::util::MemoryPoolMt<Packet> m_PacketsPool;
i2p::util::MemoryPool<SSU2SentPacket> m_SentPacketsPool;
boost::asio::deadline_timer m_TerminationTimer, m_ResendTimer, boost::asio::deadline_timer m_TerminationTimer, m_ResendTimer,
m_IntroducersUpdateTimer, m_IntroducersUpdateTimerV6; m_IntroducersUpdateTimer, m_IntroducersUpdateTimerV6;
std::shared_ptr<SSU2Session> m_LastSession; std::shared_ptr<SSU2Session> m_LastSession;

12
libi2pd/SSU2Session.cpp
View file

@ -282,7 +282,7 @@ namespace transport
if (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize) if (!m_SendQueue.empty () && m_SentPackets.size () <= m_WindowSize)
{ {
auto ts = i2p::util::GetMillisecondsSinceEpoch (); auto ts = i2p::util::GetMillisecondsSinceEpoch ();
auto packet = std::make_shared<SentPacket>(); auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
size_t ackBlockSize = CreateAckBlock (packet->payload, m_MaxPayloadSize); size_t ackBlockSize = CreateAckBlock (packet->payload, m_MaxPayloadSize);
bool ackBlockSent = false; bool ackBlockSent = false;
packet->payloadSize += ackBlockSize; packet->payloadSize += ackBlockSize;
@ -304,7 +304,7 @@ namespace transport
else else
{ {
// create new packet and copy ack block // create new packet and copy ack block
auto newPacket = std::make_shared<SentPacket>(); auto newPacket = m_Server.GetSentPacketsPool ().AcquireShared ();
memcpy (newPacket->payload, packet->payload, ackBlockSize); memcpy (newPacket->payload, packet->payload, ackBlockSize);
newPacket->payloadSize = ackBlockSize; newPacket->payloadSize = ackBlockSize;
// complete current packet // complete current packet
@ -361,7 +361,7 @@ namespace transport
uint32_t msgID; uint32_t msgID;
memcpy (&msgID, msg->GetHeader () + I2NP_HEADER_MSGID_OFFSET, 4); memcpy (&msgID, msg->GetHeader () + I2NP_HEADER_MSGID_OFFSET, 4);
auto ts = i2p::util::GetMillisecondsSinceEpoch (); auto ts = i2p::util::GetMillisecondsSinceEpoch ();
auto packet = std::make_shared<SentPacket>(); auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
if (extraSize >= 8) if (extraSize >= 8)
{ {
packet->payloadSize = CreateAckBlock (packet->payload, extraSize); packet->payloadSize = CreateAckBlock (packet->payload, extraSize);
@ -371,7 +371,7 @@ namespace transport
uint32_t packetNum = SendData (packet->payload, packet->payloadSize); uint32_t packetNum = SendData (packet->payload, packet->payloadSize);
packet->sendTime = ts; packet->sendTime = ts;
m_SentPackets.emplace (packetNum, packet); m_SentPackets.emplace (packetNum, packet);
packet = std::make_shared<SentPacket>(); packet = m_Server.GetSentPacketsPool ().AcquireShared ();
} }
else else
extraSize -= packet->payloadSize; extraSize -= packet->payloadSize;
@ -389,7 +389,7 @@ namespace transport
while (msg->offset < msg->len) while (msg->offset < msg->len)
{ {
offset = extraSize > 0 ? (rand () % extraSize) : 0; offset = extraSize > 0 ? (rand () % extraSize) : 0;
packet = std::make_shared<SentPacket>(); packet = m_Server.GetSentPacketsPool ().AcquireShared ();
packet->payloadSize = CreateFollowOnFragmentBlock (packet->payload, m_MaxPayloadSize - offset, msg, fragmentNum, msgID); packet->payloadSize = CreateFollowOnFragmentBlock (packet->payload, m_MaxPayloadSize - offset, msg, fragmentNum, msgID);
extraSize -= offset; extraSize -= offset;
if (msg->offset >= msg->len && packet->payloadSize + 16 < m_MaxPayloadSize) // last fragment if (msg->offset >= msg->len && packet->payloadSize + 16 < m_MaxPayloadSize) // last fragment
@ -413,7 +413,7 @@ namespace transport
} }
// resend data packets // resend data packets
if (m_SentPackets.empty ()) return; if (m_SentPackets.empty ()) return;
std::map<uint32_t, std::shared_ptr<SentPacket> > resentPackets; std::map<uint32_t, std::shared_ptr<SSU2SentPacket> > resentPackets;
for (auto it = m_SentPackets.begin (); it != m_SentPackets.end (); ) for (auto it = m_SentPackets.begin (); it != m_SentPackets.end (); )
if (ts >= it->second->sendTime + it->second->numResends*m_RTO) if (ts >= it->second->sendTime + it->second->numResends*m_RTO)
{ {

18
libi2pd/SSU2Session.h
View file

@ -172,6 +172,14 @@ namespace transport
void AttachNextFragment (const uint8_t * fragment, size_t fragmentSize); void AttachNextFragment (const uint8_t * fragment, size_t fragmentSize);
}; };
struct SSU2SentPacket
{
uint8_t payload[SSU2_MAX_PACKET_SIZE];
size_t payloadSize = 0;
uint64_t sendTime; // in milliseconds
int numResends = 0;
};
// RouterInfo flags // RouterInfo flags
const uint8_t SSU2_ROUTER_INFO_FLAG_REQUEST_FLOOD = 0x01; const uint8_t SSU2_ROUTER_INFO_FLAG_REQUEST_FLOOD = 0x01;
const uint8_t SSU2_ROUTER_INFO_FLAG_GZIP = 0x02; const uint8_t SSU2_ROUTER_INFO_FLAG_GZIP = 0x02;
@ -192,14 +200,6 @@ namespace transport
} h; } h;
}; };
struct SentPacket
{
uint8_t payload[SSU2_MAX_PACKET_SIZE];
size_t payloadSize = 0;
uint64_t sendTime; // in milliseconds
int numResends = 0;
};
struct HandshakePacket struct HandshakePacket
{ {
Header header; Header header;
@ -327,7 +327,7 @@ namespace transport
uint8_t m_KeyDataSend[64], m_KeyDataReceive[64]; uint8_t m_KeyDataSend[64], m_KeyDataReceive[64];
uint32_t m_SendPacketNum, m_ReceivePacketNum; uint32_t m_SendPacketNum, m_ReceivePacketNum;
std::set<uint32_t> m_OutOfSequencePackets; // packet nums > receive packet num std::set<uint32_t> m_OutOfSequencePackets; // packet nums > receive packet num
std::map<uint32_t, std::shared_ptr<SentPacket> > m_SentPackets; // packetNum -> packet std::map<uint32_t, std::shared_ptr<SSU2SentPacket> > m_SentPackets; // packetNum -> packet
std::map<uint32_t, std::shared_ptr<SSU2IncompleteMessage> > m_IncompleteMessages; // I2NP std::map<uint32_t, std::shared_ptr<SSU2IncompleteMessage> > m_IncompleteMessages; // I2NP
std::map<uint32_t, std::pair <std::shared_ptr<SSU2Session>, uint64_t > > m_RelaySessions; // nonce->(Alice, timestamp) for Bob or nonce->(Charlie, timestamp) for Alice std::map<uint32_t, std::pair <std::shared_ptr<SSU2Session>, uint64_t > > m_RelaySessions; // nonce->(Alice, timestamp) for Bob or nonce->(Charlie, timestamp) for Alice
std::map<uint32_t, std::pair <std::shared_ptr<SSU2Session>, uint64_t > > m_PeerTests; // same as for relay sessions std::map<uint32_t, std::pair <std::shared_ptr<SSU2Session>, uint64_t > > m_PeerTests; // same as for relay sessions