

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  and is released under the following
//_/_/  
//_/_/  EXTENDED BSD License
//_/_/
//_/_/  Redistribution and use in source and binary forms, with or without
//_/_/  modification, is permitted provided that the following conditions 
//_/_/  are met:
//_/_/
//_/_/  - Redistributions of source code must retain the above copyright 
//_/_/    and collaboration notice, this list of conditions and the following 
//_/_/    disclaimer.
//_/_/
//_/_/  - Redistributions in binary form must reproduce the above copyright 
//_/_/    notice, this list of conditions and the following disclaimer in the 
//_/_/    documentation and/or other materials provided with the distribution.
//_/_/
//_/_/  - Neither the name of its copyright holders nor the names of its 
//_/_/    contributors may be used to endorse or promote products derived from 
//_/_/    this software without specific prior written permission.
//_/_/
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#if !defined(_PROCESSMEMORY_H_)
#define _PROCESSMEMORY_H_

#include "MemoryManager.h"
#include "PsyTime.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, commandline, 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_

