logic: reimplemented task system to abort ast visiting

* reimplemented interrupting TaskParseCxx by adding a listener for the MessageInterruptTask and returning a failure status code on update. This cancels the parent sequence task which results in the indexed items not getting inserted into the persistent storage
* removed the capability for interrupting from TaskScheduler
* changed task system to be closer to the standard behavior tree implementation
* changed task system to accommodate the 3 status return types: Running, Success and Failure
* made TaskGroupSequential fail once a member task fails
* made TaskGroupParallel fail once a member task fails
* split TaskParse... into one task for indexing and one task for injecting
* added TaskRunner that handles updating and resetting the managed task
* fixed numbers that are shown as parsed file count in indexing ui
* fixed deadlock that originated from interaction between TaskScheduler and MessageQueue (one thread wanted to destroy a message listener on a task while the other one wanted to send as message as a task)

src/lib/utility/scheduling/Task.cpp

@@ -14,7 +14,8 @@ void Task::dispatchNext(std::shared_ptr<Task> task)
 }
 
 Task::Task()
-	: m_state(STATE_NEW)
+	: m_enterCalled(false)
+	, m_exitCalled(false)
 {
 }
 
@@ -22,74 +23,28 @@ Task::~Task()
 {
 }
 
-Task::TaskState Task::getState() const
+Task::TaskState Task::update()
 {
-	return m_state;
-}
-
-Task::TaskState Task::processTask()
-{
-	switch (m_state)
+	if (!m_enterCalled)
 	{
-	case STATE_NEW:
-	case STATE_CANCELED:
-		enter();
-	case STATE_RUNNING:
-		{
-			TaskState newState = update();
-			if (newState == STATE_NEW || newState == STATE_CANCELED)
-			{
-				LOG_ERROR("Task can't change to state NEW or CANCELLED");
-				return m_state;
-			}
-
-			setState(newState);
-			if (m_state == STATE_FINISHED)
-			{
-				exit();
-			}
-		}
-		break;
-	case STATE_FINISHED:
-		break;
+		doEnter();
+		m_enterCalled = true;
 	}
 
-	return m_state;
-}
+	TaskState state = doUpdate();
 
-Task::TaskState Task::interruptTask()
-{
-	switch (m_state)
+	if (state != STATE_RUNNING && !m_exitCalled)
 	{
-	case STATE_NEW:
-		abort();
-		break;
-	case STATE_CANCELED:
-		break;
-	case STATE_RUNNING:
-		interrupt();
-		exit();
-		break;
-	case STATE_FINISHED:
-		revert();
-		break;
+		doExit();
+		m_exitCalled = true;
 	}
 
-	setState(STATE_CANCELED);
-	return m_state;
+	return state;
 }
 
-void Task::executeTask()
+void Task::reset()
 {
-	TaskState state;
-	do
-	{
-		state = processTask();
-	}
-	while (state != STATE_FINISHED);
-}
-
-void Task::setState(TaskState state)
-{
-	m_state = state;
+	doReset();
+	m_enterCalled = false;
+	m_exitCalled = false;
 }

src/lib/utility/scheduling/Task.h

@@ -8,10 +8,9 @@ class Task
 public:
 	enum TaskState
 	{
-		STATE_NEW,
 		STATE_RUNNING,
-		STATE_FINISHED,
-		STATE_CANCELED
+		STATE_SUCCESS,
+		STATE_FAILURE
 	};
 
 	static void dispatch(std::shared_ptr<Task> task);
@@ -20,26 +19,19 @@ public:
 	Task();
 	virtual ~Task();
 
-	TaskState getState() const;
+//	virtual TaskState getState() const = 0;
 
-	TaskState processTask();
-	TaskState interruptTask();
-
-	void executeTask();
-
-	virtual void enter() = 0;
-	virtual TaskState update() = 0;
-	virtual void exit() = 0;
-
-	virtual void interrupt() = 0;
-	virtual void revert() = 0;
-	virtual void abort() = 0;
-
-protected:
-	void setState(TaskState state);
+	TaskState update();
+	void reset();
 
 private:
-	TaskState m_state;
+	virtual void doEnter() = 0;
+	virtual Task::TaskState doUpdate() = 0;
+	virtual void doExit() = 0;
+	virtual void doReset() = 0;
+
+	bool m_enterCalled;
+	bool m_exitCalled;
 };
 
 #endif // TASK_H

src/lib/utility/scheduling/TaskDecorator.cpp

@@ -7,8 +7,3 @@ TaskDecorator::TaskDecorator()
 TaskDecorator::~TaskDecorator()
 {
 }
-
-void TaskDecorator::setTask(std::shared_ptr<Task> task)
-{
-	m_task = task;
-}

src/lib/utility/scheduling/TaskDecorator.h

@@ -12,10 +12,7 @@ public:
 	TaskDecorator();
 	virtual ~TaskDecorator();
 
-	void setTask(std::shared_ptr<Task> task);
-
-protected:
-	std::shared_ptr<Task> m_task;
+	virtual void setTask(std::shared_ptr<Task> task) = 0;
 };
 
 #endif // TASK_DECORATOR_H

src/lib/utility/scheduling/TaskGroup.cpp

@@ -7,8 +7,3 @@ TaskGroup::TaskGroup()
 TaskGroup::~TaskGroup()
 {
 }
-
-void TaskGroup::addTask(std::shared_ptr<Task> task)
-{
-	m_tasks.push_back(task);
-}

src/lib/utility/scheduling/TaskGroup.h

@@ -13,10 +13,7 @@ public:
 	TaskGroup();
 	virtual ~TaskGroup();
 
-	void addTask(std::shared_ptr<Task> task);
-
-protected:
-	std::vector<std::shared_ptr<Task>> m_tasks;
+	virtual void addTask(std::shared_ptr<Task> task) = 0;
 };
 
 #endif // TASK_GROUP_H

src/lib/utility/scheduling/TaskGroupParallel.cpp

@@ -1,6 +1,9 @@
 #include "utility/scheduling/TaskGroupParallel.h"
 
+#include "utility/ScopedFunctor.h"
+
 TaskGroupParallel::TaskGroupParallel()
+	: m_needsToStartThreads(true)
 {
 }
 
@@ -8,83 +11,88 @@ TaskGroupParallel::~TaskGroupParallel()
 {
 }
 
-void TaskGroupParallel::enter()
+void TaskGroupParallel::addTask(std::shared_ptr<Task> task)
 {
-	m_interrupt = false;
-	m_running = false;
-	m_activeTaskCount = 0;
+	m_tasks.push_back(std::make_shared<TaskInfo>(std::make_shared<TaskRunner>(task)));
 }
 
-Task::TaskState TaskGroupParallel::update()
+void TaskGroupParallel::doEnter()
 {
-	if (!m_running)
+	m_taskFailed = false;
+
+	if (m_needsToStartThreads)
 	{
+		m_needsToStartThreads = false;
+		m_activeTaskCount = 0;
 		for (size_t i = 0; i < m_tasks.size(); i++)
 		{
-			m_threads.push_back(std::thread(&TaskGroupParallel::processTaskThreaded, this, m_tasks[i]));
-
-			std::lock_guard<std::mutex> lock(m_activeTaskCountMutex);
+			m_tasks[i]->thread = std::make_shared<std::thread>(&TaskGroupParallel::processTaskThreaded, this, m_tasks[i]);
+			m_tasks[i]->active = true;
 			m_activeTaskCount++;
 		}
-		m_running = true;
+	}
+}
+
+Task::TaskState TaskGroupParallel::doUpdate()
+{
+	if (m_tasks.size() != 0 && getActveTaskCount() > 0)
+	{
+		return STATE_RUNNING;
 	}
 
-	int activeTaskCount = 0;
+	return (m_taskFailed ? STATE_FAILURE : STATE_SUCCESS);
+}
+
+void TaskGroupParallel::doExit()
+{
+	for (size_t i = 0; i < m_tasks.size(); i++)
 	{
-		std::lock_guard<std::mutex> lock(m_activeTaskCountMutex);
-		activeTaskCount = m_activeTaskCount;
+		m_tasks[i]->thread->join();
+		m_tasks[i]->thread.reset();
 	}
+}
 
-	if (activeTaskCount == 0)
+void TaskGroupParallel::doReset()
+{
+	for (size_t i = 0; i < m_tasks.size(); i++)
 	{
-		return (m_interrupt ? STATE_CANCELED : STATE_FINISHED);
-	}
-
-	return Task::STATE_RUNNING;
-}
-
-void TaskGroupParallel::exit()
-{
-	for (size_t i = 0; i < m_threads.size(); i++)
-	{
-		m_threads[i].join();
-	}
-	m_threads.clear();
-}
-
-void TaskGroupParallel::interrupt()
-{
-	m_interrupt = true;
-}
-
-void TaskGroupParallel::revert()
-{
-	m_interrupt = true;
-}
-
-void TaskGroupParallel::abort()
-{
-	m_interrupt = true;
-}
-
-
-void TaskGroupParallel::processTaskThreaded(std::shared_ptr<Task> task)
-{
-	Task::TaskState state = Task::STATE_NEW;
-	while (state != Task::STATE_FINISHED && state != Task::STATE_CANCELED)
-	{
-		if (m_interrupt)
+		m_tasks[i]->taskRunner->reset();
+		if (!m_tasks[i]->active)
 		{
-			state = task->interruptTask();
-		}
-		else
-		{
-			state = task->processTask();
+			m_tasks[i]->thread->join();
+			m_tasks[i]->thread = std::make_shared<std::thread>(&TaskGroupParallel::processTaskThreaded, this, m_tasks[i]);
+			m_tasks[i]->active = true;
+			m_activeTaskCount++;
 		}
 	}
+}
 
-	{
+void TaskGroupParallel::processTaskThreaded(std::shared_ptr<TaskInfo> taskInfo)
+{
+	ScopedFunctor functor([&](){
 		std::lock_guard<std::mutex> lock(m_activeTaskCountMutex);
-		m_activeTaskCount--; // not safe! if exception hits this thread before this point the count is not decremented.
+		m_activeTaskCount--;
+	});
+
+
+	while (true)
+	{
+		TaskState state = taskInfo->taskRunner->update();
+
+		if (state != STATE_RUNNING)
+		{
+			if (state == STATE_FAILURE)
+			{
+				m_taskFailed = true;
+			}
+			taskInfo->active = false;
+			break;
+		}
 	}
 }
+
+int TaskGroupParallel::getActveTaskCount() const
+{
+	std::lock_guard<std::mutex> lock(m_activeTaskCountMutex);
+	return m_activeTaskCount;
+}

src/lib/utility/scheduling/TaskGroupParallel.h

@@ -1,10 +1,12 @@
 #ifndef TASK_GROUP_PARALLEL_H
 #define TASK_GROUP_PARALLEL_H
 
+#include <map>
 #include <mutex>
 #include <thread>
 
 #include "utility/scheduling/TaskGroup.h"
+#include "utility/scheduling/TaskRunner.h"
 
 class TaskGroupParallel
 	: public TaskGroup
@@ -13,24 +15,34 @@ public:
 	TaskGroupParallel();
 	virtual ~TaskGroupParallel();
 
-	virtual void enter();
-	virtual TaskState update();
-	virtual void exit();
-
-	virtual void interrupt();
-	virtual void revert();
-	virtual void abort();
+	virtual void addTask(std::shared_ptr<Task> task);
 
 private:
-	void processTaskThreaded(std::shared_ptr<Task> task);
+	struct TaskInfo
+	{
+		TaskInfo(std::shared_ptr<TaskRunner> taskRunner)
+			: taskRunner(taskRunner)
+			, active(false)
+		{}
+		std::shared_ptr<TaskRunner> taskRunner;
+		std::shared_ptr<std::thread> thread;
+		volatile bool active;
+	};
 
-	volatile bool m_interrupt;
-	bool m_running;
+	virtual void doEnter();
+	virtual TaskState doUpdate();
+	virtual void doExit();
+	virtual void doReset();
 
-	std::vector<std::thread> m_threads;
+	void processTaskThreaded(std::shared_ptr<TaskInfo> taskInfo);
+	int getActveTaskCount() const;
 
+	std::vector<std::shared_ptr<TaskInfo>> m_tasks;
+	bool m_needsToStartThreads;
+
+	volatile bool m_taskFailed;
 	volatile int m_activeTaskCount;
-	std::mutex m_activeTaskCountMutex;
+	mutable std::mutex m_activeTaskCountMutex;
 };
 
 #endif // TASK_GROUP_PARALLEL_H

src/lib/utility/scheduling/TaskGroupSequential.cpp

@@ -1,7 +1,7 @@
 #include "utility/scheduling/TaskGroupSequential.h"
+#include <iostream>
 
 TaskGroupSequential::TaskGroupSequential()
-	: m_taskIndex(-1)
 {
 }
 
@@ -9,55 +9,49 @@ TaskGroupSequential::~TaskGroupSequential()
 {
 }
 
-void TaskGroupSequential::enter()
+void TaskGroupSequential::addTask(std::shared_ptr<Task> task)
 {
+	m_taskRunners.push_back(std::make_shared<TaskRunner>(task));
 }
 
-Task::TaskState TaskGroupSequential::update()
+void TaskGroupSequential::doEnter()
 {
-	if (!m_tasks.size())
+	m_taskIndex = 0;
+}
+
+Task::TaskState TaskGroupSequential::doUpdate()
+{
+	if (m_taskIndex >= int(m_taskRunners.size()))
 	{
-		return Task::STATE_FINISHED;
+		return STATE_SUCCESS;
+	}
+	else if (m_taskIndex < 0)
+	{
+		return STATE_FAILURE;
 	}
 
-	if (m_taskIndex < 0 || m_tasks[m_taskIndex]->getState() != Task::STATE_RUNNING)
+	TaskState state = m_taskRunners[m_taskIndex]->update();
+
+	if (state == STATE_SUCCESS)
 	{
 		m_taskIndex++;
 	}
-
-	std::shared_ptr<Task> task = m_tasks[m_taskIndex];
-
-	TaskState state = task->processTask();
-
-	if (state == Task::STATE_FINISHED && size_t(m_taskIndex + 1) == m_tasks.size())
+	else if (state == STATE_FAILURE)
 	{
-		return Task::STATE_FINISHED;
+		m_taskIndex = -1;
 	}
 
-	return Task::STATE_RUNNING;
+	return STATE_RUNNING;
 }
 
-void TaskGroupSequential::exit()
+void TaskGroupSequential::doExit()
 {
 }
 
-void TaskGroupSequential::interrupt()
+void TaskGroupSequential::doReset()
 {
-	for (size_t i = 0; i < m_tasks.size(); i++)
+	for (size_t i = 0; i < m_taskRunners.size(); i++)
 	{
-		m_tasks[i]->interruptTask();
+		m_taskRunners[i]->reset();
 	}
 }
-
-void TaskGroupSequential::revert()
-{
-	for (int i = m_tasks.size() - 1; i >= 0; i--)
-	{
-		m_tasks[i]->interruptTask();
-	}
-}
-
-void TaskGroupSequential::abort()
-{
-	interrupt();
-}

src/lib/utility/scheduling/TaskGroupSequential.h

@@ -2,6 +2,7 @@
 #define TASK_GROUP_SEQUENTIAL_H
 
 #include "utility/scheduling/TaskGroup.h"
+#include "utility/scheduling/TaskRunner.h"
 
 class TaskGroupSequential
 	: public TaskGroup
@@ -10,15 +11,15 @@ public:
 	TaskGroupSequential();
 	virtual ~TaskGroupSequential();
 
-	virtual void enter();
-	virtual TaskState update();
-	virtual void exit();
-
-	virtual void interrupt();
-	virtual void revert();
-	virtual void abort();
+	virtual void addTask(std::shared_ptr<Task> task);
 
 private:
+	virtual void doEnter();
+	virtual TaskState doUpdate();
+	virtual void doExit();
+	virtual void doReset();
+
+	std::vector<std::shared_ptr<TaskRunner>> m_taskRunners;
 	int m_taskIndex;
 };
 

src/lib/utility/scheduling/TaskLambda.cpp

@@ -9,28 +9,20 @@ TaskLambda::~TaskLambda()
 {
 }
 
-void TaskLambda::enter()
+void TaskLambda::doEnter()
 {
 }
 
-Task::TaskState TaskLambda::update()
+Task::TaskState TaskLambda::doUpdate()
 {
 	m_func();
-	return Task::STATE_FINISHED;
+	return STATE_SUCCESS;
 }
 
-void TaskLambda::exit()
+void TaskLambda::doExit()
 {
 }
 
-void TaskLambda::interrupt()
-{
-}
-
-void TaskLambda::revert()
-{
-}
-
-void TaskLambda::abort()
+void TaskLambda::doReset()
 {
 }

src/lib/utility/scheduling/TaskLambda.h

@@ -12,15 +12,12 @@ public:
 	TaskLambda(std::function<void()> func);
 	virtual ~TaskLambda();
 
-	virtual void enter();
-	virtual TaskState update();
-	virtual void exit();
-
-	virtual void interrupt();
-	virtual void revert();
-	virtual void abort();
-
 private:
+	virtual void doEnter();
+	virtual TaskState doUpdate();
+	virtual void doExit();
+	virtual void doReset();
+
 	std::function<void()> m_func;
 };
 

src/lib/utility/scheduling/TaskRunner.cpp

@@ -0,0 +1,34 @@
+#include "utility/scheduling/TaskRunner.h"
+
+TaskRunner::TaskRunner(std::shared_ptr<Task> task)
+	: m_task(task)
+	, m_reset(false)
+{
+}
+
+TaskRunner::~TaskRunner()
+{
+}
+
+//Task::TaskState TaskRunner::getState() const
+//{
+//	return m_task->getState();
+//}
+
+Task::TaskState TaskRunner::update()
+{
+	if (m_reset)
+	{
+		m_task->reset();
+		m_reset = false;
+	}
+
+	return m_task->update();
+}
+
+void TaskRunner::reset()
+{
+	m_reset = true;
+}
+
+

src/lib/utility/scheduling/TaskRunner.h

@@ -0,0 +1,24 @@
+#ifndef TASK_RUNNER_H
+#define TASK_RUNNER_H
+
+#include <memory>
+
+#include "utility/scheduling/Task.h"
+
+class TaskRunner
+{
+public:
+	TaskRunner(std::shared_ptr<Task> task);
+	~TaskRunner();
+
+	//Task::TaskState getState() const;
+
+	Task::TaskState update();
+	void reset();
+
+private:
+	std::shared_ptr<Task> m_task;
+	bool m_reset;
+};
+
+#endif // TASK_H

src/lib/utility/scheduling/TaskScheduler.cpp

@@ -19,29 +19,23 @@ std::shared_ptr<TaskScheduler> TaskScheduler::getInstance()
 void TaskScheduler::pushTask(std::shared_ptr<Task> task)
 {
 	std::lock_guard<std::mutex> lock(m_tasksMutex);
-	m_tasks.push_back(task);
+	m_taskRunners.push_back(std::make_shared<TaskRunner>(task));
 }
 
 void TaskScheduler::pushNextTask(std::shared_ptr<Task> task)
 {
 	std::lock_guard<std::mutex> lock(m_tasksMutex);
 
-	if (m_tasks.size() == 0)
+	if (m_taskRunners.size() == 0)
 	{
-		m_tasks.push_front(task);
+		m_taskRunners.push_front(std::make_shared<TaskRunner>(task));
 	}
 	else
 	{
-		m_tasks.insert(m_tasks.begin() + 1, task);
+		m_taskRunners.insert(m_taskRunners.begin() + 1, std::make_shared<TaskRunner>(task));
 	}
 }
 
-void TaskScheduler::interruptCurrentTask()
-{
-	std::lock_guard<std::mutex> lock(m_tasksMutex);
-	m_interruptTask = true;
-}
-
 void TaskScheduler::startSchedulerLoopThreaded()
 {
 	std::thread(&TaskScheduler::startSchedulerLoop, this).detach();
@@ -57,7 +51,7 @@ void TaskScheduler::startSchedulerLoop()
 
 		if (m_loopIsRunning)
 		{
-			LOG_ERROR("Loop is already running");
+			LOG_ERROR("Unable to start task scheduler. Loop is already running.");
 			return;
 		}
 
@@ -97,14 +91,12 @@ void TaskScheduler::stopSchedulerLoop()
 
 		if (!m_loopIsRunning)
 		{
-			LOG_WARNING("Loop is not running");
+			LOG_WARNING("Unable to stop task scheduler. Loop is not running.");
 		}
 
 		m_loopIsRunning = false;
 	}
 
-	interruptCurrentTask();
-
 	while (true)
 	{
 		{
@@ -129,7 +121,7 @@ bool TaskScheduler::loopIsRunning() const
 bool TaskScheduler::hasTasksQueued() const
 {
 	std::lock_guard<std::mutex> lock(m_tasksMutex);
-	return m_tasks.size();
+	return m_taskRunners.size();
 }
 
 std::shared_ptr<TaskScheduler> TaskScheduler::s_instance;
@@ -137,7 +129,6 @@ std::shared_ptr<TaskScheduler> TaskScheduler::s_instance;
 TaskScheduler::TaskScheduler()
 	: m_loopIsRunning(false)
 	, m_threadIsRunning(false)
-	, m_interruptTask(false)
 {
 }
 
@@ -145,41 +136,19 @@ void TaskScheduler::processTasks()
 {
 	std::lock_guard<std::mutex> lock(m_tasksMutex);
 
-	while (m_tasks.size())
+	while (m_taskRunners.size())
 	{
-		bool interrupt = m_interruptTask;
-		m_interruptTask = false;
-
-		std::shared_ptr<Task> task = m_tasks.front();
-		Task::TaskState state;
+		std::shared_ptr<TaskRunner> runner = m_taskRunners.front();
 
 		m_tasksMutex.unlock();
-		if (interrupt)
-		{
-			state = task->interruptTask();
-		}
-		else
-		{
-			state = task->processTask();
-		}
+
+		Task::TaskState state = runner->update();
+
 		m_tasksMutex.lock();
 
-		if (state == Task::STATE_FINISHED || state == Task::STATE_CANCELED)
+		if (state != Task::STATE_RUNNING)
 		{
-			m_tasks.pop_front();
+			m_taskRunners.pop_front();
 		}
 	}
-
-	m_interruptTask = false;
-}
-
-void TaskScheduler::handleMessage(MessageInterruptTasks* message)
-{
-	interruptCurrentTask();
-
-	std::lock_guard<std::mutex> lock(m_tasksMutex);
-	if (m_tasks.size())
-	{
-		MessageStatus("Stop running tasks...", false, true).dispatch();
-	}
 }

src/lib/utility/scheduling/TaskScheduler.h

@@ -8,16 +8,15 @@
 #include "utility/messaging/MessageListener.h"
 #include "utility/messaging/type/MessageInterruptTasks.h"
 #include "utility/scheduling/Task.h"
+#include "utility/scheduling/TaskRunner.h"
 
 class TaskScheduler
-	: public MessageListener<MessageInterruptTasks>
 {
 public:
 	static std::shared_ptr<TaskScheduler> getInstance();
 
 	void pushTask(std::shared_ptr<Task> task);
 	void pushNextTask(std::shared_ptr<Task> task);
-	void interruptCurrentTask();
 
 	void startSchedulerLoopThreaded();
 	void startSchedulerLoop();
@@ -35,13 +34,11 @@ private:
 
 	void processTasks();
 
-	virtual void handleMessage(MessageInterruptTasks* message);
 
 	bool m_loopIsRunning;
 	bool m_threadIsRunning;
 
-	std::deque<std::shared_ptr<Task>> m_tasks;
-	bool m_interruptTask;
+	std::deque<std::shared_ptr<TaskRunner>> m_taskRunners;
 
 	mutable std::mutex m_tasksMutex;
 	mutable std::mutex m_loopMutex;