PsySpace.cpp

Go to the documentation of this file.

#include "PsySpace.h"
 
namespace cmlabs {
 
 
TimeQueueSchedule::TimeQueueSchedule(uint32 interval, uint64 start, uint64 end) {
    this->interval = interval;
    this->start = start;
    this->end = end;
}
 
TimeQueueSchedule::~TimeQueueSchedule() {
}
 
 
TimeQueue::TimeQueue() {
    timer = new utils::Timer();
}
 
TimeQueue::~TimeQueue() {
    mutex.enter();
    std::map<uint32, TimeQueueSchedule*>::iterator it, itEnd;
    for (it = schedules.begin(), itEnd = schedules.end(); it!=itEnd; it++)
        delete(it->second);
    delete(timer);
    mutex.leave();
}
 
bool TimeQueue::addSchedule(TimeQueueSchedule* schedule) {
    mutex.enter();
    if (schedules[schedule->id]) {
        mutex.leave();
        return false;
    }
    schedules[schedule->id] = schedule;
    mutex.leave();
    return timer->addTimer(schedule->id, schedule->interval, schedule->start, schedule->end);
}
 
bool TimeQueue::removeSchedule(uint32 id) {
    mutex.enter();
    TimeQueueSchedule* schedule = schedules[id];
    if (!schedule) {
        mutex.leave();
        return true;
    }
    timer->removeTimer(id);
    schedules.erase(id);
    delete(schedule);
    mutex.leave();
    return true;
}
 
DataMessage* TimeQueue::waitForNextEvent(uint32 ms) {
 
    uint32 id;
    uint64 time;
    if (!timer->waitForTimer(ms, id, time))
        return NULL;
 
    mutex.enter();
    TimeQueueSchedule* schedule = schedules[id];
    if (!schedule) {
        mutex.leave();
        return NULL;
    }
 
    DataMessage* msg = new DataMessage(schedule->msgType, 0);
    if (schedule->msgTag)
        msg->setTag(schedule->msgTag);
    mutex.leave();
    return msg;
}
 
 
 
 
 
 
 
PsySpace::PsySpace(const char* name, bool isAdHoc, uint16 procID, bool isLocal) {
    isMaster = true;
    manager = NULL;
    masterCreatedTime = 0;
    timeQ = new TimeQueue();
    this->isAdHoc = isAdHoc;
    if (name)
        this->name = utils::TextTrimQuotes(name);
    else
        this->name = utils::StringFormat("Unnamed_Space_%u", (uint32)utils::RandomValue(MAXVALUINT32));
    this->isLocal = isLocal;
    this->procID = procID;
    finishedShuttingDown = false;
 
    memset(waitCounters, 0, 3*sizeof(uint32));
    memset(procCounters, 0, 3*sizeof(uint32));
 
    threadPoolWaitLowerThreshold = 5;
    threadPoolWaitUpperThreshold = 20;
    threadPoolWaitIncrement = 5;
    lastThreadPoolCheck = GetTimeNow();
    threadPoolCheckInterval = 5000;
 
    internalCranks["Simple"] = Internal_Simple;
    internalCranks["Ping"] = Internal_Ping;
    internalCranks["Pong"] = Internal_Pong;
    internalCranks["Shutdown"] = Internal_Shutdown;
    internalCranks["Print"] = Internal_Print;
    internalCranks["Time"] = Internal_Time;
    internalCranks["SignalPing"] = Internal_SignalPing;
    internalCranks["SignalPong"] = Internal_SignalPong;
    internalCranks["RetrieveTest"] = Internal_RetrieveTest;
    internalCranks["MessageScript"] = Internal_MessageScript;
    internalCranks["QueryTest"] = Internal_QueryTest;
    internalCranks["StatsLog"] = Internal_StatsLog;
    internalCranks["BitmapPoster"] = Internal_BitmapPoster;
    internalCranks["MessageToggler"] = Internal_MessageToggler;
    internalCranks["MessageTypeConverter"] = Internal_MessageTypeConverter;
}
 
PsySpace::~PsySpace() {
    reset();
    if (timeQ) delete(timeQ);
    timeQ = NULL;
}
 
bool PsySpace::connect(uint16 systemID, bool isMaster, const char* cmdline) {
 
    // LogPrint(0, LOG_SPACE, 1, "PsySpace '%s' starting up, connecting to local Node %u...", name.c_str(), systemID);
 
    this->isMaster = isMaster;
 
    // Connect to MemoryManager
    if (manager) delete(manager);
    manager = new MemoryManager();
    if (!manager->connect(systemID, isMaster)) {
        // LogPrint(0, LOG_SPACE, 0, "PsySpace couldn't connect to local Node %u...", systemID);
        delete(manager);
        manager = NULL;
        return false;
    }
    finishedShuttingDown = false;
 
    uint16 checkID;
    if (!isAdHoc) {
        if (!manager->processMemory->getProcessID(name.c_str(), checkID)) {
            LogPrint(0, LOG_SPACE, 0, "Could not find process '%s'...", name.c_str());
            return false;
        }
        if (!procID)
            procID = checkID;
        else if (procID != checkID) {
            LogPrint(0, LOG_SPACE, 0, "Process ID '%u' does not match shared memory id '%u'...", procID, checkID);
            return false;
        }
    }
    else {
        if (manager->processMemory->getProcessID(name.c_str(), checkID) && checkID) {
            procID = checkID;
        }
        else {
            if (!manager->processMemory->createNewProcess(name.c_str(), procID)) {
                LogPrint(0, LOG_SPACE, 0, "PsySpace couldn't connect to local Node %u...", systemID);
                return false;
            }
        }
    }
 
    // Send message on sysQID to inform node that process has started
    DataMessage* msg = new DataMessage(PsyAPI::CTRL_PROCESS_INITIALISE, procID);
    if (!manager->processMemory->addToCmdQ(0, msg)) {
        LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Command Queue...");
        delete(msg);
        return false;
    }
    delete(msg);
 
    //if (!manager->processMemory->createNewProcess(name.c_str(), procID)) {
    //  LogPrint(0, 0, 0, "Could not create Process Map\n");
    //  return false;
    //}
 
    if (cmdline)
        manager->processMemory->setProcessCommandLine(procID, cmdline);
    manager->processMemory->setProcessStatus(procID, PSYPROC_INIT);
 
    // Connect to Node
    // Send message on sysQID to inform node that process has started
    msg = new DataMessage(PsyAPI::CTRL_PROCESS_GREETING, procID);
    if (!manager->processMemory->addToCmdQ(0, msg)) {
        LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Command Queue...");
        delete(msg);
        return false;
    }
    delete(msg);
 
    if (isMaster) {
        // Logging will be done via LogPrint directly to the Node as we are in the same process
    }
    else {
        //LogSystem::SetLogReceiver(this);
    }
 
    uint64 lastSeen;
    manager->getNodeStatus(lastSeen, masterCreatedTime);
 
    // Setup Heartbeat
    manager->processMemory->setProcessStatus(procID, PSYPROC_READY);
 
    // Start SpaceThread
    if (!ThreadManager::CreateThread(PsySpaceRun, this, threadID, 0)) {
        LogPrint(procID, LOG_SPACE, 0, "Could not start PsySpace thread...");
        return false;
    }
 
    return true;
}
 
bool PsySpace::isConnected(uint32 timeoutMS) {
    if (!manager) return false;
 
    uint64 lastSeen, createTime;
    uint8 status = manager->getNodeStatus(lastSeen, createTime);
    if (masterCreatedTime && (createTime != masterCreatedTime)) {
        LogPrint(procID, LOG_SPACE, 0, "Psyclone Node process appears to have restarted...");
        reset();
        return false;
    }
    if (status != PSYCLONE_STATUS_READY) {
        LogPrint(procID, LOG_SPACE, 0, "Psyclone Node process doesn't appear to be running...");
        reset();
        return false;
    }
    if (GetTimeAgeMS(lastSeen) > (int32)timeoutMS) {
        LogPrint(procID, LOG_SPACE, 0, "Psyclone Node process appears to have stopped, resetting connection...");
        reset();
        return false;
    }
    return true;
}
 
bool PsySpace::reset() {
    if (!isMaster)
        LogSystem::SetLogReceiver(NULL);
    if (manager)
        shutdown();
    if (manager) delete(manager);
    manager = NULL;
 
    std::map<std::string, utils::Library*>::iterator it = libraries.begin(), itEnd = libraries.end();
    while (it != itEnd)
        delete((it++)->second);
    libraries.clear();
 
    std::map<uint32, PsyAPI*>::iterator ait = psyAPIs.begin(), aitEnd = psyAPIs.end();
    while (ait != aitEnd)
        delete((ait++)->second);
    psyAPIs.clear();
 
    std::map<PsyType, SignalStruct*>::iterator sit = signalList.begin(), sitEnd = signalList.end();
    while (sit != sitEnd) {
        delete(sit->second->lastSignalMsg);
        delete((sit++)->second);
    }
    signalList.clear();
    return true;
}
 
bool PsySpace::start(uint16 threadCount) {
    // Setup Threading Pool
    LogPrint(procID, LOG_SPACE, 1, "PsySpace '%s' (%u) starting up...", name.c_str(), procID);
    return setThreadPoolSize(threadCount);
}
 
bool PsySpace::shutdown() {
    if (finishedShuttingDown)
        return true;
    finishedShuttingDown = true;
    stop();
 
    //LogPrint(procID, LOG_SPACE, 1, "Space ID %u shutting down...", procID);
    LogPrint(procID, LOG_SPACE, 1, "PsySpace '%s' (%u) shutting down...", name.c_str(), procID);
 
    DataMessage* msg = new DataMessage(PsyAPI::CTRL_PROCESS_SHUTDOWN, procID);
    if (!manager->processMemory->addToCmdQ(0, msg)) {
        LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Command Queue...");
        delete(msg);
        return false;
    }
    delete(msg);
 
    // shutdown all components
    std::map<uint32, PsyAPI*>::iterator it, itEnd;
    for (it=psyAPIs.begin(), itEnd=psyAPIs.end(); it != itEnd; it++)
        it->second->stop();
 
    uint32 maxWaitCount = 20;
    while (threadPool.size()) {
        utils::Sleep(50);
        if (!(--maxWaitCount)) {
            LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' waited for the last %u threads to terminate, killing threads now...", name.c_str(), threadPool.size());
            std::map<uint32, uint8>::iterator threadI = threadPool.begin(), threadE = threadPool.end();
            while (threadI != threadE) {
                if (!ThreadManager::TerminateThread(threadI->first))
                    LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' couldn't kill thread %u...", name.c_str(), threadI->first);
                else
                    LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' killed thread %u...", name.c_str(), threadI->first);
                threadI++;
            }
            threadPool.clear();
            break;
        }
    }
    while (isRunning) {
        LogPrint(procID, LOG_SPACE,0,"PsySpace '%s' waiting for the main thread to terminate...", name.c_str());
        utils::Sleep(50);
    }
 
    manager->processMemory->setProcessStatus(procID, PSYPROC_TERMINATED);
    LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' has shut down (%u, %p)", name.c_str(), procID, this);
    return true;
}
 
bool PsySpace::hasShutdown() {
    return finishedShuttingDown;
}
 
// ***************** PsyProbe plugins *****************
bool PsySpace::addPsyProbeCustomView(uint32 compID, const char* name, const char* templateURL) {
    DataMessage* msg = new DataMessage(PsyAPI::CTRL_CREATECUSTOMPAGE, compID);
    msg->setString("Name", name);
    msg->setString("Template", templateURL);
    manager->processMemory->addToCmdQ(0, msg);
    delete(msg);
    return true;
}
 
 
bool PsySpace::logEntry(LogEntry* entry) {
    if (!entry)
        return false;
 
    DataMessage* msg = new DataMessage(CTRL_LOGPRINT, procID);
    msg->setData("LogEntry", (char*)entry, entry->size);
    manager->processMemory->addToCmdQ(0, msg);
    delete(msg);        // DataMessage is new'd above - scalar delete is correct
    // free(), not delete: LogEntry is a variable-size record malloc'd by its
    // callers (PsyAPI::logPrint:1238, LogSystemPrint in Utils.cpp:335). Releasing
    // it with operator delete is an alloc-dealloc mismatch - undefined behaviour.
    // The two allocators sit on adjacent lines here, which is how it was missed.
    free(entry);
    return true;
}
 
uint16 PsySpace::getID() {
    return procID;
}
 
bool PsySpace::setThreadPoolSize(uint16 threadCount) {
 
    uint32 id;
    int32 n;
    int32 c = threadCount - (uint32)threadPool.size();
    threadTarget = threadCount;
//  LogPrint(procID, LOG_SPACE,0,"Setting PsySpace thread pool size to %u...", threadTarget);
    lastThreadPoolCheck = GetTimeNow();
    if (c > 0) {
        LogPrint(procID, LOG_SPACE,2,"Creating %u more Pool Threads [%u -> %u]...",
            c, (uint32)threadPool.size(), threadCount);
        // create c more threads
        for (n=0; n<c; n++) {
            if (ThreadManager::CreateThread(PsySpacePoolRun, this, id))
                threadPool[id] = 1;
            else
                return false;
        }
    }
    return true;
}
 
PsyAPI* PsySpace::getCrankAPI(const char* name) {
    PsyAPI* api;
    uint16 crankID;
    if (!manager->dataMapsMemory->getCrankID(name, crankID) || !crankID)
        return NULL;
 
    uint32 compID = manager->dataMapsMemory->getCrankCompID(crankID);
    if (!compID)
        return NULL;
 
    uint16 oldCompNodeID = manager->componentMemory->getComponentNodeID(compID);
    if (!oldCompNodeID)
        return NULL;
 
    if (!pullRemoteComponentData(compID, oldCompNodeID))
        return NULL;
 
    if (!threadPoolMutex.enter(1000, __FUNCTION__))
        return NULL;
    if (!(api = psyAPIs[crankID])) {
        api = new PsyAPI(this);
        api->setCommandlineBasedir(utils::GetCommandLinePath().c_str());
        api->currentCompID = compID;
        psyAPIs[crankID] = api;
    }
    threadPoolMutex.leave();
    return api;
}
 
bool PsySpace::registerCrankCallback(const char* name, CrankFunction func) {
    uint16 crankID;
    if (!manager->dataMapsMemory->getCrankID(name, crankID) || !crankID)
        return false;
    if (!threadPoolMutex.enter(1000, __FUNCTION__))
        return false;
    cranks[crankID] = func;
    threadPoolMutex.leave();
    return true;
}
 
bool PsySpace::registerInternalCrank(const char* name, CrankFunction func) {
    if (!name || !*name || !func)
        return false;
    if (!threadPoolMutex.enter(1000, __FUNCTION__))
        return false;
    internalCranks[name] = func;
    threadPoolMutex.leave();
    return true;
}
 
uint32 PsySpace::getComponentID(const char* name) {
    uint32 compID;
    if (manager->componentMemory->getComponentID(name, compID))
        return compID;
    else
        return 0;
}
 
bool PsySpace::pullRemoteComponentData(uint32 compID, uint16 fromNodeID) {
    DataMessage* msg = new DataMessage(PsyAPI::CTRL_PULLCOMPONENTDATA, this->procID);
    msg->setInt("CompID", compID);
    msg->setInt("ProcID", this->procID);
    msg->setInt("NodeID", fromNodeID);
    DataMessage* resultMsg = NULL;
    uint8 result = query(msg, &resultMsg, 5000);
    if (resultMsg)
        delete(resultMsg);
    delete(msg);
    return (result >= QUERY_SUCCESS);
}
 
 
//PsyAPI* PsySpace::getComponentAPI(const char* name) {
//  return getComponentAPI(getComponentID(name));
//}
 
//PsyAPI* PsySpace::getComponentAPI(uint32 compID) {
//  if (!compID)
//      return false;
//}
 
//bool PsySpace::registerComponentCallback(const char* name, CrankFunction func) {
//  return registerComponentCallback(getComponentID(name), func);
//}
 
//bool PsySpace::registerComponentCallback(uint32 compID, CrankFunction func) {
//  if (!compID)
//      return false;
//}
 
 
bool PsySpace::postMessage(DataMessage* msg) {
    // Add datamessage to memory
    uint64 memID = 0;
    uint64 eol = msg->getEOL();
    if (eol && (eol < msg->getCreatedTime() - 10000))
        return false;
    if (eol) {
        if (!manager->insertMessage(msg, memID)) {
            LogPrint(procID, LOG_SPACE, 0, "Could not add message from %u to shared memory...", msg->getFrom());
            return false;
        }
        msg->data->memid = memID;
    }
    if (!manager->processMemory->addToMsgQ(0, msg)) {
        LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Message Queue...");
        return false;
    }
    manager->processMemory->addToProcessStats(procID, NULL, msg);
//  printf("************ Space Post Msg %lld us old...\n", GetTimeAge(msg->getCreatedTime()));
    return true;
}
 
 
// ***************** Signals *****************
bool PsySpace::emitSignal(const PsyType& type, DataMessage* msg) {
    // Send signal to local signal struct first
    if (!signalsMutex.enter(3000, __FUNCTION__))
        return false;
    SignalStruct* signal = signalList[type];
    if (!signal) {
        signal = new SignalStruct;
        signal->lastSignalMsg = NULL;
        signalList[type] = signal;
    }
    signalsMutex.leave();
 
    if (signal->mutex.enter(3000, __FUNCTION__)) {
        if (signal->lastSignalMsg)
            delete(signal->lastSignalMsg);
        signal->lastSignalMsg = msg;
        signal->mutex.leave();
        signal->event.signal();
    }
    else {
        return false;
    }
    // Then to all other procs and the other nodes (via Node proc 0)
//  if (!manager->processMemory->addToAllSignalQs(msg)) {
    if (!manager->processMemory->addToAllSignalQsExcept(msg, procID)) {
        LogPrint(procID, LOG_SPACE, 0, "Could not communicate with All Signal Queues...");
        return false;
    }
    manager->processMemory->addToProcessStats(procID, NULL, msg);
    return true;
}
 
DataMessage* PsySpace::waitForSignal(const PsyType& type, uint32 timeout, uint64 lastReceivedTime) {
    
    if (!signalsMutex.enter(3000, __FUNCTION__))
        return NULL;
    SignalStruct* signal = signalList[type];
    if (!signal) {
        signal = new SignalStruct;
        signal->lastSignalMsg = NULL;
        signalList[type] = signal;
    }
    signalsMutex.leave();
 
    DataMessage* msg = NULL;
 
    if (lastReceivedTime) {
        if (signal->mutex.enter(3000, __FUNCTION__)) {
            if (signal->lastSignalMsg && (signal->lastSignalMsg->getCreatedTime() > lastReceivedTime)) {
                msg = new DataMessage((char*)signal->lastSignalMsg->data, true);
                // printf("************ Space NoWait Signal %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
                signal->mutex.leave();
                //LogPrint(procID, LOG_SPACE, 0, "Space signal no wait success %s...", PrintTimeString(msg->getCreatedTime()).c_str());
                manager->processMemory->addToProcessStats(procID, msg, NULL);
                return msg;
            }
            else {
                //LogPrint(procID, LOG_SPACE, 0, "Not newer %s >= %s", PrintTimeString(lastReceivedTime).c_str(),
                //  PrintTimeString(signal->lastSignalMsg->getCreatedTime()).c_str());
                //if (lastReceivedTime > signal->lastSignalMsg->getCreatedTime())
                //  LogPrint(procID, LOG_SPACE, 0, "%llu > %llu", lastReceivedTime, signal->lastSignalMsg->getCreatedTime());
                //else if (lastReceivedTime < signal->lastSignalMsg->getCreatedTime())
                //  LogPrint(procID, LOG_SPACE, 0, "%llu < %llu", lastReceivedTime, signal->lastSignalMsg->getCreatedTime());
                signal->mutex.leave();
            }
        }
    }
 
    //if (signal->lastSignalMsg)
    //  LogPrint(procID, LOG_SPACE, 0, "Space signal wait (%s)...", PrintTimeString(signal->lastSignalMsg->getCreatedTime()).c_str());
    //else
    //  LogPrint(procID, LOG_SPACE, 0, "Space signal wait...");
 
    if (signal->event.waitNext(timeout)) {
        //LogPrint(procID, LOG_SPACE, 0, "Space signal wait success...");
        if (signal->mutex.enter(3000, __FUNCTION__) && signal->lastSignalMsg) {
            if (lastReceivedTime) {
                if (signal->lastSignalMsg->getCreatedTime() > lastReceivedTime)
                    msg = new DataMessage((char*)signal->lastSignalMsg->data, true);
            }
            else
                msg = new DataMessage((char*)signal->lastSignalMsg->data, true);
            //if (msg)
                //printf("************ Space Wait Signal %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
            //  LogPrint(procID, LOG_SPACE, 0, "Space signal wait success %s...", PrintTimeString(msg->getCreatedTime()).c_str());
        }
        signal->mutex.leave();
    }
 
    if (msg)
        manager->processMemory->addToProcessStats(procID, msg, NULL);
    return msg;
}
 
 
uint8 PsySpace::query(DataMessage* msg, DataMessage** result, uint32 timeout) {
    *result = NULL;
    if (!msg)
        return QUERY_FAILED;
 
    uint32 to = msg->getTo();
 
    uint32 reqID;
    if (!manager->dataMapsMemory->createNewRequest(msg->getFrom(), to, 0, reqID))
        return QUERY_FAILED;
//  LogPrint(procID, LOG_SPACE,0,"Created Request id %u from comp %u...", reqID, msg->getFrom());
 
    msg->setReference(reqID);
    manager->processMemory->addToProcessStats(procID, NULL, msg);
 
    uint16 otherNodeID = 0;
    uint16 otherProcID = 0;
    // If to is 0 this request is for the local Node
    if (!to) {
        manager->dataMapsMemory->setRequestStatus(reqID, REQ_PROCESSING_LOCAL);
        if (!manager->processMemory->addToReqQ(0, msg)) {
            manager->dataMapsMemory->deleteRequest(reqID);
            return QUERY_FAILED;
        }
    }
    else {
        if (!manager->componentMemory->getComponentLocation(to, otherNodeID, otherProcID))
            return QUERY_FAILED;
 
        if (otherNodeID != manager->getNodeID()) {
            manager->dataMapsMemory->setRequestStatus(reqID, REQ_PROCESSING_REMOTE);
            if (!manager->processMemory->addToReqQ(0, msg)) {
                manager->dataMapsMemory->deleteRequest(reqID);
                return QUERY_FAILED;
            }
        }
        else {
            manager->dataMapsMemory->setRequestStatus(reqID, REQ_PROCESSING_LOCAL);
            otherProcID = manager->componentMemory->getComponentProcessID(to);
            if (!manager->processMemory->addToReqQ(otherProcID, msg)) {
                manager->dataMapsMemory->deleteRequest(reqID);
                return QUERY_FAILED;
            }
        }
    }
 
    uint8 status = 0;
    uint64 time = 0;
    uint64 msgID = 0;
    uint64 msgEOL = 0;
    if (!manager->dataMapsMemory->waitForRequestReply(reqID, timeout, status, time, msgID, msgEOL)) {
        LogPrint(procID, LOG_SPACE,0,"Request reply timeout (%u) from node %u (req %u from comp %u)...",
            status, otherNodeID, reqID, msg->getFrom());
        manager->dataMapsMemory->deleteRequest(reqID);
        return QUERY_TIMEOUT;
    }
    manager->dataMapsMemory->deleteRequest(reqID);
 
    if (status < REQ_SUCCESS) {
        if (status >= REQ_FAILED)
            return QUERY_FAILED;
        if (status == REQ_FAILED_TO_SEND)
            LogPrint(procID, LOG_SPACE, 0, "Request failed as remote system cannot be reached (req %u from comp %u)...", reqID, msg->getFrom());
        else
            LogPrint(procID, LOG_SPACE,0,"Request reply failed (%u) from node %u (req %u from comp %u)...", status, otherNodeID, reqID, msg->getFrom());
        return QUERY_NOT_REACHABLE;
    }
 
    if (!msgID)
        return QUERY_SUCCESS;
 
    if (!(*result = manager->temporalMemory->getCopyOfMessage(msgID))) {
        LogPrint(procID, LOG_SPACE,0,"Couldn't getCopyOfMessage reply from node %u (req %u from comp %u)...", otherNodeID, reqID, msg->getFrom());
        return QUERY_FAILED;
    }
 
    if (*result)
        manager->processMemory->addToProcessStats(procID, *result, NULL);
    return QUERY_SUCCESS;
}
 
bool PsySpace::queryReply(uint32 id, uint8 status, DataMessage* resultMsg) {
 
    if (resultMsg)
        manager->processMemory->addToProcessStats(procID, NULL, resultMsg);
    // Check if this request was made locally
    uint32 remoteID;
    uint32 origin;
    uint32 source;
    if (!manager->dataMapsMemory->getRequestInfo(id, origin, source, remoteID)) {
        LogPrint(procID, LOG_SPACE,0,"Couldn't find request info for local id %u", id);
        return false;
    }
    // could this be a INTERSYSTEM_QUERY_REPLY
    if (remoteID && !origin) {
        DataMessage* netMsg = new DataMessage(PsyAPI::CTRL_INTERSYSTEM_QUERY_REPLY, source, origin, 10000);
        // keep the local request id here, needed by the InterSystemManager
        netMsg->setReference(id);
        netMsg->setStatus(status);
        if (resultMsg)
            netMsg->setAttachedMessage("DataMessage", resultMsg);
        manager->processMemory->addToReqQ(0, netMsg);
        //LogPrint(procID, LOG_SPACE,3,"Replied to remote request %u (local %u) (source %u, origin %u)...", remoteID, id, source, origin);
        delete(resultMsg);
        delete(netMsg);
        // dont delete this: manager->dataMapsMemory->deleteRequest(id);
        // needed by the InterSystemManager
        return true;
    }
    else if (remoteID) {
        uint16 otherNodeID = manager->componentMemory->getComponentNodeID(origin);
        if (otherNodeID && (otherNodeID != manager->getNodeID())) {
            DataMessage* netMsg = new DataMessage(PsyAPI::CTRL_QUERY_REPLY, source, origin, 10000);
            netMsg->setReference(remoteID);
            netMsg->setStatus(status);
            if (resultMsg)
                netMsg->setAttachedMessage("DataMessage", resultMsg);
            manager->processMemory->addToReqQ(0, netMsg);
            //LogPrint(procID, LOG_SPACE,3,"Replied to remote request %u (local %u) (source %u, origin %u)...", remoteID, id, source, origin);
            delete(resultMsg);
            delete(netMsg);
            return manager->dataMapsMemory->deleteRequest(id);
        }
        else {
            LogPrint(procID, LOG_SPACE,0,"Couldn't find otherNode %u for remote request %u (local %u) (source %u, origin %u)...",
                otherNodeID, remoteID, id, source, origin);
            return false;
        }
    }
 
    uint64 msgID = 0;
    uint64 msgEOL = 0;
    uint8 reqStatus = 0;
 
    if (resultMsg) {
        resultMsg->setTTL(10000000);
        msgEOL = resultMsg->getEOL();
        // decide on EOL ###################
        manager->temporalMemory->insertMessage(resultMsg, msgID);
        delete(resultMsg);
    }
 
    if (status == QUERY_SUCCESS) {
        if (msgID)
            reqStatus = REQ_SUCCESS_DATA_EOL;
        else
            reqStatus = REQ_SUCCESS;
    }
    else {
        if (msgID)
            reqStatus = REQ_FAILED_DATA_EOL;
        else
            reqStatus = REQ_FAILED;
    }
    return manager->dataMapsMemory->setRequestStatus(id, reqStatus, msgID, msgEOL);
}
 
 
 
 
 
 
 
 
 
 
 
bool PsySpace::threadPoolDispatch() {
    DataMessage* msg, *dmsg;
    uint32 threadID;
    ThreadManager::GetLocalThreadID(threadID);
    uint32 idleCount = 0;
    uint8 proc;
    char *crankName = new char[MAXVALUENAMELEN+1];
    crankName[0] = 0;
    char *crankFunction = new char[MAXVALUENAMELEN + 1];
    crankFunction[0] = 0;
    char *crankLanguage = new char[MAXVALUENAMELEN + 1];
    crankLanguage[0] = 0;
    char *libraryFilename = new char[MAXVALUENAMELEN + 1];
    libraryFilename[0] = 0;
    CrankFunction crank;
    uint32 compID;
    PsyAPI* api;
    uint32 s;
    TriggerSpec* trigger;
 
    while (shouldContinue) {
//      printf(".");
        threadPoolMutex.enter();
 
//      if (GetTimeAgeMS(lastThreadPoolCheck) > threadPoolCheckInterval) {
            idleCount = waitCounters[PROCMESSAGE] + waitCounters[PROCSIGNAL] + waitCounters[PROCREQUEST];
        //  idleCount = waitCounters[PROCMESSAGE] + waitCounters[PROCSIGNAL];
            if (idleCount < threadPoolWaitLowerThreshold)
                setThreadPoolSize((uint16)(threadPool.size() + threadPoolWaitIncrement));
        //  else if (idleCount > threadPoolWaitLowerThreshold)
        //      setThreadPoolSize(threadPool.size() - threadPoolWaitIncrement);
            else
                lastThreadPoolCheck = GetTimeNow();
//      }
 
        if (threadTarget < threadPool.size()) {
            threadPoolMutex.leave();
            break;
        }
 
    //  uint64 t = GetTimeNow();
        if ( (waitCounters[PROCMESSAGE] <= waitCounters[PROCSIGNAL]) && (waitCounters[PROCMESSAGE] <= waitCounters[PROCREQUEST]) ) {
    //  if (waitCounters[PROCMESSAGE] <= waitCounters[PROCSIGNAL]) {
            proc = PROCMESSAGE;
            waitCounters[proc]++;
            threadPoolMutex.leave();
            msg = manager->processMemory->waitForMsgQ(procID, 100);
        //  if (msg)
        //      printf("Wait for Msg counter: %llu [%u / %u]\n", GetTimeAge(msg->getCreatedTime()), waitCounters[PROCMESSAGE], waitCounters[PROCSIGNAL]);
        //      printf("************ Space waited %llu...\n", GetTimeAge(t));
        }
        else if ( (waitCounters[PROCREQUEST] <= waitCounters[PROCMESSAGE]) && (waitCounters[PROCREQUEST] <= waitCounters[PROCSIGNAL]) ) {
            proc = PROCREQUEST;
            waitCounters[proc]++;
            threadPoolMutex.leave();
            msg = manager->processMemory->waitForReqQ(procID, 100);
            //if (msg) LogPrint(procID, LOG_SPACE, 1, "Got request message, processing...");
        }
        else {
            proc = PROCSIGNAL;
            waitCounters[proc]++;
            threadPoolMutex.leave();
            if (msg = manager->processMemory->waitForSigQ(procID, 100)) {
                manager->processMemory->addToProcessStats(procID, msg, NULL);
                // printf("Wait for Signal counter: %u [%u / %u]\n", GetTimeAge(msg->getCreatedTime()), waitCounters[PROCMESSAGE], waitCounters[PROCSIGNAL]);
                if (signalsMutex.enter(3000, __FUNCTION__)) {
                    SignalStruct* signal = signalList[msg->getType()];
                    if (!signal) {
                        signal = new SignalStruct;
                        signal->lastSignalMsg = NULL;
                        signalList[msg->getType()] = signal;
                    }
                    signalsMutex.leave();
                    if (signal->mutex.enter(3000, __FUNCTION__)) {
                        if (signal->lastSignalMsg)
                            delete(signal->lastSignalMsg);
                        signal->lastSignalMsg = msg;
                        signal->mutex.leave();
                        signal->event.signal();
                        LogPrint(procID, LOG_SPACE, 5, "Space signal signal %s...", PrintTimeString(msg->getCreatedTime()).c_str());
                    }
                    else
                        delete(msg);
                }
                else
                    delete(msg);
                msg = NULL;
            }
        }
        //else {
        //  proc = PROCTIMER;
        //  waitCounters[proc]++;
        //  threadPoolMutex.leave();
        //  msg = timeQ->waitForNextEvent(30);
        //}
        while (!threadPoolMutex.enter(1000, "Threadpool run with msg")) {
        }
 
        waitCounters[proc]--;
        procCounters[proc]++;
 
        uint16 crankID;
        char* compName;
        if (msg) {
            crank = NULL;
            crankID = 0;
            dmsg = NULL;
 
//          printf("************ Space Deliver to %u (%lld us old)     %s...\n",
//              msg->getTo(), GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
            // ⚠️ BakeLive Anomaly (see BakeLiveBug.md): the four `delete(msg); continue;`
            // paths below each discard a message that was already accepted into the
            // component's shared queue, and they log at level 0/1 which the unit-test
            // framework suppresses. A message lost here is therefore invisible to every
            // instrument on either side of the process boundary.
            // These paths were fully traced during that investigation and measured
            // 0/0/0/0 on failing runs - they are NOT the anomaly - but keep this note:
            // anyone hunting silent local-delivery loss should instrument here first.
            if (!(compID = msg->getTo())) {
                LogPrint(procID, LOG_SPACE, 1, "No compID, deleting...");
                delete(msg);
                procCounters[proc]--;
                threadPoolMutex.leave();
                continue;
            }
 
            manager->processMemory->addToProcessStats(procID, msg, NULL);
 
            if (msg->getType() == PsyAPI::CTRL_TRIGGER || msg->getType() == PsyAPI::CTRL_TRIGGER_GROUP) {
                // T1.6: a CTRL_TRIGGER_GROUP joined activation carries a TriggerSpec and
                // a primary "Message" just like CTRL_TRIGGER (plus a "Members" map read
                // by waitForNewMessageGroup()); dispatch it through the same crank path.
                if (!(trigger = (TriggerSpec*) msg->getData("TriggerSpec", s))) {
                    delete(msg);
                    procCounters[proc]--;
                    threadPoolMutex.leave();
                    continue;
                }
                crankID = trigger->crankID;
                if (!crankID) {
                    //crank = internalCranks["Simple"];
                    LogPrint(procID, LOG_SPACE, 0, "Crank not found for component!");
                    delete (msg);
                    procCounters[proc]--;
                    threadPoolMutex.leave();
                    continue;
                }
 
                // Get actual data message (primary; may be absent for an empty-tag group)
                if (!(dmsg = msg->getAttachedMessageCopy("Message"))) {
                    if (msg->getType() == PsyAPI::CTRL_TRIGGER) {
                        LogPrint(procID, LOG_SPACE, 0, "Couldn't find attached input message");
                        delete (msg);
                        procCounters[proc]--;
                        threadPoolMutex.leave();
                        continue;
                    }
                    // CTRL_TRIGGER_GROUP without a primary message is still valid: the
                    // crank consumes the member map via waitForNewMessageGroup().
                }
            }
            else {
                //LogPrint(procID, LOG_SPACE, 1, "Looking for crank for direct message to component queue");
                if (!(crankID = compCrankIDs[compID])) {
                    compName = new char[MAXKEYNAMELEN+1];
                    if (manager->componentMemory->getComponentName(compID, compName, MAXKEYNAMELEN)) {
                        utils::strcpyavail(compName+strlen(compName), "ComponentCrank", 20, true);
                        if (manager->dataMapsMemory->getCrankID(compName, crankID)) {
                            compCrankIDs[compID] = crankID;
                        }
                    }
                    else
                        compName[0] = 0;
                    if (!crankID) {
                        LogPrint(procID, LOG_SPACE, 0, "Unknown Component Crank for '%s' (%u)", compName, compID);
                        delete [] compName;
                        delete (msg);
                        procCounters[proc]--;
                        threadPoolMutex.leave();
                        continue;
                    }
                    delete [] compName;
                }
            }
 
            // Load function from already loaded cranks
            if (!crank && !(crank = cranks[crankID])) {
                manager->dataMapsMemory->getCrankName(crankID, crankName, 256);
                // Check if this is a scripted module
                if (manager->dataMapsMemory->getCrankLanguage(crankID, crankLanguage, MAXVALUENAMELEN) && crankLanguage && strlen(crankLanguage)) {
                    utils::StringFormatInto(crankFunction, MAXVALUENAMELEN, "%sLink", utils::TextCapitalise(crankLanguage).c_str());
                    utils::StringFormatInto(libraryFilename, MAXVALUENAMELEN, "Psy%sLink", utils::TextCapitalise(crankLanguage).c_str());
                    LogPrint(procID, LOG_SPACE, 0, "Scripted language (%s) crank '%s' loading binary library '%s'", crankLanguage, crankName, libraryFilename);
                }
                else {
                    // Load function from library
                    if (!manager->dataMapsMemory->getCrankFunction(crankID, crankFunction, MAXVALUENAMELEN) || !crankFunction) {
                        LogPrint(procID, LOG_SPACE, 0, "Unknown Crank ID %u in Trigger Message", crankID);
                        delete (msg);
                        procCounters[proc]--;
                        threadPoolMutex.leave();
                        continue;
                    }
                    if (!manager->dataMapsMemory->getCrankLibraryFilename(crankID, libraryFilename, MAXVALUENAMELEN) || !libraryFilename) {
                        LogPrint(procID, LOG_SPACE, 0, "Crank '%s' doesn't have a valid library specified", crankFunction);
                        delete (msg);
                        procCounters[proc]--;
                        threadPoolMutex.leave();
                        continue;
                    }
                }
 
                // If crankFunction and/or libraryFilename are not yet set, accept
                // as the message will be added to the API queue for when the module
                // gets instantiated
 
                if (crankFunction && strlen(crankFunction)) {
                    if (!(crank = loadCrankFromLibrary(crankFunction, libraryFilename))) {
                        if (strlen(crankLanguage)) {
                            LogPrint(procID, LOG_SPACE, 0, "The language library '%s' is unavailable - the language name '%s' may be incorrect", libraryFilename, crankLanguage);
                        }
                        else
                            LogPrint(procID, LOG_SPACE, 0, "Couldn't find crank '%s' in library '%s'", crankFunction, libraryFilename);
                        delete (msg);
                        procCounters[proc]--;
                        threadPoolMutex.leave();
                        continue;
                    }
                    cranks[crankID] = crank;
                }
            }
 
            //printf("************ Space Trigger %llu us old, Msg %llu us old...\n", GetTimeAge(msg->getCreatedTime()), GetTimeAge(dmsg->getCreatedTime()));
 
            // Is the crank already running
            if (!(api = psyAPIs[crankID])) {
                api = new PsyAPI(this);
                api->setCommandlineBasedir(utils::GetCommandLinePath().c_str());
                psyAPIs[crankID] = api;
            }
 
            // ⚠️ The wrapper (msg) is a CTRL_TRIGGER envelope with its OWN serial, while the
            // subscriber reads getSerial() off the ATTACHED PAYLOAD (dmsg). Any future
            // correlation of a queued entry back to the message it carries must use the
            // payload's serial, not the wrapper's - conflating the two produced a
            // confidently wrong conclusion during the BakeLive Anomaly investigation.
            uint8 status;
            if (dmsg)
                status = api->addInputTrigger(crankID, compID, msg, dmsg);
            else if (msg->getType() == PsyAPI::CTRL_TRIGGER_GROUP)
                // T1.6: a group activation with no primary (e.g. empty-tag grouping)
                // still delivers as a trigger so waitForNewMessageGroup() sees a fresh
                // trigger name and reads the "Members" map off the wrapper.
                status = api->addInputTrigger(crankID, compID, msg, NULL);
            else {
                //LogPrint(procID, LOG_SPACE, 1, "Added direct message to component queue");
                status = api->addInputTrigger(crankID, compID, NULL, msg);
            }
            switch(status) {
                case CRANKAPI_RUNNING:
                    procCounters[proc]--;
                    threadPoolMutex.leave();
                    continue;
                case CRANKAPI_IDLE:
                //  threadPoolMutex.leave();
                    break;
                case CRANKAPI_INUSE:
                    // A DIFFERENT crank is mid-execution on this PsyAPI, so
                    // addInputTrigger did NOT queue the message (PsyAPI.cpp:353-356).
                    // This used to fall into `default:` alongside CRANKAPI_FAILED and be
                    // DELETED - silently destroying an already-delivered message.
                    //
                    // ⚠️ Re-queueing it here would BUSY-LOOP: PsyAPI::begin() sets
                    // startedRunning for the whole crank execution and only finish()
                    // clears it (PsyAPI.cpp:267/287), so this dispatch thread would pull
                    // the message straight back off the queue, get INUSE again, and spin
                    // hot for the entire duration of the other crank - which is worse
                    // than the drop it replaces. So: re-queue, but yield first so the
                    // running crank makes progress and the queue is not hammered.
                    // The re-queue can itself fail (queue full / no growth); if it does we
                    // ARE dropping, so say so loudly and count it rather than pretend.
                    {
                        DataMessage* requeue = dmsg ? dmsg : msg;
                        DataMessage* discard = dmsg ? msg : NULL;
                        // Release exactly once, like every other exit from this switch,
                        // then do the re-queue OUTSIDE the pool lock (addToMsgQ takes the
                        // process-memory mutex; nesting the two here would invert the
                        // lock order used elsewhere in this function).
                        procCounters[proc]--;
                        threadPoolMutex.leave();
                        utils::Sleep(1);    // yield: do not spin on a busy crank
                        if (!manager->processMemory->addToMsgQ(procID, requeue)) {
                            LogPrint(procID, LOG_SPACE, 0, "Crank API busy for crank %u and "
                                "re-queue FAILED; message DROPPED", crankID);
                            delete(requeue);
                        }
                        delete(discard);    // trigger wrapper is ours either way (NULL-safe)
                    }
                    continue;
                case CRANKAPI_FAILED:
                default:
                    LogPrint(procID, LOG_SPACE, 0, "Couldn't add trigger to Crank API");
                    delete(msg);
                    delete(dmsg);
                    procCounters[proc]--;
                    threadPoolMutex.leave();
                    continue;
            }
 
            threadPoolMutex.leave();
            // Call function
            if (crank) {
                api->begin();
                do {
                    try {
                        if (crank(api) < 0)
                            break;
                    }
                    catch (...) {
                        LogPrint(procID, LOG_SPACE, 0, "Crank %u ('%s') caused an exception", crankID, crankName);
                        break;
                    }
                } while (api->shouldContinue() && api->getInputQueueSize());
                api->finish();
            }
            threadPoolMutex.enter();
            procCounters[proc]--;
        }
        threadPoolMutex.leave();
 
    }
 
    LogPrint(procID, LOG_SPACE,3,"Thread Pool %u exited, target %u, size: %u", threadID, threadTarget, threadPool.size() - 1);
    threadPoolMutex.enter();
    std::map<uint32, uint8>::iterator i = threadPool.find(threadID);
    if (i != threadPool.end())
        threadPool.erase(i);
    threadPoolMutex.leave();
    delete [] crankName;
    delete [] crankFunction;
    delete [] libraryFilename;
    return true;
}
 
CrankFunction PsySpace::loadCrankFromLibrary(const char* crankName, const char* libraryFilename) {
    if (!crankName || !libraryFilename)
        return NULL;
 
    if (!strlen(libraryFilename)) {
        // Internal cranks
        return internalCranks[crankName];
    }
 
    utils::Library* lib = libraries[libraryFilename];
    if (!lib) {
        if (!(lib = utils::OpenLibrary(libraryFilename))) {
            LogPrint(procID, LOG_SPACE, 0, "Could not find or load library '%s'", libraryFilename);
            return NULL;
        }
        libraries[libraryFilename] = lib;
    }
 
    CrankFunction crank = (CrankFunction)lib->getFunction(crankName);
    if (!crank) {
        LogPrint(procID, LOG_SPACE, 0, "Could not load Crank '%s' from library '%s'", crankName, libraryFilename);
        return NULL;
    }
    return crank;
}
 
 
bool PsySpace::startContinuousComponent(uint32 compID) {
 
    std::map<uint32, uint32>::iterator it = continuousComponentThreads.find(compID);
    if (it != continuousComponentThreads.end())
        return false;
 
    continuousComponentThreads[threadID] = 0;
 
    uint32 threadID;
    if (!ThreadManager::CreateThread(PsySpaceContinuousRun, this, threadID))
        return false;
 
    continuousComponentThreads[threadID] = compID;
    return true;
}
 
bool PsySpace::runContinuousComponent() {
    uint32 threadID;
    DataMessage* msg = NULL;
 
    ThreadManager::GetLocalThreadID(threadID);
 
    uint32 count = 0;
    uint32 compID;
    while (!(compID = continuousComponentThreads[threadID])) {
        if (++count > 10)
            return false;
        utils::Sleep(50);
    }
 
    // Find function name and library to call
    // Load function from library
    // Check stats
    // Call function
    // Record stats usage
 
    return true;
}
 
 
bool PsySpace::run() {
    // Space Thread
    //  Update Heartbeat
    manager->processMemory->setProcessStatus(procID, PSYPROC_ACTIVE);
    //  Update Space Stats
    uint64 currentCPUTicks;
//  utils::GetProcessCPUTicks(currentCPUTicks);
    uint32 checkCount = 50, c = 0;
    uint64 lastMsgCheck;
    std::map<uint32, PsyAPI*>::iterator i, e = psyAPIs.end();
 
    DataMessage* cmsg;
    while (shouldContinue) {
        if (cmsg = manager->processMemory->waitForCmdQ(procID, 100)) {
            if (cmsg->getType() == PsyAPI::CTRL_SYSTEM_SHUTDOWN) {
                delete(cmsg);
                shutdown();
                break;
            }
            delete(cmsg);
        }
        //  Update Heartbeat
        utils::GetProcessCPUTicks(currentCPUTicks);
        manager->processMemory->setProcessStatus(procID, PSYPROC_ACTIVE, currentCPUTicks);
 
        if (++c > 50) {
            // check all apis for checkLastWaitForMessage()
            i = psyAPIs.begin();
            while (i != e) {
                if ((i->second) && (lastMsgCheck = i->second->checkLastWaitForMessage())) {
                    i->second->logPrint(1, "Component has unchecked messages, last check %s ago",
                        PrintTimeDifString(GetTimeAge(lastMsgCheck)).c_str());
                }
                i++;
            }
            c = 0;
        }
    }
 
    manager->processMemory->setProcessStatus(procID, PSYPROC_SHUTTING_DOWN);
    //LogPrint(procID, LOG_SPACE, 1, "PsySpace '%s' shutting down...", name.c_str());
    isRunning = false;
    return true;
}
 
THREAD_RET THREAD_FUNCTION_CALL PsySpaceRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((PsySpace*)arg)->run() ? 0 : 1));
}
 
 
THREAD_RET THREAD_FUNCTION_CALL PsySpacePoolRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((PsySpace*)arg)->threadPoolDispatch() ? 0 : 1));
}
 
THREAD_RET THREAD_FUNCTION_CALL PsySpaceContinuousRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((PsySpace*)arg)->runContinuousComponent() ? 0 : 1));
}
 
 
} // namespace cmlabs