source: src/Actions/ActionQueue.cpp@ 9a4949

/*
 * 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 <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),
    CurrentAction(0),
#ifndef HAVE_ACTION_THREAD
    lastActionOk(true)
#else
    lastActionOk(true),
    run_thread_isIdle(true),
    run_thread_running(false),
    run_thread(boost::bind(&ActionQueue::run, this))
#endif
{
  // 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();
#endif

  clearQueue();

  delete history;
  delete AR;
}

void ActionQueue::queueAction(const std::string &name, enum Action::QueryOptions state)
{
  const Action * const 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 {
    newaction->call();
    lastActionOk = true;
  } catch(ActionFailureException &e) {
    std::cerr << "Action " << *boost::get_error_info<ActionNameString>(e) << " has failed." << std::endl;
    World::getInstance().setExitFlag(5);
    clearQueue();
    lastActionOk = false;
    std::cerr << "ActionQueue cleared." << std::endl;
  } catch (std::exception &e) {
    pushStatus("FAIL: General exception caught, aborting.");
    World::getInstance().setExitFlag(134);
    clearQueue();
    lastActionOk = false;
    std::cerr << "ActionQueue cleared." << std::endl;
  }
  if (lastActionOk) {
    OBSERVE;
    NOTIFY(ActionQueued);
    _lastchangedaction = newaction;
  }
#else
  const bool new_run_thread_isIdle = (CurrentAction == actionqueue.size());
  mtx_queue.unlock();
  {
    boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
    run_thread_isIdle = new_run_thread_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 );
  {
    boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
    run_thread_isIdle = !((CurrentAction != actionqueue.size()) || !tempqueue.empty());
  }
  mtx_queue.unlock();
#endif
}

#ifdef HAVE_ACTION_THREAD
void ActionQueue::run()
{
  {
    boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
    run_thread_running = true;
  }
  bool Interrupted = false;
  do {
    // sleep for some time and wait for queue to fill up again
    try {
#if BOOST_VERSION < 105000
      boost::this_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();
    bool status = (CurrentAction != actionqueue.size());
    mtx_queue.unlock();
    while ((status) && (!Interrupted)) {
      //      boost::this_thread::disable_interruption di;
      // access actionqueue, hence using mutex
      mtx_queue.lock();
      LOG(0, "Calling Action " << actionqueue[CurrentAction]->getName() << " ... ");
      try {
        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();
        lastActionOk = false;
        std::cerr << "ActionQueue cleared." << std::endl;
        CurrentAction = (size_t)-1;
      } catch (std::exception &e) {
        pushStatus("FAIL: General exception caught, aborting.");
        World::getInstance().setExitFlag(134);
        clearQueue();
        std::cerr << "ActionQueue cleared." << std::endl;
        CurrentAction = (size_t)-1;
      }
      // remember action we juse executed
      const Action *lastaction = actionqueue[CurrentAction];
      // step on to next action and check for end
      CurrentAction++;
      // 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();
      status = (CurrentAction != actionqueue.size());
      // set last action
      if (lastActionOk) {
        OBSERVE;
        NOTIFY(ActionQueued);
        _lastchangedaction = lastaction;
        // unlock before we delve into Observer functions ...
        mtx_queue.unlock();
      } else {
        mtx_queue.unlock();
      }
    }
    mtx_queue.lock();
    const bool new_run_thread_isIdle = !((CurrentAction != actionqueue.size()) || !tempqueue.empty());
    mtx_queue.unlock();
    {
      boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
      run_thread_isIdle = new_run_thread_isIdle;
    }
    cond_idle.notify_one();
//    LOG(1, "DEBUG: End of ActionQueue's run() loop.");
  } while (!Interrupted);
  {
    boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
    run_thread_running = false;
  }
}
#endif

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

#ifdef HAVE_ACTION_THREAD
void ActionQueue::wait()
{
  boost::unique_lock<boost::mutex> lock(mtx_run_thread_isIdle);
  if (run_thread_running)
    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

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 * const 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()
{
  // free all actions contained in actionqueue
  for (ActionQueue_t::iterator iter = actionqueue.begin();
      !actionqueue.empty(); iter = actionqueue.begin()) {
    delete *iter;
    actionqueue.erase(iter);
  }
  // free all actions contained in tempqueue
  for (ActionQueue_t::iterator iter = tempqueue.begin();
      !tempqueue.empty(); iter = tempqueue.begin()) {
    delete *iter;
    tempqueue.erase(iter);
  }
#ifdef HAVE_ACTION_THREAD
  {
    boost::unique_lock<boost::mutex> lock(mtx_idle);
    run_thread_isIdle = true;
  }
#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();
}

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


CONSTRUCT_SINGLETON(ActionQueue)