ProcessMemory.h

Go to the documentation of this file.

 
#if !defined(_PROCESSMEMORY_H_)
#define _PROCESSMEMORY_H_
 
#include "MemoryManager.h"
#include "PsyTime.h"
#include "HTML.h"
 
namespace cmlabs {

#define SIGNALQ_ID 0    
#define MSGQ_ID 1       
#define REQQ_ID 2       
#define CMDQ_ID 3       
 
#define MAXPROC 64      
#define MAXPROC4 256    

#define PSYPROC_ERROR           0   
#define PSYPROC_CREATED         1   
#define PSYPROC_INIT            2   
#define PSYPROC_IDLE            3   
#define PSYPROC_READY           4   
#define PSYPROC_ACTIVE          5   
#define PSYPROC_SHUTTING_DOWN   6   
#define PSYPROC_TERMINATED      7   

#define CMDQ_TYPE       1   
#define MSGQ_TYPE       2   
#define SIGQ_TYPE       3   
#define REQQ_TYPE       4   
 
THREAD_RET THREAD_FUNCTION_CALL QueueTest(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL ProcessMemoryPerfTest(THREAD_ARG arg);

struct ProcessStats {
    uint64 time;        
    uint64 msgInCount;  
    uint64 msgInBytes;  
    char recentInMsg[10 * DRAFTMSGSIZE];    
    uint64 msgOutCount; 
    uint64 msgOutBytes; 
    char recentOutMsg[10 * DRAFTMSGSIZE];   
    uint64 procMemUsage;        
    uint64 currentCPUTicks;     
    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("<processstats msgincount=\"%llu\" msginbytes=\"%llu\" msgoutcount=\"%llu\" msgoutbytes=\"%llu\" time=\"%llu\" memusage=\"%llu\">\n<inmsg>\n%s</inmsg>\n<outmsg>\n%s</outmsg>\n</processstats>\n",
            msgInCount, msgInBytes, msgOutCount, msgOutBytes, time, procMemUsage, xmlIn.c_str(), xmlOut.c_str());
    }
};

struct ProcessInfoStruct {
    uint32 id;                              
    uint16 nodeID;                          
    uint64 createdTime;                     
    uint64 lastseen;                        
    uint8 status;                           
    uint32 osID;                            
    uint8 type;                             
//  stats(activity)
    char name[MAXKEYNAMELEN+1];             
    char commandline[MAXCOMMANDLINELEN+1];  
    ProcessStats stats;                     
    AveragePerfStats perfStats;             
    uint16 cmdQID;      
    uint16 msgQID;      
    uint16 sigQID;      
    uint16 reqQID;      

    std::string toXML() {
        return utils::StringFormat("<space id=\"%u\" name=\"%s\" node=\"%u\" status=\"%u\" osid=\"%u\" commandline=\"%s\" created=\"%llu\" lastseen=\"%llu\">\n%s%s</space>",
            id, name, nodeID, status, osID, html::EncodeHTML(commandline).c_str(), createdTime, lastseen, stats.toXML().c_str(), perfStats.toXML().c_str());
    }
};

struct ProcessMemoryStruct {
    uint64 size;            
    uint32 cid;             
    uint64 createdTime;     // time of creation
    uint32 count;           
    uint64 usage;           
    ProcessInfoStruct processes[MAXPROC];   
    uint64 qIndex[MAXPROC*4];   
};

#define INITIALQSIZE sizeof(MessageQueueHeader) + 2048
struct MessageQueueHeader {
    uint64 size;        
    uint32 id;          
    uint32 count;       
    uint64 startPos;    
    uint64 endPos;      
    uint64 padding;     
    char name[MAXKEYNAMELEN+1];             
};

#define CHECKPROCESSMEMORYSERIAL    if (serial != master->getProcessShmemSerial()) {if (!open()) {mutex->leave();return 0;}}
 
class MasterMemory;
class ProcessMemory {
public:
    static bool UnitTest();     
    static bool PerfTest();     

    ProcessMemory(MasterMemory* master);
    ~ProcessMemory();

    bool getMemoryUsage(uint64& alloc, uint64& usage);
    std::vector<ProcessInfoStruct>* getAllProcesses();
    bool checkProcessHeartbeats(uint32 timeoutMS, std::list<ProcessInfoStruct>& procIssues);
 
    bool open();    
    bool create(uint32 initialProcCount);
 
    // Processes
    bool createNewProcess(const char* name, uint16 &id);
    bool deleteProcess(uint16 id);
 
    // Process Info
    bool getProcessName(uint16 id, char* name, uint32 maxSize); 
    bool getProcessID(const char* name, uint16 &id);            
    uint64 getProcessCreateTime(uint16 id);                     
 
    uint8 getProcessStatus(uint16 id, uint64& lastseen);        
    uint8 getProcessStatus(uint16 id, uint64& lastseen, uint64& createTime);    
    bool getProcessCommandLine(uint16 id, char* cmdline, uint32 maxSize);       
    uint16 getProcessIDFromOSID(uint32 osid);   
    uint32 getProcessOSID(uint16 id);           

    bool setProcessStatus(uint16 id, uint8 status, uint64 currentCPUTicks = 0);
    bool setProcessCommandLine(uint16 id, const char* cmdline); 
    bool setProcessOSID(uint16 id, uint32 osid);                
    bool setProcessType(uint16 id, uint8 type);                 
 
    AveragePerfStats getProcessPerfStats(uint16 procID);        
    bool setProcessPerfStats(uint16 procID, AveragePerfStats &perfStruct);  

    bool addToProcessStats(uint16 id, DataMessage* inputMsg, DataMessage* outputMsg);
 
    // Process Queues
    bool addToCmdQ(uint16 procID, DataMessage* msg);    
    bool addToMsgQ(uint16 procID, DataMessage* msg);    
    bool addToSigQ(uint16 procID, DataMessage* msg);    
    bool addToReqQ(uint16 procID, DataMessage* msg);    
    bool addToAllSignalQs(DataMessage* msg);            
    bool addToAllSignalQsExcept(DataMessage* msg, uint16 except);   

    DataMessage* waitForCmdQ(uint16 procID, uint32 timeout);    
    DataMessage* waitForMsgQ(uint16 procID, uint32 timeout);    
    DataMessage* waitForSigQ(uint16 procID, uint32 timeout);    
    DataMessage* waitForReqQ(uint16 procID, uint32 timeout);    
 
    uint32 getCmdQCount(uint16 procID);     
    uint32 getMsgQCount(uint16 procID);     
    uint32 getSigQCount(uint16 procID);     
    uint32 getReqQCount(uint16 procID);     

    bool getQueueSizes(uint16 procID, uint64 &bytes, uint32 &count);
    bool addLocalPerformanceStats(std::list<PerfStats> &perfStats);
 
private:
    bool resize(uint64 newMemorySize);
    bool setupNextAvailableQ(uint16& qID);
    bool deleteQ(uint16 id);
    bool resizeQ(uint16 id, uint64 newSize);
 
    bool printQ(MessageQueueHeader* qHeader);
 
    MessageQueueHeader* getQHeader(uint16 procID, uint8 qType);
    bool addToQ(uint16 procID, uint8 qType, DataMessage* msg);
    DataMessage* waitForQ(uint16 procID, uint8 qType, uint32 timeout);
 
    utils::Mutex* mutex;            
    MasterMemory* master;           
    ProcessMemoryStruct* header;    
    uint64 memorySize;              
    uint16 port;                    
    uint32 serial;                  
 
    utils::Semaphore* qSemaphores[MAXPROC4];    
};
 
} // namespace cmlabs
 
#endif //_PROCESSMEMORY_H_