source: src/Actions/ActionQueue.cpp@ 23b6cf

Last change on this file since 23b6cf was 23b6cf, checked in by Frederik Heber <heber@…>, 11 years ago

THREADFIX: Fixed ActionQueue with respect to mutex and threads.

  • run_thread is now last member variable, hence started after all other variables are initialized.
  • separated accesses to queues and to run_thread_isIdle flag. Also renamed its mutex for easier association with it.
  • Property mode set to 100644
File size: 10.5 KB
Line 
1/*
2 * Project: MoleCuilder
3 * Description: creates and alters molecular systems
4 * Copyright (C) 2013 Frederik Heber. All rights reserved.
5 *
6 *
7 * This file is part of MoleCuilder.
8 *
9 * MoleCuilder is free software: you can redistribute it and/or modify
10 * it under the terms of the GNU General Public License as published by
11 * the Free Software Foundation, either version 2 of the License, or
12 * (at your option) any later version.
13 *
14 * MoleCuilder is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 * GNU General Public License for more details.
18 *
19 * You should have received a copy of the GNU General Public License
20 * along with MoleCuilder. If not, see <http://www.gnu.org/licenses/>.
21 */
22
23/*
24 * ActionQueue.cpp
25 *
26 * Created on: Aug 16, 2013
27 * Author: heber
28 */
29
30// include config.h
31#ifdef HAVE_CONFIG_H
32#include <config.h>
33#endif
34
35#include "CodePatterns/MemDebug.hpp"
36
37#include "Actions/ActionQueue.hpp"
38
39#include "CodePatterns/Assert.hpp"
40#include "CodePatterns/IteratorAdaptors.hpp"
41#include "CodePatterns/Log.hpp"
42#include "CodePatterns/Singleton_impl.hpp"
43
44#include <boost/date_time/posix_time/posix_time.hpp>
45#include <boost/version.hpp>
46#include <string>
47#include <sstream>
48#include <vector>
49
50#include "Actions/ActionExceptions.hpp"
51#include "Actions/ActionHistory.hpp"
52#include "Actions/ActionRegistry.hpp"
53#include "World.hpp"
54
55using namespace MoleCuilder;
56
57const Action* ActionQueue::_lastchangedaction = NULL;
58
59ActionQueue::ActionQueue() :
60 Observable("ActionQueue"),
61 AR(new ActionRegistry()),
62 history(new ActionHistory),
63 CurrentAction(0),
64#ifndef HAVE_ACTION_THREAD
65 lastActionOk(true)
66#else
67 lastActionOk(true),
68 run_thread_isIdle(true),
69 run_thread(boost::bind(&ActionQueue::run, this))
70#endif
71{
72 // channels of observable
73 Channels *OurChannel = new Channels;
74 NotificationChannels.insert( std::make_pair(static_cast<Observable *>(this), OurChannel) );
75 // add instance for each notification type
76 for (size_t type = 0; type < NotificationType_MAX; ++type)
77 OurChannel->addChannel(type);
78}
79
80ActionQueue::~ActionQueue()
81{
82#ifdef HAVE_ACTION_THREAD
83 stop();
84#endif
85
86 clearQueue();
87
88 delete history;
89 delete AR;
90}
91
92void ActionQueue::queueAction(const std::string &name, enum Action::QueryOptions state)
93{
94 const Action * const registryaction = AR->getActionByName(name);
95 queueAction(registryaction, state);
96}
97
98void ActionQueue::queueAction(const Action * const _action, enum Action::QueryOptions state)
99{
100 Action *newaction = _action->clone(state);
101 newaction->prepare(state);
102#ifdef HAVE_ACTION_THREAD
103 mtx_queue.lock();
104#endif
105 actionqueue.push_back( newaction );
106#ifndef HAVE_ACTION_THREAD
107 try {
108 newaction->call();
109 lastActionOk = true;
110 } catch(ActionFailureException &e) {
111 std::cerr << "Action " << *boost::get_error_info<ActionNameString>(e) << " has failed." << std::endl;
112 World::getInstance().setExitFlag(5);
113 clearQueue();
114 lastActionOk = false;
115 std::cerr << "ActionQueue cleared." << std::endl;
116 } catch (std::exception &e) {
117 pushStatus("FAIL: General exception caught, aborting.");
118 World::getInstance().setExitFlag(134);
119 clearQueue();
120 lastActionOk = false;
121 std::cerr << "ActionQueue cleared." << std::endl;
122 }
123 if (lastActionOk) {
124 OBSERVE;
125 NOTIFY(ActionQueued);
126 _lastchangedaction = newaction;
127 }
128#else
129 const bool new_run_thread_isIdle = (CurrentAction == actionqueue.size());
130 mtx_queue.unlock();
131 {
132 boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
133 run_thread_isIdle = new_run_thread_isIdle;
134 }
135#endif
136}
137
138void ActionQueue::insertAction(Action *_action, enum Action::QueryOptions state)
139{
140#ifndef HAVE_ACTION_THREAD
141 queueAction(_action, state);
142#else
143 Action *newaction = _action->clone(state);
144 newaction->prepare(state);
145 mtx_queue.lock();
146 tempqueue.push_back( newaction );
147 {
148 boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
149 run_thread_isIdle = !((CurrentAction != actionqueue.size()) || !tempqueue.empty());
150 }
151 mtx_queue.unlock();
152#endif
153}
154
155#ifdef HAVE_ACTION_THREAD
156void ActionQueue::run()
157{
158 bool Interrupted = false;
159 do {
160 // sleep for some time and wait for queue to fill up again
161 try {
162#if BOOST_VERSION < 105000
163 boost::this_thread::sleep(boost::get_system_time() + boost::posix_time::milliseconds(100));
164#else
165 boost::this_thread::sleep_for(boost::chrono::milliseconds(100));
166#endif
167 } catch(boost::thread_interrupted &e) {
168 LOG(2, "INFO: ActionQueue has received stop signal.");
169 Interrupted = true;
170 }
171// LOG(1, "DEBUG: Start of ActionQueue's run() loop.");
172 // call all currently present Actions
173 mtx_queue.lock();
174 insertTempQueue();
175 bool status = (CurrentAction != actionqueue.size());
176 mtx_queue.unlock();
177 while (status) {
178 // boost::this_thread::disable_interruption di;
179 // access actionqueue, hence using mutex
180 mtx_queue.lock();
181 LOG(0, "Calling Action " << actionqueue[CurrentAction]->getName() << " ... ");
182 try {
183 actionqueue[CurrentAction]->call();
184 pushStatus("SUCCESS: Action "+actionqueue[CurrentAction]->getName()+" successful.");
185 lastActionOk = true;
186 } catch(ActionFailureException &e) {
187 pushStatus("FAIL: Action "+*boost::get_error_info<ActionNameString>(e)+" has failed.");
188 World::getInstance().setExitFlag(5);
189 clearQueue();
190 lastActionOk = false;
191 std::cerr << "ActionQueue cleared." << std::endl;
192 CurrentAction = (size_t)-1;
193 } catch (std::exception &e) {
194 pushStatus("FAIL: General exception caught, aborting.");
195 World::getInstance().setExitFlag(134);
196 clearQueue();
197 std::cerr << "ActionQueue cleared." << std::endl;
198 CurrentAction = (size_t)-1;
199 }
200 // remember action we juse executed
201 const Action *lastaction = actionqueue[CurrentAction];
202 // step on to next action and check for end
203 CurrentAction++;
204 // insert new actions (before [CurrentAction]) if they have been spawned
205 // we must have an extra vector for this, as we cannot change actionqueue
206 // while an action instance is "in-use"
207 insertTempQueue();
208 status = (CurrentAction != actionqueue.size());
209 // set last action
210 if (lastActionOk) {
211 OBSERVE;
212 NOTIFY(ActionQueued);
213 _lastchangedaction = lastaction;
214 // unlock before we delve into Observer functions ...
215 mtx_queue.unlock();
216 } else {
217 mtx_queue.unlock();
218 }
219 }
220 mtx_queue.lock();
221 const bool new_run_thread_isIdle = !((CurrentAction != actionqueue.size()) || !tempqueue.empty());
222 mtx_queue.unlock();
223 {
224 boost::lock_guard<boost::mutex> lock(mtx_run_thread_isIdle);
225 run_thread_isIdle = new_run_thread_isIdle;
226 }
227 cond_idle.notify_one();
228// LOG(1, "DEBUG: End of ActionQueue's run() loop.");
229 } while (!Interrupted);
230}
231#endif
232
233void ActionQueue::insertTempQueue()
234{
235 if (!tempqueue.empty()) {
236 ActionQueue_t::iterator InsertionIter = actionqueue.begin();
237 std::advance(InsertionIter, CurrentAction);
238 actionqueue.insert( InsertionIter, tempqueue.begin(), tempqueue.end() );
239 tempqueue.clear();
240 }
241}
242
243#ifdef HAVE_ACTION_THREAD
244void ActionQueue::wait()
245{
246 boost::unique_lock<boost::mutex> lock(mtx_run_thread_isIdle);
247 while(!run_thread_isIdle)
248 {
249 cond_idle.wait(lock);
250 }
251}
252#endif
253
254#ifdef HAVE_ACTION_THREAD
255void ActionQueue::stop()
256{
257 // notify actionqueue thread that we wish to terminate
258 run_thread.interrupt();
259 // wait till it ends
260 run_thread.join();
261}
262#endif
263
264Action* ActionQueue::getActionByName(const std::string &name)
265{
266 return AR->getActionByName(name);
267}
268
269bool ActionQueue::isActionKnownByName(const std::string &name) const
270{
271 return AR->isActionPresentByName(name);
272}
273
274void ActionQueue::registerAction(Action *_action)
275{
276 AR->registerInstance(_action);
277}
278
279void ActionQueue::outputAsCLI(std::ostream &output) const
280{
281 for (ActionQueue_t::const_iterator iter = actionqueue.begin();
282 iter != actionqueue.end();
283 ++iter) {
284 // skip store-session in printed list
285 if ( ((*iter)->getName() != std::string("store-session"))
286 && ((*iter)->getName() != std::string("load-session"))) {
287 if (iter != actionqueue.begin())
288 output << " ";
289 (*iter)->outputAsCLI(output);
290 }
291 }
292 output << std::endl;
293}
294
295void ActionQueue::outputAsPython(std::ostream &output) const
296{
297 const std::string prefix("pyMoleCuilder");
298 output << "import " << prefix << std::endl;
299 output << "# ========================== Stored Session BEGIN ==========================" << std::endl;
300 for (ActionQueue_t::const_iterator iter = actionqueue.begin();
301 iter != actionqueue.end();
302 ++iter) {
303 // skip store-session in printed list
304 if ( ((*iter)->getName() != std::string("store-session"))
305 && ((*iter)->getName() != std::string("load-session")))
306 (*iter)->outputAsPython(output, prefix);
307 }
308 output << "# =========================== Stored Session END ===========================" << std::endl;
309}
310
311const ActionTrait& ActionQueue::getActionsTrait(const std::string &name) const
312{
313 // this const_cast is just required as long as we have a non-const getActionByName
314 const Action * const action = const_cast<ActionQueue *>(this)->getActionByName(name);
315 return action->Traits;
316}
317
318void ActionQueue::addElement(Action* _Action,ActionState::ptr _state)
319{
320 history->addElement(_Action, _state);
321}
322
323void ActionQueue::clear()
324{
325 history->clear();
326}
327
328void ActionQueue::clearQueue()
329{
330 // free all actions contained in actionqueue
331 for (ActionQueue_t::iterator iter = actionqueue.begin();
332 !actionqueue.empty(); iter = actionqueue.begin()) {
333 delete *iter;
334 actionqueue.erase(iter);
335 }
336 // free all actions contained in tempqueue
337 for (ActionQueue_t::iterator iter = tempqueue.begin();
338 !tempqueue.empty(); iter = tempqueue.begin()) {
339 delete *iter;
340 tempqueue.erase(iter);
341 }
342#ifdef HAVE_ACTION_THREAD
343 {
344 boost::unique_lock<boost::mutex> lock(mtx_idle);
345 run_thread_isIdle = true;
346 }
347#endif
348}
349
350const ActionQueue::ActionTokens_t ActionQueue::getListOfActions() const
351{
352 ActionTokens_t returnlist;
353
354 returnlist.insert(
355 returnlist.end(),
356 MapKeyConstIterator<ActionRegistry::const_iterator>(AR->getBeginIter()),
357 MapKeyConstIterator<ActionRegistry::const_iterator>(AR->getEndIter()));
358
359 return returnlist;
360}
361
362void ActionQueue::undoLast()
363{
364 history->undoLast();
365}
366
367void ActionQueue::redoLast()
368{
369 history->redoLast();
370}
371
372
373CONSTRUCT_SINGLETON(ActionQueue)
Note: See TracBrowser for help on using the repository browser.