ThreadManager.cpp

Go to the documentation of this file.

#include "ThreadManager.h"
#include "PsyTime.h"
#include "Utils.h"
#include "UnitTestFramework.h"
 
namespace cmlabs {
 
ThreadManager* ThreadManager::Singleton = NULL;
 
// Single-teardown guard. shutdown() force-terminates and joins the registered
// worker threads while holding the registry mutex. If a SECOND driver enters
// shutdown() concurrently on another thread - e.g. a signal-driven exit() runs
// CleanShutdown::~CleanShutdown() -> ThreadManager::Shutdown() on a worker while
// main is already tearing the node down - the two collide on the mutex/join and
// deadlock. This flag lets only the FIRST caller run the teardown; a concurrent
// second caller returns immediately. It is scoped to the in-flight teardown (set
// on entry, cleared on exit), NOT a permanent latch, so sequential teardowns
// (e.g. one per CMSDK network unit test in fork=0 mode) still work.
static volatile int32 g_threadMgrTearingDown = 0;
 
// Atomic test-and-set: returns true iff it flipped 0 -> 1 (i.e. WE won the race
// and now own the teardown). Mirrors the WINDOWS / __sync idiom used in Utils.cpp.
static bool TryBeginThreadMgrTeardown() {
    #if defined WINDOWS
        return (_InterlockedCompareExchange((volatile long*)&g_threadMgrTearingDown, 1, 0) == 0);
    #else
        return __sync_bool_compare_and_swap(&g_threadMgrTearingDown, 0, 1);
    #endif
}
 
static void EndThreadMgrTeardown() {
    #if defined WINDOWS
        _InterlockedExchange((volatile long*)&g_threadMgrTearingDown, 0);
    #else
        __sync_lock_test_and_set(&g_threadMgrTearingDown, 0);
    #endif
}
 
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
// Wait for a thread to have COMPLETELY finished, then release it - exactly once
bool ThreadManager::JoinThread(uint32 id) {
    if (!ThreadManager::Singleton)
        if (!CreateThreadManager())
            return false;
    return ThreadManager::Singleton->joinThread(id);
}
 
 
// Static
// Copy out the OS-level ThreadHandle for a manager slot
bool ThreadManager::GetThreadHandle(uint32 id, ThreadHandle& out) {
    if (!ThreadManager::Singleton)
        if (!CreateThreadManager())
            return false;
    return ThreadManager::Singleton->getThreadHandle(id, out);
}
 
 
// 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;
    // Make the registry lock cancel-safe: teardown force-terminates worker threads
    // via pthread_cancel, and a cancel landing on a thread that holds this lock
    // (e.g. mid createThread/addLocalThreadStats) would orphan it and hang every
    // later terminate/isRunning call for 30s. Disabling cancellation while the lock
    // is held closes that window. See utils::Mutex::setCancelSafe().
    mutex.setCancelSafe(true);
    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() {
    // Single-teardown guard: if another thread is already tearing the registry
    // down, do NOT run a second concurrent teardown (it would collide on the
    // registry mutex / thread joins and deadlock). Return false so the static
    // Shutdown() wrapper skips deleting the singleton out from under the
    // in-flight teardown. See g_threadMgrTearingDown.
    if (!TryBeginThreadMgrTeardown())
        return false;
    stop();
    if (!mutex.enter()) {
        EndThreadMgrTeardown();
        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();
    EndThreadMgrTeardown();
    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);
        if (stats->hThread) {
            utils::TerminateThread(stats->hThread);   // terminates+closes (Windows)
            stats->hThread = 0;
        }
#else
        if (stats->hThread) {
            // Never cancel+join the CALLING thread. If teardown is being driven from
            // a registered worker (e.g. exit() -> CleanShutdown -> Shutdown() runs on
            // a bake/network worker), shutdown()'s loop will reach that worker's own
            // slot; pthread_cancel+pthread_join on self is undefined / self-deadlocks.
            // Just detach the handle and mark it terminated; the thread is on its way
            // out through exit() anyway.
            if (pthread_equal(stats->hThread, pthread_self())) {
                stats->hThread = 0;
            }
            else {
                // The blocking part of TerminateThread on POSIX is pthread_cancel +
                // pthread_join. It MUST NOT run while we hold the registry mutex: the
                // target may be blocked in Mutex::enter() for this same lock, and while
                // the lock is held it cannot acquire it - so joining here would deadlock
                // main against the worker. Snapshot the handle, zero it in the slot so no
                // other caller races the same join (double-join is undefined behaviour),
                // release the mutex, then cancel+join. Re-acquire and re-fetch afterwards
                // because the storage block may have been resized meanwhile.
                ThreadHandle h = stats->hThread;
                stats->hThread = 0;
                mutex.leave();
                utils::TerminateThread(h);   // cancels + joins - NO registry lock held
                if (!mutex.enter())
                    return false;
                header = (ThreadDataHeader*) data;
                stats = GETTHREADSTATS(data, id);
            }
        }
#endif
    }
    //if (utils::IsThreadRunning(stats->hThread)) {
        //if (!utils::TerminateThread(stats->hThread)) {
        //  mutex.leave();
        //  return false;
        //}
    //}
    // Finalize idempotently: another terminate/monitor pass may have already
    // reaped this slot while the mutex was released above.
    finaliseSlot(stats, header);
    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;
}
 
 
// Copy out the OS-level ThreadHandle for a slot (by value, under the mutex)
bool ThreadManager::getThreadHandle(uint32 id, ThreadHandle& out) {
    if (!mutex.enter())
        return false;
    ThreadStats* stats = GETTHREADSTATS(data, id);
    if (stats->hThread == 0) {
        mutex.leave();
        return false;
    }
    out = stats->hThread;
    mutex.leave();
    return true;
}
 
 
// Wait for a thread to have COMPLETELY finished, then release it - exactly once.
// This is the ONLY sound "that thread is entirely gone" signal. A flag the worker sets
// itself is necessarily set BEFORE it returns from its thread function, and long before
// the runtime tears down its stack/TSD - so a reaper that frees on such a flag always
// races the thread's own exit path (observed: a worker still inside pthread_exit/TSD
// cleanup after its owning object had been freed).
bool ThreadManager::joinThread(uint32 id) {
    if (!mutex.enter())
        return false;
    ThreadDataHeader* header = (ThreadDataHeader*) data;
    ThreadStats* stats = GETTHREADSTATS(data, id);
    ThreadHandle h = stats->hThread;
    if (h == 0) {
        // Already reaped - by terminateThread(), the monitoring loop, or an earlier
        // join. Nothing left to wait for.
        mutex.leave();
        return true;
    }
    #ifndef WINDOWS
        // NEVER join the CALLING thread: teardown can be driven from a registered worker
        // (exit() -> CleanShutdown -> Shutdown()), and pthread_join on self is undefined
        // / self-deadlocks. Leave the handle for the reaper that outlives us.
        if (pthread_equal(h, pthread_self())) {
            mutex.leave();
            return true;
        }
    #endif
    // TAKE EXCLUSIVE OWNERSHIP before joining: zero the slot so no other joiner - a
    // concurrent terminateThread(), or the monitoring loop's TryReapThread() (Linux
    // only; the loop exits immediately when statColPolicy is THREAD_STATS_OFF, which
    // is the case on macOS) - can ever join the same pthread_t twice. Double-join is
    // undefined behaviour and corrupts the heap. Correctness rests on THIS, not on
    // the monitor being absent.
    stats->hThread = 0;
    // Release the registry mutex BEFORE joining: the target may itself be blocked in
    // Mutex::enter() for this very lock, and while we hold it the target can never
    // acquire it - so joining here would deadlock us against the thread we wait for.
    mutex.leave();
    #ifdef WINDOWS
        // ReapThread() only CloseHandle()s on Windows - it does NOT wait - so the wait
        // must be explicit first. Timeout 0 = infinite WaitForSingleObject.
        utils::WaitForThreadToFinish(h, 0);
        utils::ReapThread(h);
    #else
        // POSIX: pthread_join both waits and releases the thread's resources, and
        // ReapThread() zeroes its local copy of the handle afterwards.
        // UNBOUNDED on purpose: a bounded join (pthread_timedjoin_np) that expires
        // after we have zeroed the slot would leave the thread joinable by nobody -
        // silently dropping the guarantee. Callers bound the COOPERATIVE phase
        // (shouldContinue + a bounded isRunning wait + a loud log on overrun) and
        // only then call this.
        utils::ReapThread(h);
    #endif
    // Re-acquire and RE-FETCH: the storage block is malloc/memcpy'd and may have been
    // resized while the mutex was released, so the earlier pointers can be stale.
    if (!mutex.enter())
        return true;    // the thread IS joined; only the bookkeeping below is missed
    header = (ThreadDataHeader*) data;
    stats = GETTHREADSTATS(data, id);
    // Finalise idempotently: terminateThread() or a monitor pass may already have
    // marked this slot while our mutex was released. This RELEASES THE SLOT as well
    // as marking it - see finaliseSlot(). It used to be an inline copy of
    // terminateThread's block that did not free the bit, so every join leaked a
    // registry slot: the leak survived its own fix on the path that matters most,
    // since all four NetworkChannel teardown sites reap via JoinThread and on macOS
    // the monitor loop never reaps at all. We have already zeroed stats->hThread
    // above, so finaliseSlot's handle guard is always satisfied here.
    finaliseSlot(stats, header);
    mutex.leave();
    return true;
}
 
 
// Mark a slot terminated and release it back to the allocation bitfield.
// Caller must hold the registry mutex, with stats/header re-fetched after any
// release of it. Shared by terminateThread() and joinThread() so the slot-release
// cannot be present in one and missing in the other. @see finaliseSlot() docs.
void ThreadManager::finaliseSlot(ThreadStats* stats, ThreadDataHeader* header) {
    if (!stats || !header)
        return;
    if (stats->status == THREAD_TERMINATED)
        return;         // already finalised by another terminate/join/monitor pass
    stats->status = THREAD_TERMINATED;
    stats->func = NULL;
    stats->created = 0;
    header->activeCount--;
    // Release the registry slot back to the bitfield, or the process leaks one slot
    // per thread lifetime: createThread only ever SetBit(BITOCCUPIED) and nothing
    // used to clear it, so after `count` create/terminate cycles GetFirstFreeBitLoc
    // had nothing left and thread creation failed forever. resizeThreadStorage could
    // not rescue it either - it triggers on activeCount == count, and activeCount
    // DOES decrement here, so the registry looked like it had spare capacity while
    // being unable to allocate.
    //
    // ORDERING IS LOAD-BEARING: only release the bit once hThread is zero. Freeing it
    // while the OS thread is still unwinding would let a concurrent createThread hand
    // out this slot and overwrite ThreadStats under the dying thread - trading a
    // capacity bug for a use-after-free. Callers either zero the handle themselves or
    // went through TryReapThread (which zeroes it), but assume nothing: a non-zero
    // handle here means an unreaped thread, so keep the slot occupied and leak it,
    // which is the old, safe behaviour.
    if (!stats->hThread)
        utils::SetBit(stats->id, BITFREE, ((char*)data+sizeof(ThreadDataHeader)), header->bitFieldSize);
}
 
 
// 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);
}
 
// Worker for the slot-recycling test: exits immediately, so the test can churn
// many thread LIFETIMES cheaply rather than doing work.
static THREAD_RET THREAD_FUNCTION_CALL TMSlotChurnWorker(THREAD_ARG arg) {
    (void)arg;
    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;
}
 
// Registry slot recycling: create and reap MORE threads than the registry has
// slots, in ONE process, and require that creation keeps working.
//
// This is the test that would have caught the slot leak by itself. createThread()
// marks a slot BITOCCUPIED, and until the D1 fix nothing ever cleared it, so after
// `count` lifetimes GetFirstFreeBitLoc had nothing left and thread creation failed
// permanently - while activeCount kept decrementing, so the registry reported spare
// capacity. Every existing suite test forks per run and creates a handful of
// threads, which is exactly why this stayed invisible.
//
// It deliberately reaps via ThreadManager::JoinThread(), NOT terminateThread():
// the initial D1 fix released the slot only in terminateThread's finalize block,
// and JoinThread had a second, divergent copy that did not - so a terminateThread
// based test would have passed against the leak. JoinThread is also the path every
// NetworkChannel teardown uses, and on macOS it is the ONLY reaper, because the
// monitoring loop returns early when statColPolicy is THREAD_STATS_OFF.
bool ThreadManager::UnitTestSlotRecycling() {
    unittest::progress(5, "create thread manager");
    if (!ThreadManager::CreateThreadManager()) {
        unittest::fail("slot recycling: CreateThreadManager() failed");
        return false;
    }
 
    // The registry starts with 1024 slots. Churn comfortably past that so an
    // unreleased slot per lifetime must exhaust the bitfield.
    const uint32 CYCLES = 1300;
    uint32 created = 0, failedAt = 0;
    uint32 maxID = 0;
 
    unittest::progress(20, "churn thread lifetimes");
    for (uint32 n = 0; n < CYCLES; n++) {
        uint32 id = 0;
        if (!ThreadManager::CreateThread(TMSlotChurnWorker, NULL, id)) {
            failedAt = n;
            break;
        }
        created++;
        if (id > maxID)
            maxID = id;
        // Reap it. JoinThread waits for the OS thread and finalises the slot, which
        // is where the release has to happen.
        ThreadManager::JoinThread(id);
    }
 
    unittest::progress(80, "verify");
    if (failedAt) {
        unittest::fail("slot recycling: CreateThread() failed after %u lifetimes "
            "(highest slot id seen %u) - registry slots are not being released",
            failedAt, maxID);
        ThreadManager::Shutdown();
        return false;
    }
    if (created != CYCLES) {
        unittest::fail("slot recycling: completed %u of %u lifetimes", created, CYCLES);
        ThreadManager::Shutdown();
        return false;
    }
    // Slots must be REUSED, not merely available: if ids climb monotonically past
    // the registry size then something grew the storage instead of recycling.
    if (maxID >= CYCLES) {
        unittest::fail("slot recycling: highest slot id %u >= %u lifetimes - ids are "
            "not being recycled", maxID, CYCLES);
        ThreadManager::Shutdown();
        return false;
    }
    unittest::detail("%u thread lifetimes, highest slot id %u", created, maxID);
    unittest::metric("thread_lifetimes", (double)created, "", true);
    unittest::metric("highest_slot_id", (double)maxID, "", false);
 
    unittest::progress(90, "shutdown");
    if (!ThreadManager::Shutdown()) {
        unittest::fail("slot recycling: 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");
    UnitTestRunner::instance().registerTest("threadslots", ThreadManager::UnitTestSlotRecycling,
        "Thread manager releases and reuses registry slots across >1024 lifetimes", "core");
}
 
} // namespace cmlabs