Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2025-06-29 00:27:27 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge upstream

Browse source
This commit is contained in:
hunbernd 2020-01-07 15:05:21 +01:00
commit 0e8b12ff9f
130 changed files with 2297 additions and 2835 deletions
Download patch file Download diff file Expand all files Collapse all files

2
libretroshare/src/util/dnsresolver.h
View file

@ -29,7 +29,9 @@
#include <map>
#include <string>
#include "util/rsthreads.h"
#include "util/rstime.h"
struct sockaddr ;

2
libretroshare/src/util/extaddrfinder.h
View file

@ -23,8 +23,10 @@
#include <list>
#include <string>
#include "util/rsthreads.h"
#include "util/rsnet.h"
#include "util/rstime.h"
struct sockaddr ;

54
libretroshare/src/util/rserrno.cc Normal file
View file

@ -0,0 +1,54 @@
/*******************************************************************************
* libretroshare/src/util: rserrno.cc *
* *
* libretroshare: retroshare core library *
* *
* Copyright (C) 2019 Gioacchino Mazzurco <gio@eigenlab.org> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
* published by the Free Software Foundation, either version 3 of the *
* License, or (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with this program. If not, see <https://www.gnu.org/licenses/>. *
* *
*******************************************************************************/
#include <cerrno>
#define RS_INTERNAL_ERRNO_CASE(e) case e: return #e
const char* rsErrnoName(int err)
{
switch (err)
{
RS_INTERNAL_ERRNO_CASE(EINVAL);
RS_INTERNAL_ERRNO_CASE(EBUSY);
RS_INTERNAL_ERRNO_CASE(EAGAIN);
RS_INTERNAL_ERRNO_CASE(EDEADLK);
RS_INTERNAL_ERRNO_CASE(EPERM);
RS_INTERNAL_ERRNO_CASE(EBADF);
RS_INTERNAL_ERRNO_CASE(EFAULT);
RS_INTERNAL_ERRNO_CASE(ENOTSOCK);
RS_INTERNAL_ERRNO_CASE(EISCONN);
RS_INTERNAL_ERRNO_CASE(ECONNREFUSED);
RS_INTERNAL_ERRNO_CASE(ETIMEDOUT);
RS_INTERNAL_ERRNO_CASE(ENETUNREACH);
RS_INTERNAL_ERRNO_CASE(EADDRINUSE);
RS_INTERNAL_ERRNO_CASE(EINPROGRESS);
RS_INTERNAL_ERRNO_CASE(EALREADY);
RS_INTERNAL_ERRNO_CASE(ENOTCONN);
RS_INTERNAL_ERRNO_CASE(EPIPE);
RS_INTERNAL_ERRNO_CASE(ECONNRESET);
RS_INTERNAL_ERRNO_CASE(EHOSTUNREACH);
RS_INTERNAL_ERRNO_CASE(EADDRNOTAVAIL);
}
return "rsErrnoName UNKNOWN ERROR CODE";
}

24
libretroshare/src/util/rserrno.h Normal file
View file

@ -0,0 +1,24 @@
/*******************************************************************************
* libretroshare/src/util: rserrno.h *
* *
* libretroshare: retroshare core library *
* *
* Copyright (C) 2019 Gioacchino Mazzurco <gio@eigenlab.org> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
* published by the Free Software Foundation, either version 3 of the *
* License, or (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with this program. If not, see <https://www.gnu.org/licenses/>. *
* *
*******************************************************************************/
#pragma once
const char* rsErrnoName(int err);

405
libretroshare/src/util/rsthreads.cc
View file

@ -3,7 +3,9 @@
* *
* libretroshare: retroshare core library *
* *
* Copyright 2004-2007 by Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2004-2007 Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2016-2019 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2019-2020 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
@ -20,14 +22,20 @@
* *
*******************************************************************************/
#include "rsthreads.h"
#include <unistd.h> // for usleep()
#include <errno.h> // for errno
#include <iostream>
#include "util/rstime.h"
#include "util/rsdebug.h"
#include <time.h>
#include <thread>
#include <chrono>
#ifdef RSMUTEX_DEBUG
#include <cstdio>
#include <sys/time.h>
#endif
#include "rsthreads.h"
#include "util/rsdebug.h"
#include "util/rserrno.h"
#include "util/rstime.h"
#ifdef __APPLE__
int __attribute__((weak)) pthread_setname_np(const char *__buf) ;
@ -41,16 +49,14 @@ int RS_pthread_setname_np(pthread_t __target_thread, const char *__buf) {
}
#endif
#ifdef RSMUTEX_DEBUG
#include <stdio.h>
#include <sys/time.h>
#endif
/*******
* #define DEBUG_THREADS 1
* #define RSMUTEX_ABORT 1 // Catch wrong pthreads mode.
*******/
#define THREAD_DEBUG std::cerr << "[this=" << (void*)this << ", caller thread ID: " << std::hex << pthread_self() << ", thread ID: " << mTid << std::dec << "] "
#define THREAD_DEBUG RsDbg() << "[this=" << static_cast<void*>(this) \
<< ", caller thread ID: " << std::hex << pthread_self() << ", thread ID: " \
<< mTid << std::dec << "] "
#ifdef RSMUTEX_ABORT
#include <stdlib.h>
@ -60,215 +66,113 @@ int RS_pthread_setname_np(pthread_t __target_thread, const char *__buf) {
#include <iostream>
#endif
void RsThread::go()
/*static*/ void* RsThread::rsthread_init(void* p)
{
mShouldStopSemaphore.set(0) ;
mHasStoppedSemaphore.set(0) ;
RsThread* thread = reinterpret_cast<RsThread *>(p);
if(!thread) return nullptr;
runloop();
mShouldStopSemaphore.set(0);
mHasStoppedSemaphore.set(1); // last value that we modify because this is interpreted as a signal that the object can be deleted.
}
void *RsThread::rsthread_init(void* p)
{
RsThread *thread = (RsThread *) p;
if (!thread)
{
return NULL;
}
// tell the OS to free the thread resources when this function exits
// it is a replacement for pthread_join()
pthread_detach(pthread_self());
/* Using pthread_detach(...) the thread resources will be automatically
* freed when this function return, so there is no need for pthread_join()
* later. */
pthread_detach(pthread_self());
#ifdef DEBUG_THREADS
std::cerr << "[Thread ID:" << std::hex << pthread_self() << std::dec << "] thread is started. Calling runloop()..." << std::endl;
std::cerr << "[Thread ID:" << std::hex << pthread_self() << std::dec
<< "] thread is started. Calling wrapRun()..." << std::endl;
#endif
thread->go();
return NULL;
thread->wrapRun();
return nullptr;
}
RsThread::RsThread()
RsThread::RsThread() : mHasStopped(true), mShouldStop(false)
{
#ifdef WINDOWS_SYS
memset (&mTid, 0, sizeof(mTid));
memset (&mTid, 0, sizeof(mTid));
#else
mTid = 0;
#endif
// The thread is certainly not running. This avoids to lock down when calling shutdown on a thread that has never started.
#ifdef DEBUG_THREADS
THREAD_DEBUG << "[Thread ID:" << std::hex << pthread_self() << std::dec << "] thread object created. Initing stopped=1, should_stop=0" << std::endl;
#endif
mHasStoppedSemaphore.set(1) ;
mShouldStopSemaphore.set(0) ;
}
RsThread::~RsThread()
{
if(isRunning())
{
RsErr() << "Deleting a thread that is still running! Something is very wrong here and Retroshare is likely to crash because of this." << std::endl;
print_stacktrace();
while(isRunning())
{
std::cerr << "." << std::endl;
rstime::rs_usleep(1000*1000);
}
}
}
bool RsThread::isRunning()
{
// do we need a mutex for this ?
int sval = mHasStoppedSemaphore.value() ;
return !sval ;
}
bool RsThread::shouldStop()
{
int sval = mShouldStopSemaphore.value() ;
return sval > 0;
}
void RsTickingThread::shutdown()
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << "pqithreadstreamer::shutdown()" << std::endl;
#endif
int sval = mHasStoppedSemaphore.value() ;
if(sval > 0)
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << " thread not running. Quit." << std::endl;
#endif
return ;
}
ask_for_stop() ;
}
void RsThread::ask_for_stop()
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << " calling stop" << std::endl;
#endif
mShouldStopSemaphore.set(1);
}
void RsTickingThread::fullstop()
{
shutdown() ;
#ifdef DEBUG_THREADS
THREAD_DEBUG << " waiting stop" << std::endl;
#endif
if(pthread_equal(mTid,pthread_self()))
{
THREAD_DEBUG << "(WW) RsTickingThread::fullstop() called by same thread. This is unexpected." << std::endl;
return ;
}
mHasStoppedSemaphore.wait_no_relock(); // Wait for semaphore value to become 1, but does not decrement it when obtained.
#ifdef DEBUG_THREADS
THREAD_DEBUG << " finished!" << std::endl;
mTid = 0;
#endif
}
void RsThread::start(const std::string &threadName)
bool RsThread::isRunning() { return !mHasStopped; }
bool RsThread::shouldStop() { return mShouldStop; }
void RsThread::askForStop()
{
if(isRunning())
/* Call onStopRequested() only once even if askForStop() is called multiple
* times */
if(!mShouldStop.exchange(true))
RsThread::async([&](){ onStopRequested(); });
}
void RsThread::wrapRun()
{
run();
mHasStopped = true;
}
void RsThread::fullstop()
{
askForStop();
const pthread_t callerTid = pthread_self();
if(pthread_equal(mTid, callerTid))
{
std::cerr << "(EE) RsThread \"" << threadName
<< "\" is already running. Will not start twice!"
<< std::endl;
RsErr() << __PRETTY_FUNCTION__ << " called by same thread. This should "
<< "never happen! this: " << static_cast<void*>(this)
<< std::hex << ", callerTid: " << callerTid
<< ", mTid: " << mTid << std::dec << std::endl;
print_stacktrace();
return;
}
pthread_t tid;
void *data = (void *)this ;
#ifdef DEBUG_THREADS
THREAD_DEBUG << "pqithreadstreamer::start() initing should_stop=0" << std::endl;
#endif
mShouldStopSemaphore.set(0) ;
mHasStoppedSemaphore.set(0) ;
int err ;
// pthread_create is a memory barrier
// -> the new thread will see mIsRunning() = true
if( 0 == (err=pthread_create(&tid, 0, &rsthread_init, data)))
{
mTid = tid;
// set name
if(pthread_setname_np)
{
if(!threadName.empty())
{
// thread names are restricted to 16 characters including the terminating null byte
if(threadName.length() > 15)
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << "RsThread::start called with to long name '" << threadName << "' truncating..." << std::endl;
#endif
RS_pthread_setname_np(mTid, threadName.substr(0, 15).c_str());
} else {
RS_pthread_setname_np(mTid, threadName.c_str());
}
}
}
}
else
{
THREAD_DEBUG << "Fatal error: pthread_create could not create a thread. Error returned: " << err << " !!!!!!!" << std::endl;
mHasStoppedSemaphore.set(1) ;
}
// Wait for the thread being stopped
auto i = 1;
while(!mHasStopped)
{
std::this_thread::sleep_for(std::chrono::milliseconds(200));
++i;
if(!(i%5))
RsInfo() << __PRETTY_FUNCTION__ << " " << i*0.2 << " seconds passed"
<< " waiting for thread: " << mTid << " " << mFullName
<< " to stop" << std::endl;
}
}
RsTickingThread::RsTickingThread()
bool RsThread::start(const std::string& threadName)
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << "RsTickingThread::RsTickingThread()" << std::endl;
#endif
}
// Atomically check if the thread was already started and set it as running
if(mHasStopped.exchange(false))
{
mShouldStop = false;
int pError = pthread_create(
&mTid, nullptr, &rsthread_init, static_cast<void*>(this) );
if(pError)
{
RsErr() << __PRETTY_FUNCTION__ << " pthread_create could not create"
<< " new thread: " << threadName << " pError: " << pError
<< std::endl;
mHasStopped = true;
print_stacktrace();
return false;
}
RsTickingThread::~RsTickingThread()
{
fullstop();
}
void RsSingleJobThread::runloop()
{
run() ;
}
/* Store thread full name as PThread is not able to keep it entirely */
mFullName = threadName;
void RsTickingThread::runloop()
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << "RsTickingThread::runloop(). Setting stopped=0" << std::endl;
#endif
/* Set PThread thread name which is restricted to 16 characters
* including the terminating null byte */
if(pthread_setname_np && !threadName.empty())
RS_pthread_setname_np(mTid, threadName.substr(0, 15).c_str());
while(1)
{
if(shouldStop())
{
#ifdef DEBUG_THREADS
THREAD_DEBUG << "pqithreadstreamer::runloop(): asked to stop. setting hasStopped=1, and returning. Thread ends." << std::endl;
#endif
return ;
}
return true;
}
data_tick();
}
RsErr() << __PRETTY_FUNCTION__ << " attempt to start already running thread"
<< std::endl;
print_stacktrace();
return false;
}
RsQueueThread::RsQueueThread(uint32_t min, uint32_t max, double relaxFactor )
@ -278,14 +182,14 @@ RsQueueThread::RsQueueThread(uint32_t min, uint32_t max, double relaxFactor )
mLastWork = time(NULL) ;
}
void RsQueueThread::data_tick()
void RsQueueThread::threadTick()
{
bool doneWork = false;
while(workQueued() && doWork())
{
doneWork = true;
}
rstime_t now = time(NULL);
time_t now = time(NULL);
if (doneWork)
{
mLastWork = now;
@ -310,94 +214,38 @@ void RsQueueThread::data_tick()
THREAD_DEBUG << "RsQueueThread::data_tick() no work: sleeping for: " << mLastSleep << " ms" << std::endl;
#endif
}
rstime::rs_usleep(mLastSleep * 1000); // mLastSleep msec
std::this_thread::sleep_for(std::chrono::milliseconds(mLastSleep));
}
void RsMutex::unlock()
{
#ifdef RSTHREAD_SELF_LOCKING_GUARD
if(--_cnt == 0)
{
#endif
_thread_id = 0 ;
pthread_mutex_unlock(&realMutex);
#ifdef RSTHREAD_SELF_LOCKING_GUARD
}
#endif
{
_thread_id = 0;
pthread_mutex_unlock(&realMutex);
}
void RsMutex::lock()
{
#ifdef RSMUTEX_DEBUG
pthread_t owner = _thread_id ;
#endif
int retval = 0;
#ifdef RSTHREAD_SELF_LOCKING_GUARD
if(!trylock())
if(!pthread_equal(_thread_id,pthread_self()))
#endif
retval = pthread_mutex_lock(&realMutex);
switch(retval)
int err = pthread_mutex_lock(&realMutex);
if( err != 0)
{
case 0:
break;
RsErr() << __PRETTY_FUNCTION__ << "pthread_mutex_lock returned: "
<< rsErrnoName(err)
#ifdef RSMUTEX_DEBUG
<< " name: " << name
#endif
<< std::endl;
case EINVAL:
std::cerr << "RsMutex::lock() pthread_mutex_lock returned EINVAL";
std::cerr << std::endl;
break;
case EBUSY:
std::cerr << "RsMutex::lock() pthread_mutex_lock returned EBUSY";
std::cerr << std::endl;
break;
case EAGAIN:
std::cerr << "RsMutex::lock() pthread_mutex_lock returned EAGAIN";
std::cerr << std::endl;
break;
case EDEADLK:
std::cerr << "RsMutex::lock() pthread_mutex_lock returned EDEADLK";
std::cerr << std::endl;
break;
case EPERM:
std::cerr << "RsMutex::lock() pthread_mutex_lock returned EPERM";
std::cerr << std::endl;
break;
print_stacktrace();
default:
std::cerr << "RsMutex::lock() pthread_mutex_lock returned UNKNOWN ERROR";
std::cerr << std::endl;
break;
}
/* Here is some debugging code - to catch failed locking attempts.
* Major bug is it is ever triggered.
*/
#ifdef RSMUTEX_ABORT
if (retval != 0)
{
#ifdef RSMUTEX_DEBUG
std::cerr << "RsMutex::lock() name: " << name << std::endl;
#endif
std::cerr << "RsMutex::lock() pthread_mutex_lock returned an Error. Aborting()";
std::cerr << std::endl;
abort();
#endif
}
#endif
_thread_id = pthread_self() ;
#ifdef RSTHREAD_SELF_LOCKING_GUARD
++_cnt ;
#endif
_thread_id = pthread_self();
}
#ifdef RSMUTEX_DEBUG
double RsStackMutex::getCurrentTS()
{
@ -416,3 +264,18 @@ double RsStackMutex::getCurrentTS()
#endif
RsThread::~RsThread()
{
if(isRunning())
{
RsErr() << __PRETTY_FUNCTION__ << " deleting thread: " << mTid << " "
<< mFullName << " that is still "
<< "running! Something seems very wrong here and RetroShare is "
<< "likely to crash because of this." << std::endl;
print_stacktrace();
fullstop();
}
}
RsQueueThread::~RsQueueThread() = default;

432
libretroshare/src/util/rsthreads.h
View file

@ -3,7 +3,9 @@
* *
* libretroshare: retroshare core library *
* *
* Copyright 2004-2006 by Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2004-2006 Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2016-2020 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2019-2020 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
@ -26,243 +28,199 @@
#include <string>
#include <iostream>
#include <unistd.h>
#include <semaphore.h>
#include <atomic>
#include <thread>
#include <functional>
#include <util/rsmemory.h>
#include "util/rstime.h"
#include "util/rsmemory.h"
#include "util/rsdeprecate.h"
/* RsIface Thread Wrappers */
#undef RSTHREAD_SELF_LOCKING_GUARD
//#define RSMUTEX_DEBUG 300 // Milliseconds for print in the stderr
//#define RSMUTEX_DEBUG
//#define RSMUTEX_DEBUG
/**
* @brief Provide mutexes that keep track of the owner. Based on pthread mutex.
*/
class RsMutex
{
public:
public:
RsMutex(const std::string& name)
{
/* remove unused parameter warnings */
pthread_mutex_init(&realMutex, NULL);
_thread_id = 0 ;
RsMutex(const std::string& name) : _thread_id(0)
#ifdef RSMUTEX_DEBUG
this->_name = name;
#else
(void) name;
, _name(name)
#endif
{
pthread_mutex_init(&realMutex, nullptr);
#ifndef RSMUTEX_DEBUG
(void) name; // remove unused parameter warnings
#endif
}
~RsMutex()
{
pthread_mutex_destroy(&realMutex);
}
inline const pthread_t& owner() const { return _thread_id ; }
#ifdef RSMUTEX_DEBUG
void setName(const std::string &name)
{
this->_name = name;
}
#endif
~RsMutex() { pthread_mutex_destroy(&realMutex); }
void lock();
void unlock();
bool trylock() { return (0 == pthread_mutex_trylock(&realMutex)); }
inline const pthread_t& owner() const { return _thread_id; }
void lock();
void unlock();
bool trylock() { return (0 == pthread_mutex_trylock(&realMutex)); }
#ifdef RSMUTEX_DEBUG
const std::string& name() const { return _name ; }
#endif
private:
pthread_mutex_t realMutex;
pthread_t _thread_id ;
#ifdef RSTHREAD_SELF_LOCKING_GUARD
uint32_t _cnt ;
#endif
private:
pthread_mutex_t realMutex;
pthread_t _thread_id;
#ifdef RSMUTEX_DEBUG
std::string _name;
std::string _name;
#endif
};
/**
* @def RS_STACK_MUTEX(m)
* This macro allows you to trace which mutex in the code is locked and for how
* much time. You can use this as follows:
* @code
* {
* RS_STACK_MUTEX(myMutex);
* do_something();
* }
* @endcode
*/
#define RS_STACK_MUTEX(m) \
RsStackMutex __local_retroshare_stack_mutex_##m( \
m, __PRETTY_FUNCTION__, __FILE__, __LINE__ )
/**
* Provide mutexes that automatically lock/unlock on creation/destruction and
* have powerfull debugging facilities (if RSMUTEX_DEBUG is defined at
* compiletime).
* In most of the cases you should not use this directly instead
* @see RS_STACK_MUTEX(m)
*/
class RsStackMutex
{
public:
RsStackMutex(RsMutex &mtx)
: mMtx(mtx)
{
mMtx.lock();
#ifdef RSMUTEX_DEBUG
double ts = getCurrentTS() ;
_time_stamp = ts ;
_lineno = 0 ;
_info = "[no info]" ;
#endif
}
RsStackMutex(RsMutex &mtx,const char *function_name,const char *file_name,int lineno)
: mMtx(mtx)
#ifdef RSMUTEX_DEBUG
, _info(std::string(function_name)+" in file "+file_name),_lineno(lineno)
#endif
{
#ifdef RSMUTEX_DEBUG
double ts = getCurrentTS() ;
_time_stamp = ts ;
pthread_t owner = mMtx.owner() ;
#else
/* remove unused parameter warnings */
(void) function_name;
(void) file_name;
(void) lineno;
#endif
mMtx.lock();
#ifdef RSMUTEX_DEBUG
ts = getCurrentTS() ;
if(ts - _time_stamp > 1.0)
std::cerr << "Mutex " << (void*)&mMtx << " \"" << mtx.name() << "\""
<< " waited for " << ts - _time_stamp
<< " seconds in thread " << pthread_self()
<< " for locked thread " << owner << ". in " << _info << ":" << _lineno << std::endl;
_time_stamp = ts ; // This is to re-init the locking time without accounting for how much we waited.
#endif
}
~RsStackMutex()
{
mMtx.unlock();
#ifdef RSMUTEX_DEBUG
double ts = getCurrentTS() ;
if(ts - _time_stamp > 1.0)
std::cerr << "Mutex " << (void*)&mMtx << " \"" << mMtx.name() << "\""
<< " locked for " << ts - _time_stamp
<< " seconds in thread " << pthread_self()
<< ". in " << _info << ":" << _lineno << std::endl;
#endif
}
private:
RsMutex &mMtx;
#ifdef RSMUTEX_DEBUG
static double getCurrentTS() ;
double _time_stamp ;
std::string _info ;
int _lineno ;
#endif
};
// This macro allows you to trace which mutex in the code is locked for how much time.
// se this as follows:
//
// {
// RS_STACK_MUTEX(myMutex) ;
//
// do_something() ;
// }
//
#define RS_STACK_MUTEX(m) RsStackMutex __local_retroshare_mutex(m,__PRETTY_FUNCTION__,__FILE__,__LINE__)
// This class handles a Mutex-based semaphore, that makes it cross plateform.
class RsSemaphore
{
class RsSemStruct
{
public:
RsSemStruct() : mtx("Semaphore mutex"), val(0) {}
RsMutex mtx ;
uint32_t val ;
};
public:
RsSemaphore()
{
s = new RsSemStruct ;
}
~RsSemaphore()
{
delete s ;
}
RsStackMutex(RsMutex &mtx) : mMtx(mtx)
{
mMtx.lock();
#ifdef RSMUTEX_DEBUG
double ts = getCurrentTS();
_time_stamp = ts;
_lineno = 0;
_info = "[no info]";
#endif
}
void set(uint32_t i)
{
RS_STACK_MUTEX(s->mtx) ;
s->val = i ;
}
RsStackMutex(RsMutex &mtx, const char *function_name, const char *file_name,
int lineno) : mMtx(mtx)
#ifdef RSMUTEX_DEBUG
, _info(std::string(function_name)+" in file "+file_name), _lineno(lineno)
#endif
{
#ifdef RSMUTEX_DEBUG
double ts = getCurrentTS();
_time_stamp = ts;
pthread_t owner = mMtx.owner();
#else
// remove unused parameter warnings
(void) function_name; (void) file_name; (void) lineno;
#endif
void post()
{
RS_STACK_MUTEX(s->mtx) ;
++(s->val) ;
}
mMtx.lock();
uint32_t value()
{
RS_STACK_MUTEX(s->mtx) ;
return s->val ;
}
#ifdef RSMUTEX_DEBUG
ts = getCurrentTS();
// waits but does not re-locks the semaphore
void wait_no_relock()
{
static const uint32_t max_waiting_time_before_warning=1000 *5 ; // 5 secs
uint32_t tries=0;
if(ts - _time_stamp > 1.0)
std::cerr << "Mutex " << (void*)&mMtx << " \"" << mtx.name() << "\""
<< " waited for " << ts - _time_stamp
<< " seconds in thread " << pthread_self()
<< " for locked thread " << owner << ". in " << _info
<< ":" << _lineno << std::endl;
_time_stamp = ts ; // This is to re-init the locking time without accounting for how much we waited.
#endif
}
while(true)
{
usleep(1000) ;
if(++tries >= max_waiting_time_before_warning)
std::cerr << "(EE) Semaphore waiting for too long. Something is probably wrong in the code." << std::endl;
~RsStackMutex()
{
mMtx.unlock();
RS_STACK_MUTEX(s->mtx) ;
if(s->val > 0)
return ;
}
#ifdef RSMUTEX_DEBUG
double ts = getCurrentTS();
if(ts - _time_stamp > 1.0)
std::cerr << "Mutex " << (void*)&mMtx << " \"" << mMtx.name()
<< "\"" << " locked for " << ts - _time_stamp
<< " seconds in thread " << pthread_self()
<< ". in " << _info << ":" << _lineno << std::endl;
#endif
}
}
private:
RsSemStruct *s ;
RsMutex &mMtx;
#ifdef RSMUTEX_DEBUG
static double getCurrentTS();
double _time_stamp;
std::string _info;
int _lineno;
#endif
};
class RsThread;
/* to create a thread! */
pthread_t createThread(RsThread &thread);
/// @brief Offer basic threading functionalities.
class RsThread
{
public:
RsThread();
virtual ~RsThread() ;
void start(const std::string &threadName = "");
// Returns true if the thread is still running.
bool isRunning();
// Returns true if the thread received a stopping order and hasn't yet stopped.
bool shouldStop();
// Can be called to set the stopping flags. The stop will not be handled
// by RsThread itself, but in subclasses. If you derive your own subclass,
// you need to call shouldStop() in order to check for a possible stopping order.
void ask_for_stop();
RsThread();
virtual ~RsThread();
/**
* Execute given function on another thread without blocking the caller
* @brief start the thread and call run() on it.
* @param threadName string containing the name of the thread used for
* debugging purposes, @note inside PThread it is
* truncated to 16 characters including \0 at the end of
* the string.
* @return false on error, true otherwise
*/
bool start(const std::string& threadName = "");
/**
* @brief Check if thread is running.
* @return true if the thread is still running, false otherwise.
*/
bool isRunning();
/**
* @brief Check if the thread should stop.
* Expecially useful for subclasses which implement a @see run() method
* which may take lot of time before returning when not asked, to check if
* stop has been requested and therefore interrupting the execution ASAP
* returning in a coherent state.
* @return true if the thread received a stopping order.
*/
bool shouldStop();
/**
* @brief Asyncronously ask the thread to stop.
* The real stop will happen when the @see run() method finish.
*/
void askForStop();
/**
* Call @see askForStop() then wait it has really stopped before returning.
* It must not be called in the same thread, as it would not wait for the
* effective stop to occur as it would cause a deadlock.
*/
void fullstop();
/**
* Execute given function on a detached thread without blocking the caller
* execution.
* This can be generalized with variadic template, ATM it is enough to wrap
* any kind of function call or job into a lambda which get no paramethers
@ -272,60 +230,84 @@ public:
static void async(const std::function<void()>& fn)
{ std::thread(fn).detach(); }
/** @return RsThread full name */
const std::string& threadName() { return mFullName; }
protected:
virtual void runloop() =0; /* called once the thread is started. Should be overloaded by subclasses. */
void go() ; // this one calls runloop and also sets the flags correctly when the thread is finished running.
/**
* This method must be implemented by sublasses, will be called once the
* thread is started. Should return on request, use @see shouldStop() to
* check if stop has been requested.
*/
virtual void run() = 0;
RsSemaphore mHasStoppedSemaphore;
RsSemaphore mShouldStopSemaphore;
/**
* This method is meant to be overridden by subclasses with long running
* @see run() method and is executed asyncronously when @see askForStop()
* is called, any task necessary to stop the thread (aka inducing @see run()
* to return in a coherent state) should be done in the overridden version
* of this method, @see JsonApiServer for an usage example. */
virtual void onStopRequested() {}
static void *rsthread_init(void*) ;
pthread_t mTid;
private:
/** Call @see run() setting the appropriate flags around it*/
void wrapRun();
/// True if thread is stopped, false otherwise
std::atomic<bool> mHasStopped;
/// True if stop has been requested
std::atomic<bool> mShouldStop;
/// Passed as argument for pthread_create(), call start()
static void *rsthread_init(void*);
/// Store the id of the corresponding pthread
pthread_t mTid;
/// Store thread full name
std::string mFullName;
};
/**
* Provide a detached execution loop that continuously call data_tick() once the
* thread is started
*/
class RsTickingThread: public RsThread
{
public:
RsTickingThread();
virtual ~RsTickingThread();
void shutdown();
void fullstop();
void join() { fullstop() ; } // used for compatibility
virtual void data_tick() =0;
/**
* Subclasses must implement this method, it will be called in a loop once
* the thread is started, so repetitive work (like checking if data is
* available on a socket) should be done here, at the end of this method
* sleep_for(...) or similar function should be called or the CPU will
* be used as much as possible also if there is nothing to do.
*/
virtual void threadTick() = 0;
private:
virtual void runloop() ; /* called once the thread is started. Should be overloaded by subclasses. */
};
class RsSingleJobThread: public RsThread
{
public:
virtual void run() =0;
protected:
virtual void runloop() ;
/// Implement the run loop and continuously call threadTick() in it
void run() override { while(!shouldStop()) threadTick(); }
};
// TODO: Used just one time, is this really an useful abstraction?
class RsQueueThread: public RsTickingThread
{
public:
RsQueueThread(uint32_t min, uint32_t max, double relaxFactor );
virtual ~RsQueueThread() { return; }
RsQueueThread(uint32_t min, uint32_t max, double relaxFactor);
~RsQueueThread() override;
protected:
virtual bool workQueued() = 0;
virtual bool doWork() = 0;
virtual bool workQueued() = 0;
virtual bool doWork() = 0;
virtual void data_tick() ;
void threadTick() override; /// @see RsTickingThread
private:
uint32_t mMinSleep; /* ms */
uint32_t mMaxSleep; /* ms */
uint32_t mLastSleep; /* ms */
rstime_t mLastWork; /* secs */
time_t mLastWork; /* secs */
float mRelaxFactor;
};