Merge pull request #329 from jolavillette/fix/shutdown-streamer-thread-teardown-race

Stop per-peer streamer threads at shutdown to avoid teardown crash

Author: csoler

src/pqi/pqiperson.cc

@@ -431,12 +431,25 @@ int pqiperson::fullstopthreads()
 			  << PeerId().toStdString() << std::endl;
 #endif
 
-	RS_STACK_MUTEX(mPersonMtx);
+	/* Snapshot the child connections under mPersonMtx, then release it BEFORE
+	 * joining their threads. A child's streamer thread may still be delivering a
+	 * received item that synchronously replies on the same stack through
+	 * pqiperson::SendItem(), which locks mPersonMtx; holding mPersonMtx across the
+	 * (blocking) fullstop() join would deadlock -- exactly the reason
+	 * pqipersongrp::fullstopAllThreads() already drops coreMtx before waiting.
+	 * Observed at shutdown as an endless RsThread::waitWhileStopping() on a
+	 * "pqi <peer>" thread that was answering an RTT ping. */
+	std::list<pqiconnect *> children;
+	{
+		RS_STACK_MUTEX(mPersonMtx);
+		for(std::map<uint32_t, pqiconnect *>::iterator it = kids.begin(); it != kids.end(); ++it)
+			children.push_back(it->second);
+	}
 
-	std::map<uint32_t, pqiconnect *>::iterator it;
-	for(it = kids.begin(); it != kids.end(); ++it)
-		(it->second)->fullstop(); // WAIT FOR THREAD TO STOP.
+	for(std::list<pqiconnect *>::iterator it = children.begin(); it != children.end(); ++it)
+		(*it)->fullstop(); // WAIT FOR THREAD TO STOP.
 
+	RS_STACK_MUTEX(mPersonMtx);
 	activepqi = NULL;
 	active = false;
 	lastHeartbeatReceived = 0;

src/pqi/pqipersongrp.cc

@@ -460,6 +460,53 @@ int     pqipersongrp::removePeer(const RsPeerId& id)
 	return 1;
 }
 
+int     pqipersongrp::fullstopAllThreads()
+{
+	/* Stop every peer's network I/O threads at shutdown. Unlike removePeer() this
+	 * does NOT delete the persons, it only makes sure no pqithreadstreamer is left
+	 * running once RetroShare tears down its static objects (otherwise those
+	 * detached threads keep deserialising incoming packets and crash/flood the log
+	 * when the static SmallObject allocator and its mutex get destroyed).
+	 *
+	 * Two phases on purpose:
+	 *  1) while holding coreMtx, signal every peer to stop and close its sockets.
+	 *     reset() calls askForStop() on the connection threads and closes the
+	 *     sockets; both are non-blocking, so holding coreMtx here is safe (same as
+	 *     removePeer()).
+	 *  2) release coreMtx, THEN wait for the threads to really stop.
+	 * We must NOT hold coreMtx while waiting: a peer thread that is delivering a
+	 * received item may still need coreMtx to queue a reply
+	 * (pqihandler::queueOutRsItem), so it would never finish and we would wait for
+	 * it forever -> deadlock. */
+
+	std::list<pqiperson *> persons;
+
+	{
+		RsStackMutex stack(coreMtx); /**************** LOCKED MUTEX ****************/
+
+		for(std::map<RsPeerId, SearchModule *>::iterator it = mods.begin(); it != mods.end(); ++it)
+		{
+			// mods also holds non-pqiperson interfaces (e.g. the pqiloopback for
+			// our own node) which have no network I/O threads. Only pqiperson has
+			// connection threads to stop, so skip anything that is not one -- a
+			// blind cast would crash on the loopback.
+			pqiperson *p = dynamic_cast<pqiperson *>(it->second->pqi);
+			if(!p)
+				continue;
+			p -> stoplistening();
+			p -> reset(); // askForStop() the threads + close the sockets (non-blocking)
+			persons.push_back(p);
+		}
+	}
+
+	// coreMtx released: peer threads that still need it can now make progress and
+	// notice the stop request. Only now do we wait for them to actually stop.
+	for(std::list<pqiperson *>::iterator it = persons.begin(); it != persons.end(); ++it)
+		(*it) -> fullstopthreads(); // WAIT FOR THREADS TO STOP.
+
+	return 1;
+}
+
 int pqipersongrp::tagHeartbeatRecvd(const RsPeerId& id)
 {
         std::map<RsPeerId, SearchModule *>::iterator it;

src/pqi/pqipersongrp.h

@@ -66,6 +66,7 @@ virtual void    statusChanged();
 	/******************* Peer Control **************************/
 virtual int addPeer(const RsPeerId& id); /* can be overloaded for testing */
 int     removePeer(const RsPeerId& id);
+int     fullstopAllThreads(); /* stop every peer's I/O threads (used at shutdown) */
 int     connectPeer(const RsPeerId& id
 #ifdef WINDOWS_SYS
 ///////////////////////////////////////////////////////////

src/rsserver/p3face-config.cc

@@ -103,6 +103,14 @@ void RsServer::rsGlobalShutDown()
 
 	fullstop();
 
+	/* Stop the per-peer network I/O threads (pqithreadstreamer). They are
+	 * otherwise never stopped at shutdown and keep deserialising incoming
+	 * packets, which crashes/floods the log once the static SmallObject
+	 * allocator and its mutex are destroyed during process teardown.
+	 * Must run after fullstop() so the RsServer tick thread is no longer
+	 * iterating the peer list concurrently. */
+	if(pqih) pqih->fullstopAllThreads();
+
 #ifdef RS_JSONAPI
 	if(rsJsonApi) rsJsonApi->fullstop();
 #endif