MemoryManager.h

#if !defined(_MEMORYMANAGER_H_)
#define _MEMORYMANAGER_H_

#include "MemoryController.h"
#include "Utils.h"
#include "ThreadManager.h"
#include "DataMessage.h"

namespace cmlabs {

#define PSYCLONE_STATUS_NONE    0
#define PSYCLONE_STATUS_ENDED   1
#define PSYCLONE_STATUS_INIT    2
#define PSYCLONE_STATUS_READY   3

#define DRAFTMSGSIZE        4096

struct PsycloneIndexStruct {
    uint32 size;
    uint32 cid;
    uint32 count;
    // PsycloneIndexEntries
};

struct PsycloneIndexEntry {
    uint16 port;
    uint8 status;
    uint64 heartbeat;
    uint64 instanceID;
};

#define PSYCLONE_INDEX_MAXCOUNT 128
#define PSYCLONE_INDEX_SIZE     (sizeof(PsycloneIndexStruct) + (PSYCLONE_INDEX_MAXCOUNT * sizeof(PsycloneIndexEntry)))

class PsycloneIndex {
public:
    static bool GetStatus(uint16 port, uint8& status, uint64& heartbeat);

    PsycloneIndex();
    ~PsycloneIndex();
    bool init(uint16 port, uint64 instID);
    bool setStatus(uint8 status);
private:
    PsycloneIndexStruct* index;
    PsycloneIndexEntry* entry;
};

struct PerfStats {
    uint32 compID;
    uint16 spaceID;
    uint16 nodeID;
    uint32 osID;
    uint8 type;
    uint8 status;    // Component status (idle, running, etc.)

    uint64 currentCPUTicks;

    uint64 currentMemoryBytes;

    uint64 totalInputBytes;
    uint64 totalInputCount;
    uint64 totalOutputBytes;
    uint64 totalOutputCount;

    uint64 runCount;
    uint64 totalQueueBytes;
    uint32 totalQueueCount;

    uint64 firstRunStartTime;
    uint64 currentRunStartTime;
    uint64 totalCycleCount;
    uint64 totalRunCount;
    uint32 migrationCount;
};

struct AveragePerfStats {
    uint64 time;
    uint32 periodsMS[3];

    PerfStats lastStats;

    double percentOfSystemCPU[3];
    double percentOfOSCPU[3];
    double localSystemCPUUsage[3];
    double localSystemMemoryUsage[3];
    double computerCPUUsage[3];
    double computerMemoryUsage[3];

    uint64 maxMemoryBytes;
    double memoryBytesAverage[3];

    double dataInputBytesPerSec[3];
    double dataInputCountPerSec[3];
    double dataOutputBytesPerSec[3];
    double dataOutputCountPerSec[3];

    double dataQueueBytesAverage[3];
    uint64 dataQueueCountAverage[3];

    double cycleCountPerSec[3];

    std::string toXML() {
        return utils::StringFormat("<performance time=\"%llu\" ms1=\"%u\" ms2=\"%u\" ms3=\"%u\" maxmemory=\"%llu\" "
            "percentofsystemcpu1=\"%f\" percentofsystemcpu2=\"%f\" percentofsystemcpu3=\"%f\" "
            "percentofoscpu1=\"%f\" percentofoscpu2=\"%f\" percentofoscpu3=\"%f\" "
            "memorybytesaverage1=\"%f\" memorybytesaverage2=\"%f\" memorybytesaverage3=\"%f\" "
            "datainputbytespersec1=\"%f\" datainputbytespersec2=\"%f\" datainputbytespersec3=\"%f\" "
            "datainputcountpersec1=\"%f\" datainputcountpersec2=\"%f\" datainputcountpersec3=\"%f\" "
            "dataoutputbytespersec1=\"%f\" dataoutputbytespersec2=\"%f\" dataoutputbytespersec3=\"%f\" "
            "dataoutputcountpersec1=\"%f\" dataoutputcountpersec2=\"%f\" dataoutputcountpersec3=\"%f\" "
            "dataqueuebytesaverage1=\"%f\" dataqueuebytesaverage2=\"%f\" dataqueuebytesaverage3=\"%f\" "
            "dataqueuecountaverage1=\"%f\" dataqueuecountaverage2=\"%f\" dataqueuecountaverage3=\"%f\" "
            "cyclecountpersec1=\"%f\" cyclecountpersec2=\"%f\" cyclecountpersec3=\"%f\" "
            "localsystemcpuusage1=\"%f\" localsystemcpuusage2=\"%f\" localsystemcpuusage3=\"%f\" "
            "localsystemmemoryusage1=\"%f\" localsystemmemoryusage2=\"%f\" localsystemmemoryusage3=\"%f\" "
            "computercpuusage1=\"%f\" computercpuusage2=\"%f\" computercpuusage3=\"%f\" "
            "computermemoryusage1=\"%f\" computermemoryusage2=\"%f\" computermemoryusage3=\"%f\" "
            "/>\n",
            time, periodsMS[0], periodsMS[1], periodsMS[2], maxMemoryBytes,
            percentOfSystemCPU[0], percentOfSystemCPU[1], percentOfSystemCPU[2],
            percentOfOSCPU[0], percentOfOSCPU[1], percentOfOSCPU[2],
            memoryBytesAverage[0], memoryBytesAverage[1], memoryBytesAverage[2],
            dataInputBytesPerSec[0], dataInputBytesPerSec[1], dataInputBytesPerSec[2],
            dataInputCountPerSec[0], dataInputCountPerSec[1], dataInputCountPerSec[2],
            dataOutputBytesPerSec[0], dataOutputBytesPerSec[1], dataOutputBytesPerSec[2],
            dataOutputCountPerSec[0], dataOutputCountPerSec[1], dataOutputCountPerSec[2],
            dataQueueBytesAverage[0], dataQueueBytesAverage[1], dataQueueBytesAverage[2],
            dataQueueCountAverage[0], dataQueueCountAverage[1], dataQueueCountAverage[2],
            cycleCountPerSec[0], cycleCountPerSec[1], cycleCountPerSec[2],
            localSystemCPUUsage[0], localSystemCPUUsage[1], localSystemCPUUsage[2],
            localSystemMemoryUsage[0], localSystemMemoryUsage[1], localSystemMemoryUsage[2],
            computerCPUUsage[0], computerCPUUsage[1], computerCPUUsage[2],
            computerMemoryUsage[0], computerMemoryUsage[1], computerMemoryUsage[2]
            );
    }
};


struct MemoryMasterStruct {
    uint64 size;
    uint32 cid;
    uint16 nodeID;
    uint16 status;          // 0-10, where 10 means ready
    uint64 createdTime;     // time of creation
    // uint64 timesync;     // time sync
    uint64 currentTMC;
    int64 localSyncAdjustment;
    uint64 memAlloc;        //
    uint64 memUsage;        //
    uint64 currentCPUTicks;     //
    uint32 componentCount;  //
    uint32 throughputCount; //
    uint64 throughputSize;  //
    uint32 dynamicShmemSerial;
    uint32 processShmemSerial;
    uint32 componentShmemSerial;
    uint32 datamapsShmemSerial;
    uint32 dynamicShmemLocks;
    uint32 processShmemLocks;
    uint32 componentShmemLocks;
    // uint32 datamapsShmemLocks; // nothing ever locks in here beyond the simple mutex
    uint64 dynamicShmemSize;
    uint64 processShmemSize;
    uint64 componentShmemSize;
    uint64 datamapsShmemSize;
};

class MemoryManager;
class MasterMemory : public MemoryController {
    friend class MemoryManager;
public:
    MasterMemory();
    ~MasterMemory();
    bool create(uint16 port);
    bool open(uint16 port);

    uint16 getNodeID();
    bool setNodeID(uint16 id);
    uint64 getCreatedTime();

    uint32 getDynamicShmemSerial();
    uint32 getProcessShmemSerial();
    uint32 getComponentShmemSerial();
    uint32 getDataMapsShmemSerial();
    uint32 incrementDynamicShmemSerial();
    uint32 incrementProcessShmemSerial();
    uint32 incrementComponentShmemSerial();
    uint32 incrementDataMapsShmemSerial();

    uint64 getDynamicShmemSize();
    uint64 getProcessShmemSize();
    uint64 getComponentShmemSize();
    uint64 getDataMapsShmemSize();
    bool setDynamicShmemSize(uint64 size);
    bool setProcessShmemSize(uint64 size);
    bool setComponentShmemSize(uint64 size);
    bool setDataMapsShmemSize(uint64 size);

    bool setID(uint16 id);
    uint16 getID();

    uint16 port;

private:
    utils::Mutex* mutex;
    MemoryMasterStruct* master;
};

class TemporalMemory;
class ProcessMemory;
class ComponentMemory;
class DataMapsMemory;

class MemoryManager {
public:
    static MemoryManager* Singleton;

    //static bool GetMemoryUsage(uint32& total, uint32& usage, uint32& sysTotal, uint32& sysUsage, uint32& staticTotal, uint32& staticUsage, uint32& dynamicTotal, uint32& dynamicUsage);
    //static bool InsertMemoryBlock(const char* data, uint32 size, uint64 eol, uint64& id);
    //static bool OverwriteMemoryBlock(uint64 id, const char* data, uint32 size, uint64 eol);
    //static char* GetAndLockMemoryBlock(uint64 id, uint32& size, uint64& eol);
    //static bool UnlockMemoryBlock(uint64 id);
    //static char* GetCopyMemoryBlock(uint64 id, uint32& size, uint64& eol);

    static bool UnitTest();
    static bool ShmUnitTest();

    friend THREAD_RET THREAD_FUNCTION_CALL MemoryManagement(THREAD_ARG arg);

    MemoryManager();
    ~MemoryManager();

    bool getMemoryUsage(uint64& sysAlloc, uint64& sysUsage, uint64& dataAlloc, uint64& dataUsage);

    uint8 getNodeStatus(uint64& lastseen, uint64& createdTime);

    // Connect to existing PageMaster in shared memory
    bool connect(uint16 sysID, bool isMaster);

    // Create a new PageMaster in shared memory
    bool create(uint16 sysID, uint32 slotCount = 100000, uint16 binCount = 2, uint32 minBlockSize = 1024, uint32 maxBlockSize = 64*1024, uint64 initSize = 50000000L, uint64 maxSize = 1000000000L, bool force = false);
        //uint16 sysID, uint32 slotCount = 100, uint32 slotDuration = 100000, uint64 initSize = 10000000, uint64 maxSize = 100000000, uint32 growSteps = 8, bool force = false);

    // Insert new block of memory and return full id
    bool insertMessage(DataMessage* msg, uint64& id);
    // Get copy of block of memory
    DataMessage* getCopyOfMessage(uint64 id);

    std::string typeToText(PsyType type);
    std::string subTypeToText(uint16 subtype);
    std::string contextToText(PsyContext context);
    std::string subContextToText(uint16 subcontext);
    std::string getComponentName(uint32 compID);

    uint16 getNodeID();
    bool setNodeID(uint16 id);

    DataMessage* retrieveAllSystemIDs();


    bool getTimeSyncData(uint64& tmc, int64& adjust);
    bool setTimeSyncData(uint64 tmc, int64 adjust);

    TemporalMemory* temporalMemory;     // Shared memory
    ProcessMemory* processMemory;       // Shared memory
    ComponentMemory* componentMemory;   // Shared memory
    DataMapsMemory* dataMapsMemory;     // Shared memory

private:
    // Continuous management of eol pages
    uint32 runManager();

    // Resize the EOL Page
    bool resizeEOLPage(uint32 slotCount, uint32 pagesPerSlot, uint32 slotDuration);

    PsycloneIndex* psycloneIndex;       // Shared memory
    MasterMemory* masterMemory;         // Shared memory


    bool shouldContinue;
    bool isRunning;
    bool isLocalMaster;
    uint32 managementThreadID;
    uint16 thisInstance;
};














#define MAXREGISTERCOUNT 64

struct CMRegister {
    uint16 size; // size of memory allocation
    uint16 count; // number of entries
    // MAXREGISTERCOUNT x uint16 sysIDs
};

THREAD_RET THREAD_FUNCTION_CALL MemoryManagement(THREAD_ARG arg);

#define NODE_PROCESS_ID     1
#define MAXINSTANCES        256


struct MemoryID {
    uint32 pageID; // contains [8] PageSerial [24] PageLoc
    uint16 blockID; // index containing block offset
    uint16 sysID; // global node id
};

struct MemoryBlock {
    uint32 size;
//  char* data;
};

struct PagePoolHeader {
    uint32 size; // total size of pool structure in bytes
    // uint32 size, uint64 time, uint32 pageID
};

struct PagePoolEntry {
    uint32 dataSize; // page data size
    uint64 time; // time of entry into pool
    uint32 pageID; // ID of page
};

struct EOLHeader {
    uint32 slotCount;
    uint32 pagesPerSlot;
    uint32 startSlot;
    uint64 startTime;
    uint32 slotDuration;
    // slot0: page0,eol0 , page1,eol1 ..., pageN,eolN
    // slot1: page0,eol0 , page1,eol1 ..., pageN,eolN
};

struct EOLEntry {
    uint32 pageID;
    uint64 eol;
};

#define MP_GetBlockIndexStartLoc(p) ((char*)p + sizeof(MemoryPage))
#define MP_GetBlockDataStartLoc(p) (MP_GetBlockIndexStartLoc(p) + (p->blockTableSize*sizeof(uint32)) )
//#define MP_GetBlockOffsetLoc(p,n) (uint32*)(MP_GetBlockIndexStartLoc(p) + (n * sizeof(uint32)))
//#define MP_GetBlockOffset(p,n) (*(MP_GetBlockOffsetLoc(p,n)))
//#define MP_GetBlockLoc(p,n) MP_GetBlockDataStartLoc(p) + MP_GetBlockOffset(p,n)

struct MemoryPage {
    uint32 size; // total size of page in bytes
    uint64 eol; // current number of blocks in use -> first free location
    uint32 dataSize; // total size of data space in bytes
    uint32 dataUsage; // current usage of page in bytes
    uint16 count; // current number of blocks in use -> first free location
    uint16 blockTableSize; // maximum entries in index table
//  uint16 blockTable[1024]; // index table containing block offsets
//  char* data; // MemoryBlock0, MemoryBlock1, ..., MemoryBlockN
};

struct MemoryPageEntry {
    uint8 pageSerial; // running serial number for this page location (0-255)
    uint8 cacheSerial; // running serial number for this page cache (0-255)
    uint32 pageSize;
    uint32 pageID;
    uint64 lockID;
    uint64 lockTime;
    uint16 lockCount;
};

struct MemoryPageCache {
    uint32 size; // total size of structure in bytes
    uint32 count; // total number of cache entries allocated
    // MemoryPageCacheEntries...
};

struct MemoryPageCacheEntry {
    uint8 pageSerial; // running serial number for this page location (0-255)
    uint8 cacheSerial; // running serial number for this page cache (0-255)
    uint32 pageSize;
    uint32 pageID;
    MemoryPage* page;
};

#define MP_SYSMEMID MP_CalcID(0, 0, pageMaster->nodeID)
#define MP_CalcID(pid,bid,nid) ((uint64)pid | (((uint64)bid) << 32) | (((uint64)nid) << 48))
#define MP_SetBlockID(id,var) var |= (((uint64)id) << 32)
#define MP_SetNodeID(id,var) var |= (((uint64)id) << 48)
#define MP_SetPageID(id,var) var |= ((uint64)id)
#define MP_GetBlockID(id) (*(uint16*)((char*)&id + sizeof(uint32)))
#define MP_GetNodeID(id) (*(uint16*)((char*)&id + sizeof(uint32) + sizeof(uint16)))
#define MP_GetPageID(id) (*(uint32*)(char*)&id)
#define MP_GetBitFieldLoc(m) ((char*)m + sizeof(MemoryPageMaster))
#define MP_GetPageTableLoc(m) (MP_GetBitFieldLoc(m) + m->bitFieldSize )
#define MP_GetPageEntryLoc(m,pid) (MP_GetPageTableLoc(m) + (pid * sizeof(MemoryPageEntry)) )
#define MP_GetPageLoc(m,pid) ((MemoryPageEntry*)MP_GetPageEntryLoc(m,pid))->page

#define MP_SYSTEMPAGE 0
#define MP_STATICPAGE 1
#define MP_DYNAMICPAGE 2
#define MP_GetPageType(id,eol) ( ((id < RESERVEDPAGECOUNT) ? MP_SYSTEMPAGE : (eol > 0 ? MP_DYNAMICPAGE : MP_STATICPAGE) ) )

struct MemoryPageMaster {
    uint32 size; // total size of PageMaster in bytes
    uint16 sysID; // global node id
    uint16 status; // 0-10, where 10 means ready
    uint64 createdTime; // time of creation
    uint8 instance; // Instance number on each computer
    uint64 timesync; // time sync
    uint32 pagesSize; // total size of all pages in bytes {system, static, EOL}
    uint32 dataSize[3]; // total size of all pages data spaces in bytes {system, static, EOL}
    uint32 dataUsage[3]; // current usage of all pages in bytes {system, static, EOL}
    uint32 count[3]; // current number of pages in use, for info only {system, static, EOL}
    uint32 lastCreatedStaticPage; // id of last created static page
    uint32 bitFieldSize; // number of bytes in the bitfield
    uint32 pageTableSize; // maximum entries in page table and number of bits in bit field
    // char* pageBitfield; // which pages are in use -> 111010001110101010....011010
    // MemoryPageEntry* pageTable; // {ID24 offset -> address of MemoryPageEntry}
};

// SystemBlock for system data

// structure for managing ttl pages

class MemoryManagerX {
public:
    static MemoryManagerX* Singleton;

    static char* GetAndLockSystemBlock(uint32 pageID, uint32& size);
    static bool UnlockSystemBlock(uint32 pageID);
    // Creates a new System Page with the next id
    static char* CreateAndLockNewSystemPage(uint32 size, uint32& id);
    // Resize an existing System Page
    static char* ResizeSystemPage(uint32 pageID, uint32 newSize);
    static bool GetMemoryUsage(uint32& total, uint32& usage, uint32& sysTotal, uint32& sysUsage, uint32& staticTotal, uint32& staticUsage, uint32& dynamicTotal, uint32& dynamicUsage);
    // Destroys an exising System Page by id
    static bool DestroySystemPage(uint32 pageID);

    static bool InsertMemoryBlock(const char* data, uint32 size, uint64 eol, uint64& id);
    static bool OverwriteMemoryBlock(uint64 id, const char* data, uint32 size, uint64 eol);
    static char* GetAndLockMemoryBlock(uint64 id, uint32& size, uint64& eol);
    static bool UnlockMemoryBlock(uint64 id);
    static char* GetCopyMemoryBlock(uint64 id, uint32& size, uint64& eol);

    static bool UnitTest();

    friend THREAD_RET THREAD_FUNCTION_CALL MemoryManagement(THREAD_ARG arg);

    MemoryManagerX();
    ~MemoryManagerX();

    // Connect to existing PageMaster in shared memory
    bool connect(uint16 sysID);

    // Create a new PageMaster in shared memory
    bool create(uint16 sysID, uint32 staticPageTableSize = 100, uint32 slotCount = 360, uint32 slotDuration = 60000000, uint32 pagesPerSlot = 10, bool force = false);

    // Either look up location of or create a new PageMaster in shared memory
    // bool init(uint16 nodeID, uint32 pageTableSize, uint32 slotCount = 600, uint32 slotDuration = 1000000, uint32 bufferSlots = 10, uint32 pagesPerSlot = 1024);

    // Insert new block of memory and return full id
    bool insertMemoryBlock(const char* data, uint32 size, uint64 eol, uint64& id);

    // Overwrite existing block of memory
    bool overwriteMemoryBlock(uint64 id, const char* data, uint32 size, uint64 eol);

    char* getAndLockMemoryBlock(uint64 id, uint32& size, uint64& eol);
    char* getLockedMemoryBlock(MemoryPage* page, uint64 id, uint32& size, uint64& eol);
    bool unlockMemoryBlock(uint64 id);
    char* getCopyMemoryBlock(uint64 id, uint32& size, uint64& eol);

    // Return a system memory block
    char* getSystemBlock(uint32 pageID, uint32& size);

    // Return a system memory block and lock it
    char* getAndLockSystemBlock(uint32 pageID, uint32& size);
    char* getLockedSystemBlock(uint32 pageID, uint32& size);
    char* getLockedSystemBlock(MemoryPage* page, uint32& size);

    // Unlock System Block
    bool unlockSystemBlock(uint32 pageID);

    // Creates a new System Page with the next id
    char* createAndLockNewSystemPage(uint32 size, uint32& id);
    // Resize an existing System Page
    char* resizeSystemPage(uint32 pageID, uint32 size);

private:
    // Create new MemoryPage
    MemoryPage* createPage(uint32 size, uint16 tableSize, uint64 eol, uint32& pageID); // if either parameter is 0, the default values will be used

    // Insert new block of memory into a specific page and return full id
    bool insertMemoryBlockIntoPage(uint32 pageID, const char* data, uint32 size, uint64 eol, uint64& id);

    // Create a new Page Mutex
    // Mutex createPageMutex(uint32 id);
    // Get an existing Page Mutex
    // bool getPageMutex(uint32 id, NMutex &mutex);
    // Destroy a Page Mutex
    // bool destroyPageMutex(uint32 id);

//  utils::Semaphore* getPageSemaphore(uint32 pageID);
//  utils::Mutex* getPageMutex(uint32 pageID);

    bool lockPage(uint32 pageID);
    bool unlockPage(uint32 pageID);
//  bool lockPage(MemoryPage* page);
//  bool unlockPage(MemoryPage* page);
    bool lockPage(MemoryPageEntry* entry);
    bool unlockPage(MemoryPageEntry* entry);

    // Find an available page appropriate for the block size and ttl
    MemoryPage* findPageForBlock(uint32 size, uint64 eol, uint32& id);

    // Free up a page and put it into the pool
    bool freePageIntoPool(uint32 pageID);

    // Remove page from stats, but do not delete
    bool removePageFromStats(uint32 pageID, uint32& dataSize);

    // Insert a page into the idle pool
    bool insertPageIntoPool(uint32 pageID, uint32 dataSize);

    // Get page with required size from the idle pool
    bool getPageFromPool(uint32 size, uint32& pageID);

    // Destroy all Memory Pages
    bool destroyAllPages();

    // Destroy a Memory Page
    bool destroyPage(uint32 pageID, bool updateStats = true);

    // Continuous management of eol pages
    uint32 runManager();

    // Return string representing the bitfield
    char* getBitFieldAsString(uint32& size);
    // Print string representing the bitfield
    bool printBitFieldAsString(char* title);

    // Get the percentage in use of a page
    bool getPageUsage(uint32 pageID, double& val);
    // Get the percentage in use of a page
    double getPageUsage(uint32 pageID);
    // Get the number of bytes free of a page
    bool getPageFree(uint32 pageID, uint32& bytes);
    // Get the total size in bytes a page
    bool getPageSize(uint32 pageID, uint32& bytes);

    // Creates a System Page
    char* createSystemPage(uint32 id, uint32 size, uint64& memID);

    // Create the Manager System Pages
    bool createSystemPages();

    // Initialise the System Page
    bool initSystemPage(char* data, uint32 size);

    // Initialise the Page Pool
    bool initPagePool(char* data, uint32 size);

    // Resize the EOL Page
    bool resizeEOLPage(uint32 slotCount, uint32 pagesPerSlot, uint32 slotDuration, uint32 bufferSlots);

    // Insert page into EOL Structure
    bool insertPageIntoEOL(uint32 pageID, uint64 eol, uint64& slotEOL);

    // Initialise the Queue Page
    bool initQueuePage(char* data, uint32 size);

    // Calculate the new page size based on incoming block size
    static uint16 CalcBlockTableSize(uint32 size);

    MemoryPage* getCachedPage(uint32 pageID);
    MemoryPage* getCachedPage(MemoryPageEntry *entry);
    bool updateCachedPage(MemoryPageEntry *entry, MemoryPage *page);

    MemoryPageMaster* pageMaster; // Shared memory location of PageMaster, looked up or created
    utils::Mutex *pageMasterMutex; // Handle to global mutex for PageMaster access
    MemoryPageCache* pageCache; // Non-shared memory location of cached page addresses
    bool shouldContinue;
    bool isRunning;
    bool isMaster;
    uint32 managementThreadID;
    uint16 thisInstance;
};

} // namespace cmlabs

#include "TemporalMemory.h"
#include "ComponentMemory.h"
#include "DataMapsMemory.h"
#include "ProcessMemory.h"

#endif //_MEMORYMANAGER_H_