#include <Executor.hxx>

Collaboration diagram for YACS::ENGINE::Executor:

Public Member Functions
	Executor ()

	~Executor ()

void	RunA (Scheduler *graph, int debug=0, bool fromScratch=true)
	Execute a graph waiting for completion. More...

void	RunW (Scheduler *graph, int debug=0, bool fromScratch=true)

void	RunB (Scheduler *graph, int debug=0, bool fromScratch=true)
	Execute a graph with breakpoints or step by step. More...

void	setKeepGoingProperty (bool newVal)

bool	getKeepGoingProperty () const

void	setDPLScopeSensitive (bool newVal)

bool	getDPLScopeSensitive () const

YACS::ExecutionMode	getCurrentExecMode ()

YACS::ExecutorState	getExecutorState ()

void	setExecMode (YACS::ExecutionMode mode)
	Dynamically set the current mode of execution. More...

void	setListOfBreakPoints (std::list< std::string > listOfBreakPoints)
	define a list of nodes names as breakpoints in the graph More...

std::list< std::string >	getTasksToLoad ()
	Get the list of tasks to load, to define a subset to execute in step by step mode. More...

bool	setStepsToExecute (std::list< std::string > listToExecute)
	Define a subset of task to execute in step by step mode. More...

bool	resumeCurrentBreakPoint ()
	wake up executor when in pause More...

bool	isNotFinished ()

void	stopExecution ()
	stops the execution as soon as possible More...

bool	saveState (const std::string &xmlFile)
	save the current state of execution in an xml file More...

bool	loadState ()
	not yet implemented More...

int	getNbOfThreads ()
	number of running tasks More...

int	getNumberOfRunningTasks () const

void	displayDot (Scheduler *graph)
	Display the graph state as a dot display, public method. More...

void	setStopOnError (bool dumpRequested=false, std::string xmlFile="")
	ask to stop execution on the first node found in error More...

void	unsetStopOnError ()
	ask to do not stop execution on nodes found in error More...

void	waitPause ()
	suspend pilot execution until Executor is in pause or waiting tasks completion mode. More...

YACS::BASES::Mutex &	getTheMutexForSchedulerUpdate ()

Static Public Attributes
static int	_maxThreads

static size_t	_threadStackSize

Protected Member Functions
bool	checkBreakPoints ()
	Wait reactivation in modes Step By step or with BreakPoints. More...

void	waitResume ()
	in modes Step By step or with BreakPoint, wait until pilot resumes the execution More...

void	loadTask (Task task, const Executor execInst)
	Perform loading of a Task. More...

void	loadTasks (const std::vector< Task * > &tasks, const Executor *execInst)

void	loadParallelTasks (const std::vector< Task * > &tasks, const Executor *execInst)

void	launchTasks (const std::vector< Task * > &tasks)
	Execute a list of tasks possibly connected through datastream links. More...

void	launchTask (Task *task)
	Execute a Task in a thread. More...

void	wakeUp ()
	must be used protected by _mutexForSchedulerUpdate! More...

void	sleepWhileNoEventsFromAnyRunningTask ()
	wait until a running task ends More...

void	notifyEndOfThread (YACS::BASES::Thread *thread)
	not implemented More...

void	traceExec (Task *task, const std::string &message, const std::string &placement)

void	_displayDot (Scheduler *graph)
	Display the graph state as a dot display. More...

virtual void	sendEvent (const std::string &event)
	emit notification to all observers registered with the dispatcher More...

Static Protected Member Functions
static void	FilterTasksConsideringContainers (std::vector< Task * > &tsks)

static std::string	ComputePlacement (Task *zeTask)

static void *	functionForTaskLoad (void *)

static void *	functionForTaskExecution (void *)
	Function to perform execution of a task in a thread. More...

Protected Attributes
Scheduler *	_mainSched

ComposedNode *	_root

int	_nbOfConcurrentThreads

YACS::BASES::Mutex	_mutexForNbOfConcurrentThreads

YACS::BASES::Condition	_condForNewTasksToPerform

YACS::BASES::Semaphore	_semForMaxThreads

YACS::BASES::Condition	_condForStepByStep

YACS::BASES::Condition	_condForPilot

YACS::BASES::Mutex	_mutexForSchedulerUpdate

YACS::BASES::Mutex	_mutexForTrace

bool	_toContinue

bool	_isOKToEnd

bool	_stopOnErrorRequested

bool	_dumpOnErrorRequested

bool	_errorDetected

bool	_isRunningunderExternalControl

bool	_isWaitingEventsFromRunningTasks

int	_numberOfRunningTasks

std::set< Task * >	_runningTasks

int	_numberOfEndedTasks

int	_semThreadCnt

YACS::ExecutorState	_executorState

YACS::ExecutionMode	_execMode

std::list< std::string >	_listOfBreakPoints

std::list< std::string >	_listOfTasksToLoad

std::vector< Task * >	_tasks

std::vector< Task * >	_tasksSave

std::list< YACS::BASES::Thread * >	_groupOfAllThreadsCreated

std::ofstream	_trace

std::string	_dumpErrorFile

bool	_keepGoingOnFail

bool	_DPLScopeSensitive
	specifies if scope DynParaLoop is active or not. False by default. More...

timeval	_start

Detailed Description

Threaded Executor.

Definition at line 55 of file Executor.hxx.

Constructor & Destructor Documentation

Executor::Executor ( )

Definition at line 78 of file Executor.cxx.

References _dumpOnErrorRequested, _errorDetected, _execMode, _executorState, _isOKToEnd, _isRunningunderExternalControl, _maxThreads, _root, _semThreadCnt, _stopOnErrorRequested, _toContinue, YACS::CONTINUE, DEBTRACE, and YACS::NOTYETINITIALIZED.

                   :_nbOfConcurrentThreads(0), _semForMaxThreads(_maxThreads),_keepGoingOnFail(false),_DPLScopeSensitive(false)
 {
   _root=0;
   _toContinue = true;
   _isOKToEnd = false;
   _stopOnErrorRequested = false;
   _dumpOnErrorRequested = false;
   _errorDetected = false;
   _isRunningunderExternalControl=false;
   _executorState = YACS::NOTYETINITIALIZED;
   _execMode = YACS::CONTINUE;
   _semThreadCnt = _maxThreads;
   DEBTRACE("Executor initialized with max threads = " << _maxThreads);
 }

Executor::~Executor ( )

Definition at line 93 of file Executor.cxx.

References _groupOfAllThreadsCreated.

 {
   for(list<Thread *>::iterator iter=_groupOfAllThreadsCreated.begin();iter!=_groupOfAllThreadsCreated.end();iter++)
     delete *iter;
 }

Member Function Documentation

void Executor::_displayDot ( Scheduler * graph )

protected

Display the graph state as a dot display.

Parameters

graph : the node to display

Definition at line 700 of file Executor.cxx.

References isfile().

Referenced by displayDot(), RunA(), and RunB().

 {
    std::ofstream g("titi");
    ((ComposedNode*)graph)->writeDot(g);
    g.close();
    const char displayScript[]="display.sh";
    if(isfile(displayScript))
      system("sh display.sh");
    else
      system("dot -Tpng titi|display -delay 5");
 }

bool Executor::checkBreakPoints ( )

protected

Wait reactivation in modes Step By step or with BreakPoints.

Check mode of execution (set by main thread):

YACS::CONTINUE : the graph execution continues.
YACS::STEPBYSTEP : wait on condition (pilot thread, Executor::resumeCurrentBreakPoint)
YACS::STOPBEFORENODES : if there are ready nodes in a list of breakpoints, wait on condition (pilot thread, Executor::resumeCurrentBreakPoint) else continue the graph execution.
Returns
true if end of executor thread is requested

Definition at line 723 of file Executor.cxx.

References _condForPilot, _execMode, _executorState, _isOKToEnd, _listOfBreakPoints, _listOfTasksToLoad, _mainSched, _mutexForSchedulerUpdate, _tasks, _tasksSave, YACS::CONTINUE, DEBTRACE, getNbOfThreads(), YACS::ENGINE::Scheduler::getTaskName(), YACS::PAUSED, sendEvent(), YACS::STEPBYSTEP, YACS::STOPBEFORENODES, YACS::WAITINGTASKS, and waitResume().

Referenced by RunB().

 {
   DEBTRACE("Executor::checkBreakPoints()");
   vector<Task *>::iterator iter;
   bool endRequested = false;
 
   switch (_execMode)
     {
     case YACS::CONTINUE:
       {
         break;
       }
     case YACS::STOPBEFORENODES:
       {
         bool stop = false;
         { // --- Critical section
           YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
           _tasksSave = _tasks;
           for (iter=_tasks.begin(); iter!=_tasks.end(); iter++)
             {
               string nodeToLoad = _mainSched->getTaskName(*iter);
               if (find(_listOfBreakPoints.begin(), _listOfBreakPoints.end(), nodeToLoad)
                   != _listOfBreakPoints.end())
                 {
                   stop = true;
                   break;
                 }
             }
           if (stop)
             {
               _listOfTasksToLoad.clear();
               for (iter=_tasks.begin(); iter!=_tasks.end(); iter++)
                 {
                   string nodeToLoad = _mainSched->getTaskName(*iter);
                   _listOfTasksToLoad.push_back(nodeToLoad);
                 }
               if (getNbOfThreads())
                 _executorState = YACS::WAITINGTASKS; // will be paused after completion of running tasks
               else
                 _executorState = YACS::PAUSED;
               sendEvent("executor");
               _condForPilot.notify_all();
             }
           if (stop && !_isOKToEnd) waitResume(); // wait until pilot calls resumeCurrentBreakPoint(), mutex released during wait 
           if (_isOKToEnd) endRequested = true;
         } // --- End of critical section
           if (stop) DEBTRACE("wake up from waitResume");
         break;
       }
     default:
     case YACS::STEPBYSTEP:
       {
         { // --- Critical section
           YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
           _tasksSave = _tasks;
           _listOfTasksToLoad.clear();
           for (iter=_tasks.begin(); iter!=_tasks.end(); iter++)
             {
               string nodeToLoad = _mainSched->getTaskName(*iter);
               _listOfTasksToLoad.push_back(nodeToLoad);
             }
           if (getNbOfThreads())
             _executorState = YACS::WAITINGTASKS; // will be paused after completion of running tasks
           else
             _executorState = YACS::PAUSED;
           sendEvent("executor");
           _condForPilot.notify_all();
           if (!_isOKToEnd)
             waitResume(); // wait until pilot calls resumeCurrentBreakPoint(), mutex released during wait
                           // or, if no pilot, wait until no more running tasks (stop on error)
           if (_isOKToEnd) endRequested = true;
         } // --- End of critical section
         DEBTRACE("wake up from waitResume");
         break;
       }
     }
   DEBTRACE("endRequested: " << endRequested);
   return endRequested;
 }

std::string Executor::ComputePlacement ( Task * zeTask )

staticprotected

Definition at line 1367 of file Executor.cxx.

References YACS::ENGINE::Task::getContainer(), and YACS::ENGINE::Container::getFullPlacementId().

Referenced by functionForTaskExecution(), launchTask(), launchTasks(), and loadTask().

 {
   std::string placement("---");
   if(!zeTask)
     return placement;
   if(zeTask->getContainer())
     placement=zeTask->getContainer()->getFullPlacementId(zeTask);
   return placement;
 }

void Executor::displayDot ( Scheduler * graph )

Display the graph state as a dot display, public method.

Definition at line 689 of file Executor.cxx.

References _displayDot(), and _isRunningunderExternalControl.

Referenced by main().

 {
   _isRunningunderExternalControl=true;
   _displayDot(graph);
 }

void Executor::FilterTasksConsideringContainers ( std::vector< Task * > & tsks )

staticprotected

This method takes in input a list of tasks and selects from that lists a part of it considering only the containers. If tasks have no container instance subclass of HomogeneousPoolContainer this method will let the tsks untouched.

Parameters

[in,out] tsks - list of tasks to be

Definition at line 1315 of file Executor.cxx.

References YACS::ENGINE::HomogeneousPoolContainer::allocateFor(), YACS::ENGINE::Task::getContainer(), YACS::ENGINE::HomogeneousPoolContainer::getNumberOfFreePlace(), CORBAEngineTest::i, gui.GraphViewer::m, and PMMLBasicsTestLauncher::ret.

Referenced by RunB().

 {
   std::map<HomogeneousPoolContainer *, std::vector<Task *> > m;
   for(std::vector<Task *>::const_iterator it=tsks.begin();it!=tsks.end();it++)
     {
       Task *cur(*it);
       if(!cur)
         continue;
       Container *cont(cur->getContainer());
       if(!cont)
         {
           m[(HomogeneousPoolContainer *)NULL].push_back(cur);
           continue;
         }
       HomogeneousPoolContainer *contC(dynamic_cast<HomogeneousPoolContainer *>(cont));
       if(!contC)
         {
           m[(HomogeneousPoolContainer *)NULL].push_back(cur);
           continue;
         }
       m[contC].push_back(cur);
     }
   //
   std::vector<Task *> ret;
   for(std::map<HomogeneousPoolContainer *, std::vector<Task *> >::const_iterator it=m.begin();it!=m.end();it++)
     {
       HomogeneousPoolContainer *curhpc((*it).first);
       const std::vector<Task *>& curtsks((*it).second);
       if(!curhpc)
         {
           ret.insert(ret.end(),curtsks.begin(),curtsks.end());
         }
       else
         {
           // start of critical section for container curhpc
           YACS::BASES::AutoLocker<Container> alckForCont(curhpc);
           std::vector<const Task *> vecOfTaskSharingSameHPContToBeRunSimutaneously;
           std::size_t sz(curhpc->getNumberOfFreePlace());
           std::vector<Task *>::const_iterator it2(curtsks.begin());
           for(std::size_t i=0;i<sz && it2!=curtsks.end();i++,it2++)
             {
               vecOfTaskSharingSameHPContToBeRunSimutaneously.push_back(*it2);
               ret.push_back(*it2);
             }
           curhpc->allocateFor(vecOfTaskSharingSameHPContToBeRunSimutaneously);
           //end of critical section
         }
     }
   //
   tsks=ret;
 }

void * Executor::functionForTaskExecution ( void * arg )

staticprotected

Function to perform execution of a task in a thread.

Parameters

arg	: 3 elements (a Task, a Scheduler, an Executor)

Calls Task::execute

Calls Task::finished when the task is finished

Calls (notify with event YACS::FINISH) Scheduler::notifyFrom when the task is finished

Calls Executor::wakeUp and Executor::notifyEndOfThread

Definition at line 1137 of file Executor.cxx.

References _condForPilot, _condForStepByStep, _errorDetected, _execMode, _executorState, _isOKToEnd, _isRunningunderExternalControl, _mutexForSchedulerUpdate, _numberOfRunningTasks, _runningTasks, _semForMaxThreads, _semThreadCnt, _stopOnErrorRequested, YACS::ABORT, YACS::ENGINE::Task::aborted(), YACS::ENGINE::Node::applyDPLScope(), ComputePlacement(), YACS::CONTINUE, DEBTRACE, YACS::ENGINE::Task::disconnectService(), threadargs::execInst, YACS::ENGINE::Task::execute(), YACS::FINISH, YACS::ENGINE::Task::finished(), YACS::ENGINE::Task::getContainer(), getDPLScopeSensitive(), YACS::ENGINE::Task::getState(), YACS::ENGINE::Node::getStateName(), YACS::ENGINE::Scheduler::notifyFrom(), YACS::PAUSED, YACS::ENGINE::HomogeneousPoolContainer::release(), threadargs::sched, sendEvent(), YACS::STEPBYSTEP, threadargs::task, traceExec(), YACS::WAITINGTASKS, wakeUp(), and YACS::Exception::what().

Referenced by launchTask().

 {
   DEBTRACE("Executor::functionForTaskExecution(void *arg)");
 
   struct threadargs *args = (struct threadargs *) arg;
   Task *task=args->task;
   Scheduler *sched=args->sched;
   Executor *execInst=args->execInst;
   delete args;
   execInst->traceExec(task, "state:"+Node::getStateName(task->getState()),ComputePlacement(task));
 
   Thread::detach();
 
   // Execute task
 
   if(execInst->getDPLScopeSensitive())
     {
       Node *node(dynamic_cast<Node *>(task));
       ComposedNode *gfn(dynamic_cast<ComposedNode *>(sched));
       if(node!=0 && gfn!=0)
         node->applyDPLScope(gfn);
     }
 
   YACS::Event ev=YACS::FINISH;
   try
     {
       execInst->traceExec(task, "start execution",ComputePlacement(task));
       task->execute();
       execInst->traceExec(task, "end execution OK",ComputePlacement(task));
     }
   catch(Exception& ex)
     {
       std::cerr << "YACS Exception during execute" << std::endl;
       std::cerr << ex.what() << std::endl;
       ev=YACS::ABORT;
       string message = "end execution ABORT, ";
       message += ex.what();
       execInst->traceExec(task, message,ComputePlacement(task));
     }
   catch(...) 
     {
       // Execution has failed
       std::cerr << "Execution has failed: unknown reason" << std::endl;
       ev=YACS::ABORT;
       execInst->traceExec(task, "end execution ABORT, unknown reason",ComputePlacement(task));
     }
 
   // Disconnect task
   try
     {
       DEBTRACE("task->disconnectService()");
       task->disconnectService();
       execInst->traceExec(task, "disconnectService",ComputePlacement(task));
     }
   catch(...) 
     {
       // Disconnect has failed
       std::cerr << "disconnect has failed" << std::endl;
       ev=YACS::ABORT;
       execInst->traceExec(task, "disconnectService failed, ABORT",ComputePlacement(task));
     }
   //
 
   std::string placement(ComputePlacement(task));
 
   // container management for HomogeneousPoolOfContainer
 
   HomogeneousPoolContainer *contC(dynamic_cast<HomogeneousPoolContainer *>(task->getContainer()));
   if(contC)
     {
       YACS::BASES::AutoLocker<Container> alckCont(contC);
       contC->release(task);
     }
 
   DEBTRACE("End task->execute()");
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&execInst->_mutexForSchedulerUpdate);
     try
       {
         if (ev == YACS::FINISH) task->finished();
         if (ev == YACS::ABORT)
           {
             execInst->_errorDetected = true;
             if (execInst->_stopOnErrorRequested)
               {
                 execInst->_execMode = YACS::STEPBYSTEP;
                 execInst->_isOKToEnd = true;
               }
             task->aborted();
           }
         execInst->traceExec(task, "state:"+Node::getStateName(task->getState()),placement);
         sched->notifyFrom(task,ev,execInst);
       }
     catch(Exception& ex)
       {
         //notify has failed : it is supposed to have set state
         //so no need to do anything
         std::cerr << "Error during notification" << std::endl;
         std::cerr << ex.what() << std::endl;
       }
     catch(...)
       {
         //notify has failed : it is supposed to have set state
         //so no need to do anything
         std::cerr << "Notification failed" << std::endl;
       }
     execInst->_numberOfRunningTasks--;
     execInst->_runningTasks.erase(task);
     DEBTRACE("_numberOfRunningTasks: " << execInst->_numberOfRunningTasks 
              << " _execMode: " << execInst->_execMode
              << " _executorState: " << execInst->_executorState);
     if ((execInst->_numberOfRunningTasks == 0) && (execInst->_execMode != YACS::CONTINUE)) // no more running tasks
       {
         if (execInst->_executorState == YACS::WAITINGTASKS)
           {
             execInst->_executorState = YACS::PAUSED;
             execInst->sendEvent("executor");
             execInst->_condForPilot.notify_all();
             if (execInst->_errorDetected &&
                 execInst->_stopOnErrorRequested &&
                 !execInst->_isRunningunderExternalControl)
               execInst->_condForStepByStep.notify_all(); // exec thread may be on waitResume
           }
       }
     DEBTRACE("before _semForMaxThreads.post " << execInst->_semThreadCnt);
     execInst->_semForMaxThreads.post();
     execInst->_semThreadCnt += 1;
     DEBTRACE("after _semForMaxThreads.post " << execInst->_semThreadCnt);
     if (execInst->_executorState != YACS::PAUSED)  execInst->wakeUp();
 
   } // --- End of critical section (change state)
 
   //execInst->notifyEndOfThread(0);
   Thread::exit(0);
   return 0;
 }

void * Executor::functionForTaskLoad ( void * arg )

staticprotected

This thread is NOT supposed to be detached !

Definition at line 1112 of file Executor.cxx.

References DEBTRACE, threadargs::execInst, loadTask(), threadargs::sched, and threadargs::task.

Referenced by loadParallelTasks().

 {
   DEBTRACE("Executor::functionForTaskLoad(void *arg)");
   struct threadargs *args = (struct threadargs *) arg;
   Task *task=args->task;
   Scheduler *sched=args->sched;
   Executor *execInst=args->execInst;
   delete args;
   execInst->loadTask(task,execInst);// no throw of this method - all throw are catched !
   return 0;
 }

YACS::ExecutionMode Executor::getCurrentExecMode ( )

Definition at line 374 of file Executor.cxx.

References _execMode, and _isRunningunderExternalControl.

 {
   _isRunningunderExternalControl=true;
   return _execMode;
 }

bool YACS::ENGINE::Executor::getDPLScopeSensitive ( ) const

inline

Definition at line 105 of file Executor.hxx.

Referenced by functionForTaskExecution().

105 { return _DPLScopeSensitive; }

YACS::ExecutorState Executor::getExecutorState ( )

Definition at line 381 of file Executor.cxx.

References _executorState, and _isRunningunderExternalControl.

Referenced by main().

 {
   _isRunningunderExternalControl=true;
   return _executorState;
 }

bool YACS::ENGINE::Executor::getKeepGoingProperty ( ) const

inline

Definition at line 103 of file Executor.hxx.

Referenced by YACS::ENGINE::ForEachLoop::updateStateOnFailedEventFrom().

103 { return _keepGoingOnFail; }

int Executor::getNbOfThreads ( )

number of running tasks

Definition at line 1100 of file Executor.cxx.

References _groupOfAllThreadsCreated, _isRunningunderExternalControl, _mutexForNbOfConcurrentThreads, and PMMLBasicsTestLauncher::ret.

Referenced by checkBreakPoints().

 {
   int ret;
   YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForNbOfConcurrentThreads);
   _isRunningunderExternalControl=true;
   ret = _groupOfAllThreadsCreated.size();
   return ret;
 }

int YACS::ENGINE::Executor::getNumberOfRunningTasks ( ) const

inline

Definition at line 118 of file Executor.hxx.

118 { return _numberOfRunningTasks; }

std::list< std::string > Executor::getTasksToLoad ( )

Get the list of tasks to load, to define a subset to execute in step by step mode.

If the executor is not in mode YACS::WAITINGTASKS nor YACS::PAUSED, the list is empty. Use Executor::waitPause to wait.

Definition at line 508 of file Executor.cxx.

References _executorState, _isRunningunderExternalControl, _listOfTasksToLoad, _mutexForSchedulerUpdate, DEBTRACE, YACS::FINISHED, YACS::INITIALISED, YACS::NOTYETINITIALIZED, YACS::PAUSED, YACS::RUNNING, YACS::STOPPED, and YACS::WAITINGTASKS.

 {
   DEBTRACE("Executor::getTasksToLoad()");
   list<string> listOfNodesToLoad;
   listOfNodesToLoad.clear();
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _isRunningunderExternalControl=true;
     switch (_executorState)
       {
       case YACS::WAITINGTASKS:
       case YACS::PAUSED:
         {
           listOfNodesToLoad = _listOfTasksToLoad;
           break;
         }
       case YACS::NOTYETINITIALIZED:
       case YACS::INITIALISED:
       case YACS::RUNNING:
       case YACS::FINISHED:
       case YACS::STOPPED:
       default:
         {
           break;
         }
       }
   } // --- End of critical section
   return listOfNodesToLoad;
 }

YACS::BASES::Mutex& YACS::ENGINE::Executor::getTheMutexForSchedulerUpdate ( )

inline

Definition at line 125 of file Executor.hxx.

Referenced by YACS::ENGINE::ForEachLoop::getPassedResults(), and YACS::ENGINE::schemaSaveState().

125 { return _mutexForSchedulerUpdate; }

bool Executor::isNotFinished ( )

Definition at line 388 of file Executor.cxx.

References _isRunningunderExternalControl, and _toContinue.

Referenced by main().

 {
   _isRunningunderExternalControl=true;
   return _toContinue;
 }

void Executor::launchTask ( Task * task )

protected

Execute a Task in a thread.

Parameters

task	: Task to execute

Calls Scheduler::notifyFrom of main node (_mainSched) to notify start

Calls Executor::functionForTaskExecution in Thread

Definition at line 999 of file Executor.cxx.

References _mainSched, _maxThreads, _mutexForSchedulerUpdate, _numberOfRunningTasks, _runningTasks, _semForMaxThreads, _semThreadCnt, _threadStackSize, YACS::ENGINE::Task::begin(), ComputePlacement(), DEBTRACE, threadargs::execInst, functionForTaskExecution(), YACS::ENGINE::Task::getCoupledTasks(), YACS::ENGINE::Task::getState(), threadargs::sched, threadargs::task, YACS::TOACTIVATE, and traceExec().

Referenced by launchTasks().

 {
   DEBTRACE("Executor::launchTask(Task *task)");
   struct threadargs *args;
   if(task->getState() != YACS::TOACTIVATE)return;
 
   DEBTRACE("before _semForMaxThreads.wait " << _semThreadCnt);
   if(_semThreadCnt == 0)
     {
       // --- Critical section
       YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
       //check if we have enough threads to run
       std::set<Task*> tmpSet=_runningTasks;
       std::set<Task*>::iterator it = tmpSet.begin();
       std::string status="running";
       std::set<Task*> coupledSet;
       while( it != tmpSet.end() )
         {
           Task* tt=*it;
           coupledSet.clear();
           tt->getCoupledTasks(coupledSet);
           status="running";
           for (std::set<Task*>::iterator iter = coupledSet.begin(); iter != coupledSet.end(); ++iter)
             {
               if((*iter)->getState() == YACS::TOACTIVATE)status="toactivate";
               tmpSet.erase(*iter);
             }
           if(status=="running")break;
           it = tmpSet.begin();
         }
 
       if(status=="toactivate")
         {
           std::cerr << "WARNING: maybe you need more threads to run your schema (current value="<< _maxThreads << ")" << std::endl;
           std::cerr << "If it is the case, set the YACS_MAX_THREADS environment variable to a bigger value (export YACS_MAX_THREADS=xxx)" << std::endl;
         }
       // --- End of critical section
     }
 
   _semForMaxThreads.wait();
   _semThreadCnt -= 1;
 
   args= new threadargs;
   args->task = task;
   args->sched = _mainSched;
   args->execInst = this;
 
   traceExec(task, "launch",ComputePlacement(task));
 
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _numberOfRunningTasks++;
     _runningTasks.insert(task);
     task->begin(); //change state to ACTIVATED
   } // --- End of critical section
   Thread(functionForTaskExecution, args, _threadStackSize);
 }

void Executor::launchTasks ( const std::vector< Task * > & tasks )

protected

Execute a list of tasks possibly connected through datastream links.

Parameters

tasks : a list of tasks to execute

Definition at line 898 of file Executor.cxx.

References _mainSched, _mutexForSchedulerUpdate, YACS::ABORT, YACS::ENGINE::Task::aborted(), ComputePlacement(), YACS::ENGINE::Task::disconnectService(), YACS::ERROR, YACS::ENGINE::Task::getState(), YACS::ENGINE::Node::getStateName(), launchTask(), YACS::ENGINE::Scheduler::notifyFrom(), CORBAEngineTest::state, gui.Appli::t, YACS::TOLOAD, YACS::TORECONNECT, traceExec(), and YACS::Exception::what().

Referenced by RunA(), and RunB().

 {
   //First phase, make datastream connections
   for(vector<Task *>::const_iterator iter=tasks.begin();iter!=tasks.end();iter++)
     {
       YACS::StatesForNode state=(*iter)->getState();
       if(state != YACS::TOLOAD && state != YACS::TORECONNECT)continue;
       try
         {
           (*iter)->connectService();
           traceExec(*iter, "connectService",ComputePlacement(*iter));
           {//Critical section
             YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
             (*iter)->connected();
           }//End of critical section
         }
       catch(Exception& ex) 
         {
           std::cerr << ex.what() << std::endl;
           try
             {
               (*iter)->disconnectService();
               traceExec(*iter, "disconnectService",ComputePlacement(*iter));
             }
           catch(...) 
             {
               // Disconnect has failed
               traceExec(*iter, "disconnectService failed, ABORT",ComputePlacement(*iter));
             }
           {//Critical section
             YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
             (*iter)->aborted();
             _mainSched->notifyFrom(*iter,YACS::ABORT,this);
           }//End of critical section
         }
       catch(...) 
         {
           std::cerr << "Problem in connectService" << std::endl;
           try
             {
               (*iter)->disconnectService();
               traceExec(*iter, "disconnectService",ComputePlacement(*iter));
             }
           catch(...) 
             {
               // Disconnect has failed
               traceExec(*iter, "disconnectService failed, ABORT",ComputePlacement(*iter));
             }
           {//Critical section
             YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
             (*iter)->aborted();
             _mainSched->notifyFrom(*iter,YACS::ABORT,this);
           }//End of critical section
         }
       if((*iter)->getState() == YACS::ERROR)
         {
           //try to put all coupled tasks in error
           std::set<Task*> coupledSet;
           (*iter)->getCoupledTasks(coupledSet);
           for (std::set<Task*>::iterator it = coupledSet.begin(); it != coupledSet.end(); ++it)
             {
               Task* t=*it;
               if(t == *iter)continue;
               if(t->getState() == YACS::ERROR)continue;
               try
                 {
                   t->disconnectService();
                   traceExec(t, "disconnectService",ComputePlacement(*iter));
                 }
               catch(...)
                 {
                   // Disconnect has failed
                   traceExec(t, "disconnectService failed, ABORT",ComputePlacement(*iter));
                 }
               {//Critical section
                 YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
                 t->aborted();
                 _mainSched->notifyFrom(t,YACS::ABORT,this);
               }//End of critical section
               traceExec(t, "state:"+Node::getStateName(t->getState()),ComputePlacement(*iter));
             }
         }
       traceExec(*iter, "state:"+Node::getStateName((*iter)->getState()),ComputePlacement(*iter));
     }
 
   //Second phase, execute each task in a thread
   for(vector<Task *>::const_iterator iter=tasks.begin();iter!=tasks.end();iter++)
     {
       launchTask(*iter);
     }
 }

void Executor::loadParallelTasks	(	const std::vector< Task * > &	tasks,
		const Executor *	execInst
	)

protected

Definition at line 876 of file Executor.cxx.

References _mainSched, _tasks, _threadStackSize, DEBTRACE, threadargs::execInst, functionForTaskLoad(), threadargs::sched, and threadargs::task.

Referenced by RunB().

 {
   std::vector<Thread> ths(tasks.size());
   std::size_t ithread(0);
   for(std::vector<Task *>::const_iterator iter = _tasks.begin(); iter != _tasks.end(); iter++, ithread++)
     {
       DEBTRACE("Executor::loadParallelTasks(Task *task)");
       struct threadargs *args(new threadargs);
       args->task = (*iter);
       args->sched = _mainSched;
       args->execInst = this;
       ths[ithread].go(functionForTaskLoad, args, _threadStackSize);
     }
   for(ithread=0;ithread<tasks.size();ithread++)
     ths[ithread].join();
 }

bool Executor::loadState ( )

not yet implemented

Definition at line 669 of file Executor.cxx.

References _isRunningunderExternalControl, and DEBTRACE.

 {
   DEBTRACE("Executor::loadState()");
   _isRunningunderExternalControl=true;
   return true;
 }

void Executor::loadTask	(	Task *	task,
		const Executor *	execInst
	)

protected

Perform loading of a Task.

Parameters

task	: Task to load

Definition at line 824 of file Executor.cxx.

References _mainSched, _mutexForSchedulerUpdate, YACS::ABORT, YACS::ENGINE::Task::aborted(), ComputePlacement(), DEBTRACE, YACS::ENGINE::Task::getState(), YACS::ENGINE::Node::getStateName(), YACS::ENGINE::Task::initService(), YACS::ENGINE::Task::load(), YACS::ENGINE::Scheduler::notifyFrom(), YACS::START, YACS::TOLOAD, traceExec(), and YACS::Exception::what().

Referenced by functionForTaskLoad(), loadTasks(), and RunA().

 {
   DEBTRACE("Executor::loadTask(Task *task)");
   if(task->getState() != YACS::TOLOAD)
     return;
   traceExec(task, "state:TOLOAD", ComputePlacement(task));
   {//Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _mainSched->notifyFrom(task,YACS::START,execInst);
   }//End of critical section
   try
     {
       traceExec(task, "load", ComputePlacement(task));
       task->load();
       traceExec(task, "initService", ComputePlacement(task));
       task->initService();
     }
   catch(Exception& ex) 
     {
       std::cerr << ex.what() << std::endl;
       {//Critical section
         YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
         task->aborted();
         _mainSched->notifyFrom(task,YACS::ABORT,execInst);
         traceExec(task, "state:"+Node::getStateName(task->getState()), ComputePlacement(task));
       }//End of critical section
     }
   catch(...) 
     {
       std::cerr << "Load failed" << std::endl;
       {//Critical section
         YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
         task->aborted();
         _mainSched->notifyFrom(task,YACS::ABORT,execInst);
         traceExec(task, "state:"+Node::getStateName(task->getState()), ComputePlacement(task));
       }//End of critical section
     }
 }

void Executor::loadTasks	(	const std::vector< Task * > &	tasks,
		const Executor *	execInst
	)

protected

Definition at line 870 of file Executor.cxx.

References _tasks, and loadTask().

 {
   for(std::vector<Task *>::const_iterator iter = _tasks.begin(); iter != _tasks.end(); iter++)
     loadTask(*iter,execInst);
 }

void Executor::notifyEndOfThread ( YACS::BASES::Thread * thread )

protected

not implemented

Definition at line 1075 of file Executor.cxx.

 {
   /*_mutexForNbOfConcurrentThreads.lock();
   _groupOfAllThreadsCreated.remove(thread);
   delete thread;
   _mutexForNbOfConcurrentThreads.unlock();*/
 }

bool Executor::resumeCurrentBreakPoint ( )

wake up executor when in pause

When Executor is in state paused or waiting for task completion, the thread running loop RunB waits on condition _condForStepByStep. Thread RunB is waken up.

Returns: true when actually wakes up executor

Definition at line 449 of file Executor.cxx.

References _condForStepByStep, _executorState, _isRunningunderExternalControl, _mutexForSchedulerUpdate, DEBTRACE, YACS::FINISHED, YACS::PAUSED, PMMLBasicsTestLauncher::ret, YACS::RUNNING, sendEvent(), YACS::STOPPED, and YACS::WAITINGTASKS.

Referenced by main(), and stopExecution().

 {
   DEBTRACE("Executor::resumeCurrentBreakPoint()");
   bool ret = false;
   //bool doDump = false;
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _isRunningunderExternalControl=true;
     DEBTRACE("_executorState: " << _executorState);
     switch (_executorState)
       {
       case YACS::WAITINGTASKS:
       case YACS::PAUSED:
         {
           _condForStepByStep.notify_all();
           _executorState = YACS::RUNNING;
           sendEvent("executor");
           ret = true;
           //if (_dumpOnErrorRequested && _errorDetected) doDump =true;
           break;
         }
       case YACS::FINISHED:
       case YACS::STOPPED:
         {
           //if (_dumpOnErrorRequested && _errorDetected) doDump =true;
           DEBTRACE("Graph Execution finished or stopped !");
           break;
         }
       default :
         {
           // debug: no easy way to verify if main loop is acutally waiting on condition
         }
       }
     DEBTRACE("---");
     //if (doDump) saveState(_dumpErrorFile);
   } // --- End of critical section
   return ret;
 }

void Executor::RunA	(	Scheduler *	graph,
		int	debug = `0`,
		bool	fromScratch = `true`
	)

Execute a graph waiting for completion.

Parameters

graph	: schema to execute
debug	: display the graph with dot if debug == 1
fromScratch	: if true the graph is reinitialized

Calls Scheduler::getNextTasks and Scheduler::selectRunnableTasks to select tasks to execute

Calls Executor::launchTask to execute a selected Task.

Completion when graph is finished (Scheduler::isFinished)

Definition at line 112 of file Executor.cxx.

References _displayDot(), _execMode, _isWaitingEventsFromRunningTasks, _mainSched, _mutexForSchedulerUpdate, _numberOfEndedTasks, _numberOfRunningTasks, _root, _runningTasks, _toContinue, YACS::CONTINUE, DEBTRACE, YACS::ENGINE::Scheduler::exUpdateState(), YACS::ENGINE::Scheduler::getNextTasks(), CORBAEngineTest::i, YACS::ENGINE::Scheduler::init(), YACS::ENGINE::Scheduler::isFinished(), launchTasks(), loadTask(), YACS::ENGINE::Scheduler::selectRunnableTasks(), and sleepWhileNoEventsFromAnyRunningTask().

 {
   DEBTRACE("Executor::RunW debug: " << debug << " fromScratch: " << fromScratch);
   _mainSched=graph;
   _root = dynamic_cast<ComposedNode *>(_mainSched);
   if (!_root) throw Exception("Executor::Run, Internal Error!");
   bool isMore;
   int i=0;
   if(debug>1)_displayDot(graph);
   if (fromScratch)
     {
       graph->init();
       graph->exUpdateState();
     }
   if(debug>1)_displayDot(graph);
   vector<Task *> tasks;
   vector<Task *>::iterator iter;
   _toContinue=true;
   _execMode = YACS::CONTINUE;
   _isWaitingEventsFromRunningTasks = false;
   _numberOfRunningTasks = 0;
   _runningTasks.clear();
   _numberOfEndedTasks=0;
   while(_toContinue)
     {
       sleepWhileNoEventsFromAnyRunningTask();
 
       if(debug>2)_displayDot(graph);
 
       {//Critical section
         YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
         tasks=graph->getNextTasks(isMore);
         graph->selectRunnableTasks(tasks);
       }//End of critical section
 
       if(debug>2)_displayDot(graph);
 
       for(iter=tasks.begin();iter!=tasks.end();iter++)
         loadTask(*iter,this);
 
       if(debug>1)_displayDot(graph);
 
       launchTasks(tasks);
 
       if(debug>1)_displayDot(graph);
 
       {//Critical section
         YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
         _toContinue=!graph->isFinished();
       }//End of critical section
       DEBTRACE("_toContinue: " << _toContinue);
 
       if(debug>0)_displayDot(graph);
 
       i++;
     }
 }

void Executor::RunB	(	Scheduler *	graph,
		int	debug = `0`,
		bool	fromScratch = `true`
	)

Execute a graph with breakpoints or step by step.

To be launch in a thread (main thread controls the progression).

Parameters

graph	: schema to execute
debug	: display the graph with dot if debug >0
fromScratch	: if false, state from a previous partial exection is already loaded

Calls Scheduler::getNextTasks and Scheduler::selectRunnableTasks to select tasks to execute

Calls Executor::checkBreakPoints to verify if a pause is requested

Calls Executor::launchTask to execute a selected Task

Completion when graph is finished (Scheduler::isFinished)

States of execution:

YACS::NOTYETINITIALIZED
YACS::INITIALISED
YACS::RUNNING (to next breakpoint or step)
YACS::WAITINGTASKS (a breakpoint or step as been reached, but there are still running tasks)
YACS::PAUSED (a breakpoint or step as been reached, no more running tasks)
YACS::FINISHED (no more ready tasks, nore running tasks)
YACS::STOPPED (stopped by user before end)

Modes of Execution:

YACS::CONTINUE (normal run without breakpoints)
YACS::STEPBYSTEP (pause at each loop)
YACS::STOPBEFORENODES (pause when a node is reached)

A breakpoint is defined by a node name. The breakpoint is reached when the node becomes ready. Step by Step means execution node by node or group of node by group of nodes. At a given step, the user decides to launch all the ready nodes or only a subset (Caution: some nodes must run in parallel). The next event (end of task) may give a new set of ready nodes, and define a new step.

The graph execution may be controled by a pilot which sends requests. Requests are asynchronous. Requests are taken into account only on certain states, otherwise return the status IgnoredRequest.

Executor::getCurrentExecMode
Executor::getExecutorState
Executor::setExecMode : change the execution mode for next loop
Executor::setListOfBreakPoints : must be set before setting YACS::STOPBEFORENODES
Executor::getTasksToLoad : when paused or waiting tasks, get the list of next tasks
Executor::setStepsToExecute : define a subset of the list given by Executor::getTasksToLoad
Executor::resumeCurrentBreakPoint : when paused or waiting tasks, resumes execution
Executor::isNotFinished
Executor::stopExecution : stop execution asap, i.e. set STEPBYSTEP and wait PAUSED
Executor::saveState : dump the current state of execution in an xml file
Executor::loadState : Not yet implemented
Executor::getNbOfThreads
Executor::displayDot
Executor::setStopOnError : ask to stop execution if a node is found in ERROR state

If the pilot wants to wait the state YACS::PAUSED or YACS::WAITINGTASKS, synchronisation is obtained with:

Executor::waitPause

TO BE VALIDATED:

Pilot may connect to executor during execution, or deconnect.
Several Pilots may be connected at the same time (for observation...)

Definition at line 231 of file Executor.cxx.

References _condForPilot, _displayDot(), _dumpErrorFile, _dumpOnErrorRequested, _errorDetected, _executorState, _isOKToEnd, _isWaitingEventsFromRunningTasks, _mainSched, _mutexForSchedulerUpdate, _mutexForTrace, _numberOfEndedTasks, _numberOfRunningTasks, _root, _runningTasks, _start, _tasks, _toContinue, _trace, checkBreakPoints(), DEBTRACE, YACS::ENGINE::Scheduler::exUpdateState(), FilterTasksConsideringContainers(), YACS::FINISHED, YACS::ENGINE::Scheduler::getName(), YACS::ENGINE::Scheduler::getNextTasks(), YACS::ENGINE::Scheduler::init(), YACS::INITIALISED, YACS::ENGINE::Scheduler::isFinished(), launchTasks(), loadParallelTasks(), YACS::NOTYETINITIALIZED, YACS::RUNNING, saveState(), YACS::ENGINE::Scheduler::selectRunnableTasks(), sendEvent(), sleepWhileNoEventsFromAnyRunningTask(), YACS::STOPPED, and YACS::Exception::what().

Referenced by executorFunc().

 {
   DEBTRACE("Executor::RunB debug: "<< graph->getName() <<" "<< debug<<" fromScratch: "<<fromScratch);
 
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _mainSched = graph;
     _root = dynamic_cast<ComposedNode *>(_mainSched);
     if (!_root) throw Exception("Executor::Run, Internal Error!");
     _executorState = YACS::NOTYETINITIALIZED;
     sendEvent("executor");
     _toContinue=true;
     _isOKToEnd = false;
     _errorDetected = false;
     _isWaitingEventsFromRunningTasks = false;
     _numberOfRunningTasks = 0;
     _runningTasks.clear();
     _numberOfEndedTasks = 0;
     string tracefile = "traceExec_";
     tracefile += _mainSched->getName();
     _trace.open(tracefile.c_str());
 #ifdef WIN32
    _start = timeGetTime();
 #else
     gettimeofday(&_start, NULL);
 #endif
 
   } // --- End of critical section
 
   if (debug > 1) _displayDot(graph);
 
   if (fromScratch)
     {
       try
         {
           graph->init();
           graph->exUpdateState();
         }
       catch(Exception& ex)
         {
           DEBTRACE("exception: "<< (ex.what()));
           _executorState = YACS::FINISHED;
           sendEvent("executor");
           throw;
         }
     }
   _executorState = YACS::INITIALISED;
   sendEvent("executor");
 
   if (debug > 1) _displayDot(graph);
 
   vector<Task *>::iterator iter;
   bool isMore;
   int problemCount=0;
   int numberAllTasks;
 
   _executorState = YACS::RUNNING;
   sendEvent("executor");
   while (_toContinue)
     {
       DEBTRACE("--- executor main loop");
       sleepWhileNoEventsFromAnyRunningTask();
       DEBTRACE("--- events...");
       if (debug > 2) _displayDot(graph);
       { // --- Critical section
         YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
         _tasks=graph->getNextTasks(isMore);
         graph->selectRunnableTasks(_tasks);
         FilterTasksConsideringContainers(_tasks);
         numberAllTasks=_numberOfRunningTasks+_tasks.size();
       } // --- End of critical section
       if (debug > 2) _displayDot(graph);
       if (_executorState == YACS::RUNNING)
         {
           if (checkBreakPoints()) break; // end of thread requested, OK to exit at once;
           if (debug > 0) _displayDot(graph);
           DEBTRACE("---");
           //loadTasks(_tasks);
           loadParallelTasks(_tasks,this);
           if (debug > 1) _displayDot(graph);
           DEBTRACE("---");
           launchTasks(_tasks);
           DEBTRACE("---");
         }
       if (debug > 1) _displayDot(graph);
       { // --- Critical section
         DEBTRACE("---");
         YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
         //It is possible that the graph is finished but it remains running tasks (it's an error but we must take it into account)
         if(_numberOfRunningTasks == 0)
           _toContinue = !graph->isFinished();
 
         DEBTRACE("_numberOfRunningTasks: " << _numberOfRunningTasks);
         DEBTRACE("_numberOfEndedTasks: " << _numberOfEndedTasks);
         DEBTRACE("_toContinue: " << _toContinue);
         if(_toContinue && numberAllTasks==0)
           {
             //Problem : no running tasks and no task to launch ??
             problemCount++;
             std::cerr << "Problem in Executor : no running tasks and no task to launch ?? problemCount=" << problemCount << std::endl;
             //Pause to give a chance to interrupt
             usleep(1000);
             if(problemCount > 25)
               {
                 // Too much problems encountered : stop execution
                 _toContinue=false;
               }
           }
 
         if (! _toContinue)
           {
             _executorState = YACS::FINISHED;
             sendEvent("executor");
             _condForPilot.notify_all();
           }
       } // --- End of critical section
       if (debug > 0) _displayDot(graph);
       DEBTRACE("_toContinue: " << _toContinue);
     }
 
   DEBTRACE("End of main Loop");
 
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     if ( _toContinue) // --- break while(): request to stop detected on checkBreakPoints()
       {
         DEBTRACE("stop requested: End soon");
         _executorState = YACS::STOPPED;
         _toContinue = false;
         sendEvent("executor");
       }
   } // --- End of critical section
   if ( _dumpOnErrorRequested && _errorDetected)
     {
       saveState(_dumpErrorFile);
     }
   {
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForTrace);
     _trace.close();
   }
   DEBTRACE("End of RunB thread");  
 }

void YACS::ENGINE::Executor::RunW	(	Scheduler *	graph,
		int	debug = `0`,
		bool	fromScratch = `true`
	)

inline

Definition at line 100 of file Executor.hxx.

Referenced by driverTest(), and main().

100 { RunB(graph, debug, fromScratch); }

bool Executor::saveState ( const std::string & xmlFile )

save the current state of execution in an xml file

Definition at line 649 of file Executor.cxx.

References _mutexForSchedulerUpdate, _root, YACS::ENGINE::ComposedNode::accept(), YACS::ENGINE::VisitorSaveState::closeFileDump(), DEBTRACE, YACS::ENGINE::VisitorSaveState::openFileDump(), and YACS::Exception::what().

Referenced by RunB().

 {
   DEBTRACE("Executor::saveState() in " << xmlFile);
   bool result = false;
   try {
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     YACS::ENGINE::VisitorSaveState vst(_root);
     vst.openFileDump(xmlFile.c_str());
     _root->accept(&vst);
     vst.closeFileDump();
     result = true;
   }
   catch(Exception& ex) {
     std::cerr << ex.what() << std::endl;
   }
   return result;
 }

void Executor::sendEvent ( const std::string & event )

protectedvirtual

emit notification to all observers registered with the dispatcher

The dispatcher is unique and can be obtained by getDispatcher()

Definition at line 1301 of file Executor.cxx.

References _root, YACS::ENGINE::Dispatcher::dispatch(), YACS::ENGINE::Dispatcher::getDispatcher(), and YASSERT.

Referenced by checkBreakPoints(), functionForTaskExecution(), resumeCurrentBreakPoint(), and RunB().

 {
   Dispatcher* disp=Dispatcher::getDispatcher();
   YASSERT(disp);
   YASSERT(_root);
   disp->dispatch(_root,event);
 }

void YACS::ENGINE::Executor::setDPLScopeSensitive ( bool newVal )

inline

Definition at line 104 of file Executor.hxx.

104 { _DPLScopeSensitive=newVal; }

void Executor::setExecMode ( YACS::ExecutionMode mode )

Dynamically set the current mode of execution.

The mode can be Continue, step by step, or stop before execution of a node defined in a list of breakpoints.

Definition at line 431 of file Executor.cxx.

References _execMode, _isRunningunderExternalControl, _mutexForSchedulerUpdate, and DEBTRACE.

Referenced by main(), and stopExecution().

 {
   DEBTRACE("Executor::setExecMode(YACS::ExecutionMode mode) " << mode);
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _isRunningunderExternalControl=true;
     _execMode = mode;
   } // --- End of critical section
 }

void YACS::ENGINE::Executor::setKeepGoingProperty ( bool newVal )

inline

Definition at line 102 of file Executor.hxx.

102 { _keepGoingOnFail=newVal; }

void Executor::setListOfBreakPoints ( std::list< std::string > listOfBreakPoints )

define a list of nodes names as breakpoints in the graph

Definition at line 492 of file Executor.cxx.

References _isRunningunderExternalControl, _listOfBreakPoints, _mutexForSchedulerUpdate, and DEBTRACE.

 {
   DEBTRACE("Executor::setListOfBreakPoints(std::list<std::string> listOfBreakPoints)");
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _isRunningunderExternalControl=true;
     _listOfBreakPoints = listOfBreakPoints;
   } // --- End of critical section
 }

bool Executor::setStepsToExecute ( std::list< std::string > listToExecute )

Define a subset of task to execute in step by step mode.

Behaviour is unpredictable if the list is not a subset of the list given by Executor::getTasksToLoad in the current step. If some nodes must run in parallel, they must stay together in the list.

Definition at line 546 of file Executor.cxx.

References _executorState, _isRunningunderExternalControl, _mainSched, _mutexForSchedulerUpdate, _tasks, _tasksSave, DEBTRACE, YACS::FINISHED, YACS::ENGINE::Scheduler::getTaskName(), YACS::INITIALISED, YACS::NOTYETINITIALIZED, YACS::PAUSED, PMMLBasicsTestLauncher::ret, YACS::RUNNING, YACS::STOPPED, and YACS::WAITINGTASKS.

 {
   DEBTRACE("Executor::setStepsToExecute(std::list<std::string> listToExecute)");
   bool ret = true;
   vector<Task *>::iterator iter;
   vector<Task *> restrictedTasks;
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _isRunningunderExternalControl=true;
     switch (_executorState)
       {
       case YACS::WAITINGTASKS:
       case YACS::PAUSED:
         {
           for (iter=_tasksSave.begin(); iter!=_tasksSave.end(); iter++)
             {
               string readyNode = _mainSched->getTaskName(*iter);
               if (find(listToExecute.begin(), listToExecute.end(), readyNode)
                   != listToExecute.end())
                 {
                   restrictedTasks.push_back(*iter);
                   DEBTRACE("node to execute " << readyNode);
                 }
             }
           _tasks.clear();
           for (iter=restrictedTasks.begin(); iter!=restrictedTasks.end(); iter++)
             {
               _tasks.push_back(*iter);
             }
           break;
         }
       case YACS::NOTYETINITIALIZED:
       case YACS::INITIALISED:
       case YACS::RUNNING:
       case YACS::FINISHED:
       case YACS::STOPPED:
       default:
         {
           break;
         }
       }
     } // --- End of critical section
 
   _tasks.clear();
   for (iter=restrictedTasks.begin(); iter!=restrictedTasks.end(); iter++)
     {
       _tasks.push_back(*iter);
     }
   for (iter=_tasks.begin(); iter!=_tasks.end(); iter++)
     {
       string readyNode = _mainSched->getTaskName(*iter);
       DEBTRACE("selected node to execute " << readyNode);
     }
 
   return ret;
 }

void Executor::setStopOnError	(	bool	dumpRequested = `false`,
		std::string	xmlFile = `""`
	)

ask to stop execution on the first node found in error

Parameters

dumpRequested	produce a state dump when an error is found
xmlFile	name of file used for state dump

Definition at line 400 of file Executor.cxx.

References _dumpErrorFile, _dumpOnErrorRequested, _mutexForSchedulerUpdate, _stopOnErrorRequested, and DEBTRACE.

Referenced by main().

 {
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _dumpErrorFile=xmlFile;
     _stopOnErrorRequested=true;
     _dumpOnErrorRequested = dumpRequested;
     if (dumpRequested && xmlFile.empty())
       throw YACS::Exception("dump on error requested and no filename given for dump");
     DEBTRACE("_dumpErrorFile " << _dumpErrorFile << " " << _dumpOnErrorRequested);
   } // --- End of critical section
 }

void Executor::sleepWhileNoEventsFromAnyRunningTask ( )

protected

wait until a running task ends

Definition at line 1059 of file Executor.cxx.

References _condForNewTasksToPerform, _isWaitingEventsFromRunningTasks, _mutexForSchedulerUpdate, _numberOfEndedTasks, _numberOfRunningTasks, and DEBTRACE.

Referenced by RunA(), and RunB().

 {
   DEBTRACE("Executor::sleepWhileNoEventsFromAnyRunningTask()");
 //   _semForNewTasksToPerform.wait(); //----utiliser pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
   YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
   if (_numberOfRunningTasks > 0 && _numberOfEndedTasks==0)
     {
       _isWaitingEventsFromRunningTasks = true;
       _condForNewTasksToPerform.wait(_mutexForSchedulerUpdate); // mutex released during wait
     }
   _numberOfEndedTasks=0;
   DEBTRACE("---");
 }

void Executor::stopExecution ( )

stops the execution as soon as possible

Definition at line 639 of file Executor.cxx.

References _isOKToEnd, resumeCurrentBreakPoint(), setExecMode(), and YACS::STEPBYSTEP.

 {
   setExecMode(YACS::STEPBYSTEP);
   //waitPause();
   _isOKToEnd = true;
   resumeCurrentBreakPoint();
 }

void Executor::traceExec	(	Task *	task,
		const std::string &	message,
		const std::string &	placement
	)

protected

Definition at line 1274 of file Executor.cxx.

References _mainSched, _mutexForTrace, _start, _trace, YACS::ENGINE::Task::getContainer(), YACS::ENGINE::Container::getName(), and YACS::ENGINE::Scheduler::getTaskName().

Referenced by functionForTaskExecution(), launchTask(), launchTasks(), and loadTask().

 {
   string nodeName = _mainSched->getTaskName(task);
   Container *cont = task->getContainer();
   string containerName = "---";
   if (cont)
     containerName = cont->getName();
 
 #ifdef WIN32
   DWORD now = timeGetTime();
   double elapse = (now - _start)/1000.0;
 #else
   timeval now;
   gettimeofday(&now, NULL);
   double elapse = (now.tv_sec - _start.tv_sec) + double(now.tv_usec - _start.tv_usec)/1000000.0;
 #endif
   {
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForTrace);
     _trace << elapse << " " << containerName << " " << placement << " " << nodeName << " " << message << endl;
     _trace << flush;
   }
 }

void Executor::unsetStopOnError ( )

ask to do not stop execution on nodes found in error

Definition at line 417 of file Executor.cxx.

References _mutexForSchedulerUpdate, and _stopOnErrorRequested.

 {
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _stopOnErrorRequested=false;
   } // --- End of critical section
 }

void Executor::waitPause ( )

suspend pilot execution until Executor is in pause or waiting tasks completion mode.

Do nothing if execution is finished or in pause. Wait first step if Executor is running or in initialization.

Definition at line 609 of file Executor.cxx.

References _condForPilot, _executorState, _isRunningunderExternalControl, _mutexForSchedulerUpdate, DEBTRACE, YACS::FINISHED, YACS::INITIALISED, YACS::NOTYETINITIALIZED, YACS::PAUSED, YACS::RUNNING, YACS::STOPPED, and YACS::WAITINGTASKS.

 {
   DEBTRACE("Executor::waitPause()" << _executorState);
   { // --- Critical section
     YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
     _isRunningunderExternalControl=true;
     switch (_executorState)
       {
       default:
       case YACS::STOPPED:
       case YACS::FINISHED:
       case YACS::WAITINGTASKS:
       case YACS::PAUSED:
         {
           break;
         }
       case YACS::NOTYETINITIALIZED:
       case YACS::INITIALISED:
       case YACS::RUNNING:
         {
           _condForPilot.wait(_mutexForSchedulerUpdate); // wait until executor is PAUSED or WAITINGTASKS
           break;
         }
       }
   } // --- End of critical section
   DEBTRACE("---");
 }

void Executor::waitResume ( )

protected

in modes Step By step or with BreakPoint, wait until pilot resumes the execution

With the condition Mutex, the mutex is released atomically during the wait. Pilot calls Executor::resumeCurrentBreakPoint to resume execution. Must be called while mutex is locked.

Definition at line 811 of file Executor.cxx.

References _condForStepByStep, _mutexForSchedulerUpdate, and DEBTRACE.

Referenced by checkBreakPoints().

 {
   DEBTRACE("Executor::waitResume()");
   _condForStepByStep.wait(_mutexForSchedulerUpdate); // wait until pilot calls resumeCurrentBreakPoint()
   DEBTRACE("---");
 }

void Executor::wakeUp ( )

protected

must be used protected by _mutexForSchedulerUpdate!

Definition at line 1086 of file Executor.cxx.

References _condForNewTasksToPerform, _isWaitingEventsFromRunningTasks, _numberOfEndedTasks, and DEBTRACE.

Referenced by functionForTaskExecution().

 {
   DEBTRACE("Executor::wakeUp() " << _isWaitingEventsFromRunningTasks);
   if (_isWaitingEventsFromRunningTasks)
     {
       _isWaitingEventsFromRunningTasks = false;
       _condForNewTasksToPerform.notify_all();
     }
   else
     _numberOfEndedTasks++;
 }