Browse Source

implement i2p.streaming.connectDelay option

pull/1019/head
orignal 7 years ago
parent
commit
a5b1b24fee
  1. 12
      libi2pd/Destination.cpp
  2. 6
      libi2pd/Destination.h
  3. 10
      libi2pd/Streaming.cpp
  4. 4
      libi2pd/Streaming.h
  5. 3
      libi2pd_client/ClientContext.cpp
  6. 4
      libi2pd_client/ClientContext.h

12
libi2pd/Destination.cpp

@ -715,8 +715,8 @@ namespace client
} }
ClientDestination::ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic, const std::map<std::string, std::string> * params): ClientDestination::ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic, const std::map<std::string, std::string> * params):
LeaseSetDestination (isPublic, params), LeaseSetDestination (isPublic, params), m_Keys (keys), m_StreamingAckDelay (DEFAULT_INITIAL_ACK_DELAY),
m_Keys (keys), m_DatagramDestination (nullptr), m_RefCounter (0), m_DatagramDestination (nullptr), m_RefCounter (0),
m_ReadyChecker(GetService()) m_ReadyChecker(GetService())
{ {
if (isPublic) if (isPublic)
@ -727,6 +727,14 @@ namespace client
m_Decryptor = m_Keys.CreateDecryptor (m_EncryptionPrivateKey); m_Decryptor = m_Keys.CreateDecryptor (m_EncryptionPrivateKey);
if (isPublic) if (isPublic)
LogPrint (eLogInfo, "Destination: Local address ", GetIdentHash().ToBase32 (), " created"); LogPrint (eLogInfo, "Destination: Local address ", GetIdentHash().ToBase32 (), " created");
// extract streaming params
if (params)
{
auto it = params->find (I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY);
if (it != params->end ())
m_StreamingAckDelay = std::stoi(it->second);
}
} }
ClientDestination::~ClientDestination () ClientDestination::~ClientDestination ()

6
libi2pd/Destination.h

@ -58,6 +58,10 @@ namespace client
const char I2CP_PARAM_MAX_TUNNEL_LATENCY[] = "latency.max"; const char I2CP_PARAM_MAX_TUNNEL_LATENCY[] = "latency.max";
const int DEFAULT_MAX_TUNNEL_LATENCY = 0; const int DEFAULT_MAX_TUNNEL_LATENCY = 0;
// streaming
const char I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY[] = "i2p.streaming.initialAckDelay";
const int DEFAULT_INITIAL_ACK_DELAY = 200; // milliseconds
typedef std::function<void (std::shared_ptr<i2p::stream::Stream> stream)> StreamRequestComplete; typedef std::function<void (std::shared_ptr<i2p::stream::Stream> stream)> StreamRequestComplete;
class LeaseSetDestination: public i2p::garlic::GarlicDestination, class LeaseSetDestination: public i2p::garlic::GarlicDestination,
@ -199,6 +203,7 @@ namespace client
void StopAcceptingStreams (); void StopAcceptingStreams ();
bool IsAcceptingStreams () const; bool IsAcceptingStreams () const;
void AcceptOnce (const i2p::stream::StreamingDestination::Acceptor& acceptor); void AcceptOnce (const i2p::stream::StreamingDestination::Acceptor& acceptor);
int GetStreamingAckDelay () const { return m_StreamingAckDelay; }
// datagram // datagram
i2p::datagram::DatagramDestination * GetDatagramDestination () const { return m_DatagramDestination; }; i2p::datagram::DatagramDestination * GetDatagramDestination () const { return m_DatagramDestination; };
@ -230,6 +235,7 @@ namespace client
uint8_t m_EncryptionPublicKey[256], m_EncryptionPrivateKey[256]; uint8_t m_EncryptionPublicKey[256], m_EncryptionPrivateKey[256];
std::shared_ptr<i2p::crypto::CryptoKeyDecryptor> m_Decryptor; std::shared_ptr<i2p::crypto::CryptoKeyDecryptor> m_Decryptor;
int m_StreamingAckDelay;
std::shared_ptr<i2p::stream::StreamingDestination> m_StreamingDestination; // default std::shared_ptr<i2p::stream::StreamingDestination> m_StreamingDestination; // default
std::map<uint16_t, std::shared_ptr<i2p::stream::StreamingDestination> > m_StreamingDestinationsByPorts; std::map<uint16_t, std::shared_ptr<i2p::stream::StreamingDestination> > m_StreamingDestinationsByPorts;
i2p::datagram::DatagramDestination * m_DatagramDestination; i2p::datagram::DatagramDestination * m_DatagramDestination;

10
libi2pd/Streaming.cpp

@ -62,6 +62,7 @@ namespace stream
m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_ResendTimer (m_Service), m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_ResendTimer (m_Service),
m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port), m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port),
m_WindowSize (MIN_WINDOW_SIZE), m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO), m_WindowSize (MIN_WINDOW_SIZE), m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO),
m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()),
m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0) m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0)
{ {
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); RAND_bytes ((uint8_t *)&m_RecvStreamID, 4);
@ -73,7 +74,8 @@ namespace stream
m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_LocalDestination (local), m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_LocalDestination (local),
m_ReceiveTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service), m_ReceiveTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service),
m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_WindowSize (MIN_WINDOW_SIZE), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_WindowSize (MIN_WINDOW_SIZE),
m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO), m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0) m_RTT (INITIAL_RTT), m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()),
m_LastWindowSizeIncreaseTime (0), m_NumResendAttempts (0)
{ {
RAND_bytes ((uint8_t *)&m_RecvStreamID, 4); RAND_bytes ((uint8_t *)&m_RecvStreamID, 4);
} }
@ -161,7 +163,7 @@ namespace stream
{ {
m_IsAckSendScheduled = true; m_IsAckSendScheduled = true;
auto ackTimeout = m_RTT/10; auto ackTimeout = m_RTT/10;
if (ackTimeout > ACK_SEND_TIMEOUT) ackTimeout = ACK_SEND_TIMEOUT; if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay;
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(ackTimeout)); m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(ackTimeout));
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer, m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer,
shared_from_this (), std::placeholders::_1)); shared_from_this (), std::placeholders::_1));
@ -199,7 +201,7 @@ namespace stream
{ {
// wait for SYN // wait for SYN
m_IsAckSendScheduled = true; m_IsAckSendScheduled = true;
m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(ACK_SEND_TIMEOUT)); m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(SYN_TIMEOUT));
m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer, m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer,
shared_from_this (), std::placeholders::_1)); shared_from_this (), std::placeholders::_1));
} }
@ -805,7 +807,7 @@ namespace stream
{ {
if (m_LastReceivedSequenceNumber < 0) if (m_LastReceivedSequenceNumber < 0)
{ {
LogPrint (eLogWarning, "Streaming: SYN has not been received after ", ACK_SEND_TIMEOUT, " milliseconds after follow on, terminate rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); LogPrint (eLogWarning, "Streaming: SYN has not been received after ", SYN_TIMEOUT, " milliseconds after follow on, terminate rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID);
m_Status = eStreamStatusReset; m_Status = eStreamStatusReset;
Close (); Close ();
return; return;

4
libi2pd/Streaming.h

@ -42,13 +42,13 @@ namespace stream
const size_t STREAMING_MTU = 1730; const size_t STREAMING_MTU = 1730;
const size_t MAX_PACKET_SIZE = 4096; const size_t MAX_PACKET_SIZE = 4096;
const size_t COMPRESSION_THRESHOLD_SIZE = 66; const size_t COMPRESSION_THRESHOLD_SIZE = 66;
const int ACK_SEND_TIMEOUT = 200; // in milliseconds
const int MAX_NUM_RESEND_ATTEMPTS = 6; const int MAX_NUM_RESEND_ATTEMPTS = 6;
const int WINDOW_SIZE = 6; // in messages const int WINDOW_SIZE = 6; // in messages
const int MIN_WINDOW_SIZE = 1; const int MIN_WINDOW_SIZE = 1;
const int MAX_WINDOW_SIZE = 128; const int MAX_WINDOW_SIZE = 128;
const int INITIAL_RTT = 8000; // in milliseconds const int INITIAL_RTT = 8000; // in milliseconds
const int INITIAL_RTO = 9000; // in milliseconds const int INITIAL_RTO = 9000; // in milliseconds
const int SYN_TIMEOUT = 200; // how long we wait for SYN after follow-on, in milliseconds
const size_t MAX_PENDING_INCOMING_BACKLOG = 128; const size_t MAX_PENDING_INCOMING_BACKLOG = 128;
const int PENDING_INCOMING_TIMEOUT = 10; // in seconds const int PENDING_INCOMING_TIMEOUT = 10; // in seconds
const int MAX_RECEIVE_TIMEOUT = 30; // in seconds const int MAX_RECEIVE_TIMEOUT = 30; // in seconds
@ -242,7 +242,7 @@ namespace stream
std::mutex m_SendBufferMutex; std::mutex m_SendBufferMutex;
SendBufferQueue m_SendBuffer; SendBufferQueue m_SendBuffer;
int m_WindowSize, m_RTT, m_RTO; int m_WindowSize, m_RTT, m_RTO, m_AckDelay;
uint64_t m_LastWindowSizeIncreaseTime; uint64_t m_LastWindowSizeIncreaseTime;
int m_NumResendAttempts; int m_NumResendAttempts;
}; };

3
libi2pd_client/ClientContext.cpp

@ -431,6 +431,7 @@ namespace client
options[I2CP_PARAM_TAGS_TO_SEND] = GetI2CPOption (section, I2CP_PARAM_TAGS_TO_SEND, DEFAULT_TAGS_TO_SEND); options[I2CP_PARAM_TAGS_TO_SEND] = GetI2CPOption (section, I2CP_PARAM_TAGS_TO_SEND, DEFAULT_TAGS_TO_SEND);
options[I2CP_PARAM_MIN_TUNNEL_LATENCY] = GetI2CPOption(section, I2CP_PARAM_MIN_TUNNEL_LATENCY, DEFAULT_MIN_TUNNEL_LATENCY); options[I2CP_PARAM_MIN_TUNNEL_LATENCY] = GetI2CPOption(section, I2CP_PARAM_MIN_TUNNEL_LATENCY, DEFAULT_MIN_TUNNEL_LATENCY);
options[I2CP_PARAM_MAX_TUNNEL_LATENCY] = GetI2CPOption(section, I2CP_PARAM_MAX_TUNNEL_LATENCY, DEFAULT_MAX_TUNNEL_LATENCY); options[I2CP_PARAM_MAX_TUNNEL_LATENCY] = GetI2CPOption(section, I2CP_PARAM_MAX_TUNNEL_LATENCY, DEFAULT_MAX_TUNNEL_LATENCY);
options[I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY] = GetI2CPOption(section, I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY, DEFAULT_INITIAL_ACK_DELAY);
} }
void ClientContext::ReadI2CPOptionsFromConfig (const std::string& prefix, std::map<std::string, std::string>& options) const void ClientContext::ReadI2CPOptionsFromConfig (const std::string& prefix, std::map<std::string, std::string>& options) const
@ -447,7 +448,7 @@ namespace client
if (i2p::config::GetOption(prefix + I2CP_PARAM_MIN_TUNNEL_LATENCY, value)) if (i2p::config::GetOption(prefix + I2CP_PARAM_MIN_TUNNEL_LATENCY, value))
options[I2CP_PARAM_MIN_TUNNEL_LATENCY] = value; options[I2CP_PARAM_MIN_TUNNEL_LATENCY] = value;
if (i2p::config::GetOption(prefix + I2CP_PARAM_MAX_TUNNEL_LATENCY, value)) if (i2p::config::GetOption(prefix + I2CP_PARAM_MAX_TUNNEL_LATENCY, value))
options[I2CP_PARAM_MAX_TUNNEL_LATENCY] = value; options[I2CP_PARAM_MAX_TUNNEL_LATENCY] = value;
} }
void ClientContext::ReadTunnels () void ClientContext::ReadTunnels ()

4
libi2pd_client/ClientContext.h

@ -90,8 +90,8 @@ namespace client
template<typename Section, typename Type> template<typename Section, typename Type>
std::string GetI2CPOption (const Section& section, const std::string& name, const Type& value) const; std::string GetI2CPOption (const Section& section, const std::string& name, const Type& value) const;
template<typename Section> template<typename Section>
void ReadI2CPOptions (const Section& section, std::map<std::string, std::string>& options) const; void ReadI2CPOptions (const Section& section, std::map<std::string, std::string>& options) const; // for tunnels
void ReadI2CPOptionsFromConfig (const std::string& prefix, std::map<std::string, std::string>& options) const; void ReadI2CPOptionsFromConfig (const std::string& prefix, std::map<std::string, std::string>& options) const; // for HTTP and SOCKS proxy
void CleanupUDP(const boost::system::error_code & ecode); void CleanupUDP(const boost::system::error_code & ecode);
void ScheduleCleanupUDP(); void ScheduleCleanupUDP();

Loading…
Cancel
Save