Merge pull request #18 from orignal/master

Merge pull request from orignal/master

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;

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<CryptoPP::AES>::Encryption m_Encryption;
 			CryptoPP::AutoSeededRandomPool m_Rnd;

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 () << "<BR>";
 		}	
 
@@ -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 () << "<BR>";
 		}	
 		

LeaseSet.cpp

@@ -57,13 +57,13 @@ namespace data
 			LogPrint ("LeaseSet verification failed");
 	}	
 
-	std::set<Lease> LeaseSet::GetNonExpiredLeases () const
+	const std::vector<Lease> LeaseSet::GetNonExpiredLeases () const
 	{
 		auto ts = i2p::util::GetMillisecondsSinceEpoch ();
-		std::set<Lease> leases;
+		std::vector<Lease> leases;
 		for (auto& it: m_Leases)
 			if (ts < it.endDate)
-				leases.insert (it);
+				leases.push_back (it);
 		return leases;	
 	}	
 

LeaseSet.h

@@ -4,7 +4,6 @@
 #include <inttypes.h>
 #include <string.h>
 #include <vector>
-#include <set>
 #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<Lease>& GetLeases () const { return m_Leases; };
-			std::set<Lease> GetNonExpiredLeases () const;
+			const std::vector<Lease> GetNonExpiredLeases () const;
 			bool HasExpiredLeases () const;
 			bool HasNonExpiredLeases () const;
 			const uint8_t * GetEncryptionPublicKey () const { return m_EncryptionKey; };

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 ());
+					std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
-					if (floodfill)
+					// request 3 closests floodfills
+					for (int i = 0; i < 3; i++)
 					{	
-						std::vector<i2p::tunnel::TunnelMessageBlock> msgs;
+						auto floodfill = GetClosestFloodfill (destination, dest->GetExcludedPeers ());
-						// DatabaseLookup message
+						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

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<IdentHash>& 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; };

Streaming.cpp

@@ -1,5 +1,6 @@
 #include <fstream>
 #include <algorithm>
+#include <boost/bind.hpp>
 #include <cryptopp/gzip.h>
 #include "Log.h"
 #include "RouterInfo.h"
@@ -14,12 +15,14 @@ namespace i2p
 {
 namespace stream
 {
-	Stream::Stream (StreamingDestination * local, const i2p::data::LeaseSet& remote):
+	Stream::Stream (boost::asio::io_service& service, StreamingDestination * local, 
-		m_SendStreamID (0), m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), 
+		const i2p::data::LeaseSet& remote): m_Service (service), m_SendStreamID (0), 
-		m_IsOpen (false), m_LeaseSetUpdated (true), m_LocalDestination (local), 
+		m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), m_IsOpen (false), 
-		m_RemoteLeaseSet (remote), m_OutboundTunnel (nullptr)
+		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 ();
+			auto ts = i2p::util::GetMillisecondsSinceEpoch ();
-			if (!leases.empty ())
+			if (ts >= m_CurrentRemoteLease.endDate)
+				UpdateCurrentRemoteLease ();
+			if (ts < m_CurrentRemoteLease.endDate)
 			{	
-				auto& lease = *leases.begin (); // TODO:
+				m_OutboundTunnel->SendTunnelDataMsg (m_CurrentRemoteLease.tunnelGateway, m_CurrentRemoteLease.tunnelID, msg);
-				m_OutboundTunnel->SendTunnelDataMsg (lease.tunnelGateway, lease.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)
+		return destinations.CreateClientStream (remote);
-			sharedLocalDestination = new StreamingDestination ();
-		return sharedLocalDestination->CreateNewStream (remote);
 	}
 		
 	void DeleteStream (Stream * stream)
 	{
-		if (sharedLocalDestination)
+		destinations.DeleteClientStream (stream);
-			sharedLocalDestination->DeleteStream (stream);
 	}	
 
 	void StartStreaming ()
 	{
-		if (!sharedLocalDestination)
+		destinations.Start ();
-			sharedLocalDestination = new StreamingDestination ();
 	}
 		
 	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
+			// then forward to streaming engine thread
-			// TODO: we have onle one destination, might be more
+			destinations.HandleNextPacket (destination, uncompressed);
-			if (sharedLocalDestination)
-				sharedLocalDestination->HandleNextPacket (uncompressed);
 		}	
 		else
 			LogPrint ("Data: protocol ", buf[9], " is not supported");

Streaming.h

@@ -5,6 +5,8 @@
 #include <string>
 #include <map>
 #include <set>
+#include <thread>
+#include <boost/asio.hpp>
 #include <cryptopp/dsa.h>
 #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<Packet> m_ReceiveQueue;
 			std::set<Packet *, PacketCmp> 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);
 }		
 }	

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++;
+			if (!it->IsFailed ())
-		}	
-		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)
 			{
 				tunnel = it;
-				minSent = it->GetNumSentBytes ();
+				i++;
-			}		
+			}
-		return tunnel;*/
+		}	
+		return tunnel;
 	}	
 
 	TunnelPool * Tunnels::CreateTunnelPool (i2p::data::LocalDestination * localDestination)

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<CryptoPP::AES>::Decryption m_ECBDecryption;
 			CryptoPP::CBC_Mode<CryptoPP::AES>::Decryption m_CBCDecryption;

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);
+			if (!it->IsFailed ())
-			i++;	
+			{
+				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);
+			it.second.first->SetFailed (true);
-			TunnelExpired (it.second.second);
+			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 ();
+			bool failed = false;
-			m_Tests[msgID] = std::make_pair (*it1, *it2);
+			if ((*it1)->IsFailed ())
-			(*it1)->SendTunnelDataMsg ((*it2)->GetNextIdentHash (), (*it2)->GetNextTunnelID (),
+			{	
-				CreateDeliveryStatusMsg (msgID));
+				failed = true;
-			it1++; it2++;
+				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++;
+			}	
 		}
 	}