Merge pull request #245 from jolavillette/ExposeOutqueueSizeBytes

core: expose peer output queue size in bytes for improved network monitoring

Author: csoler

src/pqi/pqi_base.h

@@ -52,13 +52,18 @@ class RsBwRates
 {
 	public:
 	RsBwRates()
-	:mRateIn(0), mRateOut(0), mMaxRateIn(0), mMaxRateOut(0), mQueueIn(0), mQueueOut(0) {return;}
+        : mRateIn(0), mRateOut(0), mMaxRateIn(0), mMaxRateOut(0), 
+          mQueueIn(0), mQueueOut(0), mQueueOutBytes(0),
+          mTotalIn(0), mTotalOut(0) { return; }
 	float mRateIn;
 	float mRateOut;
 	float mMaxRateIn;
 	float mMaxRateOut;
 	int   mQueueIn;
 	int   mQueueOut;
+	uint32_t mQueueOutBytes;
+	uint64_t mTotalIn;
+	uint64_t mTotalOut;
 };
 
 
@@ -239,6 +244,8 @@ class PQInterface: public RateInterface
 		uint64_t traf_in;
 		uint64_t traf_out;
 
+		virtual float getQueueSize_seconds() { return 0.0f; }
+
 	private:
 
 		RsPeerId peerId;

src/pqi/pqihandler.cc

@@ -287,40 +287,59 @@ int     pqihandler::ExtractTrafficInfo(std::list<RSTrafficClue>& out_lst,std::li
 }
 
 // NEW extern fn to extract rates.
-int     pqihandler::ExtractRates(std::map<RsPeerId, RsBwRates> &ratemap, RsBwRates &total)
+
+int pqihandler::ExtractRates(std::map<RsPeerId, RsBwRates> &ratemap, RsBwRates &total)
 {
+	/* Initialize standard totals */
 	total.mMaxRateIn = getMaxRate(true);
 	total.mMaxRateOut = getMaxRate(false);
 	total.mRateIn = 0;
 	total.mRateOut = 0;
 	total.mQueueIn = 0;
 	total.mQueueOut = 0;
+	total.mQueueOutBytes = 0;
+	total.mTotalIn = 0;
+	total.mTotalOut = 0;
 
 	/* Lock once rates have been retrieved */
 	RS_STACK_MUTEX(coreMtx); /**************** LOCKED MUTEX ****************/
 
 	std::map<RsPeerId, SearchModule *>::iterator it;
+
 	for(it = mods.begin(); it != mods.end(); ++it)
 	{
 		SearchModule *mod = (it -> second);
 
 		RsBwRates peerRates;
+		
+		/* Call the relay in pqiperson (which calls pqistreamer) */
 		mod -> pqi -> getRates(peerRates);
 
+		/* Accumulate standard statistics */
 		total.mRateIn  += peerRates.mRateIn;
 		total.mRateOut += peerRates.mRateOut;
 		total.mQueueIn  += peerRates.mQueueIn;
 		total.mQueueOut += peerRates.mQueueOut;
+		total.mQueueOutBytes += peerRates.mQueueOutBytes;
 
+		/* Accumulate cumulative statistics into global totals */
+		total.mTotalIn += peerRates.mTotalIn;
+		total.mTotalOut += peerRates.mTotalOut;
+
+		/* Store individual peer rates in the result map */
 		ratemap[it->first] = peerRates;
 
+		/* Clean debug message for this layer */
+		//RsDbg() << "OUTQUEUEBYTES [pqihandler] Collected Peer: " << it->first << " | Bytes: " << peerRates.mQueueOutBytes;
+
+		/* Debug message for extraction layer */
+		//RsDbg() << "BWSUM Handler [ExtractRates] Peer: " << it->first << " | In: " << peerRates.mTotalIn << " | Out: " << peerRates.mTotalOut;
+
 	}
 
 	return 1;
 }
 
-
-
 // internal fn to send updates
 int     pqihandler::UpdateRates()
 {
@@ -464,12 +483,16 @@ int     pqihandler::UpdateRates()
 	{
 		SearchModule *mod = (it -> second);
 
-		// for our down bandwidth we use the calculated value without taking into account the max up provided by peers via BwCtrl
-		// this is harmless as they will control their up bw on their side
+		// attn: the bwctrl service only passes to the peers a max in value, therefore we are not aware of our peers max out
+		// see p3bwctrl and rsconfig.h
+
+		// for our down bandwidth we use the calculated value as is, because BwCtrl does not provide a max out from our peers
+		// this down limit will be passed to our peers via BwCtrl
 		mod -> pqi -> setMaxRate(true, in_max_bw);
 
-		// for our up bandwidth we take into account the max down provided by peers via BwCtrl
-		// because we don't want to clog our outqueues, the TCP buffers, and the peers inbound queues
+		// for our up bandwidth we take into account the max in provided by peers via BwCtrl
+		// we don't want to clog our outqueues, the TCP buffers, and the peers inbound queues
+		// attn: this up limit will not passed to our peers via BwCtrl
 		mod -> pqi -> setMaxRate(false, out_max_bw);
 		if ((rateMap_it = rateMap.find(mod->pqi->PeerId())) != rateMap.end())
 			if (rateMap_it->second.mAllowedOut > 0)

src/pqi/pqiperson.cc

@@ -559,16 +559,24 @@ int	pqiperson::connect(uint32_t type, const sockaddr_storage &raddr,
 	return 1;
 }
 
-
 void pqiperson::getRates(RsBwRates &rates)
 {
 	RS_STACK_MUTEX(mPersonMtx);
 
-	// get the rate from the active one.
+	/* Check if the peer connection is established and active */
 	if ((!active) || (activepqi == NULL))
+	{
 		return;
+	}
 
+	/* Forward the request to the active streamer (pqiconnect) */
 	activepqi->getRates(rates);
+
+	/* Clean debug message for the relay layer */
+	//RsDbg() << "OUTQUEUEBYTES [Person] Peer: " << PeerId() << " | Relaying rate request";
+
+	/* Debug to confirm the relay layer receives cumulative totals */
+	//RsDbg() << "BWSUM Relay [Person] Peer: " << PeerId() << " | In: " << rates.mTotalIn << " | Out: " << rates.mTotalOut;
 }
 
 int pqiperson::gatherStatistics(std::list<RSTrafficClue>& out_lst,

src/pqi/pqistreamer.cc

@@ -1372,16 +1372,33 @@ int     pqistreamer::getQueueSize_bytes(bool in)
         }
 }
 
-void    pqistreamer::getRates(RsBwRates &rates)
+void pqistreamer::getRates(RsBwRates &rates)
 {
+	/* Call base RateInterface to get basic bandwidth numbers */
 	RateInterface::getRates(rates);
 
-// no mutex is needed here because this is atomic
+	// No mutex is needed here for mIncomingSize as it is atomic
 	rates.mQueueIn = mIncomingSize;
 
 	{
 		RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
+		
+		/* Fetch standard item count */
 		rates.mQueueOut = locked_out_queue_size();
+
+		/* Extract the actual queue size in bytes */
+		/* locked_compute_out_pkt_size() returns the sum of serialized sizes of all items in queue */
+		rates.mQueueOutBytes = (uint32_t)locked_compute_out_pkt_size();
+
+		/* Populate cumulative totals from internal variables */
+     		rates.mTotalIn = (uint64_t)mTotalRead;
+		rates.mTotalOut = (uint64_t)mTotalSent;
+
+		/* Debug message */
+		//RsDbg() << "BWSUM Source [Streamer] Peer: " << PeerId() << " | In: " << rates.mTotalIn << " | Out: " << rates.mTotalOut;
+
+		/* Debus message */ 
+		//RsDbg() << "OUTQUEUEBYTES [Streamer] Peer: " << PeerId() << " | Bytes: " << rates.mQueueOutBytes;
 	}
 }
 

src/pqi/pqistreamer.h

@@ -80,6 +80,10 @@ class pqistreamer: public PQInterface
 		virtual float getMaxRate(bool b) ;
 		virtual float getMaxRate_locked(bool b);
 
+		/* Public getters for cumulative traffic */
+		uint64_t getTotalRead() const { return (uint64_t)mTotalRead; }
+		uint64_t getTotalSent() const { return (uint64_t)mTotalSent; }
+
 	protected:
        		virtual int reset() ;
 
@@ -152,8 +156,8 @@ class pqistreamer: public PQInterface
 		uint32_t mIncomingSize_bytes; // size of Incoming in btyes
 
 		// data for network stats.
-		int mTotalRead;
-		int mTotalSent;
+		uint64_t mTotalRead;
+		uint64_t mTotalSent;
 
 		// these are representative (but not exact)
 		int mCurrRead;

src/retroshare/rsconfig.h

@@ -149,7 +149,9 @@ struct RsConfigDataRates : RsSerializable
 	    mAllocTs(0),
 	    mRateOut(0), mRateMaxOut(0), mAllowedOut(0),
 	    mAllowedTs(0),
-	    mQueueIn(0), mQueueOut(0)
+	    mQueueIn(0), mQueueOut(0),
+	    mQueueOutBytes(0),
+	    mTotalIn(0), mTotalOut(0)
 	{}
 
 	/* all in kB/s */
@@ -167,6 +169,10 @@ struct RsConfigDataRates : RsSerializable
 
 	int	mQueueIn;
 	int	mQueueOut;
+	uint32_t mQueueOutBytes;
+
+	uint64_t mTotalIn;  // Total bytes received (cumulative)
+	uint64_t mTotalOut; // Total bytes sent (cumulative)
 
 	// RsSerializable interface
 	void serial_process(RsGenericSerializer::SerializeJob j, RsGenericSerializer::SerializeContext &ctx) {
@@ -184,6 +190,10 @@ struct RsConfigDataRates : RsSerializable
 
 		RS_SERIAL_PROCESS(mQueueIn);
 		RS_SERIAL_PROCESS(mQueueOut);
+		RS_SERIAL_PROCESS(mQueueOutBytes);
+
+		RS_SERIAL_PROCESS(mTotalIn);
+		RS_SERIAL_PROCESS(mTotalOut);
 	}
 };
 

src/services/p3bwctrl.cc

@@ -110,6 +110,9 @@ bool p3BandwidthControl::checkAvailableBandwidth()
 	std::map<RsPeerId, RsBwRates> rateMap;
 	RsBwRates total;
 
+	/* VERIFICATION DEBUG */
+	//RsDbg() << "OUTQUEUEBYTES [p3bwctrl] checkAvailableBandwidth: triggering ExtractRates";
+
 	mPg->ExtractRates(rateMap, total);
 	std::map<RsPeerId, RsBwRates>::iterator it;
 	std::map<RsPeerId, BwCtrlData>::iterator bit;
@@ -162,6 +165,12 @@ bool p3BandwidthControl::checkAvailableBandwidth()
 		bit->second.mRates = it->second;
 		bit->second.mRateUpdateTs = now;
 
+		/* Debug message */
+		//RsDbg() << "OUTQUEUEBYTES [p3bwctrl] Bridge active for Peer: " << bit->first << " | Bytes: " << bit->second.mRates.mQueueOutBytes;
+
+		/* Debug to confirm data reached the Bandwidth Control service */
+		//RsDbg() << "BWSUM Service [checkAvailableBandwidth] Peer: " << bit->first << " | In: " << bit->second.mRates.mTotalIn << " | Out: " << bit->second.mRates.mTotalOut;
+
 		if (updatePeer)
 		{
 #define ALLOC_FACTOR (1.0)
@@ -225,6 +234,7 @@ bool p3BandwidthControl::processIncoming()
 	return true;
 }
 
+
 int  p3BandwidthControl::getTotalBandwidthRates(RsConfigDataRates &rates)
 {
 	RsStackMutex stack(mBwMtx); /****** LOCKED MUTEX *******/
@@ -242,10 +252,22 @@ int  p3BandwidthControl::getTotalBandwidthRates(RsConfigDataRates &rates)
 
 	rates.mQueueIn = mTotalRates.mQueueIn;
 	rates.mQueueOut = mTotalRates.mQueueOut;
+	rates.mQueueOutBytes = mTotalRates.mQueueOutBytes;
+
+	/* Copy global cumulative totals to the output structure */
+	rates.mTotalIn = mTotalRates.mTotalIn;
+	rates.mTotalOut = mTotalRates.mTotalOut;
+
+	/* DEBUG OPTIONNEL pour vérifier le total global */
+	//RsDbg() << "OUTQUEUEBYTES [p3bwctrl] GLOBAL TOTAL Bridge: " << rates.mQueueOutBytes;
+	
+	/* Debug message for global totals */
+	//RsDbg() << "BWSUM Final API [Global Total] | In: " << rates.mTotalIn << " | Out: " << rates.mTotalOut;
 
 	return 1;
 }
 
+
 int p3BandwidthControl::getAllBandwidthRates(std::map<RsPeerId, RsConfigDataRates> &ratemap)
 {
 	RsStackMutex stack(mBwMtx); /****** LOCKED MUTEX *******/
@@ -268,10 +290,19 @@ int p3BandwidthControl::getAllBandwidthRates(std::map<RsPeerId, RsConfigDataRate
 
         	rates.mQueueIn = bit->second.mRates.mQueueIn;
         	rates.mQueueOut = bit->second.mRates.mQueueOut;
+		rates.mQueueOutBytes = bit->second.mRates.mQueueOutBytes;
+
+		/* Copy individual cumulative totals to the API structure */
+		rates.mTotalIn = bit->second.mRates.mTotalIn;
+		rates.mTotalOut = bit->second.mRates.mTotalOut;
+
+		/* Debug message */
+		//RsDbg() << "OUTQUEUEBYTES [p3bwctrl] Final API Exit for Peer: " << bit->first << " | Bytes: " << rates.mQueueOutBytes;
+		//RsDbg() << "BWSUM Final API [Peer] " << bit->first << " | In: " << rates.mTotalIn << " | Out: " << rates.mTotalOut;
 
 		ratemap[bit->first] = rates;
 	}			
-	return true ;
+	return 1 ;
 
 
 }
@@ -281,11 +312,6 @@ int     p3BandwidthControl::ExtractTrafficInfo(std::list<RSTrafficClue>& out_sta
     return mPg->ExtractTrafficInfo(out_stats,in_stats) ;
 }
 
-
-
-
-
-
 int p3BandwidthControl::printRateInfo_locked(std::ostream &out)
 {
 	out << "p3BandwidthControl::printRateInfo_locked()";