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;
 
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