From 4aec00a0c209dd7d81c2b7304c054ae5a437063a Mon Sep 17 00:00:00 2001 From: drbob Date: Thu, 19 Jan 2012 16:23:57 +0000 Subject: [PATCH] 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 | 2 + libretroshare/src/dht/p3bitdht.h | 17 +++++ libretroshare/src/dht/p3bitdht_interface.cc | 71 +++++++++++++++++++++ libretroshare/src/dht/p3bitdht_peernet.cc | 23 +++++++ libretroshare/src/pqi/pqinetwork.cc | 4 ++ libretroshare/src/pqi/pqiperson.cc | 28 ++++++-- libretroshare/src/pqi/pqissl.cc | 6 +- libretroshare/src/pqi/pqistreamer.cc | 7 ++ libretroshare/src/pqi/pqistreamer.h | 10 ++- libretroshare/src/retroshare/rsdht.h | 2 + libretroshare/src/tcponudp/udprelay.cc | 43 +++++++++++++ libretroshare/src/tcponudp/udprelay.h | 8 +++ 12 files changed, 212 insertions(+), 9 deletions(-) diff --git a/libretroshare/src/dht/p3bitdht.cc b/libretroshare/src/dht/p3bitdht.cc index 167fcdd81..5b1a90d76 100644 --- a/libretroshare/src/dht/p3bitdht.cc +++ b/libretroshare/src/dht/p3bitdht.cc @@ -150,6 +150,8 @@ p3BitDht::p3BitDht(std::string id, pqiConnectCb *cb, p3NetMgr *nm, #endif setupRelayDefaults(); + + clearDataRates(); } p3BitDht::~p3BitDht() diff --git a/libretroshare/src/dht/p3bitdht.h b/libretroshare/src/dht/p3bitdht.h index bed2b379f..969639241 100644 --- a/libretroshare/src/dht/p3bitdht.h +++ b/libretroshare/src/dht/p3bitdht.h @@ -165,6 +165,8 @@ virtual int getRelayProxies(std::list &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& 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) **************************** ************************************************************************************************/ diff --git a/libretroshare/src/dht/p3bitdht_interface.cc b/libretroshare/src/dht/p3bitdht_interface.cc index ae28ab032..8753768ff 100644 --- a/libretroshare/src/dht/p3bitdht_interface.cc +++ b/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) ********* ************************************************************************************************/ diff --git a/libretroshare/src/dht/p3bitdht_peernet.cc b/libretroshare/src/dht/p3bitdht_peernet.cc index 1ec63604e..d7f6e6513 100644 --- a/libretroshare/src/dht/p3bitdht_peernet.cc +++ b/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; } diff --git a/libretroshare/src/pqi/pqinetwork.cc b/libretroshare/src/pqi/pqinetwork.cc index 64d0260cf..1467bd732 100644 --- a/libretroshare/src/pqi/pqinetwork.cc +++ b/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!"); diff --git a/libretroshare/src/pqi/pqiperson.cc b/libretroshare/src/pqi/pqiperson.cc index 7349a54fc..07f00cddd 100644 --- a/libretroshare/src/pqi/pqiperson.cc +++ b/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); @@ -113,17 +112,34 @@ int pqiperson::receiveHeartbeat() int pqiperson::tick() { //if lastHeartbeatReceived is 0, it might be not activated so don't do a net reset. - if (active && (lastHeartbeatReceived != 0) && + 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()); - this->reset(); - } - int activeTick = 0; +#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; { std::ostringstream out; diff --git a/libretroshare/src/pqi/pqissl.cc b/libretroshare/src/pqi/pqissl.cc index bb3879a2d..b3524b693 100644 --- a/libretroshare/src/pqi/pqissl.cc +++ b/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()); @@ -1676,7 +1678,9 @@ int pqissl::readdata(void *data, int len) { out << "pqissl::readdata() " << PeerId(); out << " SSL_read() SSL_ERROR_SYSCALL"; - out << "SOCKET_DEAD -> calling reset()"; + 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 */ diff --git a/libretroshare/src/pqi/pqistreamer.cc b/libretroshare/src/pqi/pqistreamer.cc index 456848f32..c29ac57c0 100644 --- a/libretroshare/src/pqi/pqistreamer.cc +++ b/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 diff --git a/libretroshare/src/pqi/pqistreamer.h b/libretroshare/src/pqi/pqistreamer.h index e01477b84..188c207f5 100644 --- a/libretroshare/src/pqi/pqistreamer.h +++ b/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 ; - // pthread_t thread_id; + time_t mLastIncomingTs; + + RsMutex streamerMtx ; // WHAT IS THIS PROTECTING. XXX + // pthread_t thread_id;A + + }; diff --git a/libretroshare/src/retroshare/rsdht.h b/libretroshare/src/retroshare/rsdht.h index b8ae12794..cdd43ebaf 100644 --- a/libretroshare/src/retroshare/rsdht.h +++ b/libretroshare/src/retroshare/rsdht.h @@ -176,6 +176,8 @@ virtual int getRelayProxies(std::list &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 &ids) = 0; diff --git a/libretroshare/src/tcponudp/udprelay.cc b/libretroshare/src/tcponudp/udprelay.cc index ba9b61597..6647ab5c9 100644 --- a/libretroshare/src/tcponudp/udprelay.cc +++ b/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 @@ -723,6 +759,8 @@ int UdpRelayReceiver::recvPkt(void *data, int size, struct sockaddr_in &from) /* do accounting */ 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)); diff --git a/libretroshare/src/tcponudp/udprelay.h b/libretroshare/src/tcponudp/udprelay.h index fd41dcd03..60a5672e5 100644 --- a/libretroshare/src/tcponudp/udprelay.h +++ b/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 mPeers; /* indexed by */ + 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 */