PsyAPI.cpp

Go to the documentation of this file.

#include "PsyAPI.h"
 
namespace cmlabs {
 
struct PsyType PsyAPI::CTRL_SYSTEM_READY =          { { 1,10200,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PROCESS_INITIALISE =    { { 1,10201,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PROCESS_GREETING =      { { 1,10202,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PROCESS_SHUTDOWN =      { { 1,10203,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_CONTEXT_CHANGE =        { { 1,10297,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_SYSTEM_SHUTDOWN =       { { 1,10298,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_SYSTEM_SHUTTINGDOWN =   { { 1,10299,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_TRIGGER =               { { 1,10300,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_TRIGGER_GROUP =         { { 1,10307,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_QUERY =                 { { 1,10301,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_QUERY_REPLY =           { { 1,10302,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PULLCOMPONENTDATA =     { { 1,10303,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_CREATECUSTOMPAGE =      { { 1,10304,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_ADDSUBSCRIPTION =       { { 1,10305,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_RETRIEVESYSTEMIDS =     { { 1,10306,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_INTERSYSTEM_QUERY =     { { 1,10998,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_INTERSYSTEM_QUERY_REPLY = { { 1,10999,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
 
PsyAPI::PsyAPI(PsySpace* space) {
    this->space = space;
    autoDelete = true;
    startedRunning = 0;
    shouldContinueRunning = false;
    currentCrankID = 0;
    lastCPUTicks = 0;
    lastWallTime = 0;
    chainCPUTicks = 0;
    chainWallTime = 0;
    chainCount = 0;
    currentCompID = 0;
    msgReceivedCount = 0;
    msgInputCount = 0;
    msgSentCount = 0;
    msgPostedCount = 0;
    commandlineBasedir[0] = 0;
    currentTriggerName[0] = 0;
    currentTriggerMsg = NULL;
    currentMsg = NULL;
    currentSignalMsg = NULL;
    currentTriggerSpec = NULL;
}
 
PsyAPI::~PsyAPI() {
    apiMutex.enter(2000);
    shouldContinueRunning = false;
    this->space = NULL;
    delete(currentTriggerMsg);
    currentTriggerMsg = NULL;
    delete(currentMsg);
    currentMsg = NULL;
    delete(currentSignalMsg);
    currentSignalMsg = NULL;
    // T1.6: free any owned grouped-set member messages.
    std::map<std::string, DataMessage*>::iterator git, gitEnd;
    for (git = currentGroupMap.begin(), gitEnd = currentGroupMap.end(); git != gitEnd; git++)
        if (git->second) delete(git->second);
    currentGroupMap.clear();
    apiMutex.leave();
}
 
PsyAPI* PsyAPI::fromPython(unsigned long long ptr) {
    return (PsyAPI*)ptr;
}
 
const char* PsyAPI::fromPythonAddressOf(unsigned long long ptr) {
    return (const char*)ptr;
}
 
bool PsyAPI::isRunning() {
    if (!apiMutex.enter(1000))
        return false;
    if (startedRunning) {
        apiMutex.leave();
        return true;
    }
    else {
        apiMutex.leave();
        return false;
    }
}
 
bool PsyAPI::shouldContinue() {
    if (!apiMutex.enter(1000))
        return true;
    if (shouldContinueRunning) {
        apiMutex.leave();
        return true;
    }
    else {
        apiMutex.leave();
        return false;
    }
}
 
std::string PsyAPI::getCommandlineBasedir() {
    return commandlineBasedir;
}
 
bool PsyAPI::setCommandlineBasedir(const char* cmdlineBasedir) {
    utils::strcpyavail(commandlineBasedir, cmdlineBasedir, MAXCOMMANDLINELEN, true);
    return true;
}
 
 
bool PsyAPI::getModuleName(char* name, uint32 maxSize) {
    return space->manager->componentMemory->getComponentName(currentCompID, name, maxSize);
}
 
std::string PsyAPI::getModuleName() {
    return space->manager->componentMemory->getComponentNameString(currentCompID);
}
 
std::string PsyAPI::getOtherModuleName(uint32 id) {
    return space->manager->componentMemory->getComponentNameString(id);
}
 
bool PsyAPI::getCurrentScriptLanguage(char* language, uint32 maxSize) {
    return space->manager->dataMapsMemory->getCrankLanguage(currentCrankID, language, maxSize);
}
 
std::string PsyAPI::getCurrentScriptLanguage() {
    char* language = new char[1024];
    if (!space->manager->dataMapsMemory->getCrankLanguage(currentCrankID, language, 1024)) {
        delete[] language;
        return "";
    }
    std::string lang = language;
    delete[] language;
    return lang;
}
 
bool PsyAPI::getCurrentScript(char* script, uint32 maxSize) {
    return space->manager->dataMapsMemory->getCrankScript(currentCrankID, script, 1024);
}
 
std::string PsyAPI::getCurrentScript() {
    char* script = new char[MAXSCRIPTLEN];
    if (!space->manager->dataMapsMemory->getCrankScript(currentCrankID, script, MAXSCRIPTLEN)) {
        delete[] script;
        return "";
    }
    std::string scr = script;
    delete[] script;
    return scr;
}
 
bool PsyAPI::getCurrentScriptFilename(char* file, uint32 maxSize) {
    return space->manager->dataMapsMemory->getCrankLibraryFilename(currentCrankID, file, 1024);
}
 
std::string PsyAPI::getCurrentScriptFilename() {
    char* file = new char[1024];
    if (!space->manager->dataMapsMemory->getCrankLibraryFilename(currentCrankID, file, 1024)) {
        delete[] file;
        return "";
    }
    std::string filename = file;
    delete[] file;
    return filename;
}
 
bool PsyAPI::hasCurrentRetrieveName(const char* name) {
    if (!apiMutex.enter(1000, __FUNCTION__))
        return false;
    if (currentTriggerSpec) {
        if (currentTriggerSpec->getRetrieveSpec(name)) {
            apiMutex.leave();
            return true;
        }
    }
    apiMutex.leave();
    return false;
}
 
bool PsyAPI::hasCurrentSignalName(const char* name) {
    if (!apiMutex.enter(1000, __FUNCTION__))
        return false;
    if (currentTriggerSpec) {
        if (currentTriggerSpec->getSignalSpec(name)) {
            apiMutex.leave();
            return true;
        }
    }
    apiMutex.leave();
    return false;
}
 
 
bool PsyAPI::hasCurrentQueryName(const char* name) {
    if (!apiMutex.enter(1000, __FUNCTION__))
        return false;
    if (currentTriggerSpec) {
        if (currentTriggerSpec->getQuerySpec(name)) {
            apiMutex.leave();
            return true;
        }
    }
    apiMutex.leave();
    return false;
}
 
bool PsyAPI::hasCurrentPostName(const char* name) {
    if (!apiMutex.enter(1000, __FUNCTION__))
        return false;
    if (currentTriggerSpec) {
        if (currentTriggerSpec->getPostSpec(name)) {
            apiMutex.leave();
            return true;
        }
    }
    apiMutex.leave();
    return false;
}
 
std::set<std::string> PsyAPI::getCurrentPostNames() {
    std::set<std::string> postNames;
    if (!apiMutex.enter(1000, __FUNCTION__))
        return postNames;
    if (currentTriggerSpec) {
        std::list<PostSpec*>* postSpecs = currentTriggerSpec->getPostSpecs(NULL);
        if (postSpecs && postSpecs->size()) {
            std::list<PostSpec*>::iterator it = postSpecs->begin(), itEnd = postSpecs->end();
            while (it != itEnd) {
                postNames.insert((*it)->name);
                it++;
            }
        }
        delete(postSpecs);
    }
    apiMutex.leave();
    return postNames;
}
 
uint32 PsyAPI::getInputQueueSize() {
    if (!apiMutex.enter(1000, __FUNCTION__))
        return 0;
    uint32 qSize = (uint32)inputQueue.size();
    apiMutex.leave();
//  printf("Pointer: %llu  Q: %u\n", (uint64)this, qSize);
    return qSize;
}
 
std::string PsyAPI::typeToText(PsyType type) {
    return space->manager->typeToText(type);
}
 
std::string PsyAPI::contextToText(PsyContext context) {
    return space->manager->contextToText(context);
}
 
bool PsyAPI::begin() {
    if (!apiMutex.enter(1000))
        return false;
 
    shouldContinueRunning = true;
    startedRunning = GetTimeNow();
    utils::GetCPUTicks(lastCPUTicks);
    lastWallTime = startedRunning;
 
    space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_STARTING,
        0, // cpu usage
        NULL, // no input message here, set in waitForNewMessage
        NULL, // no output message here
        1, // run count 1, only incremented at start
        0 // cycle count 0, only incremented at waitForNewMessage
        );
    apiMutex.leave();
    return true;
}
 
bool PsyAPI::finish() {
 
    if (!apiMutex.enter(1000))
        return false;
 
    startedRunning = 0;
    currentCrankID = 0;
    currentCompID = 0;
 
    uint64 currentCPUTicks;
    utils::GetCPUTicks(currentCPUTicks);
    space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_FINISHED,
        currentCPUTicks - lastCPUTicks, // cpu usage
        NULL, // no input message here
        NULL, // no output message here
        0, // run count 0, only incremented at start
        0 // cycle count 0, only incremented at waitForMessage
        );
    //utils::GetCPUUsage(cpuUsageBefore);
    lastCPUTicks = currentCPUTicks;
    lastWallTime = GetTimeNow();
    apiMutex.leave();
    return true;
}
 
bool PsyAPI::stop() {
    apiMutex.enter(1000);
    shouldContinueRunning = false;
    apiMutex.leave();
    return true;
}
 
 
 
// ***************** PsyProbe plugins *****************
bool PsyAPI::addPsyProbeCustomView(const char* name, const char* templateURL) {
    return space->addPsyProbeCustomView(currentCompID, name, templateURL);
}
 
 
 
uint64 PsyAPI::checkLastWaitForMessage() {
    if (!apiMutex.enter(1000))
        return CRANKAPI_FAILED;
 
    // If currently running
    if (startedRunning && currentCrankID) {
        if (inputQueue.size() && (GetTimeAgeMS(lastWallTime) > 5000)) {
            apiMutex.leave();
            return lastWallTime;
        }
    }
    apiMutex.leave();
    return 0;
}
 
uint8 PsyAPI::addInputTrigger(uint32 crankID, uint32 compID, DataMessage* trigger, DataMessage* msg) {
    
//  printf("************ API Add Input Trigger %llu us old, Msg %llu us old...\n", GetTimeAge(trigger->getCreatedTime()), GetTimeAge(msg->getCreatedTime()));
    if (!apiMutex.enter(1000))
        return CRANKAPI_FAILED;
 
    msgReceivedCount++;
    if (startedRunning && currentCrankID) {
        if (crankID == currentCrankID) {
            // just add the trigger and msg and return
            inputQueue.push(TriggerAndMessage(trigger, msg));
            inputQueueSemaphore.signal();
            apiMutex.leave();
            return CRANKAPI_RUNNING;
        }
        else {
            apiMutex.leave();
            return CRANKAPI_INUSE;
        }
    }
 
    // reset measurements
    currentCrankID = crankID;
    currentCompID = compID;
    // CLAIM the API here, inside apiMutex, not later in begin(). The caller (the
    // thread-pool dispatcher) releases its pool lock between this IDLE return and
    // api->begin(), and the test above is "startedRunning && currentCrankID". If only
    // begin() set startedRunning, a second pool thread delivering the next message for
    // the same crank in that gap ALSO got CRANKAPI_IDLE, and two threads then ran the
    // same crank on the same PsyAPI at once: one thread's waitForNewMessage()
    // delete(currentMsg) freed the message the other had just popped (ASan:
    // heap-use-after-free in addComponentStats <- waitForNewMessage, freed at the
    // delete(currentMsg) line; test=udpservice crashed ~1 in 12 under a UDP burst).
    // begin() re-stamps the time; finish() still clears it.
    startedRunning = GetTimeNow();
 
    // add the trigger and msg
    inputQueue.push(TriggerAndMessage(trigger, msg));
    inputQueueSemaphore.signal();
 
    apiMutex.leave();
    return CRANKAPI_IDLE;
}
 
DataMessage* PsyAPI::waitForNewMessage(uint32 ms) {
    const char* triggerName;
    return waitForNewMessage(ms, triggerName);
}
 
std::string PsyAPI::getCurrentTriggerName() {
    return currentTriggerName;
}
 
PsyContext PsyAPI::getCurrentTriggerContext() {
    if (this->currentTriggerSpec)
        return this->currentTriggerSpec->context;
    else
        return NOCONTEXT;
}
 
 
DataMessage* PsyAPI::waitForNewMessage(uint32 ms, const char* &triggerName) {
    
    DataMessage* msg = NULL;
    uint32 s;
    uint64 currentCPUTicks;
    
//  if (!apiMutex.enter(ms ? ms : 10, __FUNCTION__))sig
    if (!apiMutex.enter(1000, __FUNCTION__))
        return NULL;
    
    if (!inputQueue.size()) {
        apiMutex.leave();
        // possible queue.size without mutex issue?
        if (!inputQueueSemaphore.wait(ms) || !inputQueue.size()) {
            utils::GetCPUTicks(currentCPUTicks);
            space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_RUNNING,
                currentCPUTicks - lastCPUTicks, // cpu usage
                NULL, // input message if any
                NULL, // no output message here
                0, // run count 0, only incremented at start
                1 // cycle count 
                );
            lastCPUTicks = currentCPUTicks;
            lastWallTime = GetTimeNow();
            chainCPUTicks = 0;
            chainWallTime = 0;
            chainCount = 0;
            return NULL;
        }
//      if (!apiMutex.enter(ms, __FUNCTION__))
        if (!apiMutex.enter(1000, __FUNCTION__))
            return NULL;
    }
 
    delete(currentMsg);
    //delete(currentTriggerMsg);
 
    // COPY, not a reference: the old code bound `TriggerAndMessage& tm` to
    // inputQueue.front() and then immediately pop()ed, which destroys the element
    // the reference names - every later read of tm (the while condition, the
    // tm.first/tm.second extraction below) was a dangling read into a freed queue
    // slot. ASan reports it as container-overflow. It survived because
    // std::queue/deque usually leaves the popped slot's bytes intact, so the
    // pointers still looked valid; under memory pressure or a different allocator
    // it hands back whatever now occupies the slot and the two DataMessage* are
    // then freed. TriggerAndMessage is a pair of pointers, so copying costs
    // nothing. Found by Isaac on macOS.
    TriggerAndMessage tm = inputQueue.front();
    inputQueue.pop();
 
    while ((inputQueue.size() > 100) && (tm.second->getPolicy() & MESSAGE_NON_GUARANTEED)) {
        delete(tm.first);
        delete(tm.second);
        tm = inputQueue.front();
        inputQueue.pop();
    }
 
    DataMessage* msgFirst = tm.first;
    DataMessage* msgSecond = tm.second;
 
    // If this message is a trigger message (has tm.first) replace currentTriggerXXX
    if (msgFirst) {
        delete(currentTriggerMsg);
        currentTriggerMsg = msgFirst;
        currentTriggerSpec = (TriggerSpec*)currentTriggerMsg->getData("TriggerSpec", s);
        utils::strcpyavail(currentTriggerName, currentTriggerSpec->name, MAXKEYNAMELEN, true);
        triggerName = currentTriggerName;
    }
    else {
        // this is a temporary query message, keep the previous currentTriggerXXX around
        triggerName = NULL;
    }
 
    //currentTriggerMsg = msgFirst;
    //if (currentTriggerMsg) {
    //  currentTriggerSpec = (TriggerSpec*) currentTriggerMsg->getData("TriggerSpec", s);
    //  utils::strcpyavail(currentTriggerName, currentTriggerSpec->name, MAXKEYNAMELEN, true);
    //  triggerName = currentTriggerName;
    //}
    //else {
    //  triggerName = NULL;
    //  currentTriggerSpec = NULL;
    //}
    msg = msgSecond;
    if (autoDelete)
        currentMsg = msg;
 
//  printf("************ API Trigger %llu us old, Msg %llu us old...\n", GetTimeAge(currentTriggerMsg->getCreatedTime()), GetTimeAge(msg->getCreatedTime()));
 
    if (msgFirst && msgSecond) {
        chainCPUTicks = msgSecond->data->chaincputime;
        chainWallTime = msgSecond->data->chainwalltime;
        chainCount = msgSecond->data->chaincount;
    }
    else {
        chainCPUTicks = 0;
        chainWallTime = 0;
        chainCount = 0;
    }
 
    // calc stats
    apiMutex.leave();
 
    // Add stats to component
    utils::GetCPUTicks(currentCPUTicks);
    space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_RUNNING,
        currentCPUTicks - lastCPUTicks, // cpu usage
        msg, // input message if any
        NULL, // no output message here
        0, // run count 0, only incremented at start
        1 // cycle count 
        );
    lastCPUTicks = currentCPUTicks;
    lastWallTime = GetTimeNow();
 
    msgInputCount++;
    return msg;
}
 
 
// T1.6: wait for a joined <triggergroup> set. Reuses waitForNewMessage() to dequeue
// (which sets currentTriggerMsg + currentTriggerName), then reads the member map that
// deliverTriggerGroup() attached under "Members". If the dequeued delivery is not a
// group activation, returns an empty map (and the group name is empty).
std::map<std::string, DataMessage*> PsyAPI::waitForNewMessageGroup(uint32 ms, std::string& groupName) {
    // Free any member messages we own from the previous grouped activation.
    std::map<std::string, DataMessage*>::iterator git, gitEnd;
    for (git = currentGroupMap.begin(), gitEnd = currentGroupMap.end(); git != gitEnd; git++)
        if (git->second) delete(git->second);
    currentGroupMap.clear();
    groupName.clear();
 
    const char* triggerName = NULL;
    DataMessage* primary = waitForNewMessage(ms, triggerName);
    // currentTriggerMsg is the delivery wrapper; it persists across timeout polls, so
    // only treat this as a group delivery when waitForNewMessage() actually dequeued a
    // fresh trigger this call (triggerName is non-NULL only for a fresh trigger msg).
    // Without this guard a completed group re-fires on every subsequent timeout poll.
    if (triggerName && currentTriggerMsg && currentTriggerMsg->getType() == CTRL_TRIGGER_GROUP) {
        currentGroupMap = currentTriggerMsg->getAttachedMessageMap("Members");
        groupName = triggerName;
    }
    else if (triggerName && primary) {
        // Not a group delivery: hand back the single message under its trigger name so a
        // crank calling only waitForNewMessageGroup() still sees something sensible.
        if (triggerName) groupName = triggerName;
        // Note: 'primary' is owned by the API (currentMsg); we do NOT copy it into the
        // owned map to avoid a double free. Callers mixing both APIs should prefer one.
    }
    return currentGroupMap;
}
 
bool PsyAPI::setAutoDelete(bool autodel) {
    autoDelete = autodel;
    return true;
}
 
// Get current message receive time
uint64 PsyAPI::getCurrentMessageReceiveTime() {
    uint64 t;
    if (currentTriggerMsg && (t = currentTriggerMsg->getRecvTime()) )
        return t;
    else if (currentMsg)
        return currentMsg->getRecvTime();
    else
        return 0;
}
 
int32 PsyAPI::postOutputMessage(PsyType msgType, DataMessage* msg) {
    if (msgType == NOTYPE)
        return postOutputMessage(NULL, msg);
    if (msg) {
        msg->setType(msgType);
        return postOutputMessage(NULL, msg);
    }
    DataMessage* newMsg = new DataMessage(msgType, 0);
    bool result = postOutputMessage("A non-existing postname", newMsg);
    delete(newMsg);
    return result;
}
 
 
int32 PsyAPI::postOutputMessage(const char* postName, DataMessage* msg) {
 
    uint64 currentCPUTicks;
    utils::GetCPUTicks(currentCPUTicks);
 
    if (!apiMutex.enter(1000))
        return POST_FAILED;
 
    if (msg)
        msg->addTimeUsage((uint32)(currentCPUTicks - lastCPUTicks), (uint32)GetTimeAge(lastWallTime), chainCPUTicks, chainWallTime, chainCount + 1);
 
    uint32 currentTag = 0;
    if (currentMsg)
        currentTag = currentMsg->getTag();
 
    uint32 to;
    int32 failed = 0;
    int32 success = 0;
    if (!currentTriggerSpec) {
        if (!msg) {
            apiMutex.leave();
            logPrint(1, "Posting none - out of context");
            return POST_OUTOFCONTEXT;
        }
        if (currentTag && !msg->getTag())
            msg->setTag(currentTag);
        if ( (to = msg->getTo()) || msg->getType().isValid()) {
            if (space->postMessage(msg)) {
                //logPrint(3,"Posted direct message to component %u", to);
                success++;
            }
            else {
                logPrint(0,"Failed to post direct message to component %u", to);
                failed++;
            }
        }
        else {
            apiMutex.leave();
            logPrint(1,"Posting none - out of context");
            delete(msg);
            return POST_OUTOFCONTEXT;
        }
    }
    else {
        PostSpec* postSpec;
        std::list<PostSpec*>* postSpecs = currentTriggerSpec->getPostSpecs(postName);
        if (!postSpecs || !postSpecs->size()) {
            delete(postSpecs);
            postSpecs = NULL;
            if (msg)
                postSpecs = currentTriggerSpec->getPostSpecs(msg->getType());
            if (!postSpecs || !postSpecs->size()) {
                delete(postSpecs);
                apiMutex.leave();
                logPrint(2, "Posting nothing - no post specs match for post name '%s'", postName);
                //logPrint(0, "Posting nothing - no post specs match for post name '%s'\n%s",
                //  postName, currentTriggerSpec->toXML().c_str());
                delete(msg);
                return POST_NOSPEC;
            }
        }
 
        uint64 msgTTL = 0;
        if (!msg) {
            msg = new DataMessage();
            msg->addTimeUsage((uint32)(currentCPUTicks - lastCPUTicks), (uint32)GetTimeAge(lastWallTime), chainCPUTicks, chainWallTime, chainCount + 1);
        }
        else
            msgTTL = msg->getTTL();
 
        if (currentTag && !msg->getTag())
            msg->setTag(currentTag);
 
        // uint32 msgTTL = msg->getTTL();
 
        std::list<PostSpec*>::iterator it = postSpecs->begin(), itEnd = postSpecs->end();
        while (it != itEnd) {
            if (postSpec = *it) {
                if (postSpec->contextchange.isValid()) {
                    msg->setType(CTRL_CONTEXT_CHANGE);
                    msg->setContextChange(postSpec->contextchange);
                    // msg->setData("Context", &postSpec->contextchange, sizeof(PsyContext));
                }
                else {
                    msg->setType(postSpec->type);
                }
                postSpec->addContentToMsg(msg);
                //if (strlen(postSpec->contentKey) && !msg->getString(postSpec->contentKey))
                //  msg->setString(postSpec->contentKey, postSpec->content);
                msg->setMultiple(currentCompID, postSpec->to, postSpec->tag, msgTTL ? msgTTL : postSpec->ttl, postSpec->policy, GetTimeNow());
 
                if (space->postMessage(msg)) {
                    //logPrint(3,"Posted msg delivered to %u", postSpec->to);
                    success++;
                }
                else {
                    logPrint(0,"Failed to deliver posted msg to %u", postSpec->to);
                    failed++;
                }
            }
            it++;
        }
        delete(postSpecs);
    }
 
    msgPostedCount++;
    msgSentCount += success;
 
    if (!success)
        logPrint(0,"No msgs were delivered at postOutputMessage", success);
    //else
    //  logPrint(3,"Posted msg was delivered to %u receivers", success);
 
    // Add stats to component
    space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_RUNNING,
        currentCPUTicks - lastCPUTicks, // cpu usage
        NULL, // no input message here
        msg, // output message
        0, // run count 0, only incremented at start
        0 // cycle count 0, only incremented at waitForNewMessage
        );
    // utils::GetCPUUsage(cpuUserBefore, cpuKernelBefore);
    lastCPUTicks = currentCPUTicks;
 
    apiMutex.leave();
    delete(msg);
    if (!success && failed)
        return POST_FAILED;
    else
        return success;
}
 
// ***************** Signals *****************
bool PsyAPI::emitSignal(const char* name, DataMessage* msg) {
 
    if (!apiMutex.enter(1000))
        return false;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        delete(msg);
        return false;
    }
 
    SignalSpec* signalSpec = currentTriggerSpec->getSignalSpec(name);
    if (!signalSpec) {
        apiMutex.leave();
        delete(msg);
        return false;
    }
    
    if (!msg)
        msg = new DataMessage();
    msg->setType(signalSpec->type);
    if (space->emitSignal(signalSpec->type, msg)) {
        apiMutex.leave();
        return true;
    }
    else {
        apiMutex.leave();
        delete(msg);
        return false;
    }
}
 
DataMessage* PsyAPI::waitForSignal(const char* name, uint32 timeout, uint64 lastReceivedTime) {
 
    apiMutex.enter(3000, __FUNCTION__);
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return NULL;
    }
 
    SignalSpec* signalSpec = currentTriggerSpec->getSignalSpec(name);
    if (!signalSpec) {
        apiMutex.leave();
        return NULL;
    }
    
    apiMutex.leave();
    DataMessage* msg = space->waitForSignal(signalSpec->type, timeout, lastReceivedTime);
 
    if (msg) {
        apiMutex.enter(3000, __FUNCTION__);
        if (currentSignalMsg)
            delete(currentSignalMsg);
        currentSignalMsg = msg;
        apiMutex.leave();
        return msg;
    }
    else
        return NULL;
}
 
 
 
uint8 PsyAPI::retrieve(std::list<DataMessage*> &result, const char* name, uint32 maxcount, uint32 maxage, uint32 timeout) {
    if (!apiMutex.enter(1000))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
    if (!retrieveSpec) {
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    RetrieveSpec* specCopy = new RetrieveSpec;
    memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));
 
    if (maxcount)
        specCopy->maxcount = maxcount;
    if (maxage)
        specCopy->maxage = maxage;
 
    apiMutex.leave();
    uint8 res = retrieve(result, specCopy, timeout);
    return res;
}
 
uint8 PsyAPI::retrieveTimeParam(std::list<DataMessage*> &result, const char* name, uint64 startTime, uint64 endTime, uint32 maxcount, uint32 maxage, uint32 timeout) {
    std::list<DataMessage*> list;
    if (!startTime)
        return QUERY_FAILED;
    if (!apiMutex.enter(1000))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
    if (!retrieveSpec) {
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    RetrieveSpec* specCopy = new RetrieveSpec;
    memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));
 
    specCopy->startTime = startTime;
    if (endTime)
        specCopy->endTime = endTime;
    if (maxcount)
        specCopy->maxcount = maxcount;
    if (maxage)
        specCopy->maxage = maxage;
 
    apiMutex.leave();
    uint8 res = retrieve(result, specCopy, timeout);
    return res;
}
 
uint8 PsyAPI::retrieveStringParam(std::list<DataMessage*> &result, const char* name, const char* startString, const char* endString, uint32 maxcount, uint32 maxage, uint32 timeout) {
    std::list<DataMessage*> list;
    if (!startString || !strlen(startString))
        return QUERY_FAILED;
 
    if (!apiMutex.enter(1000))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
    if (!retrieveSpec) {
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    RetrieveSpec* specCopy = new RetrieveSpec;
    memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));
 
    utils::strcpyavail(specCopy->startString, startString, MAXVALUENAMELEN, true);
    if (endString)
        utils::strcpyavail(specCopy->endString, endString, MAXVALUENAMELEN, true);
    if (maxcount)
        specCopy->maxcount = maxcount;
    if (maxage)
        specCopy->maxage = maxage;
 
    apiMutex.leave();
    uint8 res = retrieve(result, specCopy, timeout);
    return res;
}
 
uint8 PsyAPI::retrieveIntegerParam(std::list<DataMessage*> &result, const char* name, int64 startInteger, int64 endInteger, uint32 maxcount, uint32 maxage, uint32 timeout) {
    std::list<DataMessage*> list;
    if (startInteger == INT64_NOVALUE)
        return QUERY_FAILED;
 
    if (!apiMutex.enter(1000))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
    if (!retrieveSpec) {
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    RetrieveSpec* specCopy = new RetrieveSpec;
    memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));
 
    specCopy->startInt = startInteger;
    if (endInteger != INT64_NOVALUE)
        specCopy->endInt = endInteger;
    if (maxcount)
        specCopy->maxcount = maxcount;
    if (maxage)
        specCopy->maxage = maxage;
 
    apiMutex.leave();
    uint8 res = retrieve(result, specCopy, timeout);
    return res;
}
 
uint8 PsyAPI::retrieveFloatParam(std::list<DataMessage*> &result, const char* name, float64 startFloat, float64 endFloat, uint32 maxcount, uint32 maxage, uint32 timeout) {
    std::list<DataMessage*> list;
    if (startFloat == FLOAT64_NOVALUE)
        return QUERY_FAILED;
    if (!apiMutex.enter(1000))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
    if (!retrieveSpec) {
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    RetrieveSpec* specCopy = new RetrieveSpec;
    memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));
 
    specCopy->startFloat = startFloat;
    if (endFloat != FLOAT64_NOVALUE)
        specCopy->endFloat = endFloat;
    if (maxcount)
        specCopy->maxcount = maxcount;
    if (maxage)
        specCopy->maxage = maxage;
 
    apiMutex.leave();
    uint8 res = retrieve(result, specCopy, timeout);
    return res;
}
 
uint8 PsyAPI::retrieve(std::list<DataMessage*> &result, RetrieveSpec* spec, uint32 timeout) {
    if (!spec)
        return QUERY_FAILED;
    spec->origin = currentCompID;
 
    DataMessage* msg = new DataMessage(PsyAPI::CTRL_QUERY, this->currentCompID, spec->source, timeout);
    msg->setData("RetrieveSpec", (char*)spec, sizeof(RetrieveSpec));
 
    DataMessage* resultMsg = NULL;
    uint8 status = space->query(msg, &resultMsg, timeout);
    delete(msg);
 
    if (!resultMsg) {
        if (status == QUERY_FAILED)
            logPrint(0,"Query failed and did not return a message...");
        delete(spec);
        return status;
    }
 
    uint32 size;
    int64 count;
    const char* data = resultMsg->getData("ReplyData", size);
    if (!data || !resultMsg->getInt("ReplyCount", count)) {
        if (!data)
            logPrint(0,"Query did not return any ReplyData...");
        else
            logPrint(0,"Query did not return any ReplyCount...");
        delete(resultMsg);
        delete(spec);
        return QUERY_FAILED;
    }
 
    DataMessage* subMsg;
    const char* src = data;
    for (uint32 n=0; n<(uint32)count; n++) {
        if (GetObjID(src) != DATAMESSAGEID)
            break;
        subMsg = new DataMessage(src);
        result.push_back(subMsg);
        src += subMsg->getSize();
    }
 
    delete(spec);
    delete(resultMsg);
    return QUERY_SUCCESS;
}
 
bool PsyAPI::addSubscription(const char* xml) {
    if (!apiMutex.enter(1000))
        return false;
    DataMessage* msg = new DataMessage(PsyAPI::CTRL_ADDSUBSCRIPTION, this->currentCompID, 0, 5000);
    msg->setString("Subscription", xml);
 
    DataMessage* resultMsg = NULL;
    uint8 status = space->query(msg, &resultMsg, 5000);
    delete(msg);
 
    if (resultMsg)
        delete(resultMsg);
 
    apiMutex.leave();
    return (status == QUERY_SUCCESS);
}
 
 
uint8 PsyAPI::queryCatalog(DataMessage** resultMsg, const char* name, DataMessage* msg, uint32 timeout) {
    *resultMsg = NULL;
    if (!msg)
        return QUERY_FAILED;
    if (!apiMutex.enter(1000))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        logPrint(1, "Query '%s' cannot execute before a trigger message has been processed", name);
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    QuerySpec* querySpec = currentTriggerSpec->getQuerySpec(name);
    if (!querySpec) {
        logPrint(1, "Query '%s' not found in PsySpec", name);
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    if (strlen(querySpec->operation))
        msg->setString("Operation", querySpec->operation);
    if (strlen(querySpec->type))
        msg->setString("Type", querySpec->type);
    if (strlen(querySpec->ext))
        msg->setString("Ext", querySpec->ext);
    if (strlen(querySpec->subdir))
        msg->setString("Subdir", querySpec->subdir);
    if (strlen(querySpec->key))
        msg->setString("Key", querySpec->key);
    if (strlen(querySpec->value))
        msg->setString("Value", querySpec->value);
    if (querySpec->binary)
        msg->setString("Binary", "yes");
 
    if (strlen(querySpec->hostName)) {
        // This is a query for another system
        if (!querySpec->ipHost) {
            if (!utils::LookupIPAddress(querySpec->hostName, querySpec->ipHost)) {
                apiMutex.leave();
                return QUERY_COMPONENT_UNKNOWN;
            }
        }
        if (!querySpec->ipPort)
            querySpec->ipPort = 10000;
        // now we have all the info we need
        msg->setType(PsyAPI::CTRL_INTERSYSTEM_QUERY);
        msg->setFrom(currentCompID);
        msg->setTo(0);
        msg->setString("INTERSYSTEM_SOURCENAME", querySpec->sourceName);
        msg->setInt("INTERSYSTEM_ADDRESS", querySpec->ipHost);
        msg->setInt("INTERSYSTEM_PORT", querySpec->ipPort);
    }
    else {
        msg->setType(PsyAPI::CTRL_QUERY);
        msg->setFrom(currentCompID);
        msg->setTo(querySpec->source);
    }
 
    apiMutex.leave();
    uint8 status = space->query(msg, resultMsg, timeout);
    delete(msg);
 
    return status;
}
 
uint8 PsyAPI::queryRemoteCatalog(DataMessage** resultMsg, const char* componentName, const char* ipAddress, uint16 port, DataMessage* msg, uint32 timeout) {
    uint32 addr = 0;
    if (!utils::LookupIPAddress(ipAddress, addr))
        return QUERY_NOT_AVAILABLE;
    return queryRemoteCatalog(resultMsg, componentName, addr, port, msg, timeout);
}
 
uint8 PsyAPI::queryRemoteCatalog(DataMessage** resultMsg, uint32& chosenAddress, const char* componentName, uint32* ipAddresses, uint32 numAddresses, uint16 port, DataMessage* msg, uint32 timeout) {
    if (chosenAddress > numAddresses)
        return QUERY_NOT_AVAILABLE;
 
    uint8 result;
    if (chosenAddress) {
        // try this address first
        result = queryRemoteCatalog(resultMsg, componentName, ipAddresses[chosenAddress-1], port, msg, timeout);
        if (result != QUERY_NOT_REACHABLE)
            return result;
        // if it failed, go back and try all again
    }
 
    chosenAddress = 0;
    for (uint32 n = 0; n < numAddresses; n++) {
        result = queryRemoteCatalog(resultMsg, componentName, ipAddresses[n], port, msg, timeout);
        if (result != QUERY_NOT_REACHABLE) {
            chosenAddress = n + 1;
            return result;
        }
        // else continue trying next address
    }
    // and give up if none of them connects
    return QUERY_NOT_REACHABLE;
}
 
uint8 PsyAPI::queryRemoteCatalog(DataMessage** resultMsg, const char* componentName, uint32 ipAddress, uint16 port, DataMessage* msg, uint32 timeout) {
    *resultMsg = NULL;
    if (!msg)
        return QUERY_FAILED;
 
    // This is a query for another system
    // we have all the info we need
    msg->setType(PsyAPI::CTRL_INTERSYSTEM_QUERY);
    msg->setFrom(currentCompID);
    msg->setTo(0);
    msg->setString("INTERSYSTEM_SOURCENAME", componentName);
    msg->setInt("INTERSYSTEM_ADDRESS", ipAddress);
    msg->setInt("INTERSYSTEM_PORT", port);
 
    uint8 status = space->query(msg, resultMsg, timeout);
    delete(msg);
 
    return status;
}
 
 
 
 
uint8 PsyAPI::queryCatalog(char** result, uint32 &resultsize, const char* name, const char* query, const char* operation, const char* data, uint32 datasize, uint32 timeout) {
    *result = NULL;
    if (!apiMutex.enter(1000, "Replying to query"))
        return QUERY_FAILED;
 
    if (!currentTriggerSpec) {
        apiMutex.leave();
        return QUERY_FAILED;
    }
 
    QuerySpec* querySpec = currentTriggerSpec->getQuerySpec(name);
    if (!querySpec) {
        apiMutex.leave();
        return QUERY_NAME_UNKNOWN;
    }
 
    DataMessage* msg = new DataMessage();
 
    msg->setString("Name", querySpec->name);
 
    if (query && strlen(query))
        msg->setString("Query", query);
    if (data && datasize)
        msg->setData("Data", data, datasize);
    if (operation && strlen(operation))
        msg->setString("Operation", operation);
    else if (strlen(querySpec->operation))
        msg->setString("Operation", querySpec->operation);
 
    if (strlen(querySpec->type))
        msg->setString("Type", querySpec->type);
    if (strlen(querySpec->ext))
        msg->setString("Ext", querySpec->ext);
    if (strlen(querySpec->subdir))
        msg->setString("Subdir", querySpec->subdir);
    if (strlen(querySpec->key))
        msg->setString("Key", querySpec->key);
    if (strlen(querySpec->value))
        msg->setString("Value", querySpec->value);
    if (querySpec->binary)
        msg->setString("Binary", "yes");
 
    if (strlen(querySpec->hostName)) {
        // This is a query for another system
        if (!querySpec->ipHost) {
            if (!utils::LookupIPAddress(querySpec->hostName, querySpec->ipHost)) {
                apiMutex.leave();
                return QUERY_COMPONENT_UNKNOWN;
            }
        }
        if (!querySpec->ipPort)
            querySpec->ipPort = 10000;
        // now we have all the info we need
        msg->setType(PsyAPI::CTRL_INTERSYSTEM_QUERY);
        msg->setFrom(currentCompID);
        msg->setTo(0);
        msg->setString("INTERSYSTEM_SOURCENAME", querySpec->sourceName);
        msg->setInt("INTERSYSTEM_ADDRESS", querySpec->ipHost);
        msg->setInt("INTERSYSTEM_PORT", querySpec->ipPort);
    }
    else {
        msg->setType(PsyAPI::CTRL_QUERY);
        msg->setFrom(currentCompID);
        msg->setTo(querySpec->source);
    }
 
    apiMutex.leave();
    DataMessage* resultMsg = NULL;
    uint8 status = space->query(msg, &resultMsg, timeout);
    delete(msg);
 
    if (resultMsg) {
        *result = resultMsg->getDataCopy("ReplyData", resultsize);
        delete(resultMsg);
    }
 
    return status;
}
 
bool PsyAPI::queryReply(uint32 id, uint8 status, char* data, uint32 size, uint32 count) {
    DataMessage* msg = NULL;
    if (data && size) {
        msg = new DataMessage(PsyAPI::CTRL_QUERY_REPLY, currentCompID);
        msg->setData("ReplyData", data, size);
        msg->setInt("ReplyCount", count);
        // decide on EOL ###################
    }
    bool res = queryReply(id, status, msg);
    //delete(msg);
    return res;
}
 
bool PsyAPI::queryReply(uint32 id, uint8 status, DataMessage* msg) {
    if (!apiMutex.enter(1000, "Replying to query"))
        return false;
    bool res = space->queryReply(id, status, msg);
    apiMutex.leave();
    return res;
}
 
bool PsyAPI::logPrint(int level, const char *formatstring, ... ) {
    char* str;
    uint32 lenStr;
    va_list args;
    va_start(args, formatstring);
    str = utils::StringFormatVA(lenStr, formatstring, args);
    va_end(args);
    if (!str || !lenStr) {
        free(str);
        return false;
    }
 
    char* name = new char[256];
    if (!space->manager->componentMemory->getComponentName(currentCompID, name, 255)) {
        delete [] name;
        free(str);
        return false;
    }
 
    uint32 lenNewStr;
    char* newStr = utils::StringFormat(lenNewStr, "[%s] %s", name, str);
    // delete[] for `name` (new char[256] at the top of this function), free() for
    // `str` and `newStr` (StringFormat returns a vasprintf/malloc'd buffer,
    // Utils.cpp:7644). The two allocators sit two lines apart here, which is
    // exactly how this function ended up mismatched in the first place.
    delete [] name;
    free(str);
 
    LogEntry* entry = (LogEntry*) malloc(sizeof(LogEntry)+lenNewStr+1);
    entry->size = sizeof(LogEntry)+lenNewStr+1;
    entry->cid = LOGENTRYID;
    entry->time = GetTimeNow();
    entry->source = currentCompID;
    entry->subject = LOG_COMPONENT;
    entry->level = (uint8)level;
    entry->type = LOGPRINT;
    entry->setText(newStr, lenNewStr);
    free(newStr);
 
    return space->logEntry(entry);
//  return LogPrint(this->currentCompID, LOG_COMPONENT, level, str);
}
 
} // namespace cmlabs