moved semaphore based thread logic up to RsThread. This should help terminating service threads properly and possibly remove the SIGSEGV when quitting

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

libretroshare/src/pqi/pqithreadstreamer.cc

@@ -36,11 +36,8 @@
 pqithreadstreamer::pqithreadstreamer(PQInterface *parent, RsSerialiser *rss, const RsPeerId& id, BinInterface *bio_in, int bio_flags_in)
 :pqistreamer(rss, id, bio_in, bio_flags_in), mParent(parent), mThreadMutex("pqithreadstreamer"),  mTimeout(0)
 {
-	mTimeout = DEFAULT_STREAMER_TIMEOUT;
+    mTimeout = DEFAULT_STREAMER_TIMEOUT;
     mSleepPeriod = DEFAULT_STREAMER_SLEEP;
-
-    sem_init(&mShouldStopSemaphore,0,0) ;
-    sem_init(&mHasStoppedSemaphore,0,1) ;
 }
 
 bool pqithreadstreamer::RecvItem(RsItem *item)
@@ -56,128 +53,46 @@ int	pqithreadstreamer::tick()
 	return 0;
 }
 
-void pqithreadstreamer::start()
-{
-//    mToRun = true;
-
-#ifdef PQISTREAMER_DEBUG
-    std::cerr << "pqithreadstreamer::run()" << std::endl;
-    std::cerr << "  initing should_stop=0" << std::endl;
-    std::cerr << "  initing has_stopped=1" << std::endl;
-#endif
-
-    sem_init(&mShouldStopSemaphore,0,0) ;
-    sem_init(&mHasStoppedSemaphore,0,0) ;
-
-    RsThread::start();
-}
-
-void pqithreadstreamer::run()
-{
-#ifdef PQISTREAMER_DEBUG
-    std::cerr << "pqithreadstream::run()";
-    std::cerr << std::endl;
-#endif
-
-    // tell the OS to free the thread resources when this function exits
-    // it is a replacement for pthread_join()
-    pthread_detach(pthread_self());
-
-    while(1)
-    {
-        int sval =0;
-        sem_getvalue(&mShouldStopSemaphore,&sval) ;
-
-        if(sval > 0)
-        {
-#ifdef PQISTREAMER_DEBUG
-            std::cerr << "pqithreadstreamer::run(): asked to stop." << std::endl;
-            std::cerr << "  setting hasStopped=1" << std::endl;
-#endif
-            sem_post(&mHasStoppedSemaphore) ;
-            return ;
-        }
-
-        data_tick();
-    }
-}
-
-void pqithreadstreamer::shutdown()
-{
-#ifdef PQISTREAMER_DEBUG
-    std::cerr << "pqithreadstreamer::stop()" << std::endl;
-#endif
-
-    int sval =0;
-    sem_getvalue(&mHasStoppedSemaphore,&sval) ;
-
-    if(sval > 0)
-    {
-#ifdef PQISTREAMER_DEBUG
-        std::cerr << "  thread not running. Quit." << std::endl;
-#endif
-        return ;
-    }
-
-#ifdef PQISTREAMER_DEBUG
-    std::cerr << "  calling stop" << std::endl;
-#endif
-    sem_post(&mShouldStopSemaphore) ;
-}
-
-void pqithreadstreamer::fullstop()
-{
-    shutdown() ;
-
-#ifdef PQISTREAMER_DEBUG
-    std::cerr << "  waiting stop" << std::endl;
-#endif
-    sem_wait(&mHasStoppedSemaphore) ;
-#ifdef PQISTREAMER_DEBUG
-    std::cerr << "  finished!" << std::endl;
-#endif
-}
-
 int	pqithreadstreamer::data_tick()
 {
-	uint32_t recv_timeout = 0;
+    uint32_t recv_timeout = 0;
-	uint32_t sleep_period = 0;
+    uint32_t sleep_period = 0;
-	bool isactive = false;
+    bool isactive = false;
-	{
+    {
-		RsStackMutex stack(mStreamerMtx);
+        RsStackMutex stack(mStreamerMtx);
-		recv_timeout = mTimeout;
+        recv_timeout = mTimeout;
-		sleep_period = mSleepPeriod;
+        sleep_period = mSleepPeriod;
-        	isactive = mBio->isactive();
+        isactive = mBio->isactive();
-	}
+    }
 
-	if (!isactive)
+    if (!isactive)
-	{
+    {
-		usleep(DEFAULT_STREAMER_IDLE_SLEEP);
+        usleep(DEFAULT_STREAMER_IDLE_SLEEP);
-		return 0;
+        return 0;
-	}
+    }
 
     {
         RsStackMutex stack(mThreadMutex);
-    tick_recv(recv_timeout);
+        tick_recv(recv_timeout);
     }
 
-	// Push Items, Outside of Mutex.
+    // Push Items, Outside of Mutex.
-	RsItem *incoming = NULL;
+    RsItem *incoming = NULL;
-	while((incoming = GetItem()))
+    while((incoming = GetItem()))
-	{
+    {
-		RecvItem(incoming);
+        RecvItem(incoming);
-	}
+    }
 
     {
         RsStackMutex stack(mThreadMutex);
-    tick_send(0);
+        tick_send(0);
     }
 
-	if (sleep_period)
+    if (sleep_period)
-	{
+    {
-		usleep(sleep_period);
+        usleep(sleep_period);
-	}
+    }
-	return 1;
+    return 1;
 }
 
 

libretroshare/src/pqi/pqithreadstreamer.h

@@ -26,19 +26,14 @@
 #ifndef MRK_PQI_THREAD_STREAMER_HEADER
 #define MRK_PQI_THREAD_STREAMER_HEADER
 
-#include <semaphore.h>
 #include "pqi/pqistreamer.h"
 #include "util/rsthreads.h"
 
-class pqithreadstreamer: public pqistreamer, private RsThread
+class pqithreadstreamer: public pqistreamer, public RsThread
 {
 	public:
 		pqithreadstreamer(PQInterface *parent, RsSerialiser *rss, const RsPeerId& peerid, BinInterface *bio_in, int bio_flagsin);
 
-virtual void start(); 
-virtual void shutdown();
-virtual void fullstop(); 
-
 virtual bool RecvItem(RsItem *item);
 virtual int  tick();
 
@@ -55,8 +50,6 @@ protected:
 private:
 	/* thread variables */
 	RsMutex mThreadMutex;
-    sem_t mShouldStopSemaphore;
-    sem_t mHasStoppedSemaphore;
 };
 
 #endif //MRK_PQI_THREAD_STREAMER_HEADER

libretroshare/src/util/rsthreads.cc

@@ -60,6 +60,41 @@ void *RsThread::rsthread_init(void* p)
   return NULL;
 }
 
+void RsThread::shutdown()
+{
+#ifdef DEBUG_THREADS
+    std::cerr << "pqithreadstreamer::stop()" << std::endl;
+#endif
+
+    int sval =0;
+    sem_getvalue(&mHasStoppedSemaphore,&sval) ;
+
+    if(sval > 0)
+    {
+#ifdef DEBUG_THREADS
+        std::cerr << "  thread not running. Quit." << std::endl;
+#endif
+        return ;
+    }
+
+#ifdef DEBUG_THREADS
+    std::cerr << "  calling stop" << std::endl;
+#endif
+    sem_post(&mShouldStopSemaphore) ;
+}
+
+void RsThread::fullstop()
+{
+    shutdown() ;
+
+#ifdef DEBUG_THREADS
+    std::cerr << "  waiting stop" << std::endl;
+#endif
+    sem_wait(&mHasStoppedSemaphore) ;
+#ifdef DEBUG_THREADS
+    std::cerr << "  finished!" << std::endl;
+#endif
+}
 void RsThread::start()
 {
     pthread_t tid;
@@ -67,41 +102,33 @@ void RsThread::start()
 
     RS_STACK_MUTEX(mMutex) ;
 
-#if 0
+#ifdef DEBUG_THREADS
-    int ret;
+    std::cerr << "pqithreadstreamer::run()" << std::endl;
-    ret = pthread_attr_init(&tattr);
+    std::cerr << "  initing should_stop=0" << std::endl;
-    if (doDetached)
+    std::cerr << "  initing has_stopped=1" << std::endl;
-    {
-        ret = pthread_attr_setdetachstate(&tattr,PTHREAD_CREATE_DETACHED);
-    }
-    else
-    {
-        ret = pthread_attr_setdetachstate(&tattr,PTHREAD_CREATE_JOINABLE);
-    }
-
-    pthread_create(&tid, &tattr, &rsthread_init, data);
 #endif
+    sem_init(&mShouldStopSemaphore,0,0) ;
+    sem_init(&mHasStoppedSemaphore,0,0) ;
 
     int err ;
 
     // pthread_create is a memory barrier
-    // -> the new thread will see mIsRunning = true
+    // -> the new thread will see mIsRunning() = true
-    mIsRunning = true ;
+
     if( 0 == (err=pthread_create(&tid, 0, &rsthread_init, data)))
-    {
         mTid = tid;
-    }
     else
     {
         std::cerr << "Fatal error: pthread_create could not create a thread. Error returned: " << err << " !!!!!!!" << std::endl;
-        mIsRunning = false ;
+    sem_init(&mHasStoppedSemaphore,0,1) ;
     }
 }
 
 
 RsThread::RsThread () : mMutex("RsThread")
 {
-    mIsRunning = false;
+    sem_init(&mShouldStopSemaphore,0,0) ;
+    sem_init(&mHasStoppedSemaphore,0,1) ;
 
 #ifdef WINDOWS_SYS
     memset (&mTid, 0, sizeof(mTid));
@@ -110,73 +137,89 @@ RsThread::RsThread () : mMutex("RsThread")
 #endif
 }
 
-void RsThread::join() /* waits for the the mTid thread to stop */
-{
-	// do we need a mutex for this ?
-	mIsRunning = false;
-
-	void *ptr;
-	pthread_join(mTid, &ptr);
-}
-
-void RsThread::stop() 
-{
-	pthread_exit(NULL);
-}
-
 bool RsThread::isRunning()
 {
-	// do we need a mutex for this ?
+    // do we need a mutex for this ?
-	return mIsRunning;
+    int sval =0;
+    sem_getvalue(&mHasStoppedSemaphore,&sval) ;
+
+    return !sval ;
+}
+
+void RsThread::run()
+{
+#ifdef DEBUG_THREADS
+    std::cerr << "pqithreadstream::run()";
+    std::cerr << std::endl;
+#endif
+
+    // tell the OS to free the thread resources when this function exits
+    // it is a replacement for pthread_join()
+    pthread_detach(pthread_self());
+
+    while(1)
+    {
+        int sval =0;
+        sem_getvalue(&mShouldStopSemaphore,&sval) ;
+
+        if(sval > 0)
+        {
+#ifdef DEBUG_THREADS
+            std::cerr << "pqithreadstreamer::run(): asked to stop." << std::endl;
+            std::cerr << "  setting hasStopped=1" << std::endl;
+#endif
+            sem_post(&mHasStoppedSemaphore) ;
+            return ;
+        }
+
+        data_tick();
+    }
 }
 
 RsQueueThread::RsQueueThread(uint32_t min, uint32_t max, double relaxFactor )
-	:mMinSleep(min), mMaxSleep(max), mRelaxFactor(relaxFactor)
+    :mMinSleep(min), mMaxSleep(max), mRelaxFactor(relaxFactor)
 {
-	mLastSleep = (uint32_t)mMinSleep ;
+    mLastSleep = (uint32_t)mMinSleep ;
-	mLastWork = time(NULL) ;
+    mLastWork = time(NULL) ;
 }
 
-void RsQueueThread::run()
+void RsQueueThread::data_tick()
 {
-	while(isRunning())
+    bool doneWork = false;
-	{
+    while(workQueued() && doWork())
-		bool doneWork = false;
+    {
-		while(workQueued() && doWork())
+        doneWork = true;
-		{
+    }
-			doneWork = true;
+    time_t now = time(NULL);
-		}
+    if (doneWork)
-		time_t now = time(NULL);
+    {
-		if (doneWork)
+        mLastWork = now;
-		{
+        mLastSleep = (uint32_t) (mMinSleep + (mLastSleep - mMinSleep) / 2.0);
-			mLastWork = now;
-			mLastSleep = (uint32_t) (mMinSleep + (mLastSleep - mMinSleep) / 2.0);
 #ifdef DEBUG_THREADS
-			std::cerr << "RsQueueThread::run() done work: sleeping for: " << mLastSleep;
+        std::cerr << "RsQueueThread::run() done work: sleeping for: " << mLastSleep;
-			std::cerr << " ms";
+        std::cerr << " ms";
-			std::cerr << std::endl;
+        std::cerr << std::endl;
 #endif
 
-		}
+    }
-		else
+    else
-		{
+    {
-			uint32_t deltaT = now - mLastWork;
+        uint32_t deltaT = now - mLastWork;
-			double frac = deltaT / mRelaxFactor;
+        double frac = deltaT / mRelaxFactor;
 
-			mLastSleep += (uint32_t) 
+        mLastSleep += (uint32_t)
-				((mMaxSleep-mMinSleep) * (frac + 0.05));
+                        ((mMaxSleep-mMinSleep) * (frac + 0.05));
-			if (mLastSleep > mMaxSleep)
+        if (mLastSleep > mMaxSleep)
-			{
+        {
-				mLastSleep = mMaxSleep;
+            mLastSleep = mMaxSleep;
-			}
+        }
 #ifdef DEBUG_THREADS
-			std::cerr << "RsQueueThread::run() no work: sleeping for: " << mLastSleep;
+        std::cerr << "RsQueueThread::run() no work: sleeping for: " << mLastSleep;
-			std::cerr << " ms";
+        std::cerr << " ms";
-			std::cerr << std::endl;
+        std::cerr << std::endl;
 #endif
-		}
+    }
-		usleep(mLastSleep * 1000); // mLastSleep msec
+    usleep(mLastSleep * 1000); // mLastSleep msec
-	}
 }
 
 void RsMutex::unlock()

libretroshare/src/util/rsthreads.h

@@ -1,6 +1,3 @@
-#ifndef RSIFACE_THREADS_H
-#define RSIFACE_THREADS_H
-
 /*
  * "$Id: rsthreads.h,v 1.1 2007-02-19 20:08:30 rmf24 Exp $"
  *
@@ -26,11 +23,13 @@
  *
  */
 
+#pragma once
 
 #include <pthread.h>
 #include <inttypes.h>
 #include <string>
 #include <iostream>
+#include <semaphore.h>
 
 /* RsIface Thread Wrappers */
 
@@ -183,43 +182,43 @@ public:
     virtual ~RsThread() {}
 
     void start() ;
-    void join(); /* waits for the the mTid thread to stop */
+    void shutdown();
-    void stop(); /* calls pthread_exit() */
+    void fullstop();
-
+    void join() { fullstop() ; }	// used for compatibility
     bool isRunning();
 
 protected:
-    virtual void run() = 0; /* called once the thread is started. Should be overloaded by subclasses. */
+    void run() ; /* called once the thread is started. Should be overloaded by subclasses. */
+
 private:
     static void *rsthread_init(void*) ;
 
     pthread_t mTid;
     RsMutex   mMutex;
 
-    bool mIsRunning;
+    sem_t mShouldStopSemaphore;
+    sem_t mHasStoppedSemaphore;
 };
 
 
 class RsQueueThread: public RsThread
 {
-	public:
+public:
 
-	RsQueueThread(uint32_t min, uint32_t max, double relaxFactor );
+    RsQueueThread(uint32_t min, uint32_t max, double relaxFactor );
-virtual ~RsQueueThread() { return; }
+    virtual ~RsQueueThread() { return; }
 
-virtual void run();
+protected:
 
-	protected:
+    virtual bool workQueued() = 0;
+    virtual bool doWork() = 0;
+    virtual void data_tick() ;
 
-virtual bool workQueued() = 0;
+private:
-virtual bool doWork() = 0;
+    uint32_t mMinSleep; /* ms */
-
+    uint32_t mMaxSleep; /* ms */
-	private:
+    uint32_t mLastSleep; /* ms */
-	uint32_t mMinSleep; /* ms */
+    time_t   mLastWork;  /* secs */
-	uint32_t mMaxSleep; /* ms */
+    float    mRelaxFactor;
-	uint32_t mLastSleep; /* ms */
-	time_t   mLastWork;  /* secs */
-	float    mRelaxFactor; 
 };
 
-#endif