ActiveObjectAce.cpp

Go to the documentation of this file.

/*
Copyright 2007, 2008, 2009, 2010, 2011 Instituto de Sistemas e Robotica, Instituto Superior Tecnico

This file is part of MeRMaID.

MeRMaID is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

MeRMaID is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with MeRMaID.  If not, see <http://www.gnu.org/licenses/>.
*/



/**
 * @Filename ActiveObjectAce.cpp
 * @Description ActiveObjectAce base implementation
 * @Status Work in Progress
 * @Version $Id: ActiveObjectAce.cpp 1 2011-03-04 18:13:18Z jreis $
 * @Maintainer nmsra (nmsra@mega.ist.utl.pt)
 */

#include "config.h"

#include <ActiveObjectAce.hpp>
#include <ActiveObjectAceAddTaskMethod.hpp>
#include <Exception.hpp>
#include <Time.hpp>

#include <iostream>

#include <ace/Date_Time.h>
#include <ace/Future.h>
#include <ace/Method_Object.h>

#include <stdexcept>

using namespace mermaid::support::activeobject;
using mermaid::support::errorhandling::Exception;
using mermaid::support::system::Time;



ActiveObjectAce::ActiveObjectAce (ActiveObjectManager * aom) : ActiveObject (aom), ACE_Task<ACE_MT_SYNCH>()
{
  // std::cerr << "ActiveObjectAce::ActiveObjectAce()(this=" << this << ")" << std::endl;
} // ActiveObjectAce()

ActiveObjectAce::~ActiveObjectAce()
{
  std::cerr << "ActiveObjectAce::~ActiveObjectAce()" << std::endl;
} // ~ActiveObjectAce()



void ActiveObjectAce::start()
{
  // std::cerr << "ActiveObjectAce::start()(this=" << this << ")" << std::endl;
  
  if (getState() != INITIALIZED) {
    throw Exception ("ActiveObjectAce::doShutdown() : To run start(), the ActiveObject must have state = INITIALIZED, and the actual state is " + getStateString());
  }
  
  setState (RUNNING);
  this->activate();
}



void ActiveObjectAce::stop()
{
  // std::cerr << "ActiveObjectAce::stop()(this=" << this << ")" << std::endl;
  
  if (getState() != RUNNING) {
    throw Exception ("ActiveObjectAce::doShutdown() : To run stop(), the ActiveObject must have state = RUNNING, and the actual state is " + getStateString());
  }
  
  setState (STOPPING);
  activationQueue_.queue()->close();
  this->wait();
};


//Private


int ActiveObjectAce::svc (void)
{
  Time lastTime = Time::getCurrentTime();
  
  try {
    while (1) {
      //std::cerr << std::endl;
      // calculate time until next object method should be run
      shared_ptr<Task> task = taskScheduler_->peakNextTask();
      
      if (task == false) {
        // no need to worry about times here
        // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : dequeuing (empty scheduler)" << std::endl;
        shared_ptr<ACE_Method_Object> mo = shared_ptr<ACE_Method_Object> (activationQueue_.dequeue());
        
        if (mo) {
          // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : calling method object (empty sxheduler)" << std::endl;
          {
            Time delta = Time::getCurrentTime() - lastTime;
            //std::cerr << "ActiveObjectAce::svc() : time since last run: sec=" << delta.sec() << " usec=" << delta.usec() << std::endl;
            lastTime = Time::getCurrentTime();
          }
          if (mo->call() == -1) {
            break;
          }
          // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : returned from call to method object (empty scheduler)" << std::endl;
        }
        else {
          // interruption occured, active object is shutting down
          // std::cerr << "ActiveObjectAce::svc : interruption occured, shutting down" << std::endl;
          setState (STOPPED);
          return 0;
        }
      }
      else {
        // let's see how much time we have to wait
        Time scheduleTime = task->getScheduledTime();
        // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : dequeuing (tasks in scheduler)" << std::endl;
        shared_ptr<ACE_Method_Object> mo = shared_ptr<ACE_Method_Object> (activationQueue_.dequeue ( (ACE_Time_Value*) & scheduleTime));
        
        //did a timeout occur?
        if (mo == false) {
          //check if timeout was because of active object shutdown
          if (getState() != RUNNING) {
            // std::cerr << "ActiveObjectAce::svc : interruption occured, shutting down" << std::endl;
            // std::cerr << "ActiveObjectAce::svc : remaining tasks will not be run." << std::endl;
            setState (STOPPED);
            return 0;
          }
          
          //it's a timeout, we have a task to execute
          shared_ptr<Task> t = taskScheduler_->getNextTask();
          // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : running scheduled task(" << t.get() << ")" << std::endl;
          {
            Time delta = Time::getCurrentTime() - lastTime;
            // std::cerr << "ActiveObjectAce::svc() : time since last run: sec=" << delta.sec() << " usec=" << delta.usec() << std::endl;
            lastTime = Time::getCurrentTime();
          }
          
          if (t->isOkToRun() == true) {
            try {
              t->run();
            }
            catch (Exception &e) {
              std::cerr << "ActiveObjectAce::svc : caught an Exception from a Task : " << e.what() << std::endl;
            }
          }
          else {
            std::cerr << "ActiveObjectAce::svc : task not ok to run" << std::endl;
          }
          // std::cerr << "ActiveObjectAce::svc() : scheduled task(" << t.get() << ") ended running" << std::endl;
          
          //if task is to be repeated, reschedule
          if (t->hasRepeatScheduling()) {
            // std::cerr << "ActiveObjectAce::svc() : task(" << t.get() << ") has repeat schedule, rescheduling it" << std::endl;
            t->incrementScheduledTimeForRepeat();
            this->addTaskInternal (t);
          }
          else {
            // std::cerr << "ActiveObjectAce::svc() : task(" << t.get() << ") IS NOT repeatable, forgetting it" << std::endl;
          }
        }
        else {
          // not a timeout, we got a method object to execute
          // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : calling method object (tasks in scheduler)" << std::endl;
          {
            Time delta = Time::getCurrentTime() - lastTime;
            //std::cerr << "ActiveObjectAce::svc() : time since last run: sec=" << delta.sec() << " usec=" << delta.usec() << std::endl;
            lastTime = Time::getCurrentTime();
          }
          if (mo->call() == -1) {
            break;
          }
          // std::cerr << std::endl << std::endl << "ActiveObjectAce::svc() : returned from call to method object (tasks in schedeuler)" << std::endl;
        } // else (timeout)
        
      } // else (valid task)
      
    } // the big while
  } // try
  catch (Exception &e) {
    std::cerr << "ActiveObjectAce::svc() : caught an Exception: " << e.what() << std::endl;
    
    setState (STOPPED);
    
    if (getState() != STOPPED) {
      std::cerr << "ActiveObjectAce::svc() : failed to set state to STOPPED" << std::endl;
    }
  }
  catch (std::runtime_error &e) {
    std::cerr << "ActiveObjectAce::svc() : caught a runtime_error exception: " << e.what() << std::endl;
    
    setState (STOPPED);
    
    if (getState() != STOPPED) {
      std::cerr << "ActiveObjectAce::svc() : failed to set state to STOPPED" << std::endl;
    }
  }
  catch (...) {
  
    //exit svc if something nasty happens...
    std::cerr << "ActiveObjectAce::svc() : something nasty happened" << std::endl;
    
    setState (STOPPED);
    
    if (getState() != STOPPED) {
      std::cerr << "ActiveObjectAce::svc() : failed to set state to STOPPED" << std::endl;
    }
    
    //std::cerr << "ActiveObjectAce(this=" << this << ").state_ = " << state_ << std::endl;
  }
  return 0;
};



void ActiveObjectAce::addTask (shared_ptr<Task> t)
{
  // std::cerr << "ActiveObjectAce::addTask(this=" << this << ", task=" << t.get() << ")" << std::endl;
  
  ACE_Future<shared_ptr<Exception> > resultantFuture;
  
  this->activationQueue_.enqueue (new ActiveObjectAceAddTaskMethod (this, t, resultantFuture));
  
  shared_ptr<Exception> e;
  // std::cerr << "ActiveObjectAce::addTask : waiting for add task result" << std::endl;
  //! @TODO IMPORTANT: uncommenting the following line causes segfaults and/or memory corruption. The line should be uncommented and the cause of the error should be found and corrected.
  // resultantFuture.get(e);
  
  if (e) {
    throw Exception ("ActiveObjectAce::addTask : task added UNSUCCESSFULY");
  }
  else {
    // std::cerr << "ActiveObjectAce::addTask : task added successfuly" << std::endl;
  }
  
}; // addTask()



void ActiveObjectAce::addTaskInternal (shared_ptr<Task> t)
{
  // std::cerr << "ActiveObjectAce::addTaskInternal(this=" << this << ", task=" << t.get() << ")" << std::endl;
  addTaskImplementation (t);
}; // addTaskInternal()



void ActiveObjectAce::addTaskImplementation (shared_ptr<Task> t)
{
  // std::cerr << "ActiveObjectAce::addTaskImplementation(this=" << this << ", task=" << t.get() << ")" << std::endl;
  ActiveObject::addTask (t);
}; // addTaskImplementation()