ProcessMemory.cpp

Go to the documentation of this file.

 
#include "ProcessMemory.h"
#include "UnitTestFramework.h"
#include "ThreadManager.h"
 
namespace cmlabs {
 
// Cooperative stop for the ProcessMemory unit-test slave threads (QueueTest /
// ProcessMemoryPerfTest). The test sets this true and waits for the worker to
// exit before freeing its MemoryManager, so the worker never touches freed
// memory (which would otherwise hang the next test in the suite).
static volatile bool g_processTestStop = false;
static uint32        g_processTestThreadID = 0;
 
static void StopProcessTestThread() {
    g_processTestStop = true;
    for (int i = 0; i < 60 && g_processTestThreadID &&
                    ThreadManager::IsThreadRunning(g_processTestThreadID); i++)
        utils::Sleep(50);
    g_processTestThreadID = 0;
}
 
ProcessMemory::ProcessMemory(MasterMemory* master) {
    memset(qSemaphores, 0, MAXPROC4*sizeof(utils::Semaphore*));
    mutex = NULL;
    this->master = master;
    header = NULL;
    memorySize = 0;
    port = 0;
    serial = 0;
}
 
ProcessMemory::~ProcessMemory() {
    if (mutex)
        mutex->enter(5000, __FUNCTION__);
    if (memorySize)
        utils::CloseSharedMemorySegment((char*) header, memorySize);
    memset(qSemaphores, 0, MAXPROC4*sizeof(utils::Semaphore*));
    header = NULL;
    if (mutex)
        mutex->leave();
    delete(mutex);
    mutex = NULL;
}
 
bool ProcessMemory::getMemoryUsage(uint64& alloc, uint64& usage) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
//  printf(" --------- Get ProcessMap usage size: %u header: %u usage: %u serial %u --------\n", memorySize, header->size, header->usage, serial);
 
    alloc = memorySize;
    usage = header->usage;
 
    mutex->leave();
    return true;
}
 
bool ProcessMemory::checkProcessHeartbeats(uint32 timeoutMS, std::list<ProcessInfoStruct>& procIssues) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    bool issuesFound = false;
    ProcessInfoStruct* pInfo = &(header->processes[1]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (pInfo->createdTime && (pInfo->nodeID == this->master->getNodeID())) {
            //if (!pInfo->lastseen) {
            //  procIssues.push_back(*pInfo);
            //  issuesFound = true;
            //}
            if (pInfo->status > PSYPROC_ACTIVE) {
                procIssues.push_back(*pInfo);
                issuesFound = true;
            }
            else if ((pInfo->status > PSYPROC_IDLE) && (GetTimeAgeMS(pInfo->lastseen) > (int64)timeoutMS)) {
                procIssues.push_back(*pInfo);
                issuesFound = true;
            }
        }
    }
    mutex->leave();
    return (!issuesFound);
}
 
std::vector<ProcessInfoStruct>* ProcessMemory::getAllProcesses() {
    std::vector<ProcessInfoStruct>* processes = new std::vector<ProcessInfoStruct>;
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return processes;
    CHECKPROCESSMEMORYSERIAL
 
    uint16 thisNodeID = this->master->getNodeID();
    ProcessInfoStruct* pInfo = &(header->processes[1]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (pInfo->createdTime && (pInfo->nodeID == thisNodeID))
            processes->push_back(*pInfo);
    }
    mutex->leave();
    return processes;
}
 
bool ProcessMemory::addLocalPerformanceStats(std::list<PerfStats> &perfStats) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    uint16 thisNodeID = this->master->getNodeID();
 
    PerfStats stats;
    ProcessInfoStruct* pInfo = &(header->processes[1]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (pInfo->createdTime && (pInfo->nodeID == thisNodeID)) {
            memset(&stats, 0, sizeof(PerfStats));
            stats.spaceID = pInfo->id;
            stats.nodeID = pInfo->nodeID;
            stats.osID = pInfo->osID;
            // stats.type = stats.space;
            if (pInfo->type == 1) // inside node
                stats.currentCPUTicks = 0;
            else
                stats.currentCPUTicks = pInfo->stats.currentCPUTicks;
            // stats.currentMemoryBytes = pInfo->stats.procMemUsage;
            stats.currentMemoryBytes = utils::GetProcessMemoryUsage();
            stats.totalInputBytes = pInfo->stats.msgInBytes;
            stats.totalInputCount = pInfo->stats.msgInCount;
            stats.totalOutputBytes = pInfo->stats.msgOutBytes;
            stats.totalOutputCount = pInfo->stats.msgOutCount;
            // stats.runCount = 0;
            getQueueSizes(pInfo->id, stats.totalQueueBytes, stats.totalQueueCount);
            // stats.currentRunStartTime = 0;
            stats.firstRunStartTime = pInfo->createdTime;
            // stats.totalCycleCount = 0;
            // stats.totalRunCount = 0;
            // stats.migrationCount = 0;
            perfStats.push_back(stats);
        }
    }
    mutex->leave();
 
    return true;
}
 
bool ProcessMemory::getQueueSizes(uint16 procID, uint64 &bytes, uint32 &count) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
 
    bytes = 0;
    count = 0;
 
    MessageQueueHeader* qHeader;
    for (uint8 t = 1; t<5; t++) {
        qHeader = getQHeader(procID, t);
        if (qHeader) {
            count += qHeader->count;
            if (qHeader->count) {
                if (qHeader->endPos > qHeader->startPos)
                    bytes = qHeader->endPos - qHeader->startPos;
                else
                    bytes = qHeader->startPos - qHeader->endPos;
            }
        }
    }
    mutex->leave();
    return true;
}
 
bool ProcessMemory::create(uint32 initialProcCount) {
    serial = master->incrementProcessShmemSerial();
    port = master->port;
 
//  LogPrint(0,LOG_MEMORY,0," --------- Creating ProcessMemory section serial %u --------", serial);
    if (!mutex)
        mutex = new utils::Mutex(utils::StringFormat("PsycloneProcessMemoryMutex_%u", port).c_str(), true);
    if (!mutex->enter(5000, __FUNCTION__))
        return false;
    
    memorySize = sizeof(ProcessMemoryStruct) + (initialProcCount * INITIALQSIZE * 4);
 
    //header = (ProcessMemoryStruct*) utils::OpenSharedMemorySegment(utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), memorySize);
    //if (header) {
    //  utils::CloseSharedMemorySegment((char*)header, memorySize);
    //  LogPrint(0,LOG_MEMORY,2,"ProcessMemory removing stale shared memory (%u/%u)...", port, serial);
    //}
 
    header = (ProcessMemoryStruct*) utils::CreateSharedMemorySegment(utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), memorySize, true);
    if (!header) {
        LogPrint(0,LOG_MEMORY,0,"Cannot create memory");
        mutex->leave();
        return false;
    }
    memset(header, 0, (size_t)memorySize);
    header->size = memorySize;
    header->cid = PROCESSMEMORYID;
    header->createdTime = GetTimeNow();
    header->usage = sizeof(ProcessMemoryStruct);
    // this->port = port;
 
//  printf(" --------- Creating new ProcessMap header: %u usage: %u serial %u --------\n", header->size, header->usage, serial);
 
    // statically initialise base process id 0
    ProcessInfoStruct* pInfo = &(header->processes[0]);
    pInfo->createdTime = pInfo->lastseen = GetTimeNow();
    pInfo->status = PSYPROC_ACTIVE;
    pInfo->nodeID = master->getNodeID();
    utils::strcpyavail(pInfo->name, "Node", MAXKEYNAMELEN, true);
 
    uint16 cmdQID;
    uint16 msgQID;
    uint16 sigQID;
    uint16 reqQID;
 
    if (
        !setupNextAvailableQ(cmdQID) ||
        !setupNextAvailableQ(msgQID) ||
        !setupNextAvailableQ(sigQID) ||
        !setupNextAvailableQ(reqQID) ) {
        mutex->leave();
        LogPrint(0,LOG_MEMORY,0,"Cannot setup queues");
        return false;
    }
 
    // reset as header might have changed...
    pInfo = &(header->processes[0]);
    pInfo->cmdQID = cmdQID;
    pInfo->msgQID = msgQID;
    pInfo->sigQID = sigQID;
    pInfo->reqQID = reqQID;
 
    master->setProcessShmemSize(memorySize);
 
//  printf(" --------- Created new ProcessMap header: %u usage: %u serial %u --------\n", header->size, header->usage, serial);
    mutex->leave();
    return true;
}
 
bool ProcessMemory::open() {
    serial = master->getProcessShmemSerial();
    uint64 size = master->getProcessShmemSize();
    this->port = master->port;
//  printf(" --------- Opening new ProcessMemory section serial %u --------\n", serial);
 
    bool createdMutex = false;
    if (!mutex) {
        //LogPrint(0,LOG_MEMORY,0," --------- Opening new ProcessMemory section serial %u --------", serial);
        mutex = new utils::Mutex(utils::StringFormat("PsycloneProcessMemoryMutex_%u", port).c_str());
        createdMutex = true;
        if (!mutex->enter(5000, __FUNCTION__))
            return false;
    }
    //else
    //  LogPrint(0,LOG_MEMORY,0," --------- Reopening ProcessMemory section serial %u --------", serial);
 
    ProcessMemoryStruct* newHeader = (ProcessMemoryStruct*) utils::OpenSharedMemorySegment(utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), size);
    if (!newHeader) {
        LogPrint(0,LOG_MEMORY,0,"Cannot open shared memory: %s size: %u", utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), size);
        if (createdMutex)
            mutex->leave();
        return false;
    }
    if (newHeader->cid != PROCESSMEMORYID) {
        LogPrint(0,LOG_MEMORY,0,"Cannot verify shared memory: %s size: %u", utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), size);
        utils::CloseSharedMemorySegment((char*)newHeader, size);
        if (createdMutex)
            mutex->leave();
        return false;
    }
 
    memorySize = size;
    if (header) 
        utils::CloseSharedMemorySegment((char*)header, header->size);
    header = newHeader;
    if (createdMutex)
        mutex->leave();
    // else leave mutex locked as we are calling from within the object
    return true;
}
 
 
 
 
 
 
 
 
// static 
bool ProcessMemory::createNewProcess(const char* name, uint16 &id) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    if (!header->size) {
        printf(" --------- Process header error: %llu usage: %llu serial %u --------\n", header->size, header->usage, serial);
        return false;
    }
 
    // Find first empty offset, 0 is the base and always reserved
    id = 0;
    ProcessInfoStruct* pInfo = &(header->processes[1]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (!pInfo->createdTime) {
            id = n;
            break;
        }
    }
    if (!id) {
        mutex->leave();
        return false;
    }
    pInfo->id = id;
    pInfo->createdTime = pInfo->lastseen = GetTimeNow();
    pInfo->status = PSYPROC_CREATED;
    pInfo->nodeID = master->getNodeID();
    utils::strcpyavail(pInfo->name, name, MAXKEYNAMELEN, true);
 
    uint16 cmdQID;
    uint16 msgQID;
    uint16 sigQID;
    uint16 reqQID;
 
    if (
        !setupNextAvailableQ(cmdQID) ||
        !setupNextAvailableQ(msgQID) ||
        !setupNextAvailableQ(sigQID) ||
        !setupNextAvailableQ(reqQID) ) {
        mutex->leave();
        return false;
    }
 
    // reset as header might have changed...
    pInfo = &(header->processes[id]);
    pInfo->cmdQID = cmdQID;
    pInfo->msgQID = msgQID;
    pInfo->sigQID = sigQID;
    pInfo->reqQID = reqQID;
 
    mutex->leave();
    return true;
}
 
// static
bool ProcessMemory::deleteProcess(uint16 id) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    // delete queues
    deleteQ(header->processes[id].cmdQID);
    deleteQ(header->processes[id].msgQID);
    deleteQ(header->processes[id].sigQID);
    deleteQ(header->processes[id].reqQID);
    memset(&(header->processes[id]), 0, sizeof(ProcessInfoStruct));
    return true;
}
 
// static
bool ProcessMemory::getProcessName(uint16 id, char* name, uint32 maxSize) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        return false;
    }
    utils::strcpyavail(name, header->processes[id].name, maxSize, true);
    mutex->leave();
    return true;
}
 
// static
bool ProcessMemory::getProcessID(const char* name, uint16 &id) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    ProcessInfoStruct* pInfo = &(header->processes[1]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if ((stricmp(name, pInfo->name) == 0) && pInfo->createdTime) {
            id = n;
            mutex->leave();
            return true;
        }
    }
    mutex->leave();
    return false;
}
 
// static
uint64 ProcessMemory::getProcessCreateTime(uint16 id) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
    uint64 t = header->processes[id].createdTime;
    mutex->leave();
    return t;
}
 
// static
bool ProcessMemory::getProcessCommandLine(uint16 id, char* cmdline, uint32 maxSize) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        return false;
    }
    utils::strcpyavail(cmdline, header->processes[id].commandline, maxSize, true);
    mutex->leave();
    return true;
}
 
// static
uint8 ProcessMemory::getProcessStatus(uint16 id, uint64& lastseen) {
    uint64 createTime;
    return getProcessStatus(id, lastseen, createTime);
}
 
uint8 ProcessMemory::getProcessStatus(uint16 id, uint64& lastseen, uint64& createTime) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        lastseen = 0;
        return 0;
    }
    uint8 status = header->processes[id].status;
    lastseen = header->processes[id].lastseen;
    createTime = header->processes[id].createdTime;
    mutex->leave();
    return status;
}
 
// static
uint16 ProcessMemory::getProcessIDFromOSID(uint32 osid) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
    ProcessInfoStruct* pInfo = &(header->processes[1]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (pInfo->osID == osid) {
            uint32 id = n;
            mutex->leave();
            return id;
        }
    }
    mutex->leave();
    return 0;
}
 
// static
uint32 ProcessMemory::getProcessOSID(uint16 id) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        return 0;
    }
    uint32 osID = header->processes[id].osID;
    mutex->leave();
    return osID;
}
 
// static
bool ProcessMemory::setProcessOSID(uint16 id, uint32 osid) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        return false;
    }
    header->processes[id].osID = osid;
    mutex->leave();
    return true;
}
 
// static
bool ProcessMemory::setProcessType(uint16 id, uint8 type) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        return false;
    }
    header->processes[id].type = type;
    mutex->leave();
    return true;
}
 
// static
bool ProcessMemory::setProcessCommandLine(uint16 id, const char* cmdline) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    if (!header->processes[id].createdTime) {
        mutex->leave();
        return false;
    }
    utils::strcpyavail(header->processes[id].commandline, cmdline, MAXCOMMANDLINELEN, false);
    mutex->leave();
    return true;
}
 
// static
bool ProcessMemory::setProcessStatus(uint16 id, uint8 status, uint64 currentCPUTicks) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    ProcessInfoStruct* pInfo = &header->processes[id]; 
    if (!pInfo->createdTime) {
        mutex->leave();
        return false;
    }
    pInfo->status = status;
    pInfo->lastseen = GetTimeNow();
    pInfo->stats.procMemUsage = utils::GetProcessMemoryUsage();
    if (currentCPUTicks)
        pInfo->stats.currentCPUTicks = currentCPUTicks;
    mutex->leave();
    return true;
}
 
AveragePerfStats ProcessMemory::getProcessPerfStats(uint16 procID) {
    AveragePerfStats perfStats;
    memset(&perfStats, 0, sizeof(AveragePerfStats));
 
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return perfStats;
    if (serial != master->getProcessShmemSerial()) {if (!open()) {mutex->leave();return perfStats;}}
 
    ProcessInfoStruct* pInfo = &header->processes[procID]; 
    if (!pInfo->createdTime) {
        mutex->leave();
        return perfStats;
    }
    memcpy(&perfStats, &pInfo->perfStats, sizeof(AveragePerfStats));
    mutex->leave();
    return perfStats;
}
 
bool ProcessMemory::setProcessPerfStats(uint16 procID, AveragePerfStats &perfStruct) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    ProcessInfoStruct* pInfo = &header->processes[procID]; 
    if (!pInfo->createdTime) {
        mutex->leave();
        return false;
    }
    memcpy(&pInfo->perfStats, &perfStruct, sizeof(AveragePerfStats));
    mutex->leave();
    return true;
}
 
 
bool ProcessMemory::addToProcessStats(uint16 id, DataMessage* inputMsg, DataMessage* outputMsg) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
    ProcessInfoStruct* pInfo = &header->processes[id]; 
    if (!pInfo->createdTime) {
        mutex->leave();
        return false;
    }
 
    pInfo->stats.time = GetTimeNow();
    DataMessage* draft;
    if (inputMsg) {
        pInfo->stats.msgInCount++;
        pInfo->stats.msgInBytes += inputMsg->getSize();
        // Circular ring: write the newest draft to the head slot and advance, instead
        // of memmove-shifting the whole 10-deep ring (9*DRAFTMSGSIZE = 36KB) per message.
        // Readers (toXML, PsyProbe) iterate all slots, so slot order doesn't matter.
        char* inSlot = pInfo->stats.recentInMsg + ((pInfo->stats.recentInPos % 10) * DRAFTMSGSIZE);
        if (inputMsg->data->size < DRAFTMSGSIZE)
            memcpy(inSlot, inputMsg->data, inputMsg->data->size);
        else {
            draft = new DataMessage(*inputMsg, DRAFTMSGSIZE);
            memcpy(inSlot, draft->data, draft->data->size);
            delete(draft);
        }
        pInfo->stats.recentInPos = (pInfo->stats.recentInPos + 1) % 10;
    }
    if (outputMsg) {
        pInfo->stats.msgOutCount++;
        pInfo->stats.msgOutBytes += outputMsg->getSize();
        char* outSlot = pInfo->stats.recentOutMsg + ((pInfo->stats.recentOutPos % 10) * DRAFTMSGSIZE);
        if (outputMsg->data->size < DRAFTMSGSIZE)
            memcpy(outSlot, outputMsg->data, outputMsg->data->size);
        else {
            draft = new DataMessage(*outputMsg, DRAFTMSGSIZE);
            memcpy(outSlot, draft->data, draft->data->size);
            delete(draft);
        }
        pInfo->stats.recentOutPos = (pInfo->stats.recentOutPos + 1) % 10;
    }
 
    mutex->leave();
    return true;
}
 
bool ProcessMemory::addToCmdQ(uint16 procID, DataMessage* msg) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    if (!addToQ(procID, CMDQ_TYPE, msg)) {
        mutex->leave();
        return false;
    }
    mutex->leave();
    return true;
}
 
bool ProcessMemory::addToMsgQ(uint16 procID, DataMessage* msg) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    if (!addToQ(procID, MSGQ_TYPE, msg)) {
        mutex->leave();
        return false;
    }
    mutex->leave();
    return true;
}
 
bool ProcessMemory::addToSigQ(uint16 procID, DataMessage* msg) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    if (!addToQ(procID, SIGQ_TYPE, msg)) {
        mutex->leave();
        return false;
    }
    mutex->leave();
    return true;
}
 
bool ProcessMemory::addToAllSignalQs(DataMessage* msg) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    ProcessInfoStruct* pInfo = &(header->processes[0]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (pInfo->createdTime)
            addToQ(pInfo->id, SIGQ_TYPE, msg);
    }
 
    mutex->leave();
    return true;
}
 
bool ProcessMemory::addToAllSignalQsExcept(DataMessage* msg, uint16 except) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    ProcessInfoStruct* pInfo = &(header->processes[0]);
    for (uint32 n=1; n<MAXPROC; n++, pInfo++) {
        if (pInfo->createdTime && (pInfo->id != except))
            addToQ(pInfo->id, SIGQ_TYPE, msg);
    }
 
    mutex->leave();
    return true;
}
 
bool ProcessMemory::addToReqQ(uint16 procID, DataMessage* msg) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
    CHECKPROCESSMEMORYSERIAL
 
    if (!addToQ(procID, REQQ_TYPE, msg)) {
        mutex->leave();
        return false;
    }
    mutex->leave();
    return true;
}
 
 
DataMessage* ProcessMemory::waitForCmdQ(uint16 procID, uint32 timeout) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return NULL;
    CHECKPROCESSMEMORYSERIAL
    DataMessage* msg = waitForQ(procID, CMDQ_TYPE, timeout);
    mutex->leave();
    return msg;
}
 
DataMessage* ProcessMemory::waitForMsgQ(uint16 procID, uint32 timeout) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return NULL;
    CHECKPROCESSMEMORYSERIAL
    DataMessage* msg = waitForQ(procID, MSGQ_TYPE, timeout);
    mutex->leave();
    return msg;
}
 
DataMessage* ProcessMemory::waitForSigQ(uint16 procID, uint32 timeout) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return NULL;
    CHECKPROCESSMEMORYSERIAL
    DataMessage* msg = waitForQ(procID, SIGQ_TYPE, timeout);
    mutex->leave();
    return msg;
}
 
DataMessage* ProcessMemory::waitForReqQ(uint16 procID, uint32 timeout) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return NULL;
    CHECKPROCESSMEMORYSERIAL
    DataMessage* msg = waitForQ(procID, REQQ_TYPE, timeout);
    mutex->leave();
    return msg;
}
 
uint32 ProcessMemory::getCmdQCount(uint16 procID) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
 
    MessageQueueHeader* qHeader = getQHeader(procID, CMDQ_TYPE);
    if (!qHeader) {
        mutex->leave();
        return 0;
    }
    uint32 count = qHeader->count;
    mutex->leave();
    return count;
}
 
uint32 ProcessMemory::getMsgQCount(uint16 procID) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
 
    MessageQueueHeader* qHeader = getQHeader(procID, MSGQ_TYPE);
    if (!qHeader) {
        mutex->leave();
        return 0;
    }
    uint32 count = qHeader->count;
    mutex->leave();
    return count;
}
 
uint32 ProcessMemory::getSigQCount(uint16 procID) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
 
    MessageQueueHeader* qHeader = getQHeader(procID, SIGQ_TYPE);
    if (!qHeader) {
        mutex->leave();
        return 0;
    }
    uint32 count = qHeader->count;
    mutex->leave();
    return count;
}
 
uint32 ProcessMemory::getReqQCount(uint16 procID) {
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return 0;
    CHECKPROCESSMEMORYSERIAL
 
    MessageQueueHeader* qHeader = getQHeader(procID, REQQ_TYPE);
    if (!qHeader) {
        mutex->leave();
        return 0;
    }
    uint32 count = qHeader->count;
    mutex->leave();
    return count;
}
 
MessageQueueHeader* ProcessMemory::getQHeader(uint16 procID, uint8 qType) {
    if (!header->processes[procID].createdTime)
        return NULL;
    uint16 qID;
    switch(qType) {
        case CMDQ_TYPE:
            qID = header->processes[procID].cmdQID;
            break;
        case MSGQ_TYPE:
            qID = header->processes[procID].msgQID;
            break;
        case SIGQ_TYPE:
            qID = header->processes[procID].sigQID;
            break;
        case REQQ_TYPE:
            qID = header->processes[procID].reqQID;
            break;
        default:
            return NULL;
    }
    if (!header->qIndex[qID])
        return NULL;
 
    MessageQueueHeader* qHeader = (MessageQueueHeader*) (((char*)header) + header->qIndex[qID]);
    if (qHeader->id != qID)
        return NULL;
    return qHeader;
}
 
 
 
bool ProcessMemory::setupNextAvailableQ(uint16& qID) {
    qID = 0;
    uint16 nextID = 0;
    bool found = false;
    for (uint32 n=0; n<MAXPROC4; n++) {
        if (!found && !header->qIndex[n]) {
            qID = n;
            found = true;
        }
        else if (found && header->qIndex[n]) {
            nextID = n;
            break;
        }
    }
    if (!found)
        return false;
 
    // Insert a new standard size queue in place
    uint64 size = INITIALQSIZE;
    // Do we have enough space free at the end?
    if (header->size - header->usage < size) {
        LogPrint(0,LOG_MEMORY,0,"SetupQ need more memory size: %u use: %u need: %u", header->size, header->usage, size);
        // Resize whole memory block
        if (!resize(memorySize * 2)) {
            LogPrint(0,LOG_MEMORY,0,"Cannot resize ProcessMemory");
            return false;
        }
    }
 
    char* data;
    // If any active queues follow the new queue
    if (nextID) {
        // Push every subsequent queue by size
        header->qIndex[qID] = header->qIndex[nextID];
        data = ((char*)header) + header->qIndex[nextID];
        uint64 sizeMove = header->usage - header->qIndex[nextID];
        memmove(data+size, data, (size_t)sizeMove);
        // Adjust indexes
        for (uint32 n=nextID; n<MAXPROC4; n++) {
            if (header->qIndex[n])
                header->qIndex[n] += size;
        }
    }
    else {
        header->qIndex[qID] = header->usage;
        data = ((char*)header) + header->usage;
    }
 
    // We now have space, a start point and the indexes filled in
    memset(data, 0, (size_t)size); // could just be the header
    ((MessageQueueHeader*)data)->size = size;
    ((MessageQueueHeader*)data)->id = qID;
    snprintf(((MessageQueueHeader*)data)->name, MAXKEYNAMELEN, "PsycloneProcessMemoryQueue_%u_%u", port, qID);
//  snprintf(((MessageQueueHeader*)data)->name, MAXKEYNAMELEN, "%u_%u_PsycloneProcessMemoryQueue", port, qID);
 
    header->usage += size;
    return true;
}
 
bool ProcessMemory::deleteQ(uint16 id) {
    if (!header->qIndex[id])
        return false;
    MessageQueueHeader* qHeader = (MessageQueueHeader*)(((char*)header) + header->qIndex[id]);
    uint64 size = qHeader->size;
 
    // Check if any data comes after this queue
    uint16 nextID = 0;
    if (id < MAXPROC4-2) {
        for (uint32 n=id+1; n<MAXPROC4; n++) {
            if (header->qIndex[n]) {
                if (!nextID) {
                    nextID = n;
                    // move memory
                    memmove(qHeader, ((char*)qHeader)+size, (size_t)(header->usage - header->qIndex[nextID]));
                }
                header->qIndex[n] -= size;
            }
        }
    }
    
    header->qIndex[id] = 0;
    header->usage -= size;
    return true;
}
 
 
bool ProcessMemory::addToQ(uint16 procID, uint8 qType, DataMessage* msg) {
    MessageQueueHeader* qHeader = getQHeader(procID, qType);
    if (!qHeader)
        return false;
 
    // ⚠️ If you ever need to identify a queued message from inside this function (as the
    // BakeLive Anomaly investigation did), note two traps:
    //   - `msg` here is usually the CTRL_TRIGGER WRAPPER from
    //     Node::deliverTriggerMessage. Its serial was NEVER SET before that function was
    //     changed to stamp it, so any trace keyed on msg->getSerial() reads 0 for every
    //     write and cannot tell one message from another.
    //   - reading the attached PAYLOAD's serial here is not cheap: getData() only returns
    //     entries whose cid is CHARDATAID, but setAttachedMessage stores the payload as
    //     DATAMESSAGEID (setRawData), so getData("Message") returns NULL.
    //     getAttachedMessageCopy() works but ALLOCATES - unacceptable on this path.
    // The wrapper is now stamped with its payload's serial at construction, so reading
    // msg->getSerial() is the cheap correct answer. @see BakeLiveBug.md
    uint32 msgSize = msg->getSize();
    uint64 spaceToEnd, inUse = 0, inUse2 = 0;
    char* qData = ((char*)qHeader) + sizeof(MessageQueueHeader);
    uint64 qDataSize = qHeader->size - sizeof(MessageQueueHeader);
 
    //if (qHeader->endPos >= qHeader->startPos)
    //  inUse = qHeader->endPos - qHeader->startPos;
    //else
    //  inUse = qDataSize - qHeader->startPos - qHeader->padding + qHeader->endPos;
 
    if (qHeader->endPos == qHeader->startPos) {
        spaceToEnd = qDataSize;
        if (spaceToEnd >= msgSize) {
            // add msg at buffer start
            memcpy(qData, msg->data, msgSize);
            qHeader->startPos = 0;
            qHeader->endPos = msgSize;
            qHeader->count++;
            qHeader->padding = 0;
        }
        else {
            // we need to resize queue
            if (!resizeQ(qHeader->id, qHeader->size*2)) {
                // ⚠️ Full AND cannot grow: the CALLER will drop this message, and the
                // only trace is a level-0 log there which the test framework suppresses.
                // A message lost this way is invisible to every instrument. @see
                // BakeLiveBug.md - this was one of the sites traced during that hunt.
                return false;
            }
            // call this function again...
            return addToQ(procID, qType, msg);
        }
    }
    else if (qHeader->endPos > qHeader->startPos) {
        spaceToEnd = qDataSize - qHeader->endPos;
        if (spaceToEnd >= msgSize) {
            // add msg here
            memcpy(qData+qHeader->endPos, msg->data, msgSize);
            qHeader->endPos += msgSize;
            qHeader->count++;
            qHeader->padding = 0;
        }
        else if (qHeader->startPos > msgSize) {
            // fill end rest of space with 0
            memset(qData+qHeader->endPos, 0, (size_t)spaceToEnd);
            qHeader->padding = spaceToEnd;
            // add msg at buffer start
            memcpy(qData, msg->data, msgSize);
            qHeader->endPos = msgSize;
            qHeader->count++;
        }
        else {
            // we need to resize queue
            if (!resizeQ(qHeader->id, qHeader->size*2)) {
                // ⚠️ Full AND cannot grow: the CALLER will drop this message, and the
                // only trace is a level-0 log there which the test framework suppresses.
                // A message lost this way is invisible to every instrument. @see
                // BakeLiveBug.md - this was one of the sites traced during that hunt.
                return false;
            }
            // call this function again...
            return addToQ(procID, qType, msg);
        }
    }
    else {
        spaceToEnd = qHeader->startPos - qHeader->endPos;
        if (spaceToEnd > msgSize) {
            // add msg here
            memcpy(qData+qHeader->endPos, msg->data, msgSize);
            qHeader->endPos += msgSize;
            qHeader->count++;
            // NOTE: this branch, unlike the two above, does NOT reset padding. If a
            // wrap earlier set padding non-zero, it stays set while endPos now advances
            // in the wrapped region - and waitForQ's end-of-buffer test subtracts
            // padding. Recording padding here is the point of the trace.
        }
        else {
            // we need to resize queue
            if (!resizeQ(qHeader->id, qHeader->size*2)) {
                // ⚠️ Full AND cannot grow: the CALLER will drop this message, and the
                // only trace is a level-0 log there which the test framework suppresses.
                // A message lost this way is invisible to every instrument. @see
                // BakeLiveBug.md - this was one of the sites traced during that hunt.
                return false;
            }
            // call this function again...
            return addToQ(procID, qType, msg);
        }
    }
 
    //if (qHeader->endPos >= qHeader->startPos)
    //  inUse2 = qHeader->endPos - qHeader->startPos;
    //else
    //  inUse2 = qDataSize - qHeader->startPos - qHeader->padding + qHeader->endPos;
    //if (inUse2 < (qHeader->count * msgSize))
    //  int ee = 0;
 
    utils::Semaphore** sem = &qSemaphores[qHeader->id];
    if (!*sem)
        if (!(*sem = utils::GetSemaphore(qHeader->name)))
            return false;
    (*sem)->signal();
 
//  utils::SignalSemaphore(qHeader->name);
//  printf("Signal %s at %s\n", qHeader->name, PrintTimeSortableMicrosecString(GetTimeNow(), true).c_str());
    return true;
}
 
DataMessage* ProcessMemory::waitForQ(uint16 procID, uint8 qType, uint32 timeout) {
    //char* name = NULL;
    MessageQueueHeader* qHeader = getQHeader(procID, qType);
    if (!qHeader)
        return NULL;
 
    uint64 start = GetTimeNow();
    int32 remain;
    while (!qHeader->count) {
        utils::Semaphore** sem = &qSemaphores[qHeader->id];
        if (!*sem) {
            if (!(*sem = utils::GetSemaphore(qHeader->name))) {
                return NULL;
            }
        }
 
        //if (!name) {
        //  name = new char[strlen(qHeader->name)+1];
        //  utils::strcpyavail(name, qHeader->name, (uint32)strlen(qHeader->name)+1, true);
        //}
        if ((remain = timeout - GetTimeAgeMS(start)) <= 0) {
        //  printf("StartWait1 %s at %s\n", qHeader->name, PrintTimeSortableMicrosecString(start, true).c_str());
        //  printf("Timeout1 %s at %s\n", qHeader->name, PrintTimeSortableMicrosecString(GetTimeNow(), true).c_str());
        //  delete [] name;
            return NULL;
        }
        mutex->leave();
        bool result = (*sem)->wait(remain);
        if (!mutex || !mutex->enter(5000, __FUNCTION__))
            return NULL;
        CHECKPROCESSMEMORYSERIAL
 
        if (!result) {
        //  printf("StartWait2 %s at %s\n", qHeader->name, PrintTimeSortableMicrosecString(start, true).c_str());
        //  printf("Timeout2 %s at %s\n", qHeader->name, PrintTimeSortableMicrosecString(GetTimeNow(), true).c_str());
        //  delete [] name;
            // printf("Waited %.3fms (%ums)...\n", GetTimeAge(start)/1000.0, remain);
            return NULL;
        }
        
        // header might be changed
        qHeader = getQHeader(procID, qType);
        if (!qHeader) {
        //  mutex->leave();
        //  delete [] name;
            return NULL;
        }
    }
    //delete [] name;
 
    // ⚠️ LATENT (not GAP4): qHeader is computed at the top of this function, and the
    // `while (!qHeader->count)` loop above is SKIPPED ENTIRELY when the queue is
    // non-empty on entry - so the re-fetch inside that loop ("header might be
    // changed") never runs on the fast path. Meanwhile ProcessMemory::resize() does
    // not grow the mapping in place: it creates a NEW segment, memcpy's, unmaps the
    // old one and repoints `header`, so anything derived from the old header dangles.
    //
    // Adding `CHECKPROCESSMEMORYSERIAL` + a re-fetch here was TRIED as the GAP4 fix
    // and MEASURABLY DID NOT FIX IT: with the guard in place bakelive still failed
    // with byte-identical figures (published=64 distinct=58 missing=6 duped=8, same
    // six serials). So this is NOT the GAP4 mechanism, and the guard was reverted
    // rather than left in place looking like a fix. It is still a real latent hazard
    // worth closing on its own merits, with its own test - see Psyclone.md.
    DataMessage* msg = NULL;
    if (qHeader->count) {
        // ⚠️ READ-POINTER CONTINUITY was measured here during the BakeLive Anomaly hunt:
        // does every read take its message from the offset the PREVIOUS read of this queue
        // left the pointer at? Result: CLEAN, including inside genuine failing runs - the
        // pointer never rewound and never advanced by anything other than msg->getSize().
        // The instrumentation was removed with the rest of that investigation.
        // If you re-add it: record in memory only. A file append on this path once took
        // bakelive from failing 1-in-26 to 76 consecutive passes - the probe hid the bug.
        // @see BakeLiveBug.md
        msg = new DataMessage(((char*)qHeader) + sizeof(MessageQueueHeader) + qHeader->startPos, true);
        qHeader->startPos += msg->getSize();
        if (qHeader->startPos >= qHeader->size - sizeof(MessageQueueHeader) - qHeader->padding) {
            qHeader->startPos = 0;
            qHeader->padding = 0;
        }
        qHeader->count--;
//if (qHeader->count && (GetObjID(((char*)qHeader) + sizeof(MessageQueueHeader) + qHeader->startPos) != DATAMESSAGEID)) {
//  printQ(qHeader);
//  return false;
//}
    }
 
    return msg;
}
 
bool ProcessMemory::printQ(MessageQueueHeader* qHeader) {
    printf("[%u] Data Size: %llu, count: %u startpos: %llu, endpos: %llu, padding: %llu\n",
        qHeader->id, (uint64)(qHeader->size - sizeof(MessageQueueHeader)), qHeader->count, qHeader->startPos, qHeader->endPos, qHeader->padding);
    char* data = ((char*)qHeader) + sizeof(MessageQueueHeader) + qHeader->startPos;
    for (uint32 n=0; n<qHeader->count; n++) {
        if (GetObjID(data) == DATAMESSAGEID)
            printf(" [%u: %llu - %llu] Msg size: %u Time: %s Type: %u.%u.%u\n", n,
                (uint64)(data - ((char*)qHeader) - sizeof(MessageQueueHeader)),
                (uint64)(data - ((char*)qHeader) - sizeof(MessageQueueHeader) + ((DataMessageHeader*)data)->size),
                ((DataMessageHeader*)data)->size, PrintTimeString(((DataMessageHeader*)data)->time).c_str(),
                (((DataMessageHeader*)data)->type).levels[0],
                (((DataMessageHeader*)data)->type).levels[1],
                (((DataMessageHeader*)data)->type).levels[2]);
        else {
            printf(" [%u: %llu] Msg corrupted\n",
                n, (uint64)(data - ((char*)qHeader) - sizeof(MessageQueueHeader)));
            return false;
        }
        data += ((DataMessageHeader*)data)->size;
        if (data == ((char*)qHeader) + qHeader->size - qHeader->padding)
            data = ((char*)qHeader) + sizeof(MessageQueueHeader);
    }
    return true;
}
 
bool ProcessMemory::resize(uint64 newMemorySize) {
 
    if (newMemorySize > 100000000) {
        //LogPrint(0,LOG_MEMORY,0,"Memory Error: Process Memory requested size %s denied...", utils::BytifySize((double)newMemorySize).c_str());
        printf("Memory Error: Process Memory requested size %s denied...\n", utils::BytifySize((double)newMemorySize).c_str());
        fflush(stdout);
        return false;
    }
 
    serial = master->incrementProcessShmemSerial();
    //LogPrint(0,LOG_MEMORY,0," --------- Resizing ProcessMemory section %u -> %u serial %u --------", memorySize, newMemorySize, serial);
 
    if (!mutex || !mutex->enter(5000, __FUNCTION__))
        return false;
 
    //ProcessMemoryStruct* newHeader = (ProcessMemoryStruct*) utils::OpenSharedMemorySegment(utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), newMemorySize);
    //if (newHeader) {
    //  utils::CloseSharedMemorySegment((char*)newHeader, newMemorySize);
    //  //LogPrint(0,LOG_MEMORY,2,"ProcessMemory removing stale shared memory (%u/%u)...", port, serial);
    //}
 
    ProcessMemoryStruct* newHeader = (ProcessMemoryStruct*) utils::CreateSharedMemorySegment(utils::StringFormat("PsycloneProcessMemory_%u_%u", port, serial).c_str(), newMemorySize, true);
    if (!newHeader) {
        mutex->leave();
        return false;
    }
    memcpy(newHeader, header, (size_t)header->size);
    newHeader->size = newMemorySize;
 
    utils::CloseSharedMemorySegment((char*)header, memorySize);
    header = newHeader;
    memorySize = newMemorySize;
    master->setProcessShmemSize(memorySize);
    mutex->leave();
 
    LogPrint(0,LOG_MEMORY,2,"Process Memory resized to %s...", utils::BytifySize((double)newMemorySize).c_str());
    if (newMemorySize > 20000000) {
        LogPrint(0,LOG_MEMORY,2,"Memory Warning: Process Memory now %s...", utils::BytifySize((double)newMemorySize).c_str());
    }
 
    return true;
}
 
bool ProcessMemory::resizeQ(uint16 id, uint64 newSize) {
    if (!header->qIndex[id])
        return false;
    MessageQueueHeader* qHeader = (MessageQueueHeader*) (((char*)header) + header->qIndex[id]);
    if (qHeader->id != id)
        return false;
 
    uint64 difSize = newSize - qHeader->size;
    if (header->size - header->usage < difSize) {
        if (!resize(memorySize * 2))
            return false;
        return resizeQ(id, newSize);
    }
 
    uint16 nextID = 0;
    uint32 n;
    for (n=id+1; n<MAXPROC4; n++) {
        if (header->qIndex[n]) {
            nextID = n;
            break;
        }
    }
 
    // shift qID > id down by difsize
    if (nextID) {
        // Push every subsequent queue by size
        char* data = ((char*)header) + header->qIndex[nextID];
        uint64 sizeMove = header->usage - header->qIndex[nextID];
        memmove(data+difSize, data, (size_t)sizeMove);
        // Adjust indexes
        for (n=nextID; n<MAXPROC4; n++) {
            if (header->qIndex[n])
                header->qIndex[n] += difSize;
        }
    }
 
    // Now rearrange messages in queue
    int64 size = qHeader->endPos - qHeader->startPos;
    uint64 firstSize = 0;
    uint64 secondSize = 0;
    if (size == 0)
        qHeader->endPos = qHeader->startPos = 0;
    else if ((size > 0) && qHeader->startPos) {
        memmove(
            ((char*)qHeader) + sizeof(MessageQueueHeader),
            ((char*)qHeader) + sizeof(MessageQueueHeader) + qHeader->startPos,
            (size_t)size);
        qHeader->endPos = (uint64)size;
        qHeader->startPos = 0;
    }
    else if (size < 0) {
        firstSize = qHeader->size - sizeof(MessageQueueHeader) - qHeader->startPos - qHeader->padding;
        secondSize = qHeader->endPos;
 
        // is there enough space to move things around without a safety copy involving a malloc?
        if (difSize > secondSize) {
            // move first part of memory to the newly available space at the end
            memmove(
                ((char*)qHeader) + qHeader->size,
                ((char*)qHeader) + sizeof(MessageQueueHeader) + qHeader->startPos,
                (size_t)firstSize);
            // move last part of memory to the right place
            memmove(
                ((char*)qHeader) + sizeof(MessageQueueHeader) + firstSize,
                ((char*)qHeader) + sizeof(MessageQueueHeader),
                (size_t)secondSize);
            // copy first part back
            memmove(
                ((char*)qHeader) + sizeof(MessageQueueHeader),
                ((char*)qHeader) + qHeader->size,
                (size_t)firstSize);
        }
        else {
            // make safety copy of first part
            char* copy = (char*)malloc((size_t)firstSize);
            memcpy(copy, ((char*)qHeader) + sizeof(MessageQueueHeader) + qHeader->startPos, (size_t)firstSize);
            // move last part of memory to the right place
            memmove(
                ((char*)qHeader) + sizeof(MessageQueueHeader) + firstSize,
                ((char*)qHeader) + sizeof(MessageQueueHeader),
                (size_t)secondSize);
            // copy first part back from the safety copy
            memcpy(((char*)qHeader) + sizeof(MessageQueueHeader), copy, (size_t)firstSize);
            free(copy);
        }
        qHeader->endPos = firstSize + secondSize;
        qHeader->startPos = 0;
    }
 
    qHeader->size = newSize;
    qHeader->padding = 0;
    header->usage += difSize;
 
//if (qHeader->count) {
//  printQ(qHeader);
//  return false;
//}
 
    return true;
}
 
 
bool ProcessMemory::UnitTest() {
    unittest::progress(0, "init memory manager");
 
    // First create and initialise the MemoryManager
    MemoryManager* manager = new MemoryManager();
    if (!manager->create(0)) {
        unittest::fail("MemoryManager create(0) failed");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    uint32 count = 10000;
    uint32 writeCount = 0;
    uint32 readCount = 0;
    uint32 n;
    DataMessage* msg;
    char str[128];
    const char* cstr;
 
    uint32 queueTestThreadID;
    g_processTestStop = false;
    if (!ThreadManager::CreateThread(QueueTest, manager, queueTestThreadID)) {
        unittest::fail("Could not create Queue Test thread");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
    g_processTestThreadID = queueTestThreadID;
 
    uint64 time = GetTimeNow();
 
    // empty-queue wait must actually block for ~1s
    unittest::progress(10, "empty queue wait");
    if ( (msg = manager->processMemory->waitForSigQ(0, 1000)) != NULL) {
        unittest::fail("Got unexpected Message from empty Queue");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    uint64 t = 0;
    if ( (t = GetTimeAge(time)) < 950000) {
        unittest::fail("Message Queue wait 1000ms didn't actually wait more than %.3fms", (t/1000.0));
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    // Single-threaded round-trip latency (same thread adds + reads sig queue)
    unittest::progress(25, "single-thread latency");
    const uint32 latIters = 20000;
    uint64 singleTotal = 0;
    t = 0;
    for (n=0; n<latIters; n++) {
        msg = new DataMessage(CTRL_TEST, 0);
        if (!manager->processMemory->addToSigQ(0, msg)) {
            unittest::fail("Could not add Message %u to SigQueue", n);
            delete(manager);
            return false;
        }
        delete(msg);
        if ( (msg = manager->processMemory->waitForSigQ(0, 1000)) == NULL) {
            unittest::fail("Could not get Message [%u] from SigQueue", n);
            delete(manager);
            return false;
        }
        t += GetTimeAge(msg->getCreatedTime());
        delete(msg);
        if (n && n % 10000 == 0) {
            unittest::detail("Single threaded message delay[%u]: %.3fus", n, ((double)t)/(2*10000.0));
            singleTotal += t;
            t = 0;
        }
    }
    singleTotal += t;
    // each iteration measures the created-time age twice (add + read)
    double singleLatency = (double)singleTotal / (2.0 * latIters);
    unittest::metric("single_thread_latency", singleLatency, "us", false);
 
    // warm up the queue (cross-thread path via cmd queue -> slave -> sig queue)
    msg = new DataMessage(CTRL_TEST, 0);
    if (!manager->processMemory->addToCmdQ(0, msg)) {
        unittest::fail("Could not add startup Message to Queue");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
    delete(msg);
    utils::Sleep(10);
    if ( (msg = manager->processMemory->waitForSigQ(0, 1000)) == NULL) {
        unittest::fail("Could not get startup Message from Queue");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
    delete(msg);
 
    // Multi-threaded round-trip latency (slave thread relays cmd -> sig)
    unittest::progress(45, "multi-thread latency");
    uint64 multiTotal = 0;
    t = 0;
    for (n=0; n<latIters; n++) {
        msg = new DataMessage(CTRL_TEST, 0);
        if (!manager->processMemory->addToCmdQ(0, msg)) {
            unittest::fail("Could not add Message %u to CmdQueue", n);
            delete(manager);
            return false;
        }
        delete(msg);
        if ( (msg = manager->processMemory->waitForSigQ(0, 1000)) == NULL) {
            unittest::fail("Could not get Message [%u] from SigQueue", n);
            delete(manager);
            return false;
        }
        t += GetTimeAge(msg->getCreatedTime());
        delete(msg);
        if (n && n % 10000 == 0) {
            unittest::detail("Multi threaded message delay[%u]: %.3fus", n, ((double)t)/(2*10000.0));
            multiTotal += t;
            t = 0;
        }
    }
    multiTotal += t;
    double multiLatency = (double)multiTotal / (2.0 * latIters);
    unittest::metric("multi_thread_latency", multiLatency, "us", false);
 
 
 
 
 
 
    // Message queue correctness + write throughput
    unittest::progress(65, "msg queue fill");
    time = GetTimeNow();
    uint64 writeStart = GetTimeNow();
    for (n=0; n<count/4; n++) {
        msg = new DataMessage(CTRL_TEST, writeCount);
        snprintf(str, 128, "Test%2u", writeCount);
        msg->setString("TestEntry", str);
        msg->setTime("TestTime", time);
        msg->setString("TestEntry2", str);
        msg->setTime("TestTime2", time);
        if (!manager->processMemory->addToMsgQ(0, msg)) {
            unittest::fail("Could not add Message %u to MsgQueue", n);
            delete(manager);
            return false;
        }
    //  if (n == 1000)
    //      utils::PrintBinary(msg->data, msg->data->size, false, "Msg1000");
        delete(msg);
        writeCount++;
    }
 
    if ( (n = manager->processMemory->getMsgQCount(0)) != count/4) {
        unittest::fail("Queue contains %u messages instead of %u", n, count/4);
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    unittest::progress(75, "msg queue drain");
    for (n=0; n<count/8; n++) {
        if ( (msg = manager->processMemory->waitForMsgQ(0, 1000)) == NULL) {
            unittest::fail("Could not get Message %u from Queue", n);
            delete(manager);
            return false;
        }
        snprintf(str, 128, "Test%2u", readCount);
        if (strcmp(str, msg->getString("TestEntry")) != 0) {
            unittest::fail("Message %u from Queue corrupted string '%s'", readCount, msg->getString("TestEntry"));
            delete(msg);
            delete(manager);
            return false;
        }
        if (msg->getTime("TestTime") != time) {
            unittest::fail("Message %u from Queue corrupted time '%llu'", readCount, msg->getTime("TestTime"));
            delete(msg);
            delete(manager);
            return false;
        }
        cstr = msg->getString("TestEntry2");
        if (!cstr || strcmp(str, cstr) != 0) {
            unittest::fail("Message %u from Queue corrupted string '%s'", readCount, cstr);
            delete(msg);
            delete(manager);
            return false;
        }
        if (msg->getTime("TestTime2") != time) {
            unittest::fail("Message %u from Queue corrupted time '%llu'", readCount, msg->getTime("TestTime2"));
            delete(msg);
            delete(manager);
            return false;
        }
        readCount++;
        delete(msg);
    }
 
    if ( (n = manager->processMemory->getMsgQCount(0)) != count/8) {
        unittest::fail("Queue contains %u messages instead of %u", n, count/8);
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    unittest::progress(85, "msg queue refill");
    for (n=0; n<3*count/4; n++) {
        msg = new DataMessage(CTRL_TEST, writeCount);
        snprintf(str, 128, "Test%2u", writeCount);
        msg->setString("TestEntry", str);
        msg->setTime("TestTime", time);
        msg->setString("TestEntry2", str);
        msg->setTime("TestTime2", time);
        if (!manager->processMemory->addToMsgQ(0, msg)) {
            unittest::fail("Could not add Message %u to MsgQueue", n);
            delete(manager);
            return false;
        }
        delete(msg);
        writeCount++;
    }
 
    if ( (n = manager->processMemory->getMsgQCount(0)) != 7*count/8) {
        unittest::fail("Queue contains %u messages instead of %u", n, 7*count/8);
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    unittest::progress(92, "msg queue verify");
    for (n=0; n<7*count/8; n++) {
        if ( (msg = manager->processMemory->waitForMsgQ(0, 1000)) == NULL) {
            unittest::fail("Could not get Message %u from Queue", n);
            delete(manager);
            return false;
        }
        snprintf(str, 128, "Test%2u", readCount);
        if (strcmp(str, msg->getString("TestEntry")) != 0) {
            unittest::fail("Message %u from Queue corrupted string '%s' (queue has %u)", readCount, msg->getString("TestEntry"), manager->processMemory->getMsgQCount(0));
            delete(msg);
            delete(manager);
            return false;
        }
        if (msg->getTime("TestTime") != time) {
            unittest::fail("Message %u from Queue corrupted time '%llu'", readCount, msg->getTime("TestTime"));
            delete(msg);
            delete(manager);
            return false;
        }
        cstr = msg->getString("TestEntry2");
        if (!cstr || strcmp(str, cstr) != 0) {
            unittest::fail("Message %u from Queue corrupted string '%s'", readCount, cstr);
            delete(msg);
            delete(manager);
            return false;
        }
        if (msg->getTime("TestTime2") != time) {
            unittest::fail("Message %u from Queue corrupted time '%llu'", readCount, msg->getTime("TestTime2"));
            delete(msg);
            delete(manager);
            return false;
        }
        readCount++;
        delete(msg);
    }
 
    if ( (n = manager->processMemory->getMsgQCount(0)) != 0) {
        unittest::fail("Queue contains %u messages instead of 0", n);
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    // Aggregate write throughput over all enqueued messages (count/4 + 3*count/4)
    double writeUs = (double)(GetTimeNow() - writeStart);
    if (writeUs > 0.0)
        unittest::metric("msg_queue_throughput", (double)count / writeUs * 1e6, "msg/s", true);
 
 
//  printf("Testing Request Queue...\n\n");
//
//  // First create two components
//  uint16 procID1, procID2;
//  uint32 compID1 = 1, compID2 = 2;
//  uint32 qID;
//  //if (!MemoryMaps::CreateNewNode(1, "", 1)) {
//  //  printf("Error: Testing Request Queue - CreateNewNode\n");
//  //  return false;
//  //}
//  if (!MemoryMaps::CreateNewProcess("Proc1", procID1)) {
//      printf("Error: Testing Request Queue - CreateNewProcess\n");
//      return false;
//  }
//  if (!MemoryMaps::CreateNewProcess("Proc2", procID2)) {
//      printf("Error: Testing Request Queue - CreateNewProcess 2\n");
//      return false;
//  }
//  if (!MemoryQueues::CreateMessageQueue(qID)) {
//      printf("Error: Testing Request Queue - CreateMessageQueue\n");
//      return false;
//  }
//  if (!MemoryMaps::SetProcessQueueID(procID1, MSGQ_ID, qID)) {
//      printf("Error: Testing Request Queue - SetProcessQueueID\n");
//      return false;
//  }
//  if (!MemoryQueues::CreateMessageQueue(qID)) {
//      printf("Error: Testing Request Queue - CreateMessageQueue\n");
//      return false;
//  }
//  if (!MemoryMaps::SetProcessQueueID(procID2, REQQ_ID, qID)) {
//      printf("Error: Testing Request Queue - SetProcessQueueID\n");
//      return false;
//  }
//
//  ComponentData* compData1 = ComponentData::CreateComponent(compID1, "Comp1", 10*1024, 1, procID1);
//  if (compData1 == NULL) {
//      printf("Error: Testing Request Queue - CreateComponent1\n");
//      return false;
//  }
//  ComponentData* compData2 = ComponentData::CreateComponent(compID2, "Comp2", 10*1024, 1, procID2);
//  if (compData2 == NULL) {
//      printf("Error: Testing Request Queue - CreateComponent2\n");
//      return false;
//  }
//
//  DataMessage* reqMsg = new DataMessage(CTRL_TEST, compID1, compID2);
//  uint32 reqID = 0;
//  if (!MemoryQueues::AddRequest(reqMsg, reqID)) {
//      printf("Error: Testing Request Queue - AddRequest\n");
//      return false;
//  }
//  delete(reqMsg);
//
//  if (!MemoryMaps::GetProcessQueueID(procID2, REQQ_ID, qID)) {
//      printf("Error: Testing Request Queue - GetProcessQueueID\n");
//      return false;
//  }
//  reqMsg = MemoryQueues::WaitForMessageQueue(qID, 1000);
//  if (reqMsg == NULL) {
//      printf("Error: Testing Request Queue - WaitForMessageQueue\n");
//      return false;
//  }
//
//  uint32 reqID2 = (uint32)reqMsg->getReference();
//  if (reqID != reqID2) {
//      printf("Error: Testing Request Queue - getReference mismatch\n");
//      return false;
//  }
//
//  DataMessage* replyMsg = new DataMessage(CTRL_TEST, compID2, 0, 1000000000);
//  if (!MemoryQueues::AddReply(reqID, true, replyMsg)) {
//      printf("Error: Testing Request Queue - AddReply\n");
//      return false;
//  }
//  delete(reqMsg);
//  delete(replyMsg);
//  replyMsg = NULL;
//
//  uint8 status;
//  if (!MemoryQueues::WaitForReply(reqID2, 1000, status, &replyMsg)) {
//      printf("Error: Testing Request Queue - WaitForReply\n");
//      return false;
//  }
//
//  if (status != REQ_SUCCESS_DATA) {
//      printf("Error: Testing Request Queue - Status mismatch\n");
//      return false;
//  }
//
//  if (replyMsg == NULL) {
//      printf("Error: Testing Request Queue - ReplyMsg NULL\n");
//      return false;
//  }
 
    //delete(replyMsg);
    //delete(compData1);
    //delete(compData2);
 
    unittest::progress(100, "done");
    StopProcessTestThread();
    delete(manager);
    return true;
}
 
 
//bool ProcessMemory::UnitTestQueues() {
//  printf("Testing Multithreaded Memory Queues...\n\n");
//
//  // First create and initialise the MemoryManager
//  MemoryManager* manager = new MemoryManager();
//  if (!manager->create(0, 30)) {
//      fprintf(stderr, "MemoryManager init() failed...\n");
//      delete(manager);
//      return false;
//  }
//
//  uint64 t1, t2, t3;
//  DataMessage* msg;
//  int64 c = 0;
//  uint32 q1, q2;
//
//  if (!MemoryQueues::CreateMessageQueue(q1, "Q1")) {
//      LogPrint(0, 0, 0, "[1] Could not create Test Queue 1...");
//      delete(manager);
//      return false;
//  }
//  if (!MemoryQueues::CreateMessageQueue(q2, "Q2")) {
//      LogPrint(0, 0, 0, "[1] Could not create Test Queue 2...");
//      delete(manager);
//      return false;
//  }
//
//  uint32 queueTestThreadID;
//  if (!ThreadManager::CreateThread(QueueTest, NULL, queueTestThreadID)) {
//      LogPrint(0, 0, 0, "[1] Could not create Queue Test thread...");
//      delete(manager);
//      return false;
//  }
//
//  msg = new DataMessage();
//  msg->setInt("Counter", 0);
//
//  if (!MemoryQueues::AddMessageToQueue(q1, msg)) {
//      LogPrint(0, 0, 0, "[1] Could not add initial message to Q1...");
//      delete(manager);
//      return false;
//  }
//
//  uint64 start = 0;
//
//  while (true) {
//      t1 = GetTimeNow();
//      if (msg = MemoryQueues::WaitForMessageQueue(q2, 50)) {
//          t2 = GetTimeNow();
//          msg->getInt("Counter", c);
//          if (c && (c % 99999 == 0)) {
//              if (start)
//                  LogPrint(0,0,0,"[1] Average path time: %.3fus", (double)GetTimeAge(start)/99999.0);
//              start = GetTimeNow();
//          }
//          msg->setInt("Counter", c+1);
//          msg->setSendTime(GetTimeNow());
//          if (!MemoryQueues::AddMessageToQueue(q1, msg)) {
//              LogPrint(0, 0, 0, "[1] Could not add message %lld to Q1...", c+1);
//              delete(msg);
//              thread_ret_val(0);
//          }
//          t3 = GetTimeNow();
//          if (t3-t1 > 10000)
//              LogPrint(0,0,0,"[1 - %lld] Wait: %lld   Add: %lld   Total: %lld   Msg: %lld\n", c+1, t2-t1, t3-t2, t3-t1, t3-msg->getSendTime());
//          delete(msg);
//      }
//      else {
//          t3 = GetTimeNow();
//          LogPrint(0,0,0,"[1 - %lld] Timeout: %lld\n", c+1, t3-t1);
//      }
//  }
//
//  return true;
//}
//
THREAD_RET THREAD_FUNCTION_CALL QueueTest(THREAD_ARG arg) {
 
    MemoryManager* manager = (MemoryManager*) arg;
 
    uint64 t = 0;
    uint32 c = 0;
    DataMessage* msg;
 
    while (!g_processTestStop) {
 
        if (msg = manager->processMemory->waitForCmdQ(0, 1000)) {
            c++;
            if (!manager->processMemory->addToSigQ(0, msg)) {
                fprintf(stderr, "Test Slave could not add Message %u to Queue...\n", c);
                delete(msg);
                thread_ret_val(0);
            }
            delete(msg);
        }
    }
    thread_ret_val(0);
 
    //uint64 t1, t2, t3;
    //int64 c = 0;
    //uint32 q1, q2;
 
    //if (!MemoryQueues::GetMessageQueueByName(q1, "Q1")) {
    //  LogPrint(0, 0, 0, "[2] Could not Get Test Queue 1...");
    //  thread_ret_val(0);
    //}
 
    //if (!MemoryQueues::GetMessageQueueByName(q2, "Q2")) {
    //  LogPrint(0, 0, 0, "[2] Could not Get Test Queue 2...");
    //  thread_ret_val(0);
    //}
 
    //while (true) {
    //  t1 = GetTimeNow();
    //  if (msg = MemoryQueues::WaitForMessageQueue(q1, 50)) {
    //      t2 = GetTimeNow();
    //      msg->getInt("Counter", c);
    //      msg->setInt("Counter", c+1);
    //      msg->setSendTime(GetTimeNow());
    //      if (!MemoryQueues::AddMessageToQueue(q2, msg)) {
    //          LogPrint(0, 0, 0, "[2] Could not add message %lld to Q2...", c+1);
    //          delete(msg);
    //          thread_ret_val(0);
    //      }
    //      t3 = GetTimeNow();
    //      if (t3-t1 > 10000)
    //          LogPrint(0,0,0,"[2 - %lld] Wait: %lld   Add: %lld   Total: %lld   Msg: %lld\n", c+1, t2-t1, t3-t2, t3-t1, t3-msg->getSendTime());
    //      delete(msg);
    //  }
    //  else {
    //      t3 = GetTimeNow();
    //      LogPrint(0,0,0,"[2 - %lld] Timeout: %lld\n", c+1, t3-t1);
    //  }
    //}
 
    thread_ret_val(0);
}
 
THREAD_RET THREAD_FUNCTION_CALL ProcessMemoryPerfTest(THREAD_ARG arg) {
 
    MemoryManager* manager = (MemoryManager*) arg;
 
    uint64 t = 0;
    uint32 c = 0;
    DataMessage* msg;
 
    while (!g_processTestStop) {
        if (msg = manager->processMemory->waitForCmdQ(0, 1000)) {
            c++;
            if (!manager->processMemory->addToMsgQ(0, msg)) {
                fprintf(stderr, "Test Slave could not add Message %u to Queue...\n", c);
                thread_ret_val(-1);
            }
            delete(msg);
        }
    }
    thread_ret_val(0);
}
 
bool ProcessMemory::PerfTest() {
    unittest::progress(0, "init memory manager");
 
    // First create and initialise the MemoryManager
    MemoryManager* manager = new MemoryManager();
    uint32 maxPageCount = 100;
    if (!manager->create(0)) {
        unittest::fail("MemoryManager create(0) failed");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
 
    uint32 count = 10000;
    uint32 writeCount = 0;
    uint32 readCount = 0;
    uint32 n;
    DataMessage* msg;
 
    uint32 queueTestThreadID;
    g_processTestStop = false;
    if (!ThreadManager::CreateThread(ProcessMemoryPerfTest, manager, queueTestThreadID)) {
        unittest::fail("Could not create Queue Test thread");
        StopProcessTestThread();
        delete(manager);
        return false;
    }
    g_processTestThreadID = queueTestThreadID;
 
    unittest::progress(10, "dual-thread round trips");
    uint64 perfStart = GetTimeNow();
    const uint32 perfIters = 20000;
    uint64 dualTotal = 0;
    uint64 t = 0;
    for (n=0; n<perfIters; n++) {
        msg = new DataMessage(CTRL_TEST, 0);
        if (!manager->processMemory->addToCmdQ(0, msg)) {
            unittest::fail("Could not add Message %u to CmdQueue", n);
            delete(manager);
            return false;
        }
        delete(msg);
        if ( (msg = manager->processMemory->waitForMsgQ(0, 10000)) == NULL) {
            unittest::fail("Could not get Message [%u] from MsgQueue", n);
            delete(manager);
            return false;
        }
        t += GetTimeAge(msg->getCreatedTime());
        delete(msg);
        if (n && n % 10000 == 0) {
            unittest::detail("Dual threaded message delay[%u]: %.3fus", n, ((double)t)/(2*10000.0));
            dualTotal += t;
            t = 0;
            unittest::progress(10 + (int)(80ULL * n / perfIters), "dual-thread round trips");
        }
    }
    dualTotal += t;
 
    double elapsedUs = (double)(GetTimeNow() - perfStart);
    // each iteration measures created-time age twice (enqueue + dequeue)
    double dualLatency = (double)dualTotal / (2.0 * perfIters);
    unittest::metric("dual_thread_latency", dualLatency, "us", false);
    if (elapsedUs > 0.0)
        unittest::metric("dual_thread_throughput", (double)perfIters / elapsedUs * 1e6, "msg/s", true);
 
    unittest::progress(100, "done");
    StopProcessTestThread();
    delete(manager);
    return true;
}
 
void Register_ProcessMemory_Tests() {
    UnitTestRunner::instance().registerTest("processmemory", ProcessMemory::UnitTest,
        "Process memory queues: signal/command/message queue round-trips and integrity", "memory");
    UnitTestRunner::instance().registerTest("processmemory_perf", ProcessMemory::PerfTest,
        "Process memory dual-threaded queue latency and throughput", "memory");
    UnitTestRunner::instance().registerTest("processmemory_fuzz", ProcessMemory::FuzzTest,
        "Process memory circular-queue seeded oracle fuzz (single process)", "memory", false);
    // Phase 2: OFF by default like the Phase 1 fuzz (last arg false) - it spawns threads
    // and runs for tens of seconds, so it must be named explicitly, never dragged into
    // test=cmsdk.
    UnitTestRunner::instance().registerTest("processmemory_concfuzz", ProcessMemory::ConcFuzzTest,
        "Process memory ring: concurrent multi-reader / burst-drain / resize-under-read / "
        "blocking-path fuzz (set-equality oracle + read-pointer continuity)", "memory", false);
}
 
} // namespace cmlabs