finish emoving unneeded pqistreamer locks in recv part

2025-01-13 16:39:43 -05:00 · 2020-05-03 20:32:17 +02:00 · 2020-05-03 20:32:17 +02:00 · d847086911
commit d847086911
parent d35dd76ca5
2 changed files with 59 additions and 61 deletions
--- a/libretroshare/src/pqi/pqistreamer.cc
+++ b/libretroshare/src/pqi/pqistreamer.cc
@ -159,7 +159,7 @@ pqistreamer::~pqistreamer()
 	if (mRsSerialiser)
 		delete mRsSerialiser;

-	free_pend_locked() ;
+	free_pend() ;

 	// clean up incoming.
 	while (!mIncoming.empty())
@ -199,14 +199,12 @@ RsItem *pqistreamer::GetItem()
 	pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::GetItem()");
 #endif

-	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
-
 	if(mIncoming.empty())
 		return NULL; 

 	RsItem *osr = mIncoming.front() ;
-    mIncoming.pop_front() ;
-    --mIncomingSize;
+	mIncoming.pop_front() ;
+	--mIncomingSize;

 	return osr;
 }
@ -288,25 +286,26 @@ int 	pqistreamer::tick_bio()

 int 	pqistreamer::tick_recv(uint32_t timeout)
 {
-//	Only our thread manipulates mIncoming queue and related counters.
-//	The lock of pqistreamer mutex is thus not needed here.
-//	For the moment and for safety reasons it is still kept in 3 places:
-//	- in pqistreamer::tick_recv before calling free_pend_locked, as I dont know what this method actually does
-//	- in pqistreamer::handleincomingitem_locked, as it modifies mIncoming queue
-//	- in pqistreamer::inReadBytes_locked, as it modifies related counters
+//      Apart from a few exceptions that are atomic (mLastIncomingTs, mIncomingSize), only this pqi thread reads/writes mIncoming queue and related counters.
+//      The lock of pqistreamer mutex is thus not needed.
+//      The mutex lock is still needed before calling locked_addTrafficClue because this method is also used by the thread pushing packets in mOutPkts
 //
-//	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
+//      The following methods have been renamed by removing the 'locked' part of the name:
+//      - handleincoming_locked
+//      - handleincomingitem_locked
+//      - inReadBytes_locked
+//      - inAllowedBytes_locked
+//      - addPartialPacket_locked
+//	- allocate_rpend_locked
+//      - free_pend_locked;

 	if (mBio->moretoread(timeout))
 	{
-		handleincoming_locked();
+		handleincoming();
 	}
 	if(!(mBio->isactive()))
 	{
-		// lock only now, see comment above
-		RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
-
-		free_pend_locked();
+		free_pend();
 	}
 	return 1;
 }
@ -319,7 +318,7 @@ int 	pqistreamer::tick_send(uint32_t timeout)
 	/* short circuit everything is bio isn't active */
 	if (!(mBio->isactive()))
 	{
-        		free_pend_locked();
+        		free_pend();
 		return 0;
 	}

@ -413,11 +412,11 @@ int	pqistreamer::queue_outpqi_locked(RsItem *pqi,uint32_t& pktsize)
 	return 1; // keep error internal.
 }

-int 	pqistreamer::handleincomingitem_locked(RsItem *pqi,int len)
+int 	pqistreamer::handleincomingitem(RsItem *pqi,int len)
 {

 #ifdef DEBUG_PQISTREAMER
-	pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleincomingitem_locked()");
+	pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleincomingitem()");
 #endif
 	// timestamp last received packet.
 	mLastIncomingTs = time(NULL);
@ -425,9 +424,6 @@ int 	pqistreamer::handleincomingitem_locked(RsItem *pqi,int len)
 	// Use overloaded Contact function 
 	pqi -> PeerId(PeerId());

-	// lock only now, see comment in pqistreamer::tick_recv
-	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
-
 	mIncoming.push_back(pqi);
 	++mIncomingSize ;

@ -435,6 +431,8 @@ int 	pqistreamer::handleincomingitem_locked(RsItem *pqi,int len)
 	// keep info for stats for a while. Only keep the items for the last two seconds. sec n is ongoing and second n-1
 	// is a full statistics chunk that can be used in the GUI

+	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
+
 	locked_addTrafficClue(pqi,len,mCurrentStatsChunk_In) ;

 	/*******************************************************************************************/
@ -470,8 +468,8 @@ void pqistreamer::locked_addTrafficClue(const RsItem *pqi,uint32_t pktsize,std::

 rstime_t	pqistreamer::getLastIncomingTS()
 {
-	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
-
+// This is the only case where another thread (rs main for pqiperson) will access our data
+// Still a mutex lock is not needed because the operation is atomic
 	return mLastIncomingTs;
 }

@ -707,23 +705,23 @@ int	pqistreamer::handleoutgoing_locked()

 /* Handles reading from input stream.
 */
-int pqistreamer::handleincoming_locked()
+int pqistreamer::handleincoming()
 {
    int readbytes = 0;
    static const int max_failed_read_attempts = 2000 ;

 #ifdef DEBUG_PQISTREAMER
-    pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleincoming_locked()");
+    pqioutput(PQL_DEBUG_ALL, pqistreamerzone, "pqistreamer::handleincoming()");
 #endif

    if(!(mBio->isactive()))
    {
 	    mReading_state = reading_state_initial ;
-	    free_pend_locked();
+	    free_pend();
 	    return 0;
    }
    else
-	    allocate_rpend_locked();
+	    allocate_rpend();

    // enough space to read any packet.
    uint32_t maxlen = mPkt_rpend_size; 
@ -732,7 +730,7 @@ int pqistreamer::handleincoming_locked()
    // initial read size: basic packet.
    int blen = getRsPktBaseSize();	// this is valid for both packet slices and normal un-sliced packets (same header size)

-    int maxin = inAllowedBytes_locked();
+    int maxin = inAllowedBytes();

 #ifdef DEBUG_PQISTREAMER
    std::cerr << "[" << (void*)pthread_self() << "] " << "reading state = " << mReading_state << std::endl ;
@ -981,19 +979,19 @@ continue_packet:
 		    std::cerr << "Inputing partial packet " << RsUtil::BinToHex((char*)block,8) << std::endl;
 #endif
            		uint32_t packet_length = 0 ;
-		    pkt = addPartialPacket_locked(block,pktlen,slice_packet_id,is_packet_starting,is_packet_ending,packet_length) ;
+		    pkt = addPartialPacket(block,pktlen,slice_packet_id,is_packet_starting,is_packet_ending,packet_length) ;
            
            		pktlen = packet_length ;
 	    }
 	    else
 		    pkt = mRsSerialiser->deserialise(block, &pktlen);

-	    if ((pkt != NULL) && (0  < handleincomingitem_locked(pkt,pktlen)))
+	    if ((pkt != NULL) && (0  < handleincomingitem(pkt,pktlen)))
 	    {
 #ifdef DEBUG_PQISTREAMER
 		    pqioutput(PQL_DEBUG_BASIC, pqistreamerzone, "Successfully Read a Packet!");
 #endif
-		    inReadBytes_locked(pktlen);	// only count deserialised packets, because that's what is actually been transfered.
+		    inReadBytes(pktlen);	// only count deserialised packets, because that's what is actually been transfered.
 	    }
 	    else if (!is_partial_packet)
 	    {
@ -1026,7 +1024,7 @@ continue_packet:
    return 0;
 }

-RsItem *pqistreamer::addPartialPacket_locked(const void *block, uint32_t len, uint32_t slice_packet_id, bool is_packet_starting, bool is_packet_ending, uint32_t &total_len) 
+RsItem *pqistreamer::addPartialPacket(const void *block, uint32_t len, uint32_t slice_packet_id, bool is_packet_starting, bool is_packet_ending, uint32_t &total_len) 
 {
 #ifdef DEBUG_PACKET_SLICING
    std::cerr << "Receiving partial packet. size=" << len << ", ID=" << std::hex << slice_packet_id << std::dec << ", starting:" << is_packet_starting << ", ending:" << is_packet_ending ;
@ -1170,7 +1168,7 @@ int     pqistreamer::outAllowedBytes_locked()
 	return quota;
 }

-int     pqistreamer::inAllowedBytes_locked()
+int     pqistreamer::inAllowedBytes()
 {
 	double t = getCurrentTS(); // in sec, with high accuracy

@ -1208,7 +1206,7 @@ int     pqistreamer::inAllowedBytes_locked()

 #ifdef DEBUG_PQISTREAMER
 	uint64_t t_now = 1000 * getCurrentTS();
-	std::cerr << std::dec << t_now << " DEBUG_PQISTREAMER pqistreamer::inAllowedBytes_locked PeerId " << this->PeerId().toStdString() << " dt " << (int)(1000 * dt) << "ms, mAvgDtIn " << (int)(1000 * mAvgDtIn) << "ms, maxin " << (int)(maxin) << " bytes/s, mCurrRead " << mCurrRead << " bytes, quota " << (int)(quota) << " bytes" << std::endl;
+	std::cerr << std::dec << t_now << " DEBUG_PQISTREAMER pqistreamer::inAllowedBytes PeerId " << this->PeerId().toStdString() << " dt " << (int)(1000 * dt) << "ms, mAvgDtIn " << (int)(1000 * mAvgDtIn) << "ms, maxin " << (int)(maxin) << " bytes/s, mCurrRead " << mCurrRead << " bytes, quota " << (int)(quota) << " bytes" << std::endl;
 #endif

 	return quota;
@ -1245,7 +1243,7 @@ void    pqistreamer::outSentBytes_locked(uint32_t outb)
 	return;
 }

-void    pqistreamer::inReadBytes_locked(uint32_t inb)
+void    pqistreamer::inReadBytes(uint32_t inb)
 {
 #ifdef DEBUG_PQISTREAMER
 	{
@ -1255,9 +1253,6 @@ void    pqistreamer::inReadBytes_locked(uint32_t inb)
 	}
 #endif

-	// lock only now, see comment in pqistreamer::tick_recv
-	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
-
 	mTotalRead += inb;
 	mCurrRead += inb;
 	mAvgReadCount += inb;
@ -1265,7 +1260,7 @@ void    pqistreamer::inReadBytes_locked(uint32_t inb)
 	return;
 }

-void pqistreamer::allocate_rpend_locked()
+void pqistreamer::allocate_rpend()
 {
    if(mPkt_rpending)
        return;
@ -1288,17 +1283,17 @@ int pqistreamer::reset()
 #ifdef DEBUG_PQISTREAMER
 	std::cerr << "pqistreamer::reset()" << std::endl;
 #endif
-	free_pend_locked();
+	free_pend();
    
    return 1 ;
 }

-void pqistreamer::free_pend_locked()
+void pqistreamer::free_pend()
 {
 	if(mPkt_rpending)
 	{
 #ifdef DEBUG_PQISTREAMER
-        		std::cerr << "pqistreamer::free_pend_locked(): pending input packet buffer" << std::endl;
+        		std::cerr << "pqistreamer::free_pend(): pending input packet buffer" << std::endl;
 #endif
 		free(mPkt_rpending);
 		mPkt_rpending = 0;
@ -1308,7 +1303,7 @@ void pqistreamer::free_pend_locked()
 	if (mPkt_wpending)
 	{
 #ifdef DEBUG_PQISTREAMER
-        		std::cerr << "pqistreamer::free_pend_locked(): pending output packet buffer" << std::endl;
+        		std::cerr << "pqistreamer::free_pend(): pending output packet buffer" << std::endl;
 #endif
 		free(mPkt_wpending);
 		mPkt_wpending = NULL;
@ -1317,7 +1312,7 @@ void pqistreamer::free_pend_locked()

 #ifdef DEBUG_PQISTREAMER
    if(!mPartialPackets.empty())
-        		std::cerr << "pqistreamer::free_pend_locked(): " << mPartialPackets.size() << " pending input partial packets" << std::endl;
+        		std::cerr << "pqistreamer::free_pend(): " << mPartialPackets.size() << " pending input partial packets" << std::endl;
 #endif
 	// also delete any incoming partial packet
 	for(std::map<uint32_t,PartialPacketRecord>::iterator it(mPartialPackets.begin());it!=mPartialPackets.end();++it)
@ -1335,23 +1330,27 @@ int     pqistreamer::gatherStatistics(std::list<RSTrafficClue>& outqueue_lst,std

    return locked_gatherStatistics(outqueue_lst,inqueue_lst);
 }
+
 int     pqistreamer::getQueueSize(bool in)
 {
-	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
-
 	if (in)
-        return mIncomingSize;
-    else
-        return locked_out_queue_size();
+// no mutex is needed here because this is atomic
+		return mIncomingSize;
+	else
+	{
+		RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
+		return locked_out_queue_size();
+	}
 }

 void    pqistreamer::getRates(RsBwRates &rates)
 {
 	RateInterface::getRates(rates);

-	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
+// no mutex is needed here because this is atomic
+	rates.mQueueIn = mIncomingSize;

-    rates.mQueueIn = mIncomingSize;
+	RsStackMutex stack(mStreamerMtx); /**** LOCKED MUTEX ****/
 	rates.mQueueOut = locked_out_queue_size();
 }

@ -1417,4 +1416,3 @@ void *pqistreamer::locked_pop_out_data(uint32_t /*max_slice_size*/, uint32_t &si
 	return res ;
 }

-    
--- a/libretroshare/src/pqi/pqistreamer.h
+++ b/libretroshare/src/pqi/pqistreamer.h
@ -104,12 +104,12 @@ class pqistreamer: public PQInterface

 	private:
 		int queue_outpqi_locked(RsItem *i,uint32_t& serialized_size);
-		int handleincomingitem_locked(RsItem *i, int len);
+		int handleincomingitem(RsItem *i, int len);

 		// ticked regularly (manages out queues and sending
 		// via above interfaces.
 		virtual int	handleoutgoing_locked();
-		virtual int	handleincoming_locked();
+		virtual int	handleincoming();

 		// Bandwidth/Streaming Management.
 		float	outTimeSlice_locked();
@ -117,11 +117,11 @@ class pqistreamer: public PQInterface
 		int	outAllowedBytes_locked();
 		void	outSentBytes_locked(uint32_t );

-		int	inAllowedBytes_locked();
-		void	inReadBytes_locked(uint32_t );
+		int	inAllowedBytes();
+		void	inReadBytes(uint32_t );

        		// cleans up everything that's pending / half finished.
-		void free_pend_locked();
+		void free_pend();

 		// RsSerialiser - determines which packets can be serialised.
 		RsSerialiser *mRsSerialiser;
@ -129,7 +129,7 @@ class pqistreamer: public PQInterface
 		void *mPkt_wpending; // storage for pending packet to write.
        	uint32_t mPkt_wpending_size; // ... and its size.

-        void allocate_rpend_locked(); // use these two functions to allocate/free the buffer below
+        void allocate_rpend(); // use these two functions to allocate/free the buffer below
        
 		int   mPkt_rpend_size; // size of pkt_rpending.
 		void *mPkt_rpending; // storage for read in pending packets.
@ -177,7 +177,7 @@ class pqistreamer: public PQInterface
        bool mAcceptsPacketSlicing ;
        rstime_t mLastSentPacketSlicingProbe ;
        void locked_addTrafficClue(const RsItem *pqi, uint32_t pktsize, std::list<RSTrafficClue> &lst);
-        RsItem *addPartialPacket_locked(const void *block, uint32_t len, uint32_t slice_packet_id,bool packet_starting,bool packet_ending,uint32_t& total_len);
+        RsItem *addPartialPacket(const void *block, uint32_t len, uint32_t slice_packet_id,bool packet_starting,bool packet_ending,uint32_t& total_len);
        
        std::map<uint32_t,PartialPacketRecord> mPartialPackets ;
 };