diff --git a/Garlic.cpp b/Garlic.cpp index 94d09a89..b4b4623a 100644 --- a/Garlic.cpp +++ b/Garlic.cpp @@ -16,7 +16,7 @@ namespace garlic { GarlicRoutingSession::GarlicRoutingSession (const i2p::data::RoutingDestination& destination, int numTags): m_Destination (destination), m_FirstMsgID (0), m_IsAcknowledged (false), - m_NumTags (numTags), m_NextTag (-1), m_SessionTags (0) + m_NumTags (numTags), m_NextTag (-1), m_SessionTags (0), m_TagsCreationTime (0) { // create new session tags and session key m_Rnd.GenerateBlock (m_SessionKey, 32); @@ -40,6 +40,8 @@ namespace garlic { for (int i = 0; i < m_NumTags; i++) m_Rnd.GenerateBlock (m_SessionTags + i*32, 32); + m_TagsCreationTime = i2p::util::GetSecondsSinceEpoch (); + SetAcknowledged (false); } } @@ -48,6 +50,24 @@ namespace garlic I2NPMessage * m = NewI2NPMessage (); size_t len = 0; uint8_t * buf = m->GetPayload () + 4; // 4 bytes for length + + // take care about tags + if (m_NumTags > 0) + { + if (i2p::util::GetSecondsSinceEpoch () >= m_TagsCreationTime + TAGS_EXPIRATION_TIMEOUT) + { + // old tags expired create new set + LogPrint ("Garlic tags expired"); + GenerateSessionTags (); + m_NextTag = -1; + } + else if (!m_IsAcknowledged) // new set of tags was not acknowledged + { + LogPrint ("Previous garlic tags was not acknowledged. Use ElGamal"); + m_NextTag = -1; // have to use ElGamal + } + } + // create message if (m_NextTag < 0 || !m_NumTags) // new session { // create ElGamal block @@ -253,7 +273,7 @@ namespace garlic session = it->second; if (!session) { - session = new GarlicRoutingSession (destination, 4); // TODO: change it later + session = new GarlicRoutingSession (destination, 32); m_Sessions[destination.GetIdentHash ()] = session; } @@ -372,7 +392,7 @@ namespace garlic // later on we should let destination decide I2NPHeader * header = (I2NPHeader *)buf; if (header->typeID == eI2NPData) - i2p::stream::HandleDataMessage (&destination, buf + sizeof (I2NPHeader), be16toh (header->size)); + i2p::stream::HandleDataMessage (destination, buf + sizeof (I2NPHeader), be16toh (header->size)); else LogPrint ("Unexpected I2NP garlic message ", (int)header->typeID); break; diff --git a/Garlic.h b/Garlic.h index f6c61a06..03fd6fe6 100644 --- a/Garlic.h +++ b/Garlic.h @@ -35,7 +35,7 @@ namespace garlic }; #pragma pack() - + const int TAGS_EXPIRATION_TIMEOUT = 900; // 15 minutes class GarlicRoutingSession { public: @@ -66,6 +66,7 @@ namespace garlic bool m_IsAcknowledged; int m_NumTags, m_NextTag; uint8_t * m_SessionTags; // m_NumTags*32 bytes + uint32_t m_TagsCreationTime; // seconds since epoch CryptoPP::CBC_Mode::Encryption m_Encryption; CryptoPP::AutoSeededRandomPool m_Rnd; diff --git a/HTTPServer.cpp b/HTTPServer.cpp index f914fefd..1628ff7b 100644 --- a/HTTPServer.cpp +++ b/HTTPServer.cpp @@ -141,6 +141,8 @@ namespace util it->GetTunnelConfig ()->Print (s); if (it->GetTunnelPool ()) s << " " << "Pool"; + if (it->IsFailed ()) + s << " " << "Failed"; s << " " << (int)it->GetNumSentBytes () << "
"; } @@ -149,6 +151,8 @@ namespace util it.second->GetTunnelConfig ()->Print (s); if (it.second->GetTunnelPool ()) s << " " << "Pool"; + if (it.second->IsFailed ()) + s << " " << "Failed"; s << " " << (int)it.second->GetNumReceivedBytes () << "
"; } diff --git a/LeaseSet.cpp b/LeaseSet.cpp index b62a4598..1383792e 100644 --- a/LeaseSet.cpp +++ b/LeaseSet.cpp @@ -57,13 +57,13 @@ namespace data LogPrint ("LeaseSet verification failed"); } - std::set LeaseSet::GetNonExpiredLeases () const + const std::vector LeaseSet::GetNonExpiredLeases () const { auto ts = i2p::util::GetMillisecondsSinceEpoch (); - std::set leases; + std::vector leases; for (auto& it: m_Leases) if (ts < it.endDate) - leases.insert (it); + leases.push_back (it); return leases; } diff --git a/LeaseSet.h b/LeaseSet.h index 6bcc8122..9c805845 100644 --- a/LeaseSet.h +++ b/LeaseSet.h @@ -4,7 +4,6 @@ #include #include #include -#include #include "Identity.h" namespace i2p @@ -43,7 +42,7 @@ namespace data const Identity& GetIdentity () const { return m_Identity; }; const IdentHash& GetIdentHash () const { return m_IdentHash; }; const std::vector& GetLeases () const { return m_Leases; }; - std::set GetNonExpiredLeases () const; + const std::vector GetNonExpiredLeases () const; bool HasExpiredLeases () const; bool HasNonExpiredLeases () const; const uint8_t * GetEncryptionPublicKey () const { return m_EncryptionKey; }; diff --git a/NetDb.cpp b/NetDb.cpp index 3d8a4212..57b5029a 100644 --- a/NetDb.cpp +++ b/NetDb.cpp @@ -42,6 +42,11 @@ namespace data return msg; } + void RequestedDestination::ClearExcludedPeers () + { + m_ExcludedPeers.clear (); + } + #ifndef _WIN32 const char NetDb::m_NetDbPath[] = "/netDb"; #else @@ -335,19 +340,26 @@ namespace data if (inbound) { RequestedDestination * dest = CreateRequestedDestination (destination, isLeaseSet); - auto floodfill = GetClosestFloodfill (destination, dest->GetExcludedPeers ()); - if (floodfill) + std::vector msgs; + // request 3 closests floodfills + for (int i = 0; i < 3; i++) { - std::vector msgs; - // DatabaseLookup message + auto floodfill = GetClosestFloodfill (destination, dest->GetExcludedPeers ()); + if (floodfill) + { + // DatabaseLookup message + msgs.push_back (i2p::tunnel::TunnelMessageBlock + { + i2p::tunnel::eDeliveryTypeRouter, + floodfill->GetIdentHash (), 0, + dest->CreateRequestMessage (floodfill, inbound) + }); + } + } + if (msgs.size () > 0) + { + dest->ClearExcludedPeers (); dest->SetLastOutboundTunnel (outbound); - msgs.push_back (i2p::tunnel::TunnelMessageBlock - { - i2p::tunnel::eDeliveryTypeRouter, - floodfill->GetIdentHash (), 0, - dest->CreateRequestMessage (floodfill, inbound) - }); - outbound->SendTunnelDataMsg (msgs); } else diff --git a/NetDb.h b/NetDb.h index a369459b..35fbe844 100644 --- a/NetDb.h +++ b/NetDb.h @@ -30,7 +30,8 @@ namespace data const IdentHash& GetDestination () const { return m_Destination; }; int GetNumExcludedPeers () const { return m_ExcludedPeers.size (); }; const std::set& GetExcludedPeers () { return m_ExcludedPeers; }; - const RouterInfo * GetLastRouter () const { return m_LastRouter; }; + void ClearExcludedPeers (); + const RouterInfo * GetLastRouter () const { return m_LastRouter; }; const i2p::tunnel::InboundTunnel * GetLastReplyTunnel () const { return m_LastReplyTunnel; }; bool IsExploratory () const { return m_IsExploratory; }; bool IsLeaseSet () const { return m_IsLeaseSet; }; diff --git a/Streaming.cpp b/Streaming.cpp index eb942bd5..40bc8929 100644 --- a/Streaming.cpp +++ b/Streaming.cpp @@ -1,5 +1,6 @@ #include #include +#include #include #include "Log.h" #include "RouterInfo.h" @@ -14,12 +15,14 @@ namespace i2p { namespace stream { - Stream::Stream (StreamingDestination * local, const i2p::data::LeaseSet& remote): - m_SendStreamID (0), m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), - m_IsOpen (false), m_LeaseSetUpdated (true), m_LocalDestination (local), - m_RemoteLeaseSet (remote), m_OutboundTunnel (nullptr) + Stream::Stream (boost::asio::io_service& service, StreamingDestination * local, + const i2p::data::LeaseSet& remote): m_Service (service), m_SendStreamID (0), + m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), m_IsOpen (false), + m_LeaseSetUpdated (true), m_LocalDestination (local), m_RemoteLeaseSet (remote), + m_OutboundTunnel (nullptr) { m_RecvStreamID = i2p::context.GetRandomNumberGenerator ().GenerateWord32 (); + UpdateCurrentRemoteLease (); } Stream::~Stream () @@ -62,6 +65,7 @@ namespace stream { // we have received duplicate. Most likely our outbound tunnel is dead LogPrint ("Duplicate message ", receivedSeqn, " received"); + UpdateCurrentRemoteLease (); // pick another lease m_OutboundTunnel = i2p::tunnel::tunnels.GetNextOutboundTunnel (); // pick another tunnel if (m_OutboundTunnel) SendQuickAck (); // resend ack for previous message again @@ -276,11 +280,12 @@ namespace stream m_OutboundTunnel = m_LocalDestination->GetTunnelPool ()->GetNextOutboundTunnel (); if (m_OutboundTunnel) { - auto leases = m_RemoteLeaseSet.GetNonExpiredLeases (); - if (!leases.empty ()) + auto ts = i2p::util::GetMillisecondsSinceEpoch (); + if (ts >= m_CurrentRemoteLease.endDate) + UpdateCurrentRemoteLease (); + if (ts < m_CurrentRemoteLease.endDate) { - auto& lease = *leases.begin (); // TODO: - m_OutboundTunnel->SendTunnelDataMsg (lease.tunnelGateway, lease.tunnelID, msg); + m_OutboundTunnel->SendTunnelDataMsg (m_CurrentRemoteLease.tunnelGateway, m_CurrentRemoteLease.tunnelID, msg); return true; } else @@ -296,8 +301,19 @@ namespace stream } return false; } + + void Stream::UpdateCurrentRemoteLease () + { + auto leases = m_RemoteLeaseSet.GetNonExpiredLeases (); + if (!leases.empty ()) + { + uint32_t i = i2p::context.GetRandomNumberGenerator ().GenerateWord32 (0, leases.size () - 1); + m_CurrentRemoteLease = leases[i]; + } + else + m_CurrentRemoteLease.endDate = 0; + } - StreamingDestination * sharedLocalDestination = nullptr; StreamingDestination::StreamingDestination (): m_LeaseSet (nullptr) { @@ -339,9 +355,10 @@ namespace stream } } - Stream * StreamingDestination::CreateNewStream (const i2p::data::LeaseSet& remote) + Stream * StreamingDestination::CreateNewStream (boost::asio::io_service& service, + const i2p::data::LeaseSet& remote) { - Stream * s = new Stream (this, remote); + Stream * s = new Stream (service, this, remote); m_Streams[s->GetRecvStreamID ()] = s; return s; } @@ -402,14 +419,14 @@ namespace stream size += 32; // tunnel_gw *(uint32_t *)(buf + size) = htobe32 (tunnel->GetNextTunnelID ()); size += 4; // tunnel_id - uint64_t ts = tunnel->GetCreationTime () + i2p::tunnel::TUNNEL_EXPIRATION_TIMEOUT; + uint64_t ts = tunnel->GetCreationTime () + i2p::tunnel::TUNNEL_EXPIRATION_TIMEOUT - 60; // 1 minute before expiration ts *= 1000; // in milliseconds *(uint64_t *)(buf + size) = htobe64 (ts); size += 8; // end_date } Sign (buf, size, buf+ size); size += 40; // signature - + LogPrint ("Local LeaseSet of ", tunnels.size (), " leases created"); m->len += size + sizeof (I2NPDatabaseStoreMsg); FillI2NPMessageHeader (m, eI2NPDatabaseStore); return m; @@ -420,32 +437,83 @@ namespace stream CryptoPP::DSA::Signer signer (m_SigningPrivateKey); signer.SignMessage (i2p::context.GetRandomNumberGenerator (), buf, len, signature); } + + StreamingDestinations destinations; + void StreamingDestinations::Start () + { + if (!m_SharedLocalDestination) + m_SharedLocalDestination = new StreamingDestination (); + + m_IsRunning = true; + m_Thread = new std::thread (std::bind (&StreamingDestinations::Run, this)); + } + + void StreamingDestinations::Stop () + { + delete m_SharedLocalDestination; + + m_IsRunning = false; + m_Service.stop (); + if (m_Thread) + { + m_Thread->join (); + delete m_Thread; + m_Thread = 0; + } + } + + void StreamingDestinations::Run () + { + m_Service.run (); + } + + Stream * StreamingDestinations::CreateClientStream (const i2p::data::LeaseSet& remote) + { + if (!m_SharedLocalDestination) return nullptr; + return m_SharedLocalDestination->CreateNewStream (m_Service, remote); + } + + void StreamingDestinations::DeleteClientStream (Stream * stream) + { + if (m_SharedLocalDestination) + m_SharedLocalDestination->DeleteStream (stream); + else + delete stream; + } + + void StreamingDestinations::HandleNextPacket (i2p::data::IdentHash destination, Packet * packet) + { + m_Service.post (boost::bind (&StreamingDestinations::PostNextPacket, this, destination, packet)); + } + + void StreamingDestinations::PostNextPacket (i2p::data::IdentHash destination, Packet * packet) + { + // TODO: we have onle one destination, might be more + if (m_SharedLocalDestination) + m_SharedLocalDestination->HandleNextPacket (packet); + } Stream * CreateStream (const i2p::data::LeaseSet& remote) { - if (!sharedLocalDestination) - sharedLocalDestination = new StreamingDestination (); - return sharedLocalDestination->CreateNewStream (remote); + return destinations.CreateClientStream (remote); } void DeleteStream (Stream * stream) { - if (sharedLocalDestination) - sharedLocalDestination->DeleteStream (stream); + destinations.DeleteClientStream (stream); } void StartStreaming () { - if (!sharedLocalDestination) - sharedLocalDestination = new StreamingDestination (); + destinations.Start (); } void StopStreaming () { - delete sharedLocalDestination; + destinations.Stop (); } - void HandleDataMessage (i2p::data::IdentHash * destination, const uint8_t * buf, size_t len) + void HandleDataMessage (i2p::data::IdentHash destination, const uint8_t * buf, size_t len) { uint32_t length = be32toh (*(uint32_t *)buf); buf += 4; @@ -465,10 +533,8 @@ namespace stream uncompressed->len = MAX_PACKET_SIZE; } decompressor.Get (uncompressed->buf, uncompressed->len); - // then forward to streaming engine - // TODO: we have onle one destination, might be more - if (sharedLocalDestination) - sharedLocalDestination->HandleNextPacket (uncompressed); + // then forward to streaming engine thread + destinations.HandleNextPacket (destination, uncompressed); } else LogPrint ("Data: protocol ", buf[9], " is not supported"); diff --git a/Streaming.h b/Streaming.h index b9f52c1a..484479b2 100644 --- a/Streaming.h +++ b/Streaming.h @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include #include "I2PEndian.h" #include "Queue.h" @@ -67,7 +69,7 @@ namespace stream { public: - Stream (StreamingDestination * local, const i2p::data::LeaseSet& remote); + Stream (boost::asio::io_service& service, StreamingDestination * local, const i2p::data::LeaseSet& remote); ~Stream (); uint32_t GetSendStreamID () const { return m_SendStreamID; }; uint32_t GetRecvStreamID () const { return m_RecvStreamID; }; @@ -90,13 +92,17 @@ namespace stream void SavePacket (Packet * packet); void ProcessPacket (Packet * packet); + + void UpdateCurrentRemoteLease (); private: + boost::asio::io_service& m_Service; uint32_t m_SendStreamID, m_RecvStreamID, m_SequenceNumber, m_LastReceivedSequenceNumber; bool m_IsOpen, m_LeaseSetUpdated; StreamingDestination * m_LocalDestination; const i2p::data::LeaseSet& m_RemoteLeaseSet; + i2p::data::Lease m_CurrentRemoteLease; i2p::util::Queue m_ReceiveQueue; std::set m_SavedPackets; i2p::tunnel::OutboundTunnel * m_OutboundTunnel; @@ -116,7 +122,7 @@ namespace stream i2p::tunnel::TunnelPool * GetTunnelPool () const { return m_Pool; }; void Sign (uint8_t * buf, int len, uint8_t * signature) const; - Stream * CreateNewStream (const i2p::data::LeaseSet& remote); + Stream * CreateNewStream (boost::asio::io_service& service, const i2p::data::LeaseSet& remote); void DeleteStream (Stream * stream); void HandleNextPacket (Packet * packet); @@ -139,13 +145,44 @@ namespace stream CryptoPP::DSA::PrivateKey m_SigningPrivateKey; }; + class StreamingDestinations + { + public: + + StreamingDestinations (): m_IsRunning (false), m_Thread (nullptr), + m_Work (m_Service), m_SharedLocalDestination (nullptr) {}; + ~StreamingDestinations () {}; + + void Start (); + void Stop (); + + void HandleNextPacket (i2p::data::IdentHash destination, Packet * packet); + + Stream * CreateClientStream (const i2p::data::LeaseSet& remote); + void DeleteClientStream (Stream * stream); + + private: + + void Run (); + void PostNextPacket (i2p::data::IdentHash destination, Packet * packet); + + private: + + bool m_IsRunning; + std::thread * m_Thread; + boost::asio::io_service m_Service; + boost::asio::io_service::work m_Work; + + StreamingDestination * m_SharedLocalDestination; + }; + Stream * CreateStream (const i2p::data::LeaseSet& remote); void DeleteStream (Stream * stream); void StartStreaming (); void StopStreaming (); // assuming data is I2CP message - void HandleDataMessage (i2p::data::IdentHash * destination, const uint8_t * buf, size_t len); + void HandleDataMessage (i2p::data::IdentHash destination, const uint8_t * buf, size_t len); I2NPMessage * CreateDataMessage (Stream * s, uint8_t * payload, size_t len); } } diff --git a/Tunnel.cpp b/Tunnel.cpp index 0ebd6345..c32c5e0e 100644 --- a/Tunnel.cpp +++ b/Tunnel.cpp @@ -14,7 +14,8 @@ namespace i2p namespace tunnel { - Tunnel::Tunnel (TunnelConfig * config): m_Config (config), m_Pool (nullptr), m_IsEstablished (false) + Tunnel::Tunnel (TunnelConfig * config): m_Config (config), m_Pool (nullptr), + m_IsEstablished (false), m_IsFailed (false) { } @@ -130,6 +131,7 @@ namespace tunnel void InboundTunnel::HandleTunnelDataMsg (I2NPMessage * msg) { + if (IsFailed ()) SetFailed (false); // incoming messages means a tunnel is alive msg->from = this; EncryptTunnelMsg (msg); m_Endpoint.HandleDecryptedTunnelDataMsg (msg); @@ -225,11 +227,14 @@ namespace tunnel InboundTunnel * tunnel = nullptr; size_t minReceived = 0; for (auto it : m_InboundTunnels) + { + if (it.second->IsFailed ()) continue; if (!tunnel || it.second->GetNumReceivedBytes () < minReceived) { tunnel = it.second; minReceived = it.second->GetNumReceivedBytes (); - } + } + } return tunnel; } @@ -240,7 +245,7 @@ namespace tunnel for (auto it : m_InboundTunnels) { if (i >= num) break; - if (it.second->GetNextIdentHash () != i2p::context.GetRouterInfo ().GetIdentHash ()) + if (!it.second->IsFailed () && it.second->GetNextIdentHash () != i2p::context.GetRouterInfo ().GetIdentHash ()) { // exclude one hop tunnels v.push_back (it.second); @@ -254,23 +259,17 @@ namespace tunnel { CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator (); uint32_t ind = rnd.GenerateWord32 (0, m_OutboundTunnels.size () - 1), i = 0; + OutboundTunnel * tunnel = nullptr; for (auto it: m_OutboundTunnels) { if (i >= ind) return it; - else i++; - } - return nullptr; - - // TODO: implement it base on profiling - /*OutboundTunnel * tunnel = nullptr; - size_t minSent = 0; - for (auto it : m_OutboundTunnels) - if (!tunnel || it->GetNumSentBytes () < minSent) + if (!it->IsFailed ()) { tunnel = it; - minSent = it->GetNumSentBytes (); - } - return tunnel;*/ + i++; + } + } + return tunnel; } TunnelPool * Tunnels::CreateTunnelPool (i2p::data::LocalDestination * localDestination) diff --git a/Tunnel.h b/Tunnel.h index 4374c631..6711bd3d 100644 --- a/Tunnel.h +++ b/Tunnel.h @@ -37,6 +37,9 @@ namespace tunnel TunnelConfig * GetTunnelConfig () const { return m_Config; } bool IsEstablished () const { return m_IsEstablished; }; + bool IsFailed () const { return m_IsFailed; }; + void SetFailed (bool failed) { m_IsFailed = failed; } + TunnelPool * GetTunnelPool () const { return m_Pool; }; void SetTunnelPool (TunnelPool * pool) { m_Pool = pool; }; @@ -57,7 +60,7 @@ namespace tunnel TunnelConfig * m_Config; TunnelPool * m_Pool; // pool, tunnel belongs to, or null - bool m_IsEstablished; + bool m_IsEstablished, m_IsFailed; CryptoPP::ECB_Mode::Decryption m_ECBDecryption; CryptoPP::CBC_Mode::Decryption m_CBCDecryption; diff --git a/TunnelPool.cpp b/TunnelPool.cpp index d1da6e6d..f71a63c3 100644 --- a/TunnelPool.cpp +++ b/TunnelPool.cpp @@ -31,8 +31,6 @@ namespace tunnel void TunnelPool::TunnelCreated (InboundTunnel * createdTunnel) { m_InboundTunnels.insert (createdTunnel); - if (m_LocalDestination) - m_LocalDestination->UpdateLeaseSet (); } void TunnelPool::TunnelExpired (InboundTunnel * expiredTunnel) @@ -54,10 +52,10 @@ namespace tunnel void TunnelPool::TunnelExpired (OutboundTunnel * expiredTunnel) { if (expiredTunnel) - { + { expiredTunnel->SetTunnelPool (nullptr); m_OutboundTunnels.erase (expiredTunnel); - } + } if (expiredTunnel == m_LastOutboundTunnel) m_LastOutboundTunnel = nullptr; } @@ -69,8 +67,11 @@ namespace tunnel for (auto it : m_InboundTunnels) { if (i >= num) break; - v.push_back (it); - i++; + if (!it->IsFailed ()) + { + v.push_back (it); + i++; + } } return v; } @@ -82,7 +83,7 @@ namespace tunnel if (m_LastOutboundTunnel && tunnel == m_LastOutboundTunnel) { for (auto it: m_OutboundTunnels) - if (it != m_LastOutboundTunnel) + if (it != m_LastOutboundTunnel && !it->IsFailed ()) { tunnel = it; break; @@ -109,19 +110,33 @@ namespace tunnel { LogPrint ("Tunnel test ", (int)it.first, " failed"); // both outbound and inbound tunnels considered as invalid - TunnelExpired (it.second.first); - TunnelExpired (it.second.second); + it.second.first->SetFailed (true); + it.second.second->SetFailed (true); } m_Tests.clear (); auto it1 = m_OutboundTunnels.begin (); auto it2 = m_InboundTunnels.begin (); while (it1 != m_OutboundTunnels.end () && it2 != m_InboundTunnels.end ()) { - uint32_t msgID = rnd.GenerateWord32 (); - m_Tests[msgID] = std::make_pair (*it1, *it2); - (*it1)->SendTunnelDataMsg ((*it2)->GetNextIdentHash (), (*it2)->GetNextTunnelID (), - CreateDeliveryStatusMsg (msgID)); - it1++; it2++; + bool failed = false; + if ((*it1)->IsFailed ()) + { + failed = true; + it1++; + } + if ((*it2)->IsFailed ()) + { + failed = true; + it2++; + } + if (!failed) + { + uint32_t msgID = rnd.GenerateWord32 (); + m_Tests[msgID] = std::make_pair (*it1, *it2); + (*it1)->SendTunnelDataMsg ((*it2)->GetNextIdentHash (), (*it2)->GetNextTunnelID (), + CreateDeliveryStatusMsg (msgID)); + it1++; it2++; + } } }