source: src/Actions/ActionQueue.cpp@ 0ea4dd

/*
 * Project: MoleCuilder
 * Description: creates and alters molecular systems
 * Copyright (C)  2013 Frederik Heber. All rights reserved.
 * 
 *
 *   This file is part of MoleCuilder.
 *
 *    MoleCuilder is free software: you can redistribute it and/or modify
 *    it under the terms of the GNU General Public License as published by
 *    the Free Software Foundation, either version 2 of the License, or
 *    (at your option) any later version.
 *
 *    MoleCuilder 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 General Public License for more details.
 *
 *    You should have received a copy of the GNU General Public License
 *    along with MoleCuilder.  If not, see <http://www.gnu.org/licenses/>.
 */

/*
 * ActionQueue.cpp
 *
 *  Created on: Aug 16, 2013
 *      Author: heber
 */

// include config.h
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include "CodePatterns/MemDebug.hpp"

#include "Actions/ActionQueue.hpp"

#include "CodePatterns/Assert.hpp"
#include "CodePatterns/IteratorAdaptors.hpp"
#include "CodePatterns/Log.hpp"
#include "CodePatterns/Singleton_impl.hpp"

#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/version.hpp>
#include <iterator>
#include <string>
#include <sstream>
#include <vector>

#include "Actions/ActionExceptions.hpp"
#include "Actions/ActionHistory.hpp"
#include "Actions/ActionRegistry.hpp"
#include "World.hpp"

using namespace MoleCuilder;

const Action* ActionQueue::_lastchangedaction = NULL;

ActionQueue::ActionQueue() :
    Observable("ActionQueue"),
    AR(new ActionRegistry()),
    history(new ActionHistory),
    lastActionOk(true),
    CurrentAction(0),
#ifdef HAVE_ACTION_THREAD
    run_thread(boost::bind(&ActionQueue::run, this)),
    run_thread_isIdle(true),
#endif
    dryrun_flag(false)
{
  // channels of observable
  Channels *OurChannel = new Channels;
  NotificationChannels.insert( std::make_pair(static_cast<Observable *>(this), OurChannel) );
  // add instance for each notification type
  for (size_t type = 0; type < NotificationType_MAX; ++type)
    OurChannel->addChannel(type);
}

ActionQueue::~ActionQueue()
{
#ifdef HAVE_ACTION_THREAD
  stop();

  clearTempQueue();
#endif

  clearQueue();

  delete history;
  delete AR;
}

void ActionQueue::queueAction(const std::string &name, enum Action::QueryOptions state)
{
  const Action & registryaction = AR->getActionByName(name);
  queueAction(&registryaction, state);
}

void ActionQueue::queueAction(const Action * const _action, enum Action::QueryOptions state)
{
  Action *newaction = _action->clone(state);
  newaction->prepare(state);
#ifdef HAVE_ACTION_THREAD
  mtx_queue.lock();
#endif
  actionqueue.push_back( newaction );
#ifndef HAVE_ACTION_THREAD
  try {
    if (!isDryRun(newaction)) {
      CurrentAction = actionqueue.size()-1;
      newaction->call();
      CurrentAction = actionqueue.size();
    }
    lastActionOk = true;
  } catch(ActionFailureException &e) {
    std::cerr << "Action " << *boost::get_error_info<ActionNameString>(e) << " has failed." << std::endl;
    World::getInstance().setExitFlag(5);
    clearQueue(actionqueue.size()-1);
    lastActionOk = false;
    std::cerr << "Remaining Actions cleared from queue." << std::endl;
  } catch (std::exception &e) {
    pushStatus("FAIL: General exception caught, aborting.");
    World::getInstance().setExitFlag(134);
    clearQueue(actionqueue.size()-1);
    lastActionOk = false;
    std::cerr << "Remaining Actions cleared from queue." << std::endl;
  }
  if (lastActionOk) {
    OBSERVE;
    NOTIFY(ActionQueued);
    _lastchangedaction = newaction;
  }
#else
  mtx_queue.unlock();
  setRunThreadIdle(isIdle());
#endif
}

void ActionQueue::insertAction(Action *_action, enum Action::QueryOptions state)
{
#ifndef HAVE_ACTION_THREAD
  queueAction(_action, state);
#else
  Action *newaction = _action->clone(state);
  newaction->prepare(state);
  mtx_queue.lock();
  tempqueue.push_back( newaction );
  const bool tempqueue_notempty = !tempqueue.empty();
  mtx_queue.unlock();
  setRunThreadIdle( !((!isIdle()) || tempqueue_notempty) );
#endif
}

bool ActionQueue::isIdle() const
{
#ifdef HAVE_ACTION_THREAD
  boost::unique_lock<boost::mutex> lock(mtx_queue);
#endif
  bool status = (CurrentAction == actionqueue.size());
  return status;
}

#ifdef HAVE_ACTION_THREAD
void ActionQueue::run()
{
  bool Interrupted = false;
  do {
    // sleep for some time and wait for queue to fill up again
    try {
#if BOOST_VERSION < 105000
      run_thread.sleep(boost::get_system_time() + boost::posix_time::milliseconds(100));
#else
      boost::this_thread::sleep_for(boost::chrono::milliseconds(100));
#endif
    } catch(boost::thread_interrupted &e) {
      LOG(2, "INFO: ActionQueue has received stop signal.");
      Interrupted = true;
    }
//    LOG(1, "DEBUG: Start of ActionQueue's run() loop.");
    // call all currently present Actions
    mtx_queue.lock();
    insertTempQueue();
    mtx_queue.unlock();
    bool status = !isIdle();
    while (status) {
      //      boost::this_thread::disable_interruption di;
      LOG(0, "Calling Action " << actionqueue[CurrentAction]->getName() << " ... ");
      try {
        if (!isDryRun(actionqueue[CurrentAction]))
          actionqueue[CurrentAction]->call();
        pushStatus("SUCCESS: Action "+actionqueue[CurrentAction]->getName()+" successful.");
        lastActionOk = true;
      } catch(ActionFailureException &e) {
        pushStatus("FAIL: Action "+*boost::get_error_info<ActionNameString>(e)+" has failed.");
        World::getInstance().setExitFlag(5);
        clearQueue(CurrentAction);
        clearTempQueue();
        lastActionOk = false;
        std::cerr << "Remaining Actions cleared from queue." << std::endl;
      } catch (std::exception &e) {
        pushStatus("FAIL: General exception caught, aborting.");
        World::getInstance().setExitFlag(134);
        clearQueue(CurrentAction);
        clearTempQueue();
        lastActionOk = false;
        std::cerr << "Remaining Actions cleared from queue." << std::endl;
      }
      if (lastActionOk) {
        OBSERVE;
        NOTIFY(ActionQueued);
        _lastchangedaction = actionqueue[CurrentAction];
        mtx_queue.lock();
        CurrentAction++;
        mtx_queue.unlock();
      }
      // access actionqueue, hence using mutex
      mtx_queue.lock();
      // insert new actions (before [CurrentAction]) if they have been spawned
      // we must have an extra vector for this, as we cannot change actionqueue
      // while an action instance is "in-use"
      insertTempQueue();
      mtx_queue.unlock();
      status = !isIdle();
    }
    mtx_queue.lock();
    const bool tempqueue_notempty = !tempqueue.empty();
    mtx_queue.unlock();
    setRunThreadIdle( !((!isIdle()) || tempqueue_notempty) );
    cond_idle.notify_one();
//    LOG(1, "DEBUG: End of ActionQueue's run() loop.");
  } while (!Interrupted);
}

void ActionQueue::insertTempQueue()
{
  if (!tempqueue.empty()) {
    ActionQueue_t::iterator InsertionIter = actionqueue.begin();
    std::advance(InsertionIter, CurrentAction);
    actionqueue.insert( InsertionIter, tempqueue.begin(), tempqueue.end() );
    tempqueue.clear();
  }
}

void ActionQueue::wait()
{
  boost::unique_lock<boost::mutex> lock(mtx_idle);
  while(!run_thread_isIdle)
  {
      cond_idle.wait(lock);
  }
}
#endif

#ifdef HAVE_ACTION_THREAD
void ActionQueue::stop()
{
  // notify actionqueue thread that we wish to terminate
  run_thread.interrupt();
  // wait till it ends
  run_thread.join();
}
#endif

const Action& ActionQueue::getActionByName(const std::string &name)
{
  return AR->getActionByName(name);
}

bool ActionQueue::isActionKnownByName(const std::string &name) const
{
  return AR->isActionPresentByName(name);
}

void ActionQueue::registerAction(Action *_action)
{
  AR->registerInstance(_action);
}

void ActionQueue::outputAsCLI(std::ostream &output) const
{
  for (ActionQueue_t::const_iterator iter = actionqueue.begin();
      iter != actionqueue.end();
      ++iter) {
    // skip store-session in printed list
    if ( ((*iter)->getName() != std::string("store-session"))
        && ((*iter)->getName() != std::string("load-session"))) {
      if (iter != actionqueue.begin())
        output << " ";
      (*iter)->outputAsCLI(output);
    }
  }
  output << std::endl;
}

void ActionQueue::outputAsPython(std::ostream &output) const
{
  const std::string prefix("pyMoleCuilder");
  output << "import " << prefix << std::endl;
  output << "# ========================== Stored Session BEGIN ==========================" << std::endl;
  for (ActionQueue_t::const_iterator iter = actionqueue.begin();
      iter != actionqueue.end();
      ++iter) {
    // skip store-session in printed list
    if ( ((*iter)->getName() != std::string("store-session"))
        && ((*iter)->getName() != std::string("load-session")))
      (*iter)->outputAsPython(output, prefix);
  }
  output << "# =========================== Stored Session END ===========================" << std::endl;
}

const ActionTrait& ActionQueue::getActionsTrait(const std::string &name) const
{
  // this const_cast is just required as long as we have a non-const getActionByName
  const Action & action = const_cast<ActionQueue *>(this)->getActionByName(name);
  return action.Traits;
}

void ActionQueue::addElement(Action* _Action,ActionState::ptr _state)
{
  history->addElement(_Action, _state);
}

void ActionQueue::clear()
{
  history->clear();
}

void ActionQueue::clearQueue(const size_t _fromAction)
{
#ifdef HAVE_ACTION_THREAD
  mtx_queue.lock();
#endif
  LOG(1, "Removing all Actions from position " << _fromAction << " onward.");
  // free all actions still to be called contained in actionqueue
  ActionQueue_t::iterator inititer = actionqueue.begin();
  std::advance(inititer, _fromAction);
  for (ActionQueue_t::iterator iter = inititer; iter != actionqueue.end(); ++iter)
    delete *iter;
  actionqueue.erase(inititer, actionqueue.end());
  LOG(1, "There are " << actionqueue.size() << " remaining Actions.");
#ifdef HAVE_ACTION_THREAD
  CurrentAction = actionqueue.size();
  mtx_queue.unlock();
#endif
}

#ifdef HAVE_ACTION_THREAD
void ActionQueue::clearTempQueue()
{
  // free all actions contained in tempqueue
  for (ActionQueue_t::iterator iter = tempqueue.begin();
      !tempqueue.empty(); iter = tempqueue.begin()) {
    delete *iter;
    tempqueue.erase(iter);
  }
}

void ActionQueue::setRunThreadIdle(const bool _flag)
{
  {
    boost::unique_lock<boost::mutex> lock(mtx_idle);
    run_thread_isIdle = _flag;
  }
}
#endif

const ActionQueue::ActionTokens_t ActionQueue::getListOfActions() const
{
  ActionTokens_t returnlist;

  returnlist.insert(
      returnlist.end(),
      MapKeyConstIterator<ActionRegistry::const_iterator>(AR->getBeginIter()),
      MapKeyConstIterator<ActionRegistry::const_iterator>(AR->getEndIter()));

  return returnlist;
}

void ActionQueue::undoLast()
{
        history->undoLast();
}

bool ActionQueue::canUndo() const
{
  return history->hasUndo();
}

void ActionQueue::redoLast()
{
        history->redoLast();
}

bool ActionQueue::canRedo() const
{
  return history->hasRedo();
}

bool ActionQueue::isDryRun(const Action *_nextaction) const
{
  bool status = dryrun_flag;
  status &= (_nextaction->getName() != "no-dry-run");
  return status;
}

CONSTRUCT_SINGLETON(ActionQueue)