YACS::ENGINE::Executor Class Reference

Threaded Executor. More...

#include <Executor.hxx>

Collaboration diagram for YACS::ENGINE::Executor:

Public Member Functions
	Executor ()
virtual	~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	runWlm (Scheduler *graph, int debug=0, bool fromScratch=true)
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	getMaxNbOfThreads () const
void	setMaxNbOfThreads (int maxNbThreads)
int	getNbOfThreads ()
int	getNumberOfRunningTasks ()
	number of running tasks More...
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...
bool	suspendASAP ()
void	resume (bool suspended)
YACS::BASES::Mutex &	getTheMutexForSchedulerUpdate ()
void	loadTask (Task *task, const WorkloadManager::RunInfo &runInfo)
YACS::Event	runTask (Task *task)
void	makeDatastreamConnections (Task *task)
void	beginTask (Task *task)
void	endTask (Task *task, YACS::Event ev)
void	failTask (Task *task, const std::string &message)

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	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	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...
void	filterTasksConsideringContainers (std::vector< Task * > &tsks)

Static Protected Member Functions
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
std::uint32_t	_maxNbThreads = 10000
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::ofstream	_trace
std::string	_dumpErrorFile
bool	_keepGoingOnFail
bool	_DPLScopeSensitive
	specifies if scope DynParaLoop is active or not. False by default. More...
std::chrono::steady_clock::time_point	_start

Detailed Description

Threaded Executor.

Definition at line 62 of file Executor.hxx.

Constructor & Destructor Documentation

Executor()

Executor::Executor

(

)

Definition at line 82 of file Executor.cxx.

                  :_mainSched(NULL),_isWaitingEventsFromRunningTasks(false),_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;
  _numberOfRunningTasks = 0;
  _numberOfEndedTasks = 0;
  DEBTRACE("Executor initialized with max threads = " << _maxThreads);
}

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

~Executor()

Executor::~Executor ( )

virtual

Definition at line 99 of file Executor.cxx.

{
}

Member Function Documentation

_displayDot()

void Executor::_displayDot ( Scheduler *  graph )

protected

Display the graph state as a dot display.

Parameters

graph

: the node to display

Definition at line 729 of file Executor.cxx.

{
   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");
}

References isfile().

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

beginTask()

void Executor::beginTask

(

Task *

task

)

Definition at line 1464 of file Executor.cxx.

{
  // --- Critical section
  YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
  _numberOfRunningTasks++;
  _runningTasks.insert(task);
  // --- End of critical section
}

References _mutexForSchedulerUpdate, _numberOfRunningTasks, and _runningTasks.

Referenced by runWlm().

checkBreakPoints()

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 752 of file Executor.cxx.

{
  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;
}

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(), and runWlm().

ComputePlacement()

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

staticprotected

Definition at line 1412 of file Executor.cxx.

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

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

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

displayDot()

void Executor::displayDot

(

Scheduler *

graph

)

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

Definition at line 718 of file Executor.cxx.

{
  _isRunningunderExternalControl=true;
  _displayDot(graph);
}

References _displayDot(), and _isRunningunderExternalControl.

Referenced by main().

endTask()

void Executor::endTask	(	Task *	task,
		YACS::Event	ev
	)

Definition at line 1473 of file Executor.cxx.

{
  YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
  try
  {
    if (ev == YACS::FINISH) task->finished();
    if (ev == YACS::ABORT)
    {
      _errorDetected = true;
      if (_stopOnErrorRequested)
      {
        _execMode = YACS::STEPBYSTEP;
        _isOKToEnd = true;
      }
      task->aborted();
    }
    //traceExec(task, "state:"+Node::getStateName(task->getState()),placement);
    _mainSched->notifyFrom(task,ev,this);
  }
  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;
  }
  _numberOfRunningTasks--;
  _runningTasks.erase(task);
  if ((_numberOfRunningTasks == 0) && (_execMode != YACS::CONTINUE)) // no more running tasks
    {
      if (_executorState == YACS::WAITINGTASKS)
        {
          _executorState = YACS::PAUSED;
          sendEvent("executor");
          _condForPilot.notify_all();
          if (_errorDetected &&
              _stopOnErrorRequested &&
              !_isRunningunderExternalControl)
            _condForStepByStep.notify_all(); // exec thread may be on waitResume
        }
    }
  if (_executorState != YACS::PAUSED)
    wakeUp();
}

References _condForPilot, _condForStepByStep, _errorDetected, _execMode, _executorState, _isOKToEnd, _isRunningunderExternalControl, _mainSched, _mutexForSchedulerUpdate, _numberOfRunningTasks, _runningTasks, _stopOnErrorRequested, YACS::ABORT, YACS::ENGINE::Task::aborted(), YACS::CONTINUE, testCppPluginInvokation::ex, YACS::FINISH, YACS::ENGINE::Task::finished(), YACS::ENGINE::Scheduler::notifyFrom(), YACS::PAUSED, sendEvent(), YACS::STEPBYSTEP, YACS::WAITINGTASKS, and wakeUp().

Referenced by failTask(), and YACS::ENGINE::WlmTask::run().

failTask()

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

Definition at line 1524 of file Executor.cxx.

{
  ElementaryNode* elemNode = dynamic_cast<ElementaryNode*>(task);
  if(elemNode != nullptr)
  {
    StateLoader(elemNode, YACS::ERROR);
    elemNode->setErrorDetails(message);
  }
  endTask(task, YACS::ABORT);
}

References YACS::ABORT, endTask(), YACS::ERROR, YACS::ENGINE::Node::setErrorDetails(), and YACS::ENGINE::StateLoader().

Referenced by YACS::ENGINE::WlmTask::run().

filterTasksConsideringContainers()

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

protected

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 1352 of file Executor.cxx.

{
  std::map<HomogeneousPoolContainer *, std::vector<Task *>, HPCCompare > m;
  for(auto cur : tsks)
    {
      if(!cur)
        continue;
      Container *cont(cur->getContainer());
      if(!cont)
        {
          m[nullptr].push_back(cur);
          continue;
        }
      HomogeneousPoolContainer *contC(dynamic_cast<HomogeneousPoolContainer *>(cont));
      if(!contC)
        {
          m[nullptr].push_back(cur);
          continue;
        }
      m[contC].push_back(cur);
    }
  //
  std::vector<Task *> ret;
  for(auto it : m)
    {
      HomogeneousPoolContainer *curhpc(it.first);
      const std::vector<Task *>& curtsks(it.second);
      if(!curhpc)
        {
          std::uint32_t nbThreadsRunning = _runningTasks.size();
          std::uint32_t nbOfFreeSpace = _maxNbThreads - min(_maxNbThreads,nbThreadsRunning);
          std::uint32_t nbOfCandidates = static_cast<std::uint32_t>( curtsks.size() );
          std::uint32_t nbOfCandidatesToBeLaunched = std::min(nbOfCandidates,nbOfFreeSpace);
          DEBTRACE("nb threads running: " << nbThreadsRunning);
          DEBTRACE("MaxNbThreads: " << _maxNbThreads);
          DEBTRACE("nbOfFreeSpace: " << nbOfFreeSpace);
          DEBTRACE("nbOfCandidates: " << nbOfCandidates);
          DEBTRACE("nbOfCandidatesToBeLaunched: " << nbOfCandidatesToBeLaunched);
          ret.insert(ret.end(),curtsks.begin(),curtsks.begin() + nbOfCandidatesToBeLaunched);
        }
      else
        {
          // start of critical section for container curhpc
          std::lock_guard<std::mutex> alckCont(curhpc->getLocker());
          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;
}

References _maxNbThreads, _runningTasks, YACS::ENGINE::HomogeneousPoolContainer::allocateFor(), DEBTRACE, YACS::ENGINE::HomogeneousPoolContainer::getLocker(), YACS::ENGINE::HomogeneousPoolContainer::getNumberOfFreePlace(), yacsorb.CORBAEngineTest::i, and gui.GraphViewer::m.

Referenced by RunB(), and runWlm().

functionForTaskExecution()

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 1165 of file Executor.cxx.

{
  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)
    {
      std::lock_guard<std::mutex> alckCont(contC->getLocker());
      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)
 
  Thread::exit(0);
  return 0;
}

References _condForPilot, _condForStepByStep, _errorDetected, _execMode, _executorState, _isOKToEnd, _isRunningunderExternalControl, _mutexForSchedulerUpdate, _numberOfRunningTasks, _runningTasks, _semForMaxThreads, _semThreadCnt, _stopOnErrorRequested, YACS::ABORT, YACS::ENGINE::Task::aborted(), YACS::ENGINE::Node::applyDPLScope(), driver_internal::args, ComputePlacement(), YACS::CONTINUE, DEBTRACE, YACS::ENGINE::Task::disconnectService(), testCppPluginInvokation::ex, threadargs::execInst, YACS::ENGINE::Task::execute(), YACS::FINISH, YACS::ENGINE::Task::finished(), YACS::ENGINE::Task::getContainer(), getDPLScopeSensitive(), YACS::ENGINE::HomogeneousPoolContainer::getLocker(), 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, and wakeUp().

Referenced by launchTask().

functionForTaskLoad()

void * Executor::functionForTaskLoad ( void *  arg )

staticprotected

This thread is NOT supposed to be detached !

Definition at line 1140 of file Executor.cxx.

{
  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;
}

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

Referenced by loadParallelTasks().

getCurrentExecMode()

YACS::ExecutionMode Executor::getCurrentExecMode

(

)

Definition at line 373 of file Executor.cxx.

{
  _isRunningunderExternalControl=true;
  return _execMode;
}

References _execMode, and _isRunningunderExternalControl.

getDPLScopeSensitive()

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

inline

Definition at line 109 of file Executor.hxx.

{ return _DPLScopeSensitive; }

Referenced by functionForTaskExecution(), and runTask().

getExecutorState()

YACS::ExecutorState Executor::getExecutorState

(

)

Definition at line 380 of file Executor.cxx.

{
  _isRunningunderExternalControl=true;
  return _executorState;
}

References _executorState, and _isRunningunderExternalControl.

Referenced by main().

getKeepGoingProperty()

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

inline

Definition at line 107 of file Executor.hxx.

{ return _keepGoingOnFail; }

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

getMaxNbOfThreads()

int Executor::getMaxNbOfThreads

(

)

const

Definition at line 1111 of file Executor.cxx.

{
  return (int)_maxNbThreads;
}

References _maxNbThreads.

getNbOfThreads()

int Executor::getNbOfThreads

(

)

Definition at line 1121 of file Executor.cxx.

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

References _isRunningunderExternalControl, and _mutexForNbOfConcurrentThreads.

Referenced by checkBreakPoints().

getNumberOfRunningTasks()

int Executor::getNumberOfRunningTasks

(

)

number of running tasks

Definition at line 1130 of file Executor.cxx.

{
  YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
  _isRunningunderExternalControl=true;
  return _numberOfRunningTasks;
}

References _isRunningunderExternalControl, _mutexForSchedulerUpdate, and _numberOfRunningTasks.

getTasksToLoad()

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 507 of file Executor.cxx.

{
  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;
}

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

getTheMutexForSchedulerUpdate()

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

inline

Definition at line 133 of file Executor.hxx.

{ return _mutexForSchedulerUpdate; }

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

isNotFinished()

bool Executor::isNotFinished

(

)

Definition at line 387 of file Executor.cxx.

{
  _isRunningunderExternalControl=true;
  return _toContinue;
}

References _isRunningunderExternalControl, and _toContinue.

Referenced by main().

launchTask()

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 1022 of file Executor.cxx.

{
  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);
}

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

Referenced by launchTasks().

launchTasks()

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 921 of file Executor.cxx.

{
  //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);
    }
}

References _mainSched, _mutexForSchedulerUpdate, YACS::ABORT, ComputePlacement(), YACS::ERROR, testCppPluginInvokation::ex, YACS::ENGINE::Node::getStateName(), launchTask(), YACS::ENGINE::Scheduler::notifyFrom(), yacsorb.CORBAEngineTest::state, gui.Appli::t, YACS::TOLOAD, YACS::TORECONNECT, and traceExec().

Referenced by RunA(), and RunB().

loadParallelTasks()

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

protected

Definition at line 899 of file Executor.cxx.

{
  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();
}

References _mainSched, _tasks, _threadStackSize, driver_internal::args, DEBTRACE, and functionForTaskLoad().

Referenced by RunB().

loadState()

bool Executor::loadState

(

)

not yet implemented

Definition at line 698 of file Executor.cxx.

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

References _isRunningunderExternalControl, and DEBTRACE.

loadTask() [1/2]

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

protected

Perform loading of a Task.

Parameters

task	: Task to load

Definition at line 853 of file Executor.cxx.

{
  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
    }
}

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

loadTask() [2/2]

void Executor::loadTask	(	Task *	task,
		const WorkloadManager::RunInfo &	runInfo
	)

Definition at line 1423 of file Executor.cxx.

{
  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,this);
  }//End of critical section
  try
    {
      std::ostringstream container_name;
      container_name << runInfo.type.name << "-" << runInfo.index;
      task->imposeResource(runInfo.resource.name, container_name.str());
      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, this);
        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, this);
        traceExec(task, "state:"+Node::getStateName(task->getState()), ComputePlacement(task));
      }//End of critical section
    }
}

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

Referenced by functionForTaskLoad(), YACS::ENGINE::WlmTask::run(), and RunA().

makeDatastreamConnections()

void Executor::makeDatastreamConnections

(

Task *

task

)

Definition at line 1605 of file Executor.cxx.

{
  YACS::StatesForNode state=task->getState();
  if(state != YACS::TOLOAD && state != YACS::TORECONNECT)
    return;
  try
    {
      task->connectService();
      traceExec(task, "connectService",ComputePlacement(task));
      {//Critical section
        YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
        task->connected();
      }//End of critical section
    }
  catch(Exception& ex)
    {
      std::cerr << ex.what() << std::endl;
      try
        {
          (task)->disconnectService();
          traceExec(task, "disconnectService",ComputePlacement(task));
        }
      catch(...)
        {
          // Disconnect has failed
          traceExec(task, "disconnectService failed, ABORT",ComputePlacement(task));
        }
      {//Critical section
        YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
        task->aborted();
        _mainSched->notifyFrom(task,YACS::ABORT,this);
      }//End of critical section
    }
  catch(...)
    {
      std::cerr << "Problem in connectService" << std::endl;
      try
        {
          (task)->disconnectService();
          traceExec(task, "disconnectService",ComputePlacement(task));
        }
      catch(...)
        {
          // Disconnect has failed
          traceExec(task, "disconnectService failed, ABORT",ComputePlacement(task));
        }
      {//Critical section
        YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
        task->aborted();
        _mainSched->notifyFrom(task,YACS::ABORT,this);
      }//End of critical section
    }
  if(task->getState() == YACS::ERROR)
    {
      //try to put all coupled tasks in error
      std::set<Task*> coupledSet;
      task->getCoupledTasks(coupledSet);
      for (std::set<Task*>::iterator it = coupledSet.begin(); it != coupledSet.end(); ++it)
        {
          Task* t=*it;
          if(t == task)continue;
          if(t->getState() == YACS::ERROR)continue;
          try
            {
              t->disconnectService();
              traceExec(t, "disconnectService",ComputePlacement(task));
            }
          catch(...)
            {
              // Disconnect has failed
              traceExec(t, "disconnectService failed, ABORT",ComputePlacement(task));
            }
          {//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(task));
        }
    }
  traceExec(task, "state:"+Node::getStateName(task->getState()),ComputePlacement(task));
}

References _mainSched, _mutexForSchedulerUpdate, YACS::ABORT, YACS::ENGINE::Task::aborted(), ComputePlacement(), YACS::ENGINE::Task::connected(), YACS::ENGINE::Task::connectService(), YACS::ERROR, testCppPluginInvokation::ex, YACS::ENGINE::Task::getCoupledTasks(), YACS::ENGINE::Task::getState(), YACS::ENGINE::Node::getStateName(), YACS::ENGINE::Scheduler::notifyFrom(), yacsorb.CORBAEngineTest::state, gui.Appli::t, YACS::TOLOAD, YACS::TORECONNECT, and traceExec().

Referenced by YACS::ENGINE::WlmTask::run().

resume()

void Executor::resume

(

bool

suspended

)

This method is expected to be called in association with suspendASAP method. Expected to be called just after suspendASAP with output of resume as input parameter

Definition at line 660 of file Executor.cxx.

{
  if(suspended)
    _mutexForSchedulerUpdate.unLock();
}

References _mutexForSchedulerUpdate.

Referenced by gui.Appli.Browser::run(), gui.Appli.Browser::step(), and gui.Appli.Browser::susp().

resumeCurrentBreakPoint()

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 448 of file Executor.cxx.

{
  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;
}

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

Referenced by main(), and stopExecution().

RunA()

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 116 of file Executor.cxx.

{
  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++;
    }
}

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

RunB()

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 235 of file Executor.cxx.

{
  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());
    _start = std::chrono::steady_clock::now();
  } // --- 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);
        std::vector<Task *> tasks = graph->getNextTasks(isMore);
        graph->selectRunnableTasks(tasks);
        filterTasksConsideringContainers(tasks);
        _tasks = 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("---");
          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");  
}

References _condForPilot, _displayDot(), _dumpErrorFile, _dumpOnErrorRequested, _errorDetected, _executorState, _isOKToEnd, _isWaitingEventsFromRunningTasks, _mainSched, _mutexForSchedulerUpdate, _mutexForTrace, _numberOfEndedTasks, _numberOfRunningTasks, _root, _runningTasks, _start, _tasks, _toContinue, _trace, checkBreakPoints(), DEBTRACE, testCppPluginInvokation::ex, 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(), and YACS::STOPPED.

Referenced by executorFunc(), and RunW().

runTask()

YACS::Event Executor::runTask

(

Task *

task

)

Definition at line 1535 of file Executor.cxx.

{
  { // --- Critical section
    YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
    task->begin(); //change state to ACTIVATED
  }
  traceExec(task, "state:"+Node::getStateName(task->getState()),ComputePlacement(task));
 
  if(getDPLScopeSensitive())
    {
      Node *node(dynamic_cast<Node *>(task));
      ComposedNode *gfn(dynamic_cast<ComposedNode *>(_mainSched));
      if(node!=0 && gfn!=0)
        node->applyDPLScope(gfn);
    }
 
  YACS::Event ev=YACS::FINISH;
  try
    {
      traceExec(task, "start execution",ComputePlacement(task));
      task->execute();
      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();
      traceExec(task, message,ComputePlacement(task));
    }
  catch(...)
    {
      // Execution has failed
      std::cerr << "Execution has failed: unknown reason" << std::endl;
      ev=YACS::ABORT;
      traceExec(task, "end execution ABORT, unknown reason",ComputePlacement(task));
    }
 
  // Disconnect task
  try
    {
      DEBTRACE("task->disconnectService()");
      task->disconnectService();
      traceExec(task, "disconnectService",ComputePlacement(task));
    }
  catch(...)
    {
      // Disconnect has failed
      std::cerr << "disconnect has failed" << std::endl;
      ev=YACS::ABORT;
      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)
  {
    std::lock_guard<std::mutex> alckCont(contC->getLocker());
    contC->release(task);
  }
 
  return ev;
}

References _mainSched, _mutexForSchedulerUpdate, YACS::ABORT, YACS::ENGINE::Node::applyDPLScope(), YACS::ENGINE::Task::begin(), ComputePlacement(), DEBTRACE, YACS::ENGINE::Task::disconnectService(), testCppPluginInvokation::ex, YACS::ENGINE::Task::execute(), YACS::FINISH, YACS::ENGINE::Task::getContainer(), getDPLScopeSensitive(), YACS::ENGINE::HomogeneousPoolContainer::getLocker(), YACS::ENGINE::Task::getState(), YACS::ENGINE::Node::getStateName(), YACS::ENGINE::HomogeneousPoolContainer::release(), and traceExec().

Referenced by YACS::ENGINE::WlmTask::run().

RunW()

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

Definition at line 1833 of file Executor.cxx.

{
  std::string str_value = graph->getProperty("executor");
  if(str_value == "WorkloadManager"
     || str_value == "WORKLOADMANAGER"
     || str_value == "workloadmanager"
     || str_value == "WorkLoadManager")
    runWlm(graph, debug, fromScratch);
  else
    RunB(graph, debug, fromScratch);
}

References YACS::ENGINE::Scheduler::getProperty(), RunB(), and runWlm().

Referenced by driverTest(), and main().

runWlm()

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

Definition at line 1688 of file Executor.cxx.

{
  DEBTRACE("Executor::runWlm 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());
    _start = std::chrono::steady_clock::now();
  } // --- 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);
 
  bool isMore;
  int problemCount=0;
  int numberAllTasks;
 
  _executorState = YACS::RUNNING;
  sendEvent("executor");
 
  WorkloadManager::DefaultAlgorithm algo;
  WorkloadManager::WorkloadManager wlm(algo);
  WlmTask::loadResources(wlm);
  wlm.start();
 
  while (_toContinue)
    {
      DEBTRACE("--- executor main loop");
      sleepWhileNoEventsFromAnyRunningTask();
      DEBTRACE("--- events...");
      if (debug > 2) _displayDot(graph);
      { // --- Critical section
        YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForSchedulerUpdate);
        std::vector<Task *> readyTasks=graph->getNextTasks(isMore);
        graph->selectRunnableTasks(readyTasks);
        _tasks.clear();
        for(Task * t : readyTasks)
          if(_runningTasks.find(t) == _runningTasks.end())
            _tasks.push_back(t);
        // TODO: to be removed
        filterTasksConsideringContainers(_tasks);
        numberAllTasks=_numberOfRunningTasks+_tasks.size();
      } // --- End of critical section
      if (debug > 2) _displayDot(graph);
      DEBTRACE("--- events...");
      if (_executorState == YACS::RUNNING)
      {
        if (checkBreakPoints()) break; // end of thread requested, OK to exit at once;
        for(Task * task : _tasks)
        {
          beginTask(task);
          WlmTask* newTask = new WlmTask(*this, task);
          wlm.addTask(newTask);
        }
      }
      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)
        _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);
    }
 
  wlm.stop();
  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();
  }
}

References _condForPilot, _displayDot(), _dumpErrorFile, _dumpOnErrorRequested, _errorDetected, _executorState, _isOKToEnd, _isWaitingEventsFromRunningTasks, _mainSched, _mutexForSchedulerUpdate, _mutexForTrace, _numberOfEndedTasks, _numberOfRunningTasks, _root, _runningTasks, _start, _tasks, _toContinue, _trace, beginTask(), checkBreakPoints(), DEBTRACE, testCppPluginInvokation::ex, 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(), YACS::ENGINE::WlmTask::loadResources(), YACS::NOTYETINITIALIZED, YACS::RUNNING, saveState(), YACS::ENGINE::Scheduler::selectRunnableTasks(), sendEvent(), sleepWhileNoEventsFromAnyRunningTask(), YACS::STOPPED, and gui.Appli::t.

Referenced by RunW().

saveState()

bool Executor::saveState

(

const std::string &

xmlFile

)

save the current state of execution in an xml file

Definition at line 678 of file Executor.cxx.

{
  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;
}

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

Referenced by RunB(), and runWlm().

sendEvent()

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 1324 of file Executor.cxx.

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

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

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

setDPLScopeSensitive()

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

inline

Definition at line 108 of file Executor.hxx.

{ _DPLScopeSensitive=newVal; }

setExecMode()

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 430 of file Executor.cxx.

{
  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
}

References _execMode, _isRunningunderExternalControl, _mutexForSchedulerUpdate, and DEBTRACE.

Referenced by main(), and stopExecution().

setKeepGoingProperty()

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

inline

Definition at line 106 of file Executor.hxx.

{ _keepGoingOnFail=newVal; }

setListOfBreakPoints()

void Executor::setListOfBreakPoints

(

std::list< std::string >

listOfBreakPoints

)

define a list of nodes names as breakpoints in the graph

Definition at line 491 of file Executor.cxx.

{
  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
}

References _isRunningunderExternalControl, _listOfBreakPoints, _mutexForSchedulerUpdate, and DEBTRACE.

setMaxNbOfThreads()

void Executor::setMaxNbOfThreads

(

int

maxNbThreads

)

Definition at line 1116 of file Executor.cxx.

{
  _maxNbThreads = static_cast< std::uint32_t >(maxNbThreads);
}

References _maxNbThreads.

setStepsToExecute()

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 545 of file Executor.cxx.

{
  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;
}

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

setStopOnError()

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 399 of file Executor.cxx.

{
  { // --- 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
}

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

Referenced by main().

sleepWhileNoEventsFromAnyRunningTask()

void Executor::sleepWhileNoEventsFromAnyRunningTask ( )

protected

wait until a running task ends

Definition at line 1082 of file Executor.cxx.

{
  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("---");
}

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

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

stopExecution()

void Executor::stopExecution

(

)

stops the execution as soon as possible

Definition at line 668 of file Executor.cxx.

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

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

suspendASAP()

bool Executor::suspendASAP

(

)

This method can be called at any time simultaneously during a RunB call. This method will wait until the executor is locked in a consistent state of a running graph.

This method is expected to be called in association with resume method. The returned parameter is expected to be transfered to resume method.

Definition at line 643 of file Executor.cxx.

{
  // no AutoLocker here. It's not a bug.
  _mutexForSchedulerUpdate.lock();
  if(!_toContinue && _executorState==YACS::FINISHED)
    {// execution is finished
      _mutexForSchedulerUpdate.unLock();
      return false;// the executor is no more running
    }
  //general case. Leave method with locker in locked status
  return true;
}

References _executorState, _mutexForSchedulerUpdate, _toContinue, and YACS::FINISHED.

traceExec()

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

protected

Definition at line 1301 of file Executor.cxx.

{
  string nodeName = _mainSched->getTaskName(task);
  Container *cont = task->getContainer();
  string containerName = "---";
  if (cont)
    containerName = cont->getName();
 
  std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now();
  std::chrono::milliseconds millisec;
  millisec = std::chrono::duration_cast<std::chrono::milliseconds>(now -_start);
  double elapse = double(millisec.count()) / 1000.0;
  {
    YACS::BASES::AutoLocker<YACS::BASES::Mutex> alck(&_mutexForTrace);
    _trace << elapse << " " << containerName << " " << placement << " " << nodeName << " " << message << endl;
    _trace << flush;
  }
}

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

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

unsetStopOnError()

void Executor::unsetStopOnError

(

)

ask to do not stop execution on nodes found in error

Definition at line 416 of file Executor.cxx.

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

References _mutexForSchedulerUpdate, and _stopOnErrorRequested.

waitPause()

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 608 of file Executor.cxx.

{
  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("---");
}

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

waitResume()

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 840 of file Executor.cxx.

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

References _condForStepByStep, _mutexForSchedulerUpdate, and DEBTRACE.

Referenced by checkBreakPoints().

wakeUp()

void Executor::wakeUp ( )

protected

must be used protected by _mutexForSchedulerUpdate!

Definition at line 1099 of file Executor.cxx.

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

References _condForNewTasksToPerform, _isWaitingEventsFromRunningTasks, _numberOfEndedTasks, and DEBTRACE.

Referenced by endTask(), and functionForTaskExecution().

Member Data Documentation

_condForNewTasksToPerform

YACS::BASES::Condition YACS::ENGINE::Executor::_condForNewTasksToPerform

protected

Definition at line 69 of file Executor.hxx.

Referenced by sleepWhileNoEventsFromAnyRunningTask(), and wakeUp().

_condForPilot

YACS::BASES::Condition YACS::ENGINE::Executor::_condForPilot

protected

Definition at line 72 of file Executor.hxx.

Referenced by checkBreakPoints(), endTask(), functionForTaskExecution(), RunB(), runWlm(), and waitPause().

_condForStepByStep

YACS::BASES::Condition YACS::ENGINE::Executor::_condForStepByStep

protected

Definition at line 71 of file Executor.hxx.

Referenced by endTask(), functionForTaskExecution(), resumeCurrentBreakPoint(), and waitResume().

_DPLScopeSensitive

bool YACS::ENGINE::Executor::_DPLScopeSensitive

protected

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

Definition at line 97 of file Executor.hxx.

_dumpErrorFile

std::string YACS::ENGINE::Executor::_dumpErrorFile

protected

Definition at line 94 of file Executor.hxx.

Referenced by RunB(), runWlm(), and setStopOnError().

_dumpOnErrorRequested

bool YACS::ENGINE::Executor::_dumpOnErrorRequested

protected

Definition at line 79 of file Executor.hxx.

Referenced by Executor(), RunB(), runWlm(), and setStopOnError().

_errorDetected

bool YACS::ENGINE::Executor::_errorDetected

protected

Definition at line 80 of file Executor.hxx.

Referenced by endTask(), Executor(), functionForTaskExecution(), RunB(), and runWlm().

_execMode

YACS::ExecutionMode YACS::ENGINE::Executor::_execMode

protected

Definition at line 88 of file Executor.hxx.

Referenced by checkBreakPoints(), endTask(), Executor(), functionForTaskExecution(), getCurrentExecMode(), RunA(), and setExecMode().

_executorState

YACS::ExecutorState YACS::ENGINE::Executor::_executorState

protected

Definition at line 87 of file Executor.hxx.

Referenced by checkBreakPoints(), endTask(), Executor(), functionForTaskExecution(), getExecutorState(), getTasksToLoad(), resumeCurrentBreakPoint(), RunB(), runWlm(), setStepsToExecute(), suspendASAP(), and waitPause().

_isOKToEnd

bool YACS::ENGINE::Executor::_isOKToEnd

protected

Definition at line 77 of file Executor.hxx.

Referenced by checkBreakPoints(), endTask(), Executor(), functionForTaskExecution(), RunB(), runWlm(), and stopExecution().

_isRunningunderExternalControl

bool YACS::ENGINE::Executor::_isRunningunderExternalControl

protected

Definition at line 81 of file Executor.hxx.

Referenced by displayDot(), endTask(), Executor(), functionForTaskExecution(), getCurrentExecMode(), getExecutorState(), getNbOfThreads(), getNumberOfRunningTasks(), getTasksToLoad(), isNotFinished(), loadState(), resumeCurrentBreakPoint(), setExecMode(), setListOfBreakPoints(), setStepsToExecute(), and waitPause().

_isWaitingEventsFromRunningTasks

bool YACS::ENGINE::Executor::_isWaitingEventsFromRunningTasks

protected

Definition at line 82 of file Executor.hxx.

Referenced by RunA(), RunB(), runWlm(), sleepWhileNoEventsFromAnyRunningTask(), and wakeUp().

_keepGoingOnFail

bool YACS::ENGINE::Executor::_keepGoingOnFail

protected

Definition at line 95 of file Executor.hxx.

_listOfBreakPoints

std::list<std::string> YACS::ENGINE::Executor::_listOfBreakPoints

protected

Definition at line 89 of file Executor.hxx.

Referenced by checkBreakPoints(), and setListOfBreakPoints().

_listOfTasksToLoad

std::list<std::string> YACS::ENGINE::Executor::_listOfTasksToLoad

protected

Definition at line 90 of file Executor.hxx.

Referenced by checkBreakPoints(), and getTasksToLoad().

_mainSched

Scheduler* YACS::ENGINE::Executor::_mainSched

protected

Definition at line 65 of file Executor.hxx.

Referenced by checkBreakPoints(), endTask(), launchTask(), launchTasks(), loadParallelTasks(), loadTask(), makeDatastreamConnections(), RunA(), RunB(), runTask(), runWlm(), setStepsToExecute(), and traceExec().

_maxNbThreads

std::uint32_t YACS::ENGINE::Executor::_maxNbThreads = 10000

protected

Definition at line 75 of file Executor.hxx.

Referenced by filterTasksConsideringContainers(), getMaxNbOfThreads(), and setMaxNbOfThreads().

_maxThreads

int Executor::_maxThreads

static

Definition at line 131 of file Executor.hxx.

Referenced by Executor(), launchTask(), and YACS::ENGINE::Runtime::Runtime().

_mutexForNbOfConcurrentThreads

YACS::BASES::Mutex YACS::ENGINE::Executor::_mutexForNbOfConcurrentThreads

protected

Definition at line 68 of file Executor.hxx.

Referenced by getNbOfThreads().

_mutexForSchedulerUpdate

YACS::BASES::Mutex YACS::ENGINE::Executor::_mutexForSchedulerUpdate

protected

Definition at line 73 of file Executor.hxx.

Referenced by beginTask(), checkBreakPoints(), endTask(), functionForTaskExecution(), getNumberOfRunningTasks(), getTasksToLoad(), launchTask(), launchTasks(), loadTask(), makeDatastreamConnections(), resume(), resumeCurrentBreakPoint(), RunA(), RunB(), runTask(), runWlm(), saveState(), setExecMode(), setListOfBreakPoints(), setStepsToExecute(), setStopOnError(), sleepWhileNoEventsFromAnyRunningTask(), suspendASAP(), unsetStopOnError(), waitPause(), and waitResume().

_mutexForTrace

YACS::BASES::Mutex YACS::ENGINE::Executor::_mutexForTrace

protected

Definition at line 74 of file Executor.hxx.

Referenced by RunB(), runWlm(), and traceExec().

_nbOfConcurrentThreads

int YACS::ENGINE::Executor::_nbOfConcurrentThreads

protected

Definition at line 67 of file Executor.hxx.

_numberOfEndedTasks

int YACS::ENGINE::Executor::_numberOfEndedTasks

protected

Definition at line 85 of file Executor.hxx.

Referenced by Executor(), RunA(), RunB(), runWlm(), sleepWhileNoEventsFromAnyRunningTask(), and wakeUp().

_numberOfRunningTasks

int YACS::ENGINE::Executor::_numberOfRunningTasks

protected

Definition at line 83 of file Executor.hxx.

Referenced by beginTask(), endTask(), Executor(), functionForTaskExecution(), getNumberOfRunningTasks(), launchTask(), RunA(), RunB(), runWlm(), and sleepWhileNoEventsFromAnyRunningTask().

_root

ComposedNode* YACS::ENGINE::Executor::_root

protected

Definition at line 66 of file Executor.hxx.

Referenced by Executor(), RunA(), RunB(), runWlm(), saveState(), and sendEvent().

_runningTasks

std::set<Task *> YACS::ENGINE::Executor::_runningTasks

protected

Definition at line 84 of file Executor.hxx.

Referenced by beginTask(), endTask(), filterTasksConsideringContainers(), functionForTaskExecution(), launchTask(), RunA(), RunB(), and runWlm().

_semForMaxThreads

YACS::BASES::Semaphore YACS::ENGINE::Executor::_semForMaxThreads

protected

Definition at line 70 of file Executor.hxx.

Referenced by functionForTaskExecution(), and launchTask().

_semThreadCnt

int YACS::ENGINE::Executor::_semThreadCnt

protected

Definition at line 86 of file Executor.hxx.

Referenced by Executor(), functionForTaskExecution(), and launchTask().

_start

std::chrono::steady_clock::time_point YACS::ENGINE::Executor::_start

protected

Definition at line 98 of file Executor.hxx.

Referenced by RunB(), runWlm(), and traceExec().

_stopOnErrorRequested

bool YACS::ENGINE::Executor::_stopOnErrorRequested

protected

Definition at line 78 of file Executor.hxx.

Referenced by endTask(), Executor(), functionForTaskExecution(), setStopOnError(), and unsetStopOnError().

_tasks

std::vector<Task *> YACS::ENGINE::Executor::_tasks

protected

Definition at line 91 of file Executor.hxx.

Referenced by checkBreakPoints(), loadParallelTasks(), RunB(), runWlm(), and setStepsToExecute().

_tasksSave

std::vector<Task *> YACS::ENGINE::Executor::_tasksSave

protected

Definition at line 92 of file Executor.hxx.

Referenced by checkBreakPoints(), and setStepsToExecute().

_threadStackSize

size_t Executor::_threadStackSize

static

Definition at line 132 of file Executor.hxx.

Referenced by launchTask(), loadParallelTasks(), and YACS::ENGINE::Runtime::Runtime().

_toContinue

bool YACS::ENGINE::Executor::_toContinue

protected

Definition at line 76 of file Executor.hxx.

Referenced by Executor(), isNotFinished(), RunA(), RunB(), runWlm(), and suspendASAP().

_trace

std::ofstream YACS::ENGINE::Executor::_trace

protected

Definition at line 93 of file Executor.hxx.

Referenced by RunB(), runWlm(), and traceExec().

---

The documentation for this class was generated from the following files:

• src/engine/Executor.hxx
• src/engine/Executor.cxx