Attempt at fixing the Disconnection issue.

* Added DataRate accounting to Relay and Dht.
 * Rates are exposed through rsDht.h interface.
 * Added LastIncomingTS() to pqistreamer.
 * Turned HeartBeat reset() into a warning.
 * Added NoPacket in 60 sec reset().
 * Minor typos/errors corrected.



git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@4818 b45a01b8-16f6-495d-af2f-9b41ad6348cc

libretroshare/src/dht/p3bitdht.cc

@@ -150,6 +150,8 @@ p3BitDht::p3BitDht(std::string id, pqiConnectCb *cb, p3NetMgr *nm,
 #endif
 
 	setupRelayDefaults();
+
+	clearDataRates();
 }
 
 p3BitDht::~p3BitDht()

libretroshare/src/dht/p3bitdht.h

@@ -165,6 +165,8 @@ virtual int     getRelayProxies(std::list<RsDhtRelayProxy> &relayProxies);
 
 virtual std::string getUdpAddressString();
 
+virtual void    getDhtRates(float &read, float &write);
+virtual void    getRelayRates(float &read, float &write, float &relay);
 
 /***********************************************************************************************
  ********** External RsDHT Interface (defined in libretroshare/src/retroshare/rsdht.h) *********
@@ -312,6 +314,21 @@ int     pushRelayServers();
         virtual bool    loadList(std::list<RsItem *>& load);
 /*****************************************************************/
 
+	// DATA RATES: Variables (Mutex Protected).
+	private:
+
+	void 	updateDataRates();
+	void	clearDataRates();
+
+	float mRelayReadRate;
+	float mRelayWriteRate;
+	float mRelayRelayRate;
+	float mDhtReadRate;
+	float mDhtWriteRate;
+
+	time_t mLastDataRateUpdate;
+
+
 /***********************************************************************************************
  ************************** Internal Accounting (p3bitdht_peers.cc) ****************************
 ************************************************************************************************/

libretroshare/src/dht/p3bitdht_interface.cc

@@ -194,6 +194,77 @@ std::string p3BitDht::getUdpAddressString()
 	return out.str();
 }
 
+void    p3BitDht::updateDataRates()
+{
+	uint32_t relayRead = 0;
+	uint32_t relayWrite = 0;
+	uint32_t relayRelay = 0;
+	uint32_t dhtRead = 0;
+	uint32_t dhtWrite = 0;
+
+	mRelay->getDataTransferred(relayRead, relayWrite, relayRelay);
+	mUdpBitDht->getDataTransferred(dhtRead, dhtWrite);
+
+	RsStackMutex stack(dhtMtx);    /********* LOCKED *********/
+
+	time_t now = time(NULL);
+	float period = now - mLastDataRateUpdate;
+
+#define RATE_FACTOR (0.75)
+	
+	mRelayReadRate *= RATE_FACTOR;
+	mRelayReadRate += (1.0 - RATE_FACTOR) * (relayRead / period);
+
+	mRelayWriteRate *= RATE_FACTOR;
+	mRelayWriteRate += (1.0 - RATE_FACTOR) * (relayWrite / period);
+
+	mRelayRelayRate *= RATE_FACTOR;
+	mRelayRelayRate += (1.0 - RATE_FACTOR) * (relayRelay / period);
+
+	mDhtReadRate *= RATE_FACTOR;
+	mDhtReadRate += (1.0 - RATE_FACTOR) * (dhtRead / period);
+
+	mDhtWriteRate *= RATE_FACTOR;
+	mDhtWriteRate += (1.0 - RATE_FACTOR) * (dhtWrite / period);
+
+	mLastDataRateUpdate = now;
+
+}
+
+void    p3BitDht::clearDataRates()
+{
+	RsStackMutex stack(dhtMtx);    /********* LOCKED *********/
+
+	mRelayReadRate = 0;
+	mRelayWriteRate = 0;
+	mRelayRelayRate = 0;
+	mDhtReadRate  = 0;
+	mDhtWriteRate = 0;
+
+	mLastDataRateUpdate = time(NULL);
+}
+
+
+/* in kB/s */
+void    p3BitDht::getDhtRates(float &read, float &write)
+{
+	RsStackMutex stack(dhtMtx);    /********* LOCKED *********/
+
+	read = mDhtReadRate / 1024.0;
+	write = mDhtWriteRate / 1024.0;
+}
+
+void    p3BitDht::getRelayRates(float &read, float &write, float &relay)
+{
+	RsStackMutex stack(dhtMtx);    /********* LOCKED *********/
+
+	read = mRelayReadRate / 1024.0;
+	write = mRelayWriteRate / 1024.0;
+	relay = mRelayRelayRate / 1024.0;
+}
+
+
+
 /***********************************************************************************************
  ********** External RsDHT Interface (defined in libretroshare/src/retroshare/rsdht.h) *********
 ************************************************************************************************/

libretroshare/src/dht/p3bitdht_peernet.cc

@@ -1103,6 +1103,29 @@ int p3BitDht::minuteTick()
 	{
 		mRelay->checkRelays();
 
+		updateDataRates();
+
+		/* temp - testing - print dht & relay traffic */
+		float dhtRead, dhtWrite;
+		float relayRead, relayWrite, relayRelayed;
+
+		getRelayRates(relayRead, relayWrite, relayRelayed);
+		getDhtRates(dhtRead, dhtWrite);
+		
+		double denom = deltaT;
+
+		std::cerr << "p3BitDht::minuteTick() ";
+		std::cerr << "DhtRead: " << dhtRead / denom << " kB/s ";
+		std::cerr << "DhtWrite: " << dhtWrite / denom << " kB/s ";
+		std::cerr << std::endl;
+
+		std::cerr << "p3BitDht::minuteTick() ";
+		std::cerr << "RelayRead: " << relayRead / denom << " kB/s ";
+		std::cerr << "RelayWrite: " << relayWrite / denom << " kB/s ";
+		std::cerr << "RelayRelayed: " << relayRelayed / denom << " kB/s ";
+		std::cerr << std::endl;
+
+
 		RsStackMutex stack(dhtMtx); /********** LOCKED MUTEX ***************/
 		mMinuteTS = now;
 	}

libretroshare/src/pqi/pqinetwork.cc

@@ -130,6 +130,10 @@ std::string socket_errorType(int err)
 	{
 		return std::string("ECONNRESET");
 	}
+	else if (err == EHOSTUNREACH)
+	{
+		return std::string("EHOSTUNREACH");
+	}
 	//
 
 	return std::string("UNKNOWN ERROR CODE - ASK RS-DEVS TO ADD IT!");

libretroshare/src/pqi/pqiperson.cc

@@ -102,7 +102,6 @@ int 	pqiperson::status()
 
 int 	pqiperson::receiveHeartbeat()
 {
-        //pqioutput(PQL_DEBUG_ALERT, pqipersonzone, "pqiperson::receiveHeartbeat() from peer : " + PeerId());
         pqioutput(PQL_WARNING, pqipersonzone, "pqiperson::receiveHeartbeat() from peer : " + PeerId());
         lastHeartbeatReceived = time(NULL);
 
@@ -116,13 +115,30 @@ int	pqiperson::tick()
 	if (active && (lastHeartbeatReceived != 0) &&
             (time(NULL) - lastHeartbeatReceived) > HEARTBEAT_REPEAT_TIME * 5) 
 	{
+		int ageLastIncoming = time(NULL) - activepqi->getLastIncomingTS();
 		std::ostringstream out;
-		out << "pqiperson::tick() No heartbeat from the peer, assume connection is dead. calling pqissl::reset(), LastHeartbeat was: ";
+		out << "pqiperson::tick() WARNING No heartbeat from: " << PeerId();
+		//out << " assume dead. calling pqissl::reset(), LastHeartbeat was: ";
+		out << " LastHeartbeat was: ";
 		out << time(NULL) - lastHeartbeatReceived << " secs ago";
+		out << " LastIncoming was: ";
+		out << ageLastIncoming << " secs ago";
             	pqioutput(PQL_WARNING, pqipersonzone, out.str());
+
+#define NO_PACKET_TIMEOUT 60
+		
+		if (ageLastIncoming > NO_PACKET_TIMEOUT)
+		{
+			std::ostringstream out2;
+			out2 << "pqiperson::tick() " << PeerId();
+			out2 << " No Heartbeat & No Packets -> assume dead. calling pqissl::reset()";
+            		pqioutput(PQL_WARNING, pqipersonzone, out2.str());
+            	
 			this->reset();
 		}
 
+	}
+
 	int activeTick = 0;
 
 	{

libretroshare/src/pqi/pqissl.cc

@@ -1530,6 +1530,8 @@ int 	pqissl::senddata(void *data, int len)
 		{
 			out << "SSL_write() SSL_ERROR_SYSCALL ";
 			out << "SOCKET_DEAD -> calling reset()";
+			out << " errno: " << errno;
+			out << " " << socket_errorType(errno);
 			std::cerr << out.str() << std::endl;
 			rslog(RSL_ALERT, pqisslzone, out.str());
 
@@ -1677,6 +1679,8 @@ int 	pqissl::readdata(void *data, int len)
 				out << "pqissl::readdata() " << PeerId();
 				out << " SSL_read() SSL_ERROR_SYSCALL";
 				out << " SOCKET_DEAD -> calling reset()";
+				out << " errno: " << errno;
+				out << " " << socket_errorType(errno);
 				rslog(RSL_ALERT, pqisslzone, out.str());
 
 				/* extra debugging - based on SSL_get_error() man page */

libretroshare/src/pqi/pqistreamer.cc

@@ -369,6 +369,8 @@ int 	pqistreamer::handleincomingitem(RsItem *pqi)
 		pqioutput(PQL_DEBUG_ALL, pqistreamerzone, out.str());
 	}
 #endif
+	// timestamp last received packet.
+	mLastIncomingTs = time(NULL);
 
 	// Use overloaded Contact function 
 	pqi -> PeerId(PeerId());
@@ -376,6 +378,11 @@ int 	pqistreamer::handleincomingitem(RsItem *pqi)
 	return 1;
 }
 
+time_t	pqistreamer::getLastIncomingTS()
+{
+	return mLastIncomingTs;
+}
+
 int	pqistreamer::handleoutgoing()
 {
 	RsStackMutex stack(streamerMtx) ;		// lock out_pkt and out_data

libretroshare/src/pqi/pqistreamer.h

@@ -58,6 +58,8 @@ class pqistreamer: public PQInterface
 		virtual int     tick();
 		virtual int     status();
 
+		time_t  getLastIncomingTS(); 	// Time of last data packet, for checking a connection is alive.
+
 	private:
 		/* Implementation */
 
@@ -113,8 +115,12 @@ class pqistreamer: public PQInterface
 		float avgReadCount;
 		float avgSentCount;
 
-		RsMutex streamerMtx ;
+		time_t mLastIncomingTs;
-		//	pthread_t thread_id;
+	
+		RsMutex streamerMtx ; 	// WHAT IS THIS PROTECTING. XXX
+		//	pthread_t thread_id;A
+
+
 };
 
 

libretroshare/src/retroshare/rsdht.h

@@ -176,6 +176,8 @@ virtual int 	getRelayProxies(std::list<RsDhtRelayProxy> &relayProxies) = 0;
 
 virtual std::string getUdpAddressString() = 0;
 
+virtual void    getDhtRates(float &read, float &write) = 0;
+virtual void    getRelayRates(float &read, float &write, float &relay) = 0;
 
 	// Interface for controlling Relays & DHT Relay Mode 
 virtual int 	getRelayServerList(std::list<std::string> &ids) = 0;

libretroshare/src/tcponudp/udprelay.cc

@@ -67,6 +67,8 @@ UdpRelayReceiver::UdpRelayReceiver(UdpPublisher *pub)
 	mTmpSendPkt = malloc(MAX_RELAY_UDP_PACKET_SIZE);
 	mTmpSendSize = MAX_RELAY_UDP_PACKET_SIZE;
 
+	clearDataTransferred();
+
 	return;
 }
 
@@ -672,6 +674,40 @@ int UdpRelayReceiver::UdpPeersStatus(std::ostream &out)
 }
 
 
+void UdpRelayReceiver::clearDataTransferred()
+{
+	{
+	        RsStackMutex stack(relayMtx);   /********** LOCK MUTEX *********/
+
+		mWriteBytes = 0;
+		mRelayBytes = 0;
+	}
+
+	{
+	        RsStackMutex stack(udppeerMtx);   /********** LOCK MUTEX *********/
+
+		mReadBytes = 0;
+	}
+}
+
+
+void UdpRelayReceiver::getDataTransferred(uint32_t &read, uint32_t &write, uint32_t &relay)
+{
+	{
+	        RsStackMutex stack(relayMtx);   /********** LOCK MUTEX *********/
+
+		write = mWriteBytes;
+		relay = mRelayBytes;
+	}
+
+	{
+	        RsStackMutex stack(udppeerMtx);   /********** LOCK MUTEX *********/
+
+		read = mReadBytes;
+	}
+	clearDataTransferred();
+}
+
 
 
 #define UDP_RELAY_HEADER_SIZE 16
@@ -724,6 +760,8 @@ int UdpRelayReceiver::recvPkt(void *data, int size, struct sockaddr_in &from)
 			rit->second.mLastTS = time(NULL);
 			rit->second.mDataSize += size;
 
+			mRelayBytes += size;
+	
 			mPublisher->sendPkt(data, size, rit->first.mDestAddr, STD_RELAY_TTL);
 			return 1;
 		}
@@ -745,6 +783,8 @@ int UdpRelayReceiver::recvPkt(void *data, int size, struct sockaddr_in &from)
 			std::cerr << pit->first;
 			std::cerr << std::endl;
 #endif
+			mReadBytes += size;
+
 			/* remove the header */
 			void *pktdata = (void *) (((uint8_t *) data) + UDP_RELAY_HEADER_SIZE);
 			int   pktsize = size - UDP_RELAY_HEADER_SIZE;
@@ -797,6 +837,9 @@ int UdpRelayReceiver::sendPkt(const void *data, int size, const struct sockaddr_
 	std::cerr << "UdpRelayReceiver::sendPkt() to Relay: " << it->second;
 	std::cerr << std::endl;
 #endif
+
+	mWriteBytes += size;
+
 	/* add a header to packet */
 	int finalPktSize = createRelayUdpPacket(data, size, mTmpSendPkt, MAX_RELAY_UDP_PACKET_SIZE, &(it->second));
 

libretroshare/src/tcponudp/udprelay.h

@@ -159,8 +159,12 @@ virtual int sendPkt(const void *data, int size, const struct sockaddr_in &to, in
 int     status(std::ostream &out);
 int 	UdpPeersStatus(std::ostream &out);
 
+void 	getDataTransferred(uint32_t &read, uint32_t &write, uint32_t &relay);
+
 	private:
 
+	void clearDataTransferred();
+
 	int removeUdpRelay_relayLocked(UdpRelayAddrSet *addrs);
 	int installRelayClass_relayLocked(int &classIdx, uint32_t &bandwidth);
 	int removeRelayClass_relayLocked(int classIdx);
@@ -177,6 +181,7 @@ int 	UdpPeersStatus(std::ostream &out);
 	RsMutex udppeerMtx; /* for all class data (below) */
 	
 	std::map<struct sockaddr_in, UdpPeer *> mPeers; /* indexed by <dest> */
+	uint32_t mReadBytes;
 
 	RsMutex relayMtx; /* for all class data (below) */
 
@@ -187,6 +192,9 @@ int 	UdpPeersStatus(std::ostream &out);
 	void *mTmpSendPkt;
 	uint32_t mTmpSendSize;
 
+	uint32_t mWriteBytes;
+	uint32_t mRelayBytes;
+
 };
 
 /* utility functions for creating / extracting UdpRelayPackets */