Go to the documentation of this file.00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037
00038
00039
00040 #include "Logger.hpp"
00041 #include "ExecutionEngine.hpp"
00042 #include "base/TaskCore.hpp"
00043 #include "rtt-fwd.hpp"
00044 #include "os/MutexLock.hpp"
00045 #include "internal/MWSRQueue.hpp"
00046 #include "TaskContext.hpp"
00047 #include "internal/CatchConfig.hpp"
00048 #include "extras/SlaveActivity.hpp"
00049
00050 #include <boost/bind.hpp>
00051 #include <boost/ref.hpp>
00052 #ifndef USE_CPP11
00053 #include <boost/lambda/lambda.hpp>
00054 #include <boost/lambda/bind.hpp>
00055 #endif
00056 #include <functional>
00057 #include <algorithm>
00058
00059 #define ORONUM_EE_MQUEUE_SIZE 100
00060
00061 namespace RTT
00062 {
00070 using namespace std;
00071 using namespace detail;
00072 using namespace boost;
00073
00074 ExecutionEngine::ExecutionEngine( TaskCore* owner )
00075 : taskc(owner),
00076 mqueue(new MWSRQueue<DisposableInterface*>(ORONUM_EE_MQUEUE_SIZE) ),
00077 f_queue( new MWSRQueue<ExecutableInterface*>(ORONUM_EE_MQUEUE_SIZE) ),
00078 mmaster(0)
00079 {
00080 }
00081
00082 ExecutionEngine::~ExecutionEngine()
00083 {
00084 Logger::In in("~ExecutionEngine");
00085
00086
00087 std::vector<TaskCore*> copy = children;
00088 for (std::vector<TaskCore*>::iterator it = copy.begin(); it != copy.end();++it){
00089 (*it)->setExecutionEngine( 0 );
00090 }
00091 assert( children.empty() );
00092
00093 ExecutableInterface* foo;
00094 while ( f_queue->dequeue( foo ) )
00095 foo->unloaded();
00096
00097 DisposableInterface* dis;
00098 while ( mqueue->dequeue( dis ) )
00099 dis->dispose();
00100
00101 delete f_queue;
00102 delete mqueue;
00103 }
00104
00105 TaskCore* ExecutionEngine::getParent() {
00106 return taskc;
00107 }
00108
00109 void ExecutionEngine::addChild(TaskCore* tc) {
00110 children.push_back( tc );
00111 }
00112
00113 void ExecutionEngine::removeChild(TaskCore* tc) {
00114 vector<TaskCore*>::iterator it = find (children.begin(), children.end(), tc );
00115 if ( it != children.end() )
00116 children.erase(it);
00117 }
00118
00119 void ExecutionEngine::processFunctions()
00120 {
00121
00122 ExecutableInterface* foo = 0;
00123 int nbr = f_queue->size();
00124
00125 while ( f_queue->dequeue(foo) ) {
00126 assert(foo);
00127 if ( foo->execute() == false ){
00128 foo->unloaded();
00129 msg_cond.broadcast();
00130 } else {
00131 f_queue->enqueue( foo );
00132 }
00133 if ( --nbr == 0)
00134 break;
00135 }
00136 }
00137
00138 bool ExecutionEngine::runFunction( ExecutableInterface* f )
00139 {
00140 if (this->getActivity() && f) {
00141
00142 if (taskc && taskc->mTaskState == TaskCore::FatalError )
00143 return false;
00144 f->loaded(this);
00145 bool result = f_queue->enqueue( f );
00146
00147 this->getActivity()->trigger();
00148 return result;
00149 }
00150 return false;
00151 }
00152
00153 struct RemoveMsg : public DisposableInterface {
00154 ExecutableInterface* mf;
00155 ExecutionEngine* mee;
00156 bool found;
00157 RemoveMsg(ExecutableInterface* f, ExecutionEngine* ee)
00158 : mf(f),mee(ee), found(false) {}
00159 virtual void executeAndDispose() {
00160 mee->removeSelfFunction( mf );
00161 found = true;
00162 }
00163 virtual void dispose() {}
00164 virtual bool isError() const { return false;}
00165
00166 };
00167
00168 bool ExecutionEngine::removeFunction( ExecutableInterface* f )
00169 {
00170
00171 if ( !f )
00172 return false;
00173
00174 if ( !f->isLoaded() )
00175 return true;
00176
00177
00178 if ( getActivity() == 0 || !this->getActivity()->isActive() ) {
00179 if ( removeSelfFunction( f ) == false )
00180 return false;
00181 } else {
00182
00183 RemoveMsg rmsg(f,this);
00184 if ( this->process(&rmsg) )
00185 #ifdef USE_CPP11
00186 this->waitForMessages( ! bind(&ExecutableInterface::isLoaded, f) || bind(&RemoveMsg::found,boost::ref(rmsg)) );
00187 #else
00188 this->waitForMessages( ! lambda::bind(&ExecutableInterface::isLoaded, f) || lambda::bind(&RemoveMsg::found,boost::ref(rmsg)) );
00189 #endif
00190 if (!rmsg.found)
00191 return false;
00192 }
00193
00194 f->unloaded();
00195 return true;
00196 }
00197
00198 bool ExecutionEngine::removeSelfFunction(ExecutableInterface* f )
00199 {
00200
00201 if ( !f )
00202 return false;
00203 int nbr = f_queue->size();
00204 while (nbr != 0) {
00205 ExecutableInterface* foo = 0;
00206 if ( !f_queue->dequeue(foo) )
00207 return false;
00208 if ( f == foo) {
00209 return true;
00210 }
00211 f_queue->enqueue(foo);
00212 --nbr;
00213 }
00214 return true;
00215 }
00216
00217 bool ExecutionEngine::initialize() {
00218
00219 return true;
00220 }
00221
00222 bool ExecutionEngine::hasWork()
00223 {
00224 return !mqueue->isEmpty();
00225 }
00226
00227 void ExecutionEngine::processMessages()
00228 {
00229
00230
00231 DisposableInterface* com(0);
00232 {
00233 while ( mqueue->dequeue(com) ) {
00234 assert( com );
00235 com->executeAndDispose();
00236 }
00237
00238
00239
00240
00241
00242
00243 MutexLock locker( msg_lock );
00244 }
00245 if ( com )
00246 msg_cond.broadcast();
00247 }
00248
00249 bool ExecutionEngine::process( DisposableInterface* c )
00250 {
00251
00252 if (mmaster) {
00253 return mmaster->process(c);
00254 }
00255
00256 if ( c && this->getActivity() ) {
00257
00258 if (taskc && taskc->mTaskState == TaskCore::FatalError )
00259 return false;
00260
00261 bool result = mqueue->enqueue( c );
00262 this->getActivity()->trigger();
00263 msg_cond.broadcast();
00264 return result;
00265 }
00266 return false;
00267 }
00268
00269 void ExecutionEngine::waitForMessages(const boost::function<bool(void)>& pred)
00270 {
00271
00272 if (mmaster) {
00273 mmaster->waitForMessages(pred);
00274 return;
00275 }
00276
00277 if (this->getActivity()->thread()->isSelf())
00278 waitAndProcessMessages(pred);
00279 else
00280 waitForMessagesInternal(pred);
00281 }
00282
00283
00284 void ExecutionEngine::waitForFunctions(const boost::function<bool(void)>& pred)
00285 {
00286 if (this->getActivity()->thread()->isSelf())
00287 waitAndProcessFunctions(pred);
00288 else
00289 waitForMessagesInternal(pred);
00290 }
00291
00292 void ExecutionEngine::setMaster(ExecutionEngine *master)
00293 {
00294 mmaster = master;
00295 }
00296
00297 void ExecutionEngine::setActivity( base::ActivityInterface* task )
00298 {
00299 extras::SlaveActivity *slave_activity = dynamic_cast<extras::SlaveActivity *>(task);
00300 if (slave_activity && slave_activity->getMaster()) {
00301 ExecutionEngine *master = dynamic_cast<ExecutionEngine *>(slave_activity->getMaster()->getRunner());
00302 setMaster(master);
00303 } else {
00304 setMaster(0);
00305 }
00306 RTT::base::RunnableInterface::setActivity(task);
00307 }
00308
00309 void ExecutionEngine::waitForMessagesInternal(boost::function<bool(void)> const& pred)
00310 {
00311 if ( pred() )
00312 return;
00313
00314 os::MutexLock lock(msg_lock);
00315 while (!pred()) {
00316 msg_cond.wait(msg_lock);
00317 }
00318 }
00319
00320
00321 void ExecutionEngine::waitAndProcessMessages(boost::function<bool(void)> const& pred)
00322 {
00323 while ( !pred() ){
00324
00325 this->processMessages();
00326 {
00327
00328
00329 os::MutexLock lock(msg_lock);
00330 if (!pred()) {
00331 msg_cond.wait(msg_lock);
00332 } else {
00333 return;
00334 }
00335 }
00336 }
00337 }
00338
00339 void ExecutionEngine::waitAndProcessFunctions(boost::function<bool(void)> const& pred)
00340 {
00341 while ( !pred() ){
00342
00343 this->processFunctions();
00344 {
00345
00346
00347 os::MutexLock lock(msg_lock);
00348 if (!pred()) {
00349 msg_cond.wait(msg_lock);
00350 } else {
00351 return;
00352 }
00353 }
00354 }
00355 }
00356
00357 void ExecutionEngine::step() {
00358 processMessages();
00359 processFunctions();
00360 processChildren();
00361 }
00362
00363 void ExecutionEngine::processChildren() {
00364
00365 if ( taskc ) {
00366
00367 if ( taskc->mTaskState == TaskCore::Running && taskc->mTargetState == TaskCore::Running ) {
00368 TRY (
00369 taskc->prepareUpdateHook();
00370 taskc->updateHook();
00371 ) CATCH(std::exception const& e,
00372 log(Error) << "in updateHook(): switching to exception state because of unhandled exception" << endlog();
00373 log(Error) << " " << e.what() << endlog();
00374 taskc->exception();
00375 ) CATCH_ALL (
00376 log(Error) << "in updateHook(): switching to exception state because of unhandled exception" << endlog();
00377 taskc->exception();
00378 )
00379 }
00380
00381 if ( taskc->mTaskState == TaskCore::RunTimeError ) {
00382 TRY (
00383 taskc->errorHook();
00384 ) CATCH(std::exception const& e,
00385 log(Error) << "in errorHook(): switching to exception state because of unhandled exception" << endlog();
00386 log(Error) << " " << e.what() << endlog();
00387 taskc->exception();
00388 ) CATCH_ALL (
00389 log(Error) << "in errorHook(): switching to exception state because of unhandled exception" << endlog();
00390 taskc->exception();
00391 )
00392 }
00393 }
00394 if ( !this->getActivity() || ! this->getActivity()->isRunning() ) return;
00395
00396
00397 for (std::vector<TaskCore*>::iterator it = children.begin(); it != children.end();++it) {
00398 if ( (*it)->mTaskState == TaskCore::Running && (*it)->mTargetState == TaskCore::Running ){
00399 TRY (
00400 (*it)->prepareUpdateHook();
00401 (*it)->updateHook();
00402 ) CATCH(std::exception const& e,
00403 log(Error) << "in updateHook(): switching to exception state because of unhandled exception" << endlog();
00404 log(Error) << " " << e.what() << endlog();
00405 (*it)->exception();
00406 ) CATCH_ALL (
00407 log(Error) << "in updateHook(): switching to exception state because of unhandled exception" << endlog();
00408 (*it)->exception();
00409 )
00410 }
00411 if ( (*it)->mTaskState == TaskCore::RunTimeError ){
00412 TRY (
00413 (*it)->errorHook();
00414 ) CATCH(std::exception const& e,
00415 log(Error) << "in errorHook(): switching to exception state because of unhandled exception" << endlog();
00416 log(Error) << " " << e.what() << endlog();
00417 (*it)->exception();
00418 ) CATCH_ALL (
00419 log(Error) << "in errorHook(): switching to exception state because of unhandled exception" << endlog();
00420 (*it)->exception();
00421 )
00422 }
00423 if ( !this->getActivity() || ! this->getActivity()->isRunning() ) return;
00424 }
00425 }
00426
00427 bool ExecutionEngine::breakLoop() {
00428 bool ok = true;
00429 if (taskc)
00430 ok = taskc->breakUpdateHook();
00431 for (std::vector<TaskCore*>::iterator it = children.begin(); it != children.end();++it) {
00432 ok = (*it)->breakUpdateHook() && ok;
00433 }
00434 return ok;
00435 }
00436
00437 bool ExecutionEngine::stopTask(TaskCore* task) {
00438
00439
00440
00441 if ( getActivity() && this->getActivity()->stop() ) {
00442 this->getActivity()->start();
00443 return true;
00444 }
00445 return false;
00446 }
00447
00448 void ExecutionEngine::setExceptionTask() {
00449 std::string name;
00450 TaskContext* tc = dynamic_cast<TaskContext*>(taskc);
00451 if (tc)
00452 name = tc->getName();
00453 else if (taskc)
00454 name = "TaskCore";
00455 else
00456 name = "GlobalEngine";
00457 log(Error) << "in "<<name<<": unhandled exception in sent operation." << endlog();
00458 if(taskc)
00459 taskc->exception();
00460 }
00461
00462
00463 void ExecutionEngine::finalize() {
00464
00465 }
00466
00467 }
00468