Singularity-Viewer/SingularityViewer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Work in progress

Browse Source 
This work extends AIStateMachine to run multiplex() in the thread
that calls run(), cont() or set_state(). Note that all three
eventually call locked_cont(), so thats where multiplex() is called
from. Calling multiplex() means "running the state machine", as in
"calling multiplex_impl".

Currently only LLURLRequest uses this feature, and then only
for the HTTPGetResponder, and well only for the initializing,
start up and normal finish states.

A current/remaining problem is that we run into a situation where
the curl thread runs a statemachine to it's finish and kills it,
while the main thread is also 'running' it and tries to call
multiplex while the statemachine isn't running anymore.
This commit is contained in:
Aleric Inglewood
2013-02-20 23:29:34 +01:00
parent ff3910a705
commit ef35aa7954
14 changed files with 235 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
indra/aistatemachine/aistatemachine.cpp
View File

@@ -82,7 +82,8 @@ void AIStateMachine::setMaxCount(F32 StateMachineMaxTime)
void AIStateMachine::run(AIStateMachine* parent, state_type new_parent_state, bool abort_parent, bool on_abort_signal_parent)
{
DoutEntering(dc::statemachine, "AIStateMachine::run(" << (void*)parent << ", " << (parent ? parent->state_str(new_parent_state) : "NA") << ", " << abort_parent << ") [" << (void*)this << "]");
DoutEntering(dc::statemachine, "AIStateMachine::run(" << (void*)parent << ", " << (parent ? parent->state_str(new_parent_state) : "NA") <<
", " << abort_parent << ", " << on_abort_signal_parent << ") [" << (void*)this << "]");
// Must be the first time we're being run, or we must be called from a callback function.
llassert(!mParent || mState == bs_callback);
llassert(!mCallback || mState == bs_callback);
@@ -113,7 +114,7 @@ void AIStateMachine::run(AIStateMachine* parent, state_type new_parent_state, bo
// Mark that run() has been called, in case we're being called from a callback function.
mState = bs_initialize;
// Set mIdle to false and add statemachine to continued_statemachines.
// Set mIdle to false and add statemachine to continued_statemachines-- or, when boosted, run the statemachine till it finished or goes idle.
mSetStateLock.lock();
locked_cont();
}
@@ -140,7 +141,7 @@ void AIStateMachine::run(callback_type::signal_type::slot_type const& slot)
// Mark that run() has been called, in case we're being called from a callback function.
mState = bs_initialize;
// Set mIdle to false and add statemachine to continued_statemachines.
// Set mIdle to false and add statemachine to continued_statemachines-- or, when boosted, run the statemachine till it finished or goes idle.
mSetStateLock.lock();
locked_cont();
}
@@ -148,9 +149,11 @@ void AIStateMachine::run(callback_type::signal_type::slot_type const& slot)
void AIStateMachine::idle(void)
{
DoutEntering(dc::statemachine, "AIStateMachine::idle() [" << (void*)this << "]");
llassert(is_main_thread());
llassert(mInsideMultiplexImpl);
mSetStateLock.lock();
llassert(!mIdle);
mIdle = true;
mSetStateLock.unlock();
mSleep = 0;
#ifdef SHOW_ASSERT
mCalledThreadUnsafeIdle = true;
@@ -160,9 +163,12 @@ void AIStateMachine::idle(void)
void AIStateMachine::idle(state_type current_run_state)
{
DoutEntering(dc::statemachine, "AIStateMachine::idle(" << state_str(current_run_state) << ") [" << (void*)this << "]");
llassert(is_main_thread());
llassert(!mIdle);
llassert(mInsideMultiplexImpl); // Only boosted state machines can come here as non-main thread because you may only get here from multiplex_impl().
mSetStateLock.lock();
// If two different threads call multiplex_impl() (boosted state machines) then that should happen serialized:
// only one thread may call multiplex() at any given time. Hence, it is impossible that idle() is called while
// we are already idle.
llassert(!mIdle);
// Only go idle if the run state is (still) what we expect it to be.
// Otherwise assume that another thread called set_state() and continue running.
if (current_run_state == mRunState)
@@ -196,6 +202,7 @@ void AIStateMachine::locked_cont(void)
mIdle = false;
bool not_active = mActive == as_idle;
mIdleActive.unlock();
mBoost = thread_safe_impl();
// mActive is only changed in AIStateMachine::mainloop, by the main-thread, and
// here, possibly by any thread. However, after setting mIdle to false above, it
// is impossible for any thread to come here, until after the main-thread called
@@ -210,8 +217,15 @@ void AIStateMachine::locked_cont(void)
// to release mSetStateLock here, with as advantage that if we're not the main-
// thread and not_active is true, then the main-thread won't block when it has
// a timer running that times out and calls set_state().
// Finally, if mBoost is true then not_active is not going to be tested and
// mActive is not going to be changed, so then we're done with this lock too.
mSetStateLock.unlock();
if (not_active)
if (mBoost)
{
// Run statemachine until it is finished or idle again.
multiplex(0);
}
else if (not_active)
{
AIWriteAccess<csme_type> csme_w(sContinuedStateMachinesAndMainloopEnabled);
// See above: it is not possible that mActive was changed since not_active
@@ -236,9 +250,6 @@ void AIStateMachine::set_state(state_type state)
// Stop race condition of multiple threads calling cont() or set_state() here.
mSetStateLock.lock();
// Do not call set_state() unless running.
llassert(mState == bs_run || !is_main_thread());
// If this function is called from another thread than the main thread, then we have to ignore
// it if we're not idle and the state is less than the current state. The main thread must
// be able to change the state to anything (also smaller values). Note that that only can work
@@ -279,8 +290,11 @@ void AIStateMachine::set_state(state_type state)
return; // Ignore.
}
// Do not call set_state() from multiplex_impl unless running.
llassert(mState == bs_run || !mInsideMultiplexImpl);
// Do not call idle() when set_state is called from another thread; use idle(state_type) instead.
llassert(!mCalledThreadUnsafeIdle || is_main_thread());
llassert(!mCalledThreadUnsafeIdle || AIThreadID::in_main_thread());
// Change mRunState to the requested value.
if (mRunState != state)
@@ -293,7 +307,15 @@ void AIStateMachine::set_state(state_type state)
if (mIdle)
locked_cont(); // This unlocks mSetStateLock.
else
{
// The state was changed while not being idle.
// Note that if thread_safe_impl() returns true now then we don't need to call multiplex() here
// when we ended up here from multiplex(). However, if we end up here as the result of an
// external state change while the state machine is running in the main thread, then we can't
// call multiplex here; it would be too dangerous. In that case we let the main thread pick
// up this state.
mSetStateLock.unlock();
}
// If we get here then mIdle is false, so only mRunState can still be changed but we won't
// call locked_cont() anymore. When the main thread finally picks up on the state change,
@@ -454,6 +476,7 @@ void AIStateMachine::multiplex(U64 current_time)
// amount of time has passed.
if (mSleep != 0)
{
llassert(!mBoost); // Sleeping MUST be done in the main thread: thread_safe_impl() should have returned false.
if (mSleep < 0)
{
if (++mSleep)
@@ -475,11 +498,90 @@ void AIStateMachine::multiplex(U64 current_time)
{
mAborted = false;
mState = bs_run;
#ifdef SHOW_ASSERT
mInsideMultiplexImpl = true;
#endif
initialize_impl();
#ifdef SHOW_ASSERT
mInsideMultiplexImpl = false;
#endif
if (mAborted || mState != bs_run)
{
mBoost = false;
return;
}
mSetStateLock.lock();
mBoost = thread_safe_impl();
if (!mBoost && !AIThreadID::in_main_thread())
{
// Pass running on the main thread.
mIdle = true;
locked_cont();
return;
}
mSetStateLock.unlock();
}
if (!mBoost)
{
llassert(AIThreadID::in_main_thread());
#ifdef SHOW_ASSERT
mInsideMultiplexImpl = true;
#endif
multiplex_impl();
#ifdef SHOW_ASSERT
mInsideMultiplexImpl = false;
#endif
}
else
{
if (mState != bs_run)
return;
mSetStateLock.lock();
llassert(!mIdle); // We should never get here when idle.
if (!mMultiplexMutex.tryLock()) // Point A
{
// If another thread is already already calling multiplex_impl() then we can't do that too.
// However, instead of blocking we simply return: by getting here it is guaranteed that
// the other thread will do at least one more test of 'mState == bs_run && !mIdle' before
// stopping to run the statemachine.
// The reason for that is that if another thread had mMultiplexMutex at point A, while
// we had mSetStateLock, then the other thread has to be between point B and point C
// because that is the only area where mSetStateLock is unlocked while mMultiplexMutex
// is locked, and the test follows this area.
mSetStateLock.unlock();
return;
}
do
{
mSetStateLock.unlock(); // Point B
#ifdef SHOW_ASSERT
mInsideMultiplexImpl = true;
#endif
multiplex_impl();
#ifdef SHOW_ASSERT
mInsideMultiplexImpl = false;
#endif
mSetStateLock.lock(); // Point C
// The state might have changed, check if we may still run outside the main thread.
mBoost = thread_safe_impl();
if (!mBoost && !AIThreadID::in_main_thread())
{
mMultiplexMutex.unlock();
// Pass running on the main thread.
mIdle = true;
locked_cont();
return;
}
}
while (mState == bs_run && !mIdle);
mMultiplexMutex.unlock();
mSetStateLock.unlock();
}
multiplex_impl();
}
//static
@@ -590,6 +692,10 @@ void AIStateMachine::flush(void)
for (active_statemachines_type::iterator iter = active_statemachines.begin(); iter != active_statemachines.end(); ++iter)
{
AIStateMachine& statemachine(iter->statemachine());
#ifdef SHOW_ASSERT
// Make assertions happy.
statemachine.mInsideMultiplexImpl = true;
#endif
if (statemachine.abortable())
{
// We can't safely call abort() here for non-running (run() was called, but they weren't initialized yet) statemachines,

23
indra/aistatemachine/aistatemachine.h
View File

@@ -176,6 +176,7 @@
// Should return a stringified value of run_state.
//
class AIStateMachine {
protected:
//! The type of mState
enum base_state_type {
bs_initialize,
@@ -213,12 +214,15 @@ class AIStateMachine {
base_state_type mState; //!< State of the base class.
bool mIdle; //!< True if this state machine is not running.
bool mAborted; //!< True after calling abort() and before calling run().
active_type mActive; //!< Whether statemachine is idle, queued to be added to the active list, or already on the active list.
S64 mSleep; //!< Non-zero while the state machine is sleeping.
LLMutex mIdleActive; //!< Used for atomic operations on the pair mIdle / mActive.
bool mBoost; //!< True when not being run by the main thread, but by which ever thread is changing the state.
active_type mActive; //!< Whether statemachine is idle, queued to be added to the active list, or already on the active list (not used for boosted SM).
S64 mSleep; //!< Non-zero while the state machine is sleeping (not used for boosted SM).
LLMutex mIdleActive; //!< Used for atomic operations on the pair mIdle / mActive (not used for boosted SM).
LLMutex mMultiplexMutex; //!< Used for boosted state machines, locked during multiplexing.
#ifdef SHOW_ASSERT
AIThreadID mContThread; //!< Thread that last called locked_cont().
bool mCalledThreadUnsafeIdle; //!< Set to true when idle() is called.
bool mInsideMultiplexImpl; //!< Set to true while running a boosted state machine (also set while inside initialize_impl).
#endif
// Callback facilities.
@@ -250,10 +254,11 @@ class AIStateMachine {
public:
//! Create a non-running state machine.
AIStateMachine(void) : mState(bs_initialize), mIdle(true), mAborted(true), mActive(as_idle), mSleep(0), mParent(NULL), mCallback(NULL)
AIStateMachine(void) : mState(bs_initialize), mIdle(true), mAborted(true), mActive(as_idle), mSleep(0),
#ifdef SHOW_ASSERT
, mContThread(AIThreadID::none), mCalledThreadUnsafeIdle(false)
mContThread(AIThreadID::none), mCalledThreadUnsafeIdle(false),
#endif
mParent(NULL), mCallback(NULL), mRunState(0)
{ }
protected:
@@ -374,6 +379,10 @@ class AIStateMachine {
//! Return a stringified state, for debugging purposes.
char const* state_str(state_type state);
protected:
//! Return mState.
base_state_type base_state(void) const { return mState; }
private:
static void add_continued_statemachines(AIReadAccess<csme_type>& csme_r);
static void dowork(void);
@@ -412,6 +421,10 @@ class AIStateMachine {
// Handle cleaning up from initialization (or post abort) state.
virtual void finish_impl(void) = 0;
// Determine if it is thread-safe to run the current state in another thread than the main-thread.
// Only called while mSetStateLock is locked, may ONLY access mState and mRunState.
virtual bool thread_safe_impl(void) const { return false; }
// Implemenation of state_str for run states.
virtual char const* state_str_impl(state_type run_state) const = 0;
};

17
indra/aistatemachine/aitimer.cpp
View File

@@ -33,7 +33,8 @@
enum timer_state_type {
AITimer_start = AIStateMachine::max_state,
AITimer_expired
AITimer_expired,
AITimer_abort
};
char const* AITimer::state_str_impl(state_type run_state) const
@@ -42,6 +43,7 @@ char const* AITimer::state_str_impl(state_type run_state) const
{
AI_CASE_RETURN(AITimer_start);
AI_CASE_RETURN(AITimer_expired);
AI_CASE_RETURN(AITimer_abort);
}
return "UNKNOWN STATE";
}
@@ -52,6 +54,12 @@ void AITimer::initialize_impl(void)
set_state(AITimer_start);
}
void AITimer::request_abort(void)
{
mFrameTimer.cancel();
set_state(AITimer_abort);
}
void AITimer::expired(void)
{
set_state(AITimer_expired);
@@ -64,7 +72,7 @@ void AITimer::multiplex_impl(void)
case AITimer_start:
{
mFrameTimer.create(mInterval, boost::bind(&AITimer::expired, this));
idle();
idle(AITimer_start);
break;
}
case AITimer_expired:
@@ -72,6 +80,11 @@ void AITimer::multiplex_impl(void)
finish();
break;
}
case AITimer_abort:
{
abort();
break;
}
}
}

5
indra/aistatemachine/aitimer.h
View File

@@ -82,6 +82,11 @@ class AITimer : public AIStateMachine {
*/
F64 getInterval(void) const { return mInterval; }
/**
* @brief Abort the timer from another thread.
*/
void request_abort(void);
protected:
// Call finish() (or abort()), not delete.
/*virtual*/ ~AITimer() { DoutEntering(dc::statemachine, "~AITimer() [" << (void*)this << "]"); mFrameTimer.cancel(); }