

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  and is released under the following
//_/_/  
//_/_/  EXTENDED BSD License
//_/_/
//_/_/  Redistribution and use in source and binary forms, with or without
//_/_/  modification, is permitted provided that the following conditions 
//_/_/  are met:
//_/_/
//_/_/  - Redistributions of source code must retain the above copyright 
//_/_/    and collaboration notice, this list of conditions and the following 
//_/_/    disclaimer.
//_/_/
//_/_/  - Redistributions in binary form must reproduce the above copyright 
//_/_/    notice, this list of conditions and the following disclaimer in the 
//_/_/    documentation and/or other materials provided with the distribution.
//_/_/
//_/_/  - Neither the name of its copyright holders nor the names of its 
//_/_/    contributors may be used to endorse or promote products derived from 
//_/_/    this software without specific prior written permission.
//_/_/
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "ThreadManager.h"
#include "PsyTime.h"
#include "Utils.h"

namespace cmlabs {

ThreadManager* ThreadManager::Singleton = NULL;

bool ThreadManager::CreateThreadManager() {
	ThreadManager* threadManager = new ThreadManager();
	if (!threadManager->init()) {
		//fprintf(stderr, "ThreadManager init() failed...\n");
		delete(threadManager);
		return false;
	}
	return true;
}

// Static
// Create a new thread and start it
bool ThreadManager::CreateThread(THREAD_FUNCTION func, void* args, uint32& newID, uint32 reqID) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->createThread(func, args, newID, reqID);
}

// Static
// Pause the thread wherever it is now
bool ThreadManager::PauseThread(uint32 id) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->pauseThread(id);
}

// Static
// Allow the thread to continue running
bool ThreadManager::ContinueThread(uint32 id) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->continueThread(id);
}

// Static
// Terminate the thread and restart it from its base location
bool ThreadManager::InterruptThread(uint32 id) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->interruptThread(id);
}

// Static
// Terminate and destroy the thread
bool ThreadManager::TerminateThread(uint32 id) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->terminateThread(id);
}


// Static
// Terminate and destroy the thread
bool ThreadManager::IsThreadRunning(uint32 id) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->isThreadRunning(id);
}


// Static
// Add thread stats for local thread - needed for pthreads
bool ThreadManager::AddLocalThreadStats() {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->addLocalThreadStats();
}

// Static
// Add thread stats for local thread - needed for pthreads
bool ThreadManager::GetLocalThreadID(uint32& id) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return false;
	return ThreadManager::Singleton->getLocalThreadID(id);
}


// Static
// Get the statistics for local thread
ThreadStats ThreadManager::GetLocalThreadStats() {
	ThreadStats stats;
	stats.created = 0;
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return stats;
	uint32 id;
	if (!ThreadManager::Singleton->getLocalThreadID(id))
		return stats;
	return ThreadManager::Singleton->getThreadStats(id);
}

// Static
// Get the statistics for a thread
ThreadStats ThreadManager::GetThreadStats(uint32 id) {
	ThreadStats stats;
	stats.created = 0;
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return stats;
	return ThreadManager::Singleton->getThreadStats(id);
}

// Static
// Get the statistics for all threads
ThreadStats* ThreadManager::GetAllThreadStats(uint32& count) {
	if (!ThreadManager::Singleton)
		if (!CreateThreadManager())
			return NULL;
	return ThreadManager::Singleton->getAllThreadStats(count);
}

// Static
// Shutdown and delete the ThreadManager
bool ThreadManager::Shutdown() {
	if (!ThreadManager::Singleton)
		return true;
	if (!ThreadManager::Singleton->shutdown())
		return true;
	delete(ThreadManager::Singleton);
	return (ThreadManager::Singleton == NULL);
}




ThreadManager::ThreadManager() {
	// Initially allocate space for 1024 threads
	data = NULL;
	resizeThreadStorage(1024);
	shouldContinue = true;
	isRunning = false;
}

ThreadManager::~ThreadManager() {
	if (ThreadManager::Singleton == NULL)
		return;
	if (!mutex.enter())
		return;
	shutdown();
	// Deallocate data
	ThreadManager::Singleton = NULL;
	free(data);
	data = NULL;
	mutex.leave();
}

bool ThreadManager::init() {
	if (!mutex.enter())
		return false;
	// Startup the root thread for Windows only
	uint32 newID = 0;
	ThreadManager::Singleton = this;
	if ((!createThread(ThreadMonitoring, NULL, newID, 0)) || (newID != 0)) {
		mutex.leave();
		return false;
	}
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	header->activeCount = 1;
	mutex.leave();
	return true;
}

bool ThreadManager::shutdown() {
	stop();
	if (!mutex.enter())
		return false;
	// Terminate all threads
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	ThreadStats* stats = GETTHREADSTATS(data,0);
	for (uint32 n=0; n<header->count; n++) {
		if (stats->status > THREAD_TERMINATED) {
			terminateThread(stats->id);
		}
		stats += 1;
	}
	header->activeCount = 0;
	mutex.leave();
	return true;
}




// Create a new thread and start it
bool ThreadManager::createThread(THREAD_FUNCTION func, void* args, uint32& newID, uint32 reqID) {
//	fprintf(stderr, "ThreadManager creating thread enter...\n");
	if (!mutex.enter())
		return false;
//	fprintf(stderr, "ThreadManager creating thread got mutex...\n");
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	if (header->activeCount == header->count) {
		// we need to resize Thread storage
		if (!resizeThreadStorage(header->count * 4)) {
			mutex.leave();
			return false;
		}
		header = (ThreadDataHeader*)data;
	}

//	fprintf(stderr, "ThreadManager creating thread 1...\n");
	ThreadStats* stats;
	if (reqID == 0) {
		// A bit cheeky as root = 0, but it works well for the root thread as this is the first one requested anyway.
		if (!utils::GetFirstFreeBitLoc(((const char*)data+sizeof(ThreadDataHeader)), header->bitFieldSize, newID) || (newID >= header->count)) {
			mutex.leave();
			return false;
		}
		stats = GETTHREADSTATS(data, newID);
		stats->id = newID;
	}
	else {
		newID = reqID;
		stats = GETTHREADSTATS(data, newID);
		stats->id = newID;
	}

//	fprintf(stderr, "ThreadManager creating thread 2...\n");
	stats->created = GetTimeNow();
	stats->status = THREAD_INIT;
	// We assume the stats are 0 already
	// Create the thread
	if (!utils::CreateThread(func, args, stats->hThread, stats->osID)) {
		newID = 0;
		stats->created = 0;
		stats->status = THREAD_NONE;
		mutex.leave();
		return false;
	}

	// printf("Create thread %u = %u\n", stats->id, (uint32)stats->hThread);
	utils::SetBit(stats->id, BITOCCUPIED, ((char*)data+sizeof(ThreadDataHeader)), header->bitFieldSize);
	stats->func = func;
	stats->args = args;
	stats->status = THREAD_RUNNING;
	header->activeCount++;
	//fprintf(stderr, "ThreadManager creating thread ID %u, trace:\n%s", stats->osID);
	//fprintf(stderr, "ThreadManager creating thread ID %u, trace:\n%s", stats->osID, utils::PrintProgramTrace(3, 6).c_str());
	mutex.leave();
//	fprintf(stderr, "ThreadManager creating thread 4...\n");
	return true;
}

// Pause the thread wherever it is now
bool ThreadManager::pauseThread(uint32 id) {
	if (!mutex.enter())
		return false;
	ThreadStats* stats = GETTHREADSTATS(data, id);
	// Pause the thread
	if (!utils::PauseThread(stats->hThread)) {
		mutex.leave();
		return false;
	}
	stats->status = THREAD_PAUSED;
	mutex.leave();
	return true;
}

// Allow the thread to continue running
bool ThreadManager::continueThread(uint32 id) {
	if (!mutex.enter())
		return false;
	ThreadStats* stats = GETTHREADSTATS(data, id);
	// Continue the thread
	if (!utils::ContinueThread(stats->hThread)) {
		mutex.leave();
		return false;
	}
	stats->status = THREAD_RUNNING;
	mutex.leave();
	return true;
}

// Terminate the thread and restart it from its base location
bool ThreadManager::interruptThread(uint32 id) {
	if (!mutex.enter())
		return false;
	ThreadStats* stats = GETTHREADSTATS(data, id);
	stats->status = THREAD_INTERRUPTED;
	// Terminate the thread
	// Terminate the thread
	if (!utils::TerminateThread(stats->hThread)) {
		mutex.leave();
		return false;
	}
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	header->activeCount--; // it will be increased again in createThread
	// Create the thread again
	uint32 newID = 0;
	if (!createThread(stats->func, stats->args, newID, id)) {
		stats->status = THREAD_TERMINATED;
		stats->func = NULL;
		stats->created = 0;
		header->activeCount--;
		mutex.leave();
		return false;
	}
	mutex.leave();
	return false;
}

// Terminate and destroy the thread
bool ThreadManager::terminateThread(uint32 id) {
	if (!mutex.enter())
		return false;
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	ThreadStats* stats = GETTHREADSTATS(data, id);
	// Terminate the thread
	// printf("Terminate thread %u = %u\n", stats->id, (uint32)stats->hThread);
	utils::TerminateThread(stats->hThread);
	//if (utils::IsThreadRunning(stats->hThread)) {
		//if (!utils::TerminateThread(stats->hThread)) {
		//	mutex.leave();
		//	return false;
		//}
	//}
	stats->status = THREAD_TERMINATED;
	stats->func = NULL;
	stats->created = 0;
	header->activeCount--;
	mutex.leave();
	return true;
}


// Has the thread terminated or exited
bool ThreadManager::isThreadRunning(uint32 id) {
	if (!mutex.enter())
		return false;
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	ThreadStats* stats = GETTHREADSTATS(data, id);
	if (stats->hThread == 0)
		return false;
	bool res = utils::IsThreadRunning(stats->hThread);
	mutex.leave();
	return res;
}


// Get the statistics for a thread
ThreadStats ThreadManager::getThreadStats(uint32 id) {
	ThreadStats stats;
	stats.created = 0;
	if (!mutex.enter()) {
		mutex.leave();
		return stats;
	}
	stats = *GETTHREADSTATS(data, id);
	mutex.leave();
	return stats;
}

// Get the statistics for all threads
ThreadStats* ThreadManager::getAllThreadStats(uint32& count) {
	if (!mutex.enter())
		return NULL;
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	count = header->activeCount;
	ThreadStats* allStats = new ThreadStats[count];
	uint32 p=0;
	ThreadStats* stats = GETTHREADSTATS(data,0);
	for (uint32 n=0; n<header->count; n++) {
		if (stats->status > THREAD_TERMINATED)
			allStats[p++] = *stats;
		stats += 1;
		if (p >= count) break;
	}
	mutex.leave();
	return allStats;
}

// Resize the ThreadStats storage to be able to contain more threads
bool ThreadManager::resizeThreadStorage(uint32 newCount) {

	uint32 bitFieldSize = utils::Calc32BitFieldSize(newCount);
	uint32 size = sizeof(ThreadDataHeader) + bitFieldSize + newCount*sizeof(ThreadStats);
	unsigned char *newData = (unsigned char*) malloc(size);
	memset(newData, 0, size);
	ThreadDataHeader* header = (ThreadDataHeader*) newData;
	header->bitFieldSize = bitFieldSize;
	header->count = newCount;
	header->size = size;
	header->statColPolicy = utils::GetThreadStatColAbility();

	uint32 statsOffset = sizeof(header->size)+sizeof(header->count)+sizeof(header->activeCount)+sizeof(header->statColPolicy);

	if (data == NULL) {
		header->activeCount = 0;
		// Set summary stats to 0
		memset(newData+statsOffset, 0, 10*2*(sizeof(uint64)));
		// Set the bitfield to unused
		memset(newData+sizeof(ThreadDataHeader), 255, bitFieldSize);
		// Set all thread stats to 0
		ThreadStats* firstStat = GETTHREADSTATS(newData,0);
		memset(firstStat, 0, newCount*sizeof(ThreadStats));
	}
	else {
		ThreadDataHeader* oldHeader = (ThreadDataHeader*) data;
		header->activeCount = oldHeader->activeCount;		
		// Copy old summary stats 
		memcpy(newData+statsOffset, data+statsOffset, 10*2*(sizeof(uint64)));
		// Initially, set the bitfield to unused
		memset(newData+sizeof(ThreadDataHeader), 255, bitFieldSize);
		// Copy old bitfield
		memcpy(newData+sizeof(ThreadDataHeader), data+sizeof(ThreadDataHeader), oldHeader->bitFieldSize);
		ThreadStats* firstStat = GETTHREADSTATS(newData,0);
		ThreadStats* oldFirstStat = GETTHREADSTATS(data,0);
		memcpy(firstStat, oldFirstStat, oldHeader->count*sizeof(ThreadStats));
		free(data);
	}
	data = newData;
	return true;
}

// Add thread stats for local thread - needed for pthreads
bool ThreadManager::addLocalThreadStats() {
	if (!mutex.enter())
		return false;

	uint32 id = 0;
	if (!getLocalThreadID(id)) {
		mutex.leave();
		return false;
	}
	ThreadStats* stats = GETTHREADSTATS(data,id);
	if ( (stats == NULL) || (stats->created == 0) ) {
		mutex.leave();
		return false;
	}

	uint32 timeOffset = sizeof(stats->id) + sizeof(stats->created) + sizeof(stats->status)
		+ sizeof(stats->hThread) + sizeof(stats->func) + sizeof(stats->args);
	uint32 cpuUsageOffset = timeOffset + 10 * sizeof(uint64);

	ThreadDataHeader* header = NULL;
	
//	uint32 headerTimeOffset = sizeof(header->size)+sizeof(header->count)+sizeof(header->activeCount)+sizeof(header->statColPolicy);
//	uint32 headerCPUUsageOffset = headerTimeOffset + 10 * sizeof(uint64);

	// Shift the previously stored values Time
	memcpy(stats+timeOffset, stats+timeOffset+sizeof(uint64), 9*sizeof(uint64));
	// CPU Usage
	memcpy(stats+cpuUsageOffset, stats+cpuUsageOffset+sizeof(uint64), 9*sizeof(uint64));
	// Read current stats from the thread
	if (utils::GetCPUTicks(stats->currentCPUTicks[0])) {
		stats->time[0] = GetTimeNow();
		// add these to the header stat sums
		header = (ThreadDataHeader*) data;
		header->time[0] = stats->time[0];
		header->currentCPUTicks[0] += stats->currentCPUTicks[0];
		// but do not rotate
	}
	else {
		stats->time[0] = stats->currentCPUTicks[0] = 0;
		mutex.leave();
		return false;
	}

	mutex.leave();
	return true;
}

// Get the local thread ID
bool ThreadManager::getLocalThreadID(uint32& id) {
	if (!mutex.enter())
		return false;

	uint32 osID;
	if (!utils::GetCurrentThreadUniqueID(osID)) {
		mutex.leave();
		return false;
	}
	ThreadDataHeader* header = (ThreadDataHeader*) data;

	ThreadStats* stats = GETTHREADSTATS(data,0);
	for (uint32 n=0; n<header->count; n++) {
		if (stats->osID == osID) {
			id = stats->id;
			mutex.leave();
			return true;
		}
		stats += 1;
	}

	// for pthreads one has to delete the handle afterwards
	//#ifndef WINDOWS
	//	delete(hThread);
	//#endif
	mutex.leave();
	return false;
}


// Root Thread, for Windows only until pthreads support getrusage from other threads
THREAD_RET THREAD_FUNCTION_CALL ThreadMonitoring(void* arg) {

	#ifdef WINDOWS
	#else
		#ifndef Darwin
			sigset_t cancel;
			sigemptyset(&cancel);
			sigaddset(&cancel, SIGQUIT);
			pthread_sigmask(SIG_UNBLOCK, &cancel, NULL);
		#endif
	#endif

	if (!ThreadManager::Singleton)
		return 0;

	thread_ret_val(ThreadManager::Singleton->threadMonitoring());
}

int32 ThreadManager::threadMonitoring() {

	isRunning = true;

	ThreadStats* stats = NULL;
	uint32 timeOffset = sizeof(stats->id) + sizeof(stats->created) + sizeof(stats->status)
		+ sizeof(stats->hThread) + sizeof(stats->func) + sizeof(stats->args);
	uint32 cpuUsageOffset = timeOffset + 10 * sizeof(uint64);

	ThreadDataHeader* header = (ThreadDataHeader*) data;
	if (header->statColPolicy == THREAD_STATS_OFF) {
		isRunning = false;
		return 0; // do not continue as nothing is done anyway
	}

	uint32 headerTimeOffset = sizeof(header->size)+sizeof(header->count)+sizeof(header->activeCount)+sizeof(header->statColPolicy);
	uint32 headerCPUUsageOffset = headerTimeOffset + 10 * sizeof(uint64);

	uint64 sumUsage;
	uint32 interval = 1000000;

	uint64 lastCalc = 0, t;
	while (shouldContinue) {
		if ( (t = GetTimeNow()) - lastCalc > interval ) {
			lastCalc = t;

			if (!mutex.enter()) {
				isRunning = false;
				return -1;
			}

			header = (ThreadDataHeader*) data;
			stats = GETTHREADSTATS(data,0);

			if (header->statColPolicy == THREAD_STATS_ADHOC) {
				// Threads have to call the stat gathering themselves
				// so we just rotate the global stats

				// Shift the previously stored values Time
				memcpy(header+headerTimeOffset, header+headerTimeOffset+sizeof(uint64), 9*sizeof(uint64));
				// CPU Usage
				memcpy(header+headerCPUUsageOffset, header+headerCPUUsageOffset+sizeof(uint64), 9*sizeof(uint64));

				header->time[0] = header->currentCPUTicks[0] = 0;
			}
			else {
				// We have to gather the stats, sum them up and rotate all buffers
				sumUsage = 0;

				for (uint32 n=0; n<header->count; n++) {
					if (stats->status > THREAD_TERMINATED) {
						// Shift the previously stored values Time
						memcpy(stats+timeOffset, stats+timeOffset+sizeof(uint64), 9*sizeof(uint64));
						// CPU Usage
						memcpy(stats+cpuUsageOffset, stats+cpuUsageOffset+sizeof(uint64), 9*sizeof(uint64));
						// Read current stats from the thread
						if (utils::GetCPUTicks(stats->hThread, stats->currentCPUTicks[0])) {
							stats->time[0] = GetTimeNow();
							sumUsage += stats->currentCPUTicks[0];
						}
						else {
							stats->time[0] = stats->currentCPUTicks[0] = 0;
						}
					}
					stats += 1;
				}

				// Shift the previously stored values Time
				memcpy(header+headerTimeOffset, header+headerTimeOffset+sizeof(uint64), 9*sizeof(uint64));
				// CPU Usage
				memcpy(header+headerCPUUsageOffset, header+headerCPUUsageOffset+sizeof(uint64), 9*sizeof(uint64));

				header->time[0] = GetTimeNow();
				header->currentCPUTicks[0] = sumUsage;
			}

			mutex.leave();
		}

		utils::Sleep(100);
	}
			
	isRunning = false;
	return 0;
}

} // namespace cmlabs
