#ifdef _DEBUG

#define _STLP_DEBUG 1

#endif

// TnzCore includes

#include "tthreadmessage.h"

#include "tsystem.h"

#include "tatomicvar.h"

#include "tthread.h"

#include "tthreadp.h"

// STL includes

#include <set></set>

#include <deque></deque>

// tcg includes

#include "tcg/tcg_pool.h"

//Qt includes

#include <qmultimap></qmultimap>

#include <qmutablemapiterator></qmutablemapiterator>

#include <qmutex></qmutex>

#include <qwaitcondition></qwaitcondition>

#include <qmetatype></qmetatype>

#include <qcoreapplication></qcoreapplication>

//==============================================================================

//==================================================

//    Paradigms of the Executor tasks management

//--------------------------------------------------

//Basics:

//  * Tasks added by Executors are always stored in a global QMultiMap first -

//    ordering primarily being the schedulingPriority(), and insertion instant

//    (implicit) when they have the same scheduling priority.

//  * The QMultiMap needs to be reverse-iterated to do that, since values with

//    the same key are ordered from the most recent to the oldest one (see Qt's

//    manual).

//  * Worker threads are stored in a global set.

//  * When a task is added or a task has been performed, the workers list is

//    refreshed, possibly adding new Workers for some executable tasks.

//  * When a worker ends a task, it automatically takes a new one before refreshing

//    the workers list. If no task can be taken, by default the thread exits and

//    invokes its own destruction.

//  * The thread may instead be put to rest if explicitly told by the user with

//    the appropriate method.

//Default execution conditions:

//  * A task is executable if, by default, its task load added to the sum of that of

//    all other active tasks does not exceed the available resources of the system

//    (i.e.: 100 * # of cores of the machine).

//    In other words, in every instant of execution, the sum of all active task's loads

//    never exceeds the available machine resources.

//  * When such default execution condition is not met when attempting to take the

//    task, no other task is taken instead - we wait until enough resources have been

//    freed before attempting to take the same task again.

//    In other words, the default execution condition is *BLOCKING*.

//Custom execution conditions:

//  * The user may decide to impose more tight conditions for tasks added by a certain

//    Executor. Let's call such conditions 'custom' conditions.

//  * Custom conditions are always tested *AFTER* the default ones (on the same task).

//    This is necessary to enforce the global scheduling priorities mechanism.

//  * When no task of a certain executor is active, custom conditions are always considered

//    satisfied.

//  * If custom conditions are not met, we enqueue the task in an Executor-private queue

//    for later execution and remove it from the global tasks queue - making it possible

//    to perform other tasks before our custom-failed one (such operation will be called

//    "ACCUMULATION").

//    In other words, custom conditions are *NOT BLOCKING* among different Executors.

//  * Tasks in the last task's Executor-private queue are always polled before those in

//    the global queue by Workers which ended a task, and this is done in a *BLOCKING* way

//    inside the same Executor.

//  * When an Executor-private tasks queue is not empty, all the other tasks added by the

//    same executor are scheduled in the queue.

//    In other words, the order of execution is always that of global insertion, *inside

//    the same executor*.

//  * Tasks polled from an Executor-private queue (which therefore satisfied custom conditions)

//    may still fail with default execution conditions. If that happens, we put the task

//    back into the global queue with highest possible priority (timedOut) and the worker dies.

//    Tasks with this special priority are polled *before every other task*. So, again, default

//    conditions are *BLOCKING*.

//Thread-safety:

//  * Most of the following code is mutex-protected, altough it might seem not - indeed,

//    only *one* mutex is locked and unlocked all of the time. This 'transition mutex' is

//    the key to thread-safety: we're considered to lie in a 'transition state' if we are

//    operating outside the run() of some task - which covers almost the totality of the code.

//  * The transition mutex is *not* recursive. The reason is that threads con not wait on

//    QWaitConditions if the mutex is recursive. That makes it necessary (and welcome) to put

//    mutex lockers in strategic points of the code - in many low-level functions no mutex

//    is locked *because some caller already did it before*. If you're modifying the code

//    always trace back the callers of a function before inserting misleading mutex lockers.

//==============================================================================

//==================

//    TODO list

//------------------

//  * Improve dedicated threads support: make sure that tasks added to a dedicated

//    executor are directly moved to the accumulation queue. The setDedicatedThreads()

//    method must therefore react accordingly.

//  * It could be possible to implement a dependency-based mechanism...

//  * Should the hosting thread wait for worker ones upon ExecutorImp destruction??

//    It could be a problem on some forever-waiting tasks...

//  * Ricontrolla che con le ultime modifiche gli ExecutorId rimangano quando si passano

//    i puntatori in takeTask e refreshAss..

//==============================================================================

using namespace TThread;

DEFINE_CLASS_CODE(TThread::Runnable, 21)

//==============================================================================

//==========================================

//    Global init() initializer function

//------------------------------------------

void TThread::init()

{

	Executor::init();

	TThreadMessageDispatcher::init();

}

//---------------------------------------------------------------------

void TThread::shutdown()

{

	Executor::shutdown();

}

//==============================================================================

namespace TThread

{

//==============================================================================

//===========================

//    Worker Thread class

//---------------------------

//! A Worker is a specialized QThread that continuously polls Runnable

//! tasks from a global execution queue to make them work.

class Worker : public QThread

{

public:

	RunnableP m_task;

	TSmartPointerT<executorid> m_master;</executorid>

	bool m_exit;

	QWaitCondition m_waitCondition;

	Worker();

	~Worker();

	void run();

	inline void takeTask();

	inline bool canAdopt(const RunnableP &task);

	inline void adoptTask(RunnableP &task);

	inline void rest();

	inline void updateCountsOnTake();

	inline void updateCountsOnRelease();

	inline void onFinish();

};

//=====================================================================

//========================

//    ExecutorId class

//------------------------

//! Contains all the executor data that need to persist until all tasks

//! added by the executor have been processed. Upon creation of an Executor

//! object, it instantiates one ExecutorId for both storing the Executor's data

//! sensible to the underlying code of the Executor manager, and to identify all

//! tasks added through it - by copying the smart pointer to the id into each

//! added task.

//! \sa Executor and Runnable class.

class ExecutorId : public TSmartObject

{

public:

	size_t m_id;

	int m_activeTasks;

	int m_maxActiveTasks;

	int m_activeLoad;

	int m_maxActiveLoad;

	bool m_dedicatedThreads;

	bool m_persistentThreads;

	std::deque<worker *=""> m_sleepings;</worker>

	ExecutorId();

	~ExecutorId();

	inline void accumulate(const RunnableP &task);

	void newWorker(RunnableP &task);

	void refreshDedicatedList();

};

//=====================================================================

//===========================

//    Executor::Imp class

//---------------------------

//! ExecutorImp both manages the allocation of worker threads for the

//! execution of Runnable tasks and centralizes the tasks collection.

//! One process only hosts one instance of the ExecutorImp class as a

//! a global variable that needs to be allocated in an application-lasting

//! and event-looped thread - typically the main thread in GUI applications.

class ExecutorImp

{

public:

	QMultiMap<int, runnablep=""> m_tasks;</int,>

	std::set<worker *=""> m_workers; // Used just for debugging purposes</worker>

	tcg::indices_pool<> m_executorIdPool;

	std::vector<uchar> m_waitingFlagsPool;</uchar>

	int m_activeLoad;

	int m_maxLoad;

	QMutex m_transitionMutex; // Workers' transition mutex

	ExecutorImp();

	~ExecutorImp();

	inline void insertTask(int schedulingPriority, RunnableP &task);

	void refreshAssignments();

	inline bool isExecutable(RunnableP &task);

};

//=====================================================================

} // namespace TThread

//=====================================================================

//===========================

//     Global variables

//---------------------------

namespace

{

ExecutorImp *globalImp = 0;

ExecutorImpSlots *globalImpSlots = 0;

bool shutdownVar = false;

}

//=====================================================================

//=============================

//     ExecutorImp methods

//-----------------------------

ExecutorImp::ExecutorImp()

	: m_activeLoad(0), m_maxLoad(TSystem::getProcessorCount() * 100), m_transitionMutex() //NOTE: We'll wait on this mutex - so it can't be recursive

{

}

//---------------------------------------------------------------------

ExecutorImp::~ExecutorImp()

{

}

//---------------------------------------------------------------------

//A task is executable <==> its load allows it. The task load is considered

//fixed until another isExecutable() call is made again - in case it may

//change in time.

inline bool ExecutorImp::isExecutable(RunnableP &task)

{

	return m_activeLoad + task->m_load <= m_maxLoad;

}

//---------------------------------------------------------------------

inline void ExecutorImp::insertTask(int schedulingPriority, RunnableP &task)

{

	task->m_schedulingPriority = schedulingPriority;

	m_tasks.insert(schedulingPriority, task);

}

//=====================================================================

//========================

//    Runnable methods

//------------------------

Runnable::Runnable()

	: TSmartObject(m_classCode), m_id(0)

{

}

//---------------------------------------------------------------------

Runnable::~Runnable()

{

	if (m_id)

		m_id->release(); //see Executor::addTask()

}

//---------------------------------------------------------------------

//! Returns the predicted CPU load generated by the task, expressed in percentage

//! of core usage (that is, 100 is intended to fully occupy one processing core).

//! Appropriate task load calibration is an important step to take when implementing

//! a new task; for this purpose, remember some rules to follow:

//! 

//! 
In every moment, the task manager ensures that the overall sum of the active

//!     task's load does not exceed the number of machine's processing cores multiplied

//!     by 100. This condition is \a blocking with respect to the execution of any other

//!     task - meaning that when a task is about to be executed the task manager \a waits

//!     until enough CPU resources are available to make it run.

//!     In particular, observe that a task's load \b never has to exceed the total CPU

//!     resources - doing so would surely result in a block of your application. The number

//!     of available cores can be accessed via the \b TSystem::getProcessorCount() or

//!     \b QThread::idealThreadCount().

//! 
The task load is considered constant for the duration of the task. Changing its

//!     value does not affect the task manager in any way once the task has been started.

//! 
The default task load is 0, representing a very light task. If the task load

//!     is 0 the condition at point 1 always succeeds - so the task is always executed when

//!     encountered. Observe that a long succession of 0 task loads leads to the creation of

//!     a proportional number of threads simultaneously running to dispatch it; if this is

//!     a problem, consider the use of \b Executor::setMaxActiveTasks()

//!     to make only a certain number of tasks being executed at the same time.

//! 

int Runnable::taskLoad()

{

	return 0;

}

//---------------------------------------------------------------------

//! Returns the priority value used to schedule a task for execution. Tasks

//! with higher priority start before tasks with lower priority. The default

//! value returned is 5 (halfway from 0 to 10) - but any value other than

//! (std::numeric_limits<int>::max)() is acceptable.</int>

int Runnable::schedulingPriority()

{

	return 5;

}

//---------------------------------------------------------------------

//! Returns the QThread::Priority used by worker threads when they adopt

//! the task. The default value returned is QThread::Normal.

QThread::Priority Runnable::runningPriority()

{

	return QThread::NormalPriority;

}

//---------------------------------------------------------------------

inline bool Runnable::customConditions()

{

	return (m_id->m_activeTasks < m_id->m_maxActiveTasks) &&

		   (m_id->m_activeLoad + m_load <= m_id->m_maxActiveLoad);

}

/*!

  \fn void Runnable::started(RunnableP sender)

  This signal is emitted from working threads just before the run() code

  is executed. Observe that the passed smart pointer ensures the survival of

  the emitting task for the time required by connected slots execution.

  \warning The started(), finished() and exception() signals are emitted in

  a mutex-protected environment in order to provide the correct sequence of

  emissions (i.e. so that canceled() and terminated() controller-emitted signals

  are either delivered after started() and before finished() or exception(),

  or \a instead of them).

  \warning Thus, setting up blocking connections or \a direct slots that contain a

  blocking instruction or even calls to the Executor API (which would definitely

  try to relock the aforementioned mutex) is dangerous and could result in an

  application freeze.

  \warning In case it's necessary to use blocking features, they should be enforced

  through custom signals to be invoked manually in the run() method, outside the

  mutex.

  \sa \b finished and \b exception signals.

*/

/*!

  \fn void Runnable::finished(RunnableP sender)

  The \b finished signal is emitted from working threads once the run()

  code is returned without unmanaged exceptions.

  \sa \b started and \b exception signals.

*/

/*!

  \fn void Runnable::exception(RunnableP sender)

  The \b exception signal is emitted from working threads whenever an

  untrapped exception is found within the run() method.

  \sa \b started and \b finished signals.

*/

/*!

  \fn void Runnable::canceled(RunnableP sender)

  The \b canceled signal is emitted from the controller thread whenever

  a task which is currently under the task manager's control is canceled

  by the user (the signal is emitted from the thread invoking the cancel).

  Observe that tasks under execution are not stopped by the task manager

  when they are canceled, but the signal is emitted anyway - helping the

  user to stop the actual execution of the run() code in advance.

  \sa \b Executor::removeTask and \b Executor::cancelAll methods.

*/

/*!

  \fn void Runnable::terminated(RunnableP sender)

  The \b terminated signal is emitted from the controller thread when

  the Executor components are shutting down inside a call to

  Executor::shutdown(). Implementing a slot connected to this signal

  helps the user in controlling the flow of an Executor-multithreaded

  application when it is shutting down - for example, it can be imposed

  that the application must wait for the task to be finished, print logs

  or similar.

  This signal is always preceded by a canceled() signal, informing all

  active tasks that thay should begin quitting on their own in a 'soft'

  way before brute termination may occur.

  \sa \b Executor::shutdown static method and \b Runnable::canceled signal.

*/

//! Convenience slot for the started() signal - so it's not necessary to declare

//! the task in a header file for moc'ing each time. You must both reimplement

//! \b and connect it to the started() signal to make it work.

void Runnable::onStarted(RunnableP)

{

}

//---------------------------------------------------------------------

//! The analogous of onStarted() for the finished() signal.

void Runnable::onFinished(RunnableP)

{

}

//---------------------------------------------------------------------

//! The analogous of onStarted() for the exception() signal.

void Runnable::onException(RunnableP)

{

}

//---------------------------------------------------------------------

//! The analogous of onStarted() for the canceled() signal.

void Runnable::onCanceled(RunnableP)

{

}

//---------------------------------------------------------------------

//! The analogous of onStarted() for the terminated() signal.

void Runnable::onTerminated(RunnableP)

{

}

//=====================================================================

//==========================

//    ExecutorId methods

//--------------------------

ExecutorId::ExecutorId()

	: m_activeTasks(0), m_maxActiveTasks(1), m_activeLoad(0), m_maxActiveLoad((std::numeric_limits<int>::max)()), m_dedicatedThreads(false), m_persistentThreads(false)</int>

{

	QMutexLocker transitionLocker(&globalImp->m_transitionMutex);

	m_id = globalImp->m_executorIdPool.acquire();

	globalImp->m_waitingFlagsPool.resize(globalImp->m_executorIdPool.size());

}

//---------------------------------------------------------------------

ExecutorId::~ExecutorId()

{

	QMutexLocker transitionLocker(&globalImp->m_transitionMutex);

	if (m_dedicatedThreads) {

		m_persistentThreads = 0;

		refreshDedicatedList();

	}

	globalImp->m_executorIdPool.release(m_id);

}

//---------------------------------------------------------------------

//Make sure that sleeping workers are eliminated properly if the permanent

//workers count decreases.

void ExecutorId::refreshDedicatedList()

{

	//QMutexLocker transitionLocker(&globalImp->m_transitionMutex);  //Already covered

	if (!m_dedicatedThreads || !m_persistentThreads) {

		//Release all sleeping workers

		//Wake them - they will exit on their own

		unsigned int i, size = m_sleepings.size();

		for (i = 0; i < size; ++i) {

			m_sleepings[i]->m_exit = true;

			m_sleepings[i]->m_waitCondition.wakeOne();

		}

		m_sleepings.clear();

	}

}

//=====================================================================

//===========================

//      Worker methods

//---------------------------

Worker::Worker()

	: QThread(), m_task(0), m_master(0), m_exit(true)

{

}

//---------------------------------------------------------------------

Worker::~Worker()

{

}

//---------------------------------------------------------------------

void Worker::run()

{

	//Ensure atomicity of worker's state transitions

	QMutexLocker sl(&globalImp->m_transitionMutex);

	if (shutdownVar)

		return;

	for (;;) {

		//Run the taken task

		setPriority(m_task->runningPriority());

		try {

			Q_EMIT m_task->started(m_task);

			sl.unlock();

			m_task->run();

			sl.relock();

			Q_EMIT m_task->finished(m_task);

		} catch (...) {

			sl.relock(); //throw must be in the run() block

			Q_EMIT m_task->exception(m_task);

		}

		updateCountsOnRelease();

		if (shutdownVar)

			return;

		//Get the next task

		takeTask();

		if (!m_task) {

			onFinish();

			if (!m_exit && !shutdownVar) {

				//Put the worker to sleep

				m_waitCondition.wait(sl.mutex());

				//Upon thread destruction the wait condition is implicitly woken up.

				//If this is the case, m_task == 0 and we return.

				if (!m_task || shutdownVar)

					return;

			} else

				return;

		}

	}

}

//---------------------------------------------------------------------

inline void Worker::updateCountsOnTake()

{

	globalImp->m_activeLoad += m_task->m_load;

	m_task->m_id->m_activeLoad += m_task->m_load;

	++m_task->m_id->m_activeTasks;

}

//---------------------------------------------------------------------

inline void Worker::updateCountsOnRelease()

{

	globalImp->m_activeLoad -= m_task->m_load;

	m_task->m_id->m_activeLoad -= m_task->m_load;

	--m_task->m_id->m_activeTasks;

}

//---------------------------------------------------------------------

inline void Worker::onFinish()

{

	if (m_master && m_master->m_dedicatedThreads && m_master->m_persistentThreads) {

		m_exit = false;

		m_master->m_sleepings.push_back(this);

		// Unlock the mutex - since eventually invoked ~ExecutorId will relock it...

		globalImp->m_transitionMutex.unlock();

		m_master = 0; //Master may be destroyed here - and m_exit= true for all sleepings

					  //in that case

		globalImp->m_transitionMutex.lock();

	} else {

		m_exit = true;

		globalImp->m_workers.erase(this);

	}

}

//=====================================================================

//===========================

//     Executor methods

//---------------------------

Executor::Executor()

	: m_id(new ExecutorId)

{

	m_id->addRef();

}

//---------------------------------------------------------------------

Executor::~Executor()

{

	m_id->release();

}

//---------------------------------------------------------------------

//! This static method declares the use of the Executor's task manager into

//! the application code. Be sure to use it according to the following rules:

//! 

//! 
 Only QCoreApplications or QApplications may use Executors.

//! 
 This method must be invoked in a thread which performs constant Qt event

//!      processing - like the main loop of interactive GUI applications.

//! 
 No task processing is allowed after event processing stops.

//! 

void Executor::init()

{

	//If no global ExecutorImp exists, allocate it now. You may not move this

	//to a static declaration, since ExecutorImpSlots's connections must be

	//made once the QCoreApplication has been constructed.

	if (!globalImp) {

		globalImp = new ExecutorImp;

		globalImpSlots = new ExecutorImpSlots;

	}

	qRegisterMetaType<tthread::runnablep>("TThread::RunnableP");</tthread::runnablep>

}

//---------------------------------------------------------------------

//! This static method, which \b must be invoked in the controller thread, declares

//! termination of all Executor-based components, forcing the execution of tasks submitted

//! by any Executor to quit as soon as possible in a safe way.

//! When the shutdown method is invoked, the task manager first emits a canceled()

//! signal for all the tasks that were submitted to it, independently from the Executor that

//! performed the submission; then, tasks that are still active once all the cancellation signals

//! were delivered further receive a terminated() signal informing them that they must provide

//! code termination (or at least remain silent in a safe state until the application quits).

//! \b NOTE: Observe that this method does not explicitly wait for all the tasks to terminate - this depends

//! on the code connected to the terminated() signal and is under the user's responsibility (see the

//! remarks specified in started() signal descritpion); if this is the intent and the terminated slot

//! is invoked in the controller thread, you should remember to implement a local event loop in it (so that

//! event processing is still performed) and wait there until the first finished() or catched()

//! slot make it quit.

void Executor::shutdown()

{

	{

		//Updating tasks list - lock against state transitions

		QMutexLocker transitionLocker(&globalImp->m_transitionMutex);

		shutdownVar = true;

		//Cancel all tasks  - first the active ones

		std::set<worker *="">::iterator it;</worker>

		for (it = globalImp->m_workers.begin(); it != globalImp->m_workers.end(); ++it) {

			RunnableP task = (*it)->m_task;

			if (task)

				Q_EMIT task->canceled(task);

		}

		//Finally, deal with the global queue tasks

		QMutableMapIterator<int, runnablep=""> jt(globalImp->m_tasks);</int,>

		while (jt.hasNext()) {

			jt.next();

			RunnableP task = jt.value();

			Q_EMIT task->canceled(task);

			jt.remove();

		}

		//Now, send the terminate() signal to all active tasks

		for (it = globalImp->m_workers.begin(); it != globalImp->m_workers.end(); ++it) {

			RunnableP task = (*it)->m_task;

			if (task)

				Q_EMIT task->terminated(task);

		}

	}

	//Just placing a convenience processEvents() to make sure that queued slots invoked by the

	//signals above are effectively invoked in this method - without having to return to an event loop.

	QCoreApplication::processEvents();

}

//---------------------------------------------------------------------

//! Specifies the use of dedicated threads for the Executor's task group.

//! By default a worker thread attempts adoption of Runnable tasks

//! without regard to the Executor that performed the submission. This helps

//! in stabilizing the number of threads that are created and destroyed

//! by the task manager - but may be a problem in some cases.

//! Using this method the user can explicitly tell the Executor to seize the

//! ownership of worker threads assigned to its tasks, so that they will not

//! try adoption of external tasks but instead remain focused on Executor's

//! tasks only.

//! An optional \b persistent parameter may be passed, which specifies if

//! dedicated threads should remain sleeping or should rather die when no

//! processable tasks from the Executor are found.

//! This method is especially helpful in two occasions:

//! 

//! 
 The Executor's tasks use thread-specific data such as QThreadStorages,

//!      which may be recycled among different tasks.

//! 
 The Executor receives tasks at a frequent rate, but mostly ends each one before

//!      another one is submitted - resulting in a continuous thread turnover.

//! 

void Executor::setDedicatedThreads(bool dedicated, bool persistent)

{

	QMutexLocker transitionLocker(&globalImp->m_transitionMutex);