ComponentMemory.h

#if !defined(_COMPONENTMEMORY_H_)
#define _COMPONENTMEMORY_H_

#include "MemoryManager.h"
#include "PsyTime.h"
#include "Subscriptions.h"

namespace cmlabs {


#define COMPSTATUS_IDLE         1
#define COMPSTATUS_STARTING     2
#define COMPSTATUS_RUNNING      3
#define COMPSTATUS_FINISHED     4

#define NOTINUSE    65535

#define GETCOMPOFFSETP(header,id) (uint64*)(((char*)header) + sizeof(ComponentMemoryStruct) + (id*sizeof(uint64)))
#define GETCOMPOFFSET(header,id) *GETCOMPOFFSETP(header,id)
#define GETCOMPDATA(header,id) (ComponentInfoStruct*)(GETCOMPOFFSET(header,id) ? ((char*)header) + (header->indexSize * sizeof(uint64)) + GETCOMPOFFSET(header,id) : NULL)

struct ComponentMemoryStruct {
    uint64 size;
    uint32 cid;
    uint64 createdTime;     // time of creation
    uint32 indexSize;
    uint32 count;
    uint32 maxID;
    uint64 usage;
    // index * N, each a uint64
};

#define PARAM_STRING        1
#define PARAM_INTEGER       2
#define PARAM_FLOAT         3
#define PARAM_STRING_COLL   4
#define PARAM_INTEGER_COLL  5
#define PARAM_FLOAT_COLL    6
#define PARAM_TYPE_COLL     7

#define COMPONENT_CAN_MIGRATE   1

struct ComponentStats {
    uint64 time;
    uint64 msgInCount;
    uint64 msgInBytes;
    char recentInMsg[10*DRAFTMSGSIZE];
    uint64 msgOutCount;
    uint64 msgOutBytes;
    char recentOutMsg[10*DRAFTMSGSIZE];
    uint64 firstRunStartTime;
    uint64 currentRunStartTime;
    uint64 cycleCount;
    uint64 runCount;
    uint32 migrationCount;
    uint64 usageCPUTicks;
    std::string toXML() {
        std::string xmlIn, xmlOut;
        int n;
        const char* oldData = NULL;
        DataMessage* msg = new DataMessage();
        char* msgData = recentInMsg;
        for (n = 0; n < 10; n++) {
            if (((DataMessageHeader*)msgData)->time) {
                if (!oldData) oldData = msg->swapMessageData(msgData);
                else msg->swapMessageData(msgData);
                xmlIn += msg->toXML();
            }
            msgData += DRAFTMSGSIZE;
        }
        msgData = recentOutMsg;
        for (n = 0; n < 10; n++) {
            if (((DataMessageHeader*)msgData)->time) {
                if (!oldData) oldData = msg->swapMessageData(msgData);
                else msg->swapMessageData(msgData);
                xmlOut += msg->toXML();
            }
            msgData += DRAFTMSGSIZE;
        }
        if (oldData)
            msg->swapMessageData(oldData);
        delete(msg);
        return utils::StringFormat("<componentstats runcount=\"%llu\" msgincount=\"%llu\" msginbytes=\"%llu\" msgoutcount=\"%llu\" msgoutbytes=\"%llu\" time=\"%llu\">\n<inmsg>\n%s</inmsg>\n<outmsg>\n%s</outmsg>\n</componentstats>\n",
            runCount, msgInCount, msgInBytes, msgOutCount, msgOutBytes, time, xmlIn.c_str(), xmlOut.c_str());
    }
};

#define SELFTRIGGER_NONE 0
#define SELFTRIGGER_WARN 1
#define SELFTRIGGER_ALLOW 2


struct ComponentInfoStruct {
    uint64 size;
    uint32 id;
    uint8 syncStatus;
    uint8 selfTrigger;
    uint8 type;    // Component type (Whiteboard, Catalog, Module, etc.)
    uint8 status;    // Component status (idle, running, etc.)
    uint8 policy;   // Migration policy
    uint16 nodeID; // id of process component is running in currently
    uint16 procID; // id of process component is running in currently
    // Basic data
    uint64 createdTime;
    uint64 migratedTime;
    uint64 lastUpdateTime;
    char name[MAXKEYNAMELEN]; // name in text
    // Statistics
    ComponentStats stats;
    AveragePerfStats perfStats;
    // Parameters
    uint16 paramCount;
    uint32 paramSize;
    // Private Data
    uint16 dataCount;
    uint64 dataSize;

    std::string toXML(bool printStats = true) {
        if (printStats)
            return utils::StringFormat("<component id=\"%u\" status=\"%u\" name=\"%s\" node=\"%u\" proc=\"%u\" type=\"%u\" policy=\"%u\" paramcount=\"%u\" paramsize=\"%u\" datacount=\"%u\" datasize=\"%llu\" createdtime=\"%llu\" migratedtime=\"%llu\" lastupdatetime=\"%llu\" selftrigger=\"%u\">\n%s%s</component>\n",
                id, syncStatus, name, nodeID, procID, type, policy, paramCount, paramSize, dataCount, dataSize, createdTime, migratedTime, lastUpdateTime, selfTrigger, stats.toXML().c_str(), perfStats.toXML().c_str());
        else
            return utils::StringFormat("<component id=\"%u\" status=\"%u\" name=\"%s\" node=\"%u\" proc=\"%u\" type=\"%u\" policy=\"%u\" paramcount=\"%u\" paramsize=\"%u\" datacount=\"%u\" datasize=\"%llu\" createdtime=\"%llu\" migratedtime=\"%llu\" lastupdatetime=\"%llu\" selftrigger=\"%u\" />\n",
                id, syncStatus, name, nodeID, procID, type, policy, paramCount, paramSize, dataCount, dataSize, createdTime, migratedTime, lastUpdateTime, selfTrigger);
    }
    bool writeToMsg(uint32 n, DataMessage* msg) {
        if (!msg) return false;
        msg->setInt((int64)n, "ID", id);
        msg->setInt((int64)n, "Status", status);
        msg->setInt((int64)n, "Type", type);
        msg->setInt((int64)n, "Policy", policy);
        msg->setInt((int64)n, "SelfTrigger", selfTrigger);
        msg->setInt((int64)n, "Size", size);
        msg->setInt((int64)n, "NodeID", nodeID);
        msg->setInt((int64)n, "ProcID", procID);
        msg->setString((int64)n, "Name", name);
        msg->setTime((int64)n, "CreatedTime", createdTime);
        return true;
    }
};

#define COMPINFOUSAGE(info) (sizeof(ComponentInfoStruct) + info->paramSize + info->dataSize)

struct PrivateHeader {
    uint64 size;
    // name, 0, mimetype, 0, data
};

struct ParamHeader {
    uint32 size;
    uint8 type;
    // name, 0, struct
};

// -- integer --
struct ParamIntegerHeader {
    int64 val;
    int64 defaultVal;
    int64 minVal;
    int64 maxVal;
    int64 interval;
};

// -- float --
struct ParamFloatHeader {
    float64 val;
    float64 defaultVal;
    float64 minVal;
    float64 maxVal;
    float64 interval;
};

// -- integer collection --
struct ParamCollectionHeader {
    uint32 index;
    uint32 defaultIndex;
    uint32 count;
};

#define CHECKCOMPONENTMEMORYSERIAL  if (serial != master->getComponentShmemSerial()) {if (!open()) {mutex->leave();return 0;}}
#define CHECKCOMPONENTMEMORYSERIALRETURN(a)     if (serial != master->getComponentShmemSerial()) {if (!open()) {mutex->leave();return a;}}

class MasterMemory;
class ComponentMemory {
public:
    static bool UnitTest();

    ComponentMemory(MasterMemory* master);
    ~ComponentMemory();

    bool getMemoryUsage(uint64& alloc, uint64& usage);

    bool open();
    bool create(uint32 indexSize);

    std::vector<ComponentInfoStruct>* getAllComponents();
    std::vector<ComponentInfoStruct>* getAllLocalComponents();
    std::vector<ComponentInfoStruct>* getAllLocalComponents(uint8 type);
    std::vector<ComponentInfoStruct>* getAllModules();
    std::vector<ComponentInfoStruct>* getAllCatalogs();

    bool writeComponentsToMsg(DataMessage* msg);
    bool writeComponentNamesToMsg(DataMessage* msg);
    bool syncComponents(DataMessage* msg);

    bool confirmComponentID(uint32 id);
    bool cancelComponentID(uint32 id);
    uint8 lookupComponentID(const char* name, uint32 &id); // status = 0: not there, 1: in-sync, 2: ready
    bool createComponent(uint32 id, uint8 type, uint8 policy, uint8 selfTrigger, const char* name, uint64 size, uint16 nodeID, uint16 procID, uint64 time, uint32& existingID);
//  bool recordRemoteComponent(uint32 id, const char* name, uint16 nodeID, uint64 time);
//  bool reserveComponentID(uint32 id, const char* name);
    bool destroyComponent(uint32 cid);
    bool getComponentName(uint32 cid, char* name, uint32 maxNameLen);
    std::string getComponentNameString(uint32 cid);
    uint8 getComponentType(uint32 cid);
    bool getComponentID(const char* name, uint32 &cid);
    ComponentStats getComponentStats(uint32 cid);
    bool setComponentStats(uint32 cid, ComponentStats& stats);
    AveragePerfStats getComponentPerfStats(uint32 cid);
    bool setComponentPerfStats(uint32 cid, AveragePerfStats &perfStruct);

    bool isComponentLocal(uint32 cid);
    bool isComponentLocal(const char* name);
    bool getComponentLocation(uint32 cid, uint16& nodeID, uint16& procID);
    uint8 getComponentPolicy(uint32 cid);
    bool updateComponentInformation(uint32 cid, uint8 type, uint8 policy, uint8 selfTrigger, uint16 nodeID, uint16 procID);
    bool setComponentPolicy(uint32 cid, uint8 policy);
    uint8 getComponentSelfTrigger(uint32 cid);
    bool setComponentSelfTrigger(uint32 cid, uint8 selfTrigger);
    uint16 getComponentNodeID(uint32 cid);
    uint16 getComponentProcessID(uint32 cid);
    bool updateComponentLocation(uint32 cid, uint16 nodeID, uint16 procID);
    // char* GetAndRemoveComponent(uint32 cid, uint32& dataSize);
    // bool addExistingComponent(uint32 cid, uint16 nodeID, uint16 procID, char* oldData, uint32 dataSize);
    // bool updateRemoteComponent(uint32 cid, uint16 nodeID, uint16 procID, const char* name);
    bool addComponentStats(uint32 cid, uint8 status, uint64 usageCPUTicks,
        DataMessage* inputMsg, DataMessage* outputMsg, uint32 runCount, uint32 cycleCount);

    bool canComponentMigrate(uint32 cid);
    std::list<uint16>* findProcessComponents(uint16 procID);

    char* getComponentData(uint32 cid, uint64& size);
    bool setComponentData(uint32 cid, const char* data, uint64 size, bool wasMigrated);

    // Private Data
    bool setPrivateData(uint32 cid, const char* name, const char* data, uint64 size, const char* mimetype = NULL);
    uint64 getPrivateDataSize(uint32 cid, const char* name);
    std::string getPrivateDataMimetype(uint32 cid, const char* name);
    bool getPrivateData(uint32 cid, const char* name, char* data, uint64 maxSize);
    char* getPrivateDataCopy(uint32 cid, const char* name, uint64 &size);
    bool deletePrivateData(uint32 cid, const char* name);
    std::map<std::string, std::string> getPrivateDataKeysAndTypes(uint32 cid);
    std::list<std::string> getPrivateDataKeys(uint32 cid);

    // Parameters
    bool createParameter(uint32 cid, const char* name, const char* val, const char* defaultValue = NULL);
    bool createParameter(uint32 cid, const char* name, const char* val, uint32 count, uint32 defaultIndex);
    bool createParameter(uint32 cid, const char* name, std::vector<std::string> values, const char* defaultValue = NULL);
    bool createParameter(uint32 cid, const char* name, int64* val, uint32 count, uint32 defaultIndex);
    bool createParameter(uint32 cid, const char* name, std::vector<std::string> values, int64 defaultValue);
    bool createParameter(uint32 cid, const char* name, std::vector<int64> values, int64 defaultValue = 0);
    bool createParameter(uint32 cid, const char* name, float64* val, uint32 count, uint32 defaultIndex);
    bool createParameter(uint32 cid, const char* name, std::vector<std::string> values, float64 defaultValue);
    bool createParameter(uint32 cid, const char* name, std::vector<float64> values, float64 defaultValue = 0);
    bool createParameter(uint32 cid, const char* name, int64 val, int64 min = 0, int64 max = 0, int64 interval = 0);
    bool createParameter(uint32 cid, const char* name, float64 val, float64 min = 0, float64 max = 0, float64 interval = 0);

    bool hasParameter(uint32 cid, const char* name);
    bool deleteParameter(uint32 cid, const char* name);

    uint8 getParameterDataType(uint32 cid, const char* name);
    uint32 getParameterValueSize(uint32 cid, const char* name);
    bool getParameter(uint32 cid, const char* name, char* val, uint32 maxSize);
    bool getParameter(uint32 cid, const char* name, int64& val);
    bool getParameter(uint32 cid, const char* name, float64& val);
    std::string getParameterString(uint32 cid, const char* name, uint32 maxSize);
    std::string getParameterString(uint32 cid, const char* name);
    int64 getParameterInt(uint32 cid, const char* name);
    float64 getParameterFloat(uint32 cid, const char* name);
    bool getParameterAsBool(uint32 cid, const char* name);

    bool setParameter(uint32 cid, const char* name, const char* val);
    bool setParameter(uint32 cid, const char* name, int64 val);
    bool setParameter(uint32 cid, const char* name, float64 val);

    bool resetParameter(uint32 cid, const char* name);
    bool tweakParameter(uint32 cid, const char* name, int32 tweak);

    bool addLocalPerformanceStats(std::list<PerfStats> &perfStats);
    std::string printFriendly();
    std::string toXML(bool stats = true);

private:
    bool resize(uint64 newMemorySize);
    bool resizeIndex(uint32 newIndexSize);
    bool resizeComponent(uint32 id, uint64 newSize);
    bool growComponent(uint32 id, uint64 addSize);
    bool getSurroundingComponents(uint32 cid, uint32& prevID, uint32& nextID);

    char* makeRoomForPrivateData(uint32 cid, uint64 newSize);
    char* makeRoomForParameter(uint32 cid, uint64 newSize);

    ParamHeader* getParameter(uint32 cid, const char* name);
    PrivateHeader* getPrivateData(uint32 cid, const char* name);

    utils::Mutex* mutex;
    MasterMemory* master;
    ComponentMemoryStruct* header;
    uint64 memorySize;
    uint16 port;
    uint32 serial;
};

} // namespace cmlabs

#endif //_COMPONENTMEMORY_H_