ComponentMemory.h

Go to the documentation of this file.

 
#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;   
    uint32 recentInPos;  // circular write index into recentInMsg ring (newest written here, then advanced)
    uint32 recentOutPos; // circular write index into recentOutMsg ring
    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;     
    uint8 status;       
    uint8 policy;       
    uint16 nodeID;      
    uint16 procID;      
    // Basic data
    uint64 createdTime;     
    uint64 migratedTime;    
    uint64 lastUpdateTime;  
    char name[MAXKEYNAMELEN]; 
    // 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_