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

insert multiple packets to the queue using splice

Browse source
This commit is contained in:
orignal 2024-10-15 15:05:18 -04:00
parent 7104d334fd
commit ec1f41b13c
2 changed files with 33 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
libi2pd/SSU2.cpp
View file

@ -367,23 +367,22 @@ namespace transport
return; return;
} }
packet->len = bytes_transferred; packet->len = bytes_transferred;
InsertToReceivedPacketsQueue (packet);
size_t numPackets = 1;
boost::system::error_code ec; boost::system::error_code ec;
size_t moreBytes = socket.available (ec); size_t moreBytes = socket.available (ec);
if (!ec && moreBytes) if (!ec && moreBytes)
{ {
do std::list<Packet *> packets;
packets.push_back (packet);
while (moreBytes && packets.size () < 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);
if (!ec) if (!ec)
{ {
i2p::transport::transports.UpdateReceivedBytes (packet->len); i2p::transport::transports.UpdateReceivedBytes (packet->len);
numPackets++;
if (packet->len >= SSU2_MIN_RECEIVED_PACKET_SIZE) if (packet->len >= SSU2_MIN_RECEIVED_PACKET_SIZE)
InsertToReceivedPacketsQueue (packet); packets.push_back (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);
@ -396,8 +395,10 @@ namespace transport
break; break;
} }
} }
while (moreBytes && numPackets < SSU2_MAX_NUM_PACKETS_PER_BATCH); InsertToReceivedPacketsQueue (packets);
} }
else
InsertToReceivedPacketsQueue (packet);
Receive (socket); Receive (socket);
} }
else else
@ -448,8 +449,23 @@ namespace transport
m_ReceivedPacketsQueue.push_back (packet); m_ReceivedPacketsQueue.push_back (packet);
} }
if (empty) if (empty)
GetService ().post([this]() { HandleReceivedPacketsQueue (); });
}
void SSU2Server::InsertToReceivedPacketsQueue (std::list<Packet *>& packets)
{ {
GetService ().post([this]() if (packets.empty ()) return;
bool empty = false;
{
std::lock_guard<std::mutex> l(m_ReceivedPacketsQueueMutex);
empty = m_ReceivedPacketsQueue.empty ();
m_ReceivedPacketsQueue.splice (m_ReceivedPacketsQueue.end (), packets);
}
if (empty)
GetService ().post([this]() { HandleReceivedPacketsQueue (); });
}
void SSU2Server::HandleReceivedPacketsQueue ()
{ {
std::list<Packet *> receivedPackets; std::list<Packet *> receivedPackets;
{ {
@ -457,8 +473,6 @@ namespace transport
m_ReceivedPacketsQueue.swap (receivedPackets); m_ReceivedPacketsQueue.swap (receivedPackets);
} }
HandleReceivedPackets (std::move (receivedPackets)); HandleReceivedPackets (std::move (receivedPackets));
});
}
} }
void SSU2Server::AddSession (std::shared_ptr<SSU2Session> session) void SSU2Server::AddSession (std::shared_ptr<SSU2Session> session)

2
libi2pd/SSU2.h
View file

@ -135,6 +135,8 @@ namespace transport
void HandleReceivedPackets (std::list<Packet *>&& packets); void HandleReceivedPackets (std::list<Packet *>&& 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 InsertToReceivedPacketsQueue (Packet * packet); void InsertToReceivedPacketsQueue (Packet * packet);
void InsertToReceivedPacketsQueue (std::list<Packet *>& packets);
void HandleReceivedPacketsQueue ();
void ScheduleTermination (); void ScheduleTermination ();
void HandleTerminationTimer (const boost::system::error_code& ecode); void HandleTerminationTimer (const boost::system::error_code& ecode);