

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.1.1.260726 (26 July 2026)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2026
//_/_/  
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
/*! @file ThreadManager.cpp
    @brief Implementation of the ThreadManager singleton, its monitoring thread
           and its self-contained unit test.

    See ThreadManager.h for the design overview (contiguous storage block,
    statistics collection policies, platform differences). Implementation
    highlights:
    - Static entry points lazily create the singleton before delegating to the
      instance-side worker of the same (lowercase) name.
    - The occupancy bitfield is initialised to all-ones meaning "free"; slot
      allocation uses utils::GetFirstFreeBitLoc / utils::SetBit.
    - The monitoring thread (ThreadMonitoring, slot 0) wakes every 100 ms and
      does a full statistics pass roughly once per second.
*/
#include "ThreadManager.h"
#include "PsyTime.h"
#include "Utils.h"
#include "UnitTestFramework.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);
	ThreadManager::Singleton = NULL;
	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;
	// Give running threads a chance to exit on their own before resorting to
	// utils::TerminateThread(). Forcibly terminating a thread that is right now
	// exiting naturally (worker loops poll their stop flags every <=1s) can kill
	// it inside DLL_THREAD_DETACH / CRT teardown while it holds the loader or
	// heap lock. That abandoned lock then blocks every NEWLY created thread
	// before it reaches its thread function - the cause of the intermittent
	// CMSDK processmemory/network test failures on Windows.
	ThreadDataHeader* header = (ThreadDataHeader*) data;
	ThreadStats* stats;
#ifdef WINDOWS
	// The natural-exit wait is only needed on Windows, where the OS
	// TerminateThread() can abandon the loader/heap lock. On POSIX,
	// utils::TerminateThread() is pthread_cancel()+pthread_join(), which is
	// safe to apply immediately (and waiting merely shifts where the
	// cancellation lands, which destabilised the network tests on Linux).
	uint64 waitStart = GetTimeNow();
	bool anyRunning = true;
	while (anyRunning && GetTimeAgeMS(waitStart) < 2500) {
		anyRunning = false;
		stats = GETTHREADSTATS(data,0);
		for (uint32 n=0; n<header->count; n++, stats++) {
			if (stats->status > THREAD_TERMINATED &&
					!utils::TryReapThread(stats->hThread)) {
				anyRunning = true;
				break;
			}
		}
		if (anyRunning)
			utils::Sleep(50);
	}
#endif
	// Terminate whatever is genuinely still running, mark the rest terminated
	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);
	// Only force-terminate a thread that is actually still running. Killing a
	// thread that already exited (or is mid-exit) can abandon the loader/heap
	// lock and silently break all subsequent thread creation (see shutdown()).
	if (!utils::TryReapThread(stats->hThread)) {
#ifdef WINDOWS
		// brief grace period for threads that are just about to exit; killing a
		// thread inside its exit path can abandon the loader/heap lock (Windows)
		uint64 graceStart = GetTimeNow();
		while (!utils::TryReapThread(stats->hThread) && GetTimeAgeMS(graceStart) < 250)
			utils::Sleep(25);
#endif
		if (stats->hThread) {
			utils::TerminateThread(stats->hThread);   // cancels+joins (POSIX) / terminates+closes (Windows)
			stats->hThread = 0;
		}
	}
	//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) {
		mutex.leave();   // was leaking the manager mutex on this early return
		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, n;

	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

				for (n=0; n<header->count; n++) {
					if (stats->status > THREAD_TERMINATED) {
						// check if it is still actually running
						if (utils::TryReapThread(stats->hThread)) {   // reaps exactly once; zeroes handle
							stats->status = THREAD_TERMINATED;
							LogPrint(0, LOG_PROCESS, 2, "Thread finished: ID %u (OSID %u) age: %s", stats->id, stats->osID, PrintTimeDifString((uint32)GetTimeAge(stats->created)).c_str());
						}
					}
					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] = header->currentCPUTicks[0] = 0;
			}
			else {
				// We have to gather the stats, sum them up and rotate all buffers
				sumUsage = 0;

				for (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;
						}
						// check if it is still actually running
						if (utils::TryReapThread(stats->hThread)) {   // reaps exactly once; zeroes handle
							stats->status = THREAD_TERMINATED;
							LogPrint(0, LOG_PROCESS, 2, "Thread finished: ID %u (OSID %u) age: %s", stats->id, stats->osID, PrintTimeDifString((uint32)GetTimeAge(stats->created)).c_str());
						}
					}
					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;
}


// ################# Unit test #################

namespace {

// Shared state for the ThreadManager unit test. Each worker thread does a
// bounded amount of work (a fixed number of increments), accumulates its
// result into a shared total under a mutex, then marks itself finished.
struct TMTestState {
	utils::Mutex mutex;
	uint64 total;			// sum of all per-thread work, protected by mutex
	uint32 finishedCount;	// number of worker threads that ran to completion
	TMTestState() : total(0), finishedCount(0) {}
};

// Each worker performs this many increments. Kept modest so the whole test
// finishes in well under a second, but large enough to be measurable.
static const uint64 TM_ITERATIONS_PER_THREAD = 200000ULL;

static THREAD_RET THREAD_FUNCTION_CALL TMTestWorker(THREAD_ARG arg) {
	TMTestState* state = (TMTestState*)arg;
	uint64 localSum = 0;
	for (uint64 i = 0; i < TM_ITERATIONS_PER_THREAD; i++)
		localSum += 1;
	if (state->mutex.enter()) {
		state->total += localSum;
		state->finishedCount++;
		state->mutex.leave();
	}
	thread_ret_val(0);
}

} // anonymous namespace

bool ThreadManager::UnitTest() {

	// 1. Bring up the ThreadManager singleton.
	unittest::progress(5, "create thread manager");
	if (!ThreadManager::CreateThreadManager()) {
		unittest::fail("ThreadManager test: CreateThreadManager() failed");
		return false;
	}

	const uint32 NUMTHREADS = 8;
	TMTestState state;
	uint32 ids[NUMTHREADS];
	for (uint32 n = 0; n < NUMTHREADS; n++)
		ids[n] = 0;

	// 2. Start the worker threads.
	unittest::progress(20, "start worker threads");
	uint64 t0 = GetTimeNow();
	uint32 started = 0;
	for (uint32 n = 0; n < NUMTHREADS; n++) {
		if (!ThreadManager::CreateThread(TMTestWorker, &state, ids[n])) {
			unittest::fail("ThreadManager test: CreateThread() failed for worker %u", n);
			ThreadManager::Shutdown();
			return false;
		}
		unittest::detail("started worker %u with id %u", n, ids[n]);
		started++;
	}
	if (started != NUMTHREADS) {
		unittest::fail("ThreadManager test: started %u of %u threads", started, NUMTHREADS);
		ThreadManager::Shutdown();
		return false;
	}

	// 3. Wait (bounded) for every worker to finish its work. We poll the shared
	//    finished counter and also the OS-level running state, with a hard
	//    timeout so the test can never hang.
	unittest::progress(50, "wait for workers");
	const uint32 TIMEOUTMS = 5000;
	uint32 waitedMs = 0;
	bool allFinished = false;
	while (waitedMs < TIMEOUTMS) {
		uint32 done = 0;
		if (state.mutex.enter()) {
			done = state.finishedCount;
			state.mutex.leave();
		}
		if (done >= NUMTHREADS) {
			allFinished = true;
			break;
		}
		utils::Sleep(5);
		waitedMs += 5;
	}

	if (!allFinished) {
		unittest::fail("ThreadManager test: only %u of %u workers finished within %ums",
			state.finishedCount, NUMTHREADS, TIMEOUTMS);
		ThreadManager::Shutdown();
		return false;
	}

	// 4. Verify the threads actually all ran and produced the expected total.
	unittest::progress(75, "verify results");
	uint64 expected = (uint64)NUMTHREADS * TM_ITERATIONS_PER_THREAD;
	if (state.total != expected) {
		unittest::fail("ThreadManager test: total work %llu != expected %llu",
			state.total, expected);
		ThreadManager::Shutdown();
		return false;
	}
	if (state.finishedCount != NUMTHREADS) {
		unittest::fail("ThreadManager test: finishedCount %u != %u",
			state.finishedCount, NUMTHREADS);
		ThreadManager::Shutdown();
		return false;
	}

	// 5. Sanity-check the statistics API for one of the threads.
	ThreadStats stats = ThreadManager::GetThreadStats(ids[0]);
	if (stats.created == 0) {
		unittest::fail("ThreadManager test: GetThreadStats returned no creation time for id %u", ids[0]);
		ThreadManager::Shutdown();
		return false;
	}

	double us = (double)(GetTimeNow() - t0);
	if (us > 0.0) {
		unittest::metric("threads_per_sec", (double)NUMTHREADS / us * 1e6, "threads/s", true);
		unittest::metric("total_run_time", us, "us", false);
	}

	// 6. Tear everything down cleanly. No threads remain after this point.
	unittest::progress(90, "shutdown");
	if (!ThreadManager::Shutdown()) {
		unittest::fail("ThreadManager test: Shutdown() failed");
		return false;
	}

	unittest::progress(100, "done");
	return true;
}

void Register_ThreadManager_Tests() {
	UnitTestRunner::instance().registerTest("threadmanager", ThreadManager::UnitTest,
		"Thread manager create/run/join of multiple worker threads", "core");
}

} // namespace cmlabs
