Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2025-02-26 09:41:29 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

attempt to fix the deadlock situations in pqithreadstreamer. Also solves the crash when quitting

Browse Source 
git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@8009 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
csoler 2015-03-10 21:16:41 +00:00
parent 4dede651c7
commit b6ce3ab7b2
3 changed files with 81 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
libretroshare/src/pqi/pqiperson.cc
View File

@ -365,7 +365,7 @@ int pqiperson::handleNotifyEvent_locked(NetInterface *ni, int newState, const s
{ {
pqioutput(PQL_WARNING, pqipersonzone, "pqiperson::notifyEvent() Id: " + PeerId().toStdString() + " CONNECT_FAILED->marking so!"); pqioutput(PQL_WARNING, pqipersonzone, "pqiperson::notifyEvent() Id: " + PeerId().toStdString() + " CONNECT_FAILED->marking so!");
activepqi->stop(); // STOP THREAD. activepqi->shutdown(); // STOP THREAD.
active = false; active = false;
activepqi = NULL; activepqi = NULL;
} }
@ -410,7 +410,7 @@ int pqiperson::reset_locked()
std::map<uint32_t, pqiconnect *>::iterator it; std::map<uint32_t, pqiconnect *>::iterator it;
for(it = kids.begin(); it != kids.end(); ++it) for(it = kids.begin(); it != kids.end(); ++it)
{ {
(it->second) -> stop(); // STOP THREAD. (it->second) -> shutdown(); // STOP THREAD.
(it->second) -> reset(); (it->second) -> reset();
} }

109
libretroshare/src/pqi/pqithreadstreamer.cc
View File

@ -36,7 +36,9 @@ pqithreadstreamer::pqithreadstreamer(PQInterface *parent, RsSerialiser *rss, con
{ {
mTimeout = DEFAULT_STREAMER_TIMEOUT; mTimeout = DEFAULT_STREAMER_TIMEOUT;
mSleepPeriod = DEFAULT_STREAMER_SLEEP; mSleepPeriod = DEFAULT_STREAMER_SLEEP;
return;
sem_init(&mShouldStopSemaphore,0,0) ;
sem_init(&mHasStoppedSemaphore,0,1) ;
} }
bool pqithreadstreamer::RecvItem(RsItem *item) bool pqithreadstreamer::RecvItem(RsItem *item)
@ -54,7 +56,14 @@ int pqithreadstreamer::tick()
void pqithreadstreamer::start() void pqithreadstreamer::start()
{ {
mToRun = true; // mToRun = true;
std::cerr << "pqithreadstreamer::run()" << std::endl;
std::cerr << " initing should_stop=0" << std::endl;
std::cerr << " initing has_stopped=1" << std::endl;
sem_init(&mShouldStopSemaphore,0,0) ;
sem_init(&mHasStoppedSemaphore,0,0) ;
RsThread::start(); RsThread::start();
} }
@ -64,63 +73,89 @@ void pqithreadstreamer::run()
std::cerr << "pqithreadstream::run()"; std::cerr << "pqithreadstream::run()";
std::cerr << std::endl; std::cerr << std::endl;
{
RsStackMutex stack(mThreadMutex);
mRunning = true;
}
while(1) while(1)
{ {
{ {
RsStackMutex stack(mThreadMutex); int sval =0;
if (!mToRun) sem_getvalue(&mShouldStopSemaphore,&sval) ;
{
std::cerr << "pqithreadstream::run() stopping";
std::cerr << std::endl;
mRunning = false; if(sval > 0)
return; {
std::cerr << "pqithreadstreamer::run(): asked to stop." << std::endl;
std::cerr << " setting hasStopped=1" << std::endl;
sem_post(&mHasStoppedSemaphore) ;
return ;
} }
// RsStackMutex stack(mThreadMutex);
//
// if (!mToRun)
// {
// std::cerr << "pqithreadstream::run() stopping";
// std::cerr << std::endl;
//
// mRunning = false;
// return;
// }
} }
data_tick(); data_tick();
} }
} }
void pqithreadstreamer::stop() //bool pqithreadstreamer::threadrunning()
//{
// RsStackMutex stack(mThreadMutex);
// return mRunning ;
//}
void pqithreadstreamer::shutdown()
{ {
//RsStackMutex stack(mThreadMutex); std::cerr << "pqithreadstreamer::stop()" << std::endl;
std::cerr << "pqithreadstream::stop()"; int sval =0;
std::cerr << std::endl; sem_getvalue(&mHasStoppedSemaphore,&sval) ;
mToRun = false; if(sval > 0)
{
std::cerr << " thread not running. Quit." << std::endl;
return ;
}
std::cerr << " calling stop" << std::endl;
sem_post(&mShouldStopSemaphore) ;
} }
void pqithreadstreamer::fullstop() void pqithreadstreamer::fullstop()
{ {
while(1) shutdown() ;
{
{
RsStackMutex stack(mThreadMutex);
mToRun = false ; std::cerr << " waiting stop" << std::endl;
sem_wait(&mHasStoppedSemaphore) ;
std::cerr << " finished!" << std::endl;
if (!mRunning) // while(1)
{ // {
std::cerr << "pqithreadstream::fullstop() complete"; // {
std::cerr << std::endl; // RsStackMutex stack(mThreadMutex);
return; //
} // mToRun = false ;
} // // if (!mRunning)
usleep(1000); // // {
} // // std::cerr << "pqithreadstream::fullstop() complete";
// // std::cerr << std::endl;
// // return;
// // } ;
// }
// usleep(1000);
// }
} }
bool pqithreadstreamer::threadrunning() //bool pqithreadstreamer::threadrunning()
{ //{
RsStackMutex stack(mThreadMutex); // RsStackMutex stack(mThreadMutex);
return mRunning; // return mRunning;
} //}
int pqithreadstreamer::data_tick() int pqithreadstreamer::data_tick()

10
libretroshare/src/pqi/pqithreadstreamer.h
View File

@ -26,6 +26,7 @@
#ifndef MRK_PQI_THREAD_STREAMER_HEADER #ifndef MRK_PQI_THREAD_STREAMER_HEADER
#define MRK_PQI_THREAD_STREAMER_HEADER #define MRK_PQI_THREAD_STREAMER_HEADER
#include <semaphore.h>
#include "pqi/pqistreamer.h" #include "pqi/pqistreamer.h"
#include "util/rsthreads.h" #include "util/rsthreads.h"
@ -36,9 +37,9 @@ class pqithreadstreamer: public pqistreamer, public RsThread
virtual void run(); virtual void run();
virtual void start(); virtual void start();
virtual void stop(); virtual void shutdown();
virtual void fullstop(); virtual void fullstop();
virtual bool threadrunning(); //virtual bool threadrunning();
virtual bool RecvItem(RsItem *item); virtual bool RecvItem(RsItem *item);
virtual int tick(); virtual int tick();
@ -54,9 +55,8 @@ int data_tick();
private: private:
/* thread variables */ /* thread variables */
RsMutex mThreadMutex; RsMutex mThreadMutex;
bool mRunning; sem_t mShouldStopSemaphore;
bool mToRun; sem_t mHasStoppedSemaphore;
}; };
#endif //MRK_PQI_THREAD_STREAMER_HEADER #endif //MRK_PQI_THREAD_STREAMER_HEADER