133 lines
4.2 KiB
C++
133 lines
4.2 KiB
C++
#pragma once
|
|
#include <pfc/wait_queue.h>
|
|
#include <pfc/pool.h>
|
|
#include <pfc/threads.h>
|
|
#include <functional>
|
|
#include "rethrow.h"
|
|
#include <pfc/timers.h>
|
|
|
|
namespace ThreadUtils {
|
|
|
|
// Serialize access to some resource to a single thread
|
|
// Execute blocking/nonabortable methods in with proper abortability (detach on abort and move on)
|
|
class cmdThread {
|
|
public:
|
|
typedef std::function<void () > func_t;
|
|
typedef pfc::waitQueue<func_t> queue_t;
|
|
typedef std::function<void (abort_callback&) > funcAbortable_t;
|
|
|
|
protected:
|
|
std::function<void () > makeWorker() {
|
|
auto q = m_queue;
|
|
auto x = m_atExit;
|
|
return [q, x] {
|
|
for ( ;; ) {
|
|
func_t f;
|
|
if (!q->get(f)) break;
|
|
try { f(); } catch(...) {}
|
|
}
|
|
// No guard for atExit access, as nobody is supposed to be still able to call host object methods by the time we get here
|
|
for( auto i = x->begin(); i != x->end(); ++ i ) {
|
|
auto f = *i;
|
|
try { f(); } catch(...) {}
|
|
}
|
|
};
|
|
};
|
|
std::function<void () > makeWorker2( std::function<void()> updater, double interval) {
|
|
auto q = m_queue;
|
|
auto x = m_atExit;
|
|
return [=] {
|
|
pfc::lores_timer t; t.start();
|
|
for ( ;; ) {
|
|
{
|
|
bool bWorkReady = false;
|
|
double left = interval - t.query();
|
|
if ( left > 0 ) {
|
|
if (q->wait_read( left )) bWorkReady = true;
|
|
}
|
|
|
|
if (!bWorkReady) {
|
|
updater();
|
|
t.start();
|
|
continue;
|
|
}
|
|
}
|
|
|
|
func_t f;
|
|
if (!q->get(f)) break;
|
|
try { f(); } catch(...) {}
|
|
}
|
|
// No guard for atExit access, as nobody is supposed to be still able to call host object methods by the time we get here
|
|
for( auto i = x->begin(); i != x->end(); ++ i ) {
|
|
auto f = *i;
|
|
try { f(); } catch(...) {}
|
|
}
|
|
};
|
|
};
|
|
|
|
// For derived classes: create new instance without starting thread, supply thread using by yourself
|
|
class noCreate {};
|
|
cmdThread( noCreate ) {}
|
|
public:
|
|
|
|
cmdThread() {
|
|
pfc::splitThread( makeWorker() );
|
|
}
|
|
|
|
void atExit( func_t f ) {
|
|
m_atExit->push_back(f);
|
|
}
|
|
~cmdThread() {
|
|
m_queue->set_eof();
|
|
}
|
|
void runSynchronously( func_t f ) { runSynchronously_(f, nullptr); }
|
|
void runSynchronously_( func_t f, abort_callback * abortOrNull ) {
|
|
auto evt = m_eventPool.make();
|
|
evt->set_state(false);
|
|
auto rethrow = std::make_shared<ThreadUtils::CRethrow>();
|
|
auto worker2 = [f, rethrow, evt] {
|
|
rethrow->exec(f);
|
|
evt->set_state( true );
|
|
};
|
|
|
|
add ( worker2 );
|
|
|
|
if ( abortOrNull != nullptr ) {
|
|
abortOrNull->waitForEvent( * evt, -1 );
|
|
} else {
|
|
evt->wait_for(-1);
|
|
}
|
|
|
|
m_eventPool.put( evt );
|
|
|
|
rethrow->rethrow();
|
|
}
|
|
void runSynchronously( func_t f, abort_callback & abort ) {
|
|
runSynchronously_(f, &abort);
|
|
}
|
|
void runSynchronously2( funcAbortable_t f, abort_callback & abort ) {
|
|
auto subAbort = m_abortPool.make();
|
|
subAbort->reset();
|
|
auto worker = [subAbort, f] {
|
|
f(*subAbort);
|
|
};
|
|
|
|
try {
|
|
runSynchronously( worker, abort );
|
|
} catch(...) {
|
|
subAbort->set(); throw;
|
|
}
|
|
|
|
m_abortPool.put( subAbort );
|
|
}
|
|
|
|
void add( func_t f ) { m_queue->put( f ); }
|
|
private:
|
|
pfc::objPool<pfc::event> m_eventPool;
|
|
pfc::objPool<abort_callback_impl> m_abortPool;
|
|
std::shared_ptr<queue_t> m_queue = std::make_shared<queue_t>();
|
|
typedef std::list<func_t> atExit_t;
|
|
std::shared_ptr<atExit_t> m_atExit = std::make_shared< atExit_t >();
|
|
};
|
|
}
|