ProcessMemory.h

#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;

    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;                              // id, corresponds to the index in map
    uint16 nodeID;                          // Node ID
    uint64 createdTime;                     // time of creation
    uint64 lastseen;                        // last status update
    uint8 status;                           // status of the process
    uint32 osID;                            // id of OS process
    uint8 type;                             // type 0: normal, 1: inside Node
//  stats(activity)
    char name[MAXKEYNAMELEN+1];             // name in text
    char commandline[MAXCOMMANDLINELEN+1];  // commandline in text
    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];             // name for mutex
};

#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_