

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE 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/
//_/_/
//_/_/  PSYCLONE 2.0 is released under the GNU Lesser General Public License 
//_/_/  (LGPL) version 3.0, and is subject to the following
//_/_/  
//_/_/  - 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. 
//_/_/    
//_/_/  A commercial license for PSYCLONE 2.0 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.0 and/or modify it under 
//_/_/  the terms of the GNU General Public License as published by the Free 
//_/_/  Software Foundation, either version 3 of the License, or (at your 
//_/_/  option) any later version.
//_/_/  
//_/_/  You should have received a copy of the GNU General Public License
//_/_/  and the GNU Lesser General Public License along with PSYCLONE 2.0.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.0 is distributed in the hope that it will be useful,
//_/_/  but WITHOUT ANY WARRANTY; without even the implied warranty of
//_/_/  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//_/_/  GNU General Public License for more details.
//_/_/
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "Monitoring.h"

namespace	cmlabs{


// Todo: Fix MovingAverage to update when no new values are added
// Check compCom, all 0!

std::ostream& operator<<(std::ostream& os, const PerformanceMatrixResults& entry) {
    os << utils::StringFormat("<perfentry from=\"%u\" to=\"%u\" countpersec1=\"%f\" valpersec1=\"%f\" countpersec2=\"%f\" valpersec2=\"%f\" countpersec3=\"%f\" valpersec3=\"%f\" />\n",
		entry.from, entry.to, entry.countPerSec[0], entry.valuePerSec[0], entry.countPerSec[1], entry.valuePerSec[1], entry.countPerSec[2], entry.valuePerSec[2]);
    return os;
}

bool MonitoringStats::addStats(PerfStats &stats, uint64 time, uint64 osMem, OSProcStats &procStats) {
	if (this->time) {
		uint64 difTicks = stats.currentCPUTicks-lastStats.currentCPUTicks;
		double systemPercent = 0;
		if (procStats.difTicks)
			systemPercent = (double)difTicks / (double)procStats.difTicks;

		cpuTicksDif.add((double)difTicks, 1, time);
		memoryBytesTotal.add((double)stats.currentMemoryBytes, 1, time);
		if (stats.currentMemoryBytes > maxMemoryBytes)
			maxMemoryBytes = stats.currentMemoryBytes;
		if (stats.totalInputCount)
			dataInputDif.add((double)(stats.totalInputBytes-lastStats.totalInputBytes), (uint32)(stats.totalInputCount-lastStats.totalInputCount), time);
		if (stats.totalOutputCount)
			dataOutputDif.add((double)(stats.totalOutputBytes-lastStats.totalOutputBytes), (uint32)(stats.totalOutputCount-lastStats.totalOutputCount), time);
		if (!stats.compID)
			dataQueuesDif.add((double)stats.totalQueueBytes, (uint32)stats.totalQueueCount, time);
		if (stats.compID)
			cycleCountDif.add(0, (uint32)(stats.totalCycleCount-lastStats.totalCycleCount), time);

		if (stats.compID) {
			// procStats here is the parent proc (space)
			// systemMemoryPercent only for components, others are always 100%
			systemMemoryPercent.add((double)stats.currentMemoryBytes/(double)procStats.procMem, 1, time);
			// percentOfSystemCPU only for components, others are always 100%
			percentOfSystemCPU.add(systemPercent, 1, time);
			percentOfOSCPU.add(systemPercent * procStats.procCPU, 1, time);
		}
		else {
			// procStats here is the proc itself
			percentOfOSCPU.add(procStats.procCPU, 1, time);
		}
		osMemoryPercent.add((double)stats.currentMemoryBytes/(double)osMem, 1, time);
	}
	this->time = time;
	lastStats = stats;
	return true;
}

bool MonitoringStats::processData(uint32 ms1, uint32 ms2, uint32 ms3, uint64 now) {

	lastPerfStruct.lastStats = lastStats;

	// parameter time is assumed to be 'now', not last data entry time used below
	lastPerfStruct.time = this->time; 
	lastPerfStruct.periodsMS[0] = ms1;
	lastPerfStruct.periodsMS[1] = ms2;
	lastPerfStruct.periodsMS[2] = ms3;

	double val1, val2, val3;		// values not used
	uint64 count1, count2, count3;	// values not used

	percentOfSystemCPU.getAverageMulti(ms1, ms2, ms3,
		lastPerfStruct.percentOfSystemCPU[0], count1,
		lastPerfStruct.percentOfSystemCPU[1], count2,
		lastPerfStruct.percentOfSystemCPU[2], count3);
	
	percentOfOSCPU.getAverageMulti(ms1, ms2, ms3,
		lastPerfStruct.percentOfOSCPU[0], count1,
		lastPerfStruct.percentOfOSCPU[1], count2,
		lastPerfStruct.percentOfOSCPU[2], count3);

	lastPerfStruct.maxMemoryBytes = maxMemoryBytes;
	
	memoryBytesTotal.getAverageMulti(ms1, ms2, ms3,
		lastPerfStruct.memoryBytesAverage[0], count1,
		lastPerfStruct.memoryBytesAverage[1], count2,
		lastPerfStruct.memoryBytesAverage[2], count3);
	
	dataInputDif.getThroughputMulti(ms1, ms2, ms2,
		lastPerfStruct.dataInputBytesPerSec[0], lastPerfStruct.dataInputCountPerSec[0],
		lastPerfStruct.dataInputBytesPerSec[1], lastPerfStruct.dataInputCountPerSec[1],
		lastPerfStruct.dataInputBytesPerSec[2], lastPerfStruct.dataInputCountPerSec[2]);
	dataOutputDif.getThroughputMulti(ms1, ms2, ms2,
		lastPerfStruct.dataOutputBytesPerSec[0], lastPerfStruct.dataOutputCountPerSec[0],
		lastPerfStruct.dataOutputBytesPerSec[1], lastPerfStruct.dataOutputCountPerSec[1],
		lastPerfStruct.dataOutputBytesPerSec[2], lastPerfStruct.dataOutputCountPerSec[2]);

	dataQueuesDif.getAverageMulti(ms1, ms2, ms3,
		lastPerfStruct.dataQueueBytesAverage[0], lastPerfStruct.dataQueueCountAverage[0],
		lastPerfStruct.dataQueueBytesAverage[1], lastPerfStruct.dataQueueCountAverage[1],
		lastPerfStruct.dataQueueBytesAverage[2], lastPerfStruct.dataQueueCountAverage[2]);

	cycleCountDif.getThroughputMulti(ms1, ms2, ms2,
		val1, lastPerfStruct.cycleCountPerSec[0],
		val2, lastPerfStruct.cycleCountPerSec[1],
		val3, lastPerfStruct.cycleCountPerSec[2]);

	return true;
}


PerformanceMatrix::PerformanceMatrix(uint32 binTimeMS, uint16 binCount) {
	this->binTimeMS = binTimeMS;
	this->binCount = binCount;
}

PerformanceMatrix::~PerformanceMatrix() {
	mutex.enter();
	std::map<uint64, MovingAverage*>::iterator i=matrix.begin(), e=matrix.end();
	while (i != e) {
		delete(i->second);
		i++;
	}
	matrix.clear();
	mutex.leave();
}

bool PerformanceMatrix::add(uint32 from, uint32 to, double val, uint32 count, uint64 time) {
	mutex.enter();

	uint64 fromto = MERGEUINT3264(from,to);
	MovingAverage* ma = matrix[fromto];
	if (!ma)
		matrix[fromto] = ma = new MovingAverage(binTimeMS, binCount);
	if (ma->add(val, count, time)) {
		mutex.leave();
		return true;
	}
	else {
		mutex.leave();
		return false;
	}
}

bool PerformanceMatrix::getAverage(uint32 from, uint32 to, uint32 ms, double& val, uint64& count, uint64 now) {
	mutex.enter();
	MovingAverage* ma = matrix[MERGEUINT3264(from,to)];
	if (!ma)
		return false;
	if (ma->getAverage(ms, val, count, now)) {
		mutex.leave();
		return true;
	}
	else {
		mutex.leave();
		return false;
	}
}

std::map<uint64, PerformanceMatrixResults> PerformanceMatrix::getThroughputResults(uint32 ms1, uint32 ms2, uint32 ms3, uint64 now) {
	std::map<uint64, PerformanceMatrixResults> resultMatrix;
	PerformanceMatrixResults entry;

	mutex.enter();

	std::map<uint64, MovingAverage*>::iterator i=matrix.begin(), e=matrix.end();
	while (i != e) {
		entry.from = GET1UINT3264(i->first);
		entry.to = GET2UINT3264(i->first);
		i->second->getThroughputMulti(ms1, ms2, ms3, entry.valuePerSec[0], entry.countPerSec[0], entry.valuePerSec[1], entry.countPerSec[1], entry.valuePerSec[2], entry.countPerSec[2], now);
		resultMatrix[MERGEUINT3264(entry.from,entry.to)] = entry;
		i++;
	}
	mutex.leave();
	return resultMatrix;
}

std::string PerformanceMatrix::getThroughputXML(const char* tagname, uint32 ms1, uint32 ms2, uint32 ms3, uint64 now) {
	std::stringstream ss;
	ss << "<" << tagname << ">" << std::endl;
	PerformanceMatrixResults entry;

	mutex.enter();
	std::map<uint64, MovingAverage*>::iterator i=matrix.begin(), e=matrix.end();
	while (i != e) {
		entry.from = GET1UINT3264(i->first);
		entry.to = GET2UINT3264(i->first);
		i->second->getThroughputMulti(ms1, ms2, ms3, entry.valuePerSec[0], entry.countPerSec[0], entry.valuePerSec[1], entry.countPerSec[1], entry.valuePerSec[2], entry.countPerSec[2], now);
		ss << entry << std::endl;
		i++;
	}
	mutex.leave();
	ss << "</" << tagname << ">" << std::endl;
	return ss.str();
}

char* PerformanceMatrix::getThroughputBinary(uint32 &size, uint32 ms1, uint32 ms2, uint32 ms3, uint64 now) {
	PerformanceMatrixResults entry;
	mutex.enter();

	uint32 count = (uint32)matrix.size();

	// create and copy data
	PerformanceMatrixResults* data = new PerformanceMatrixResults[count];
	uint32 c = 0;

	std::map<uint64, MovingAverage*>::iterator i=matrix.begin(), e=matrix.end();
	while (i != e) {
		entry.from = GET1UINT3264(i->first);
		entry.to = GET2UINT3264(i->first);
		i->second->getThroughputMulti(ms1, ms2, ms3, entry.valuePerSec[0], entry.countPerSec[0], entry.valuePerSec[1], entry.countPerSec[1], entry.valuePerSec[2], entry.countPerSec[2], now);
		memcpy(&data[c++], &entry, sizeof(PerformanceMatrixResults));
		i++;
	}
	mutex.leave();
	size = count * sizeof(PerformanceMatrixResults);
	return (char*)data;
}

std::map<uint64, PerformanceMatrixResults> PerformanceMatrix::GetThroughputResultsFromBinary(const char* data, uint32 size) {
	std::map<uint64, PerformanceMatrixResults> resultMatrix;
	PerformanceMatrixResults* entries = (PerformanceMatrixResults*)data;
	PerformanceMatrixResults* entry;

	uint32 count = size / sizeof(PerformanceMatrixResults);
	for (uint32 c=0; c<count; c++) {
		entry = entries++;
		resultMatrix[MERGEUINT3264(entry->from,entry->to)] = *entry;
	}
	return resultMatrix;
}


std::string PerformanceMatrix::GetThroughputXMLFromBinary(const char* tagname, const char* data, uint32 size) {
	std::stringstream ss;
	ss << "<" << tagname << ">" << std::endl;
	PerformanceMatrixResults* entries = (PerformanceMatrixResults*)data;
	PerformanceMatrixResults* entry;

	uint32 count = size / sizeof(PerformanceMatrixResults);
	for (uint32 c=0; c<count; c++) {
		entry = entries++;
		ss << *entry << std::endl;
	}
	ss << "</" << tagname << ">" << std::endl;
	return ss.str();
}

std::string PerformanceMatrix::GetThroughputXMLFromResults(const char* tagname, std::map<uint64, PerformanceMatrixResults>& results) {
	std::stringstream ss;
	ss << "<" << tagname << ">" << std::endl;

	std::map<uint64, PerformanceMatrixResults>::iterator i = results.begin(), e = results.end();
	while (i != e) {
		ss << i->second << std::endl;
		i++;
	}
	ss << "</" << tagname << ">" << std::endl;
	return ss.str();
}








Monitoring::Monitoring(uint32 ms1, uint32 ms2, uint32 ms3) {
	node = NULL;
	this->ms1 = ms1;
	this->ms2 = ms2;
	this->ms3 = ms3;
	uint32 msMax = clmax(clmax(ms1, ms2), ms3);
	prevOSTicks = 0;
	prevOSIdleTicks = 0;
	prevProcTicks = 0;
	nodeCom = new PerformanceMatrix(100, (msMax/100)+50);
	compCom = new PerformanceMatrix(100, (msMax/100)+50);
	perfType = NOTYPE;
}

Monitoring::~Monitoring() {
	stop();
	delete(nodeCom);
	delete(compCom);
	node = NULL;


	std::map<uint32, MonitoringStats*>::iterator iLS=localSpaceStats.begin(), eLS=localSpaceStats.end();
	while (iLS != eLS) {
		delete(iLS->second);
		iLS++;
	}
	localSpaceStats.clear();
	
	std::map<uint32, MonitoringStats*>::iterator iLC=localComponentStats.begin(), eLC=localComponentStats.end();
	while (iLC != eLC) {
		delete(iLC->second);
		iLC++;
	}
	localComponentStats.clear();
	
	std::map<uint16, AveragePerfStats*>::iterator iON=otherNodeStats.begin(), eON=otherNodeStats.end();
	while (iON != eON) {
		delete(iON->second);
		iON++;
	}
	otherNodeStats.clear();
	
	std::map<uint32, AveragePerfStats*>::iterator iOS=otherSpaceStats.begin(), eOS=otherSpaceStats.end();
	while (iOS != eOS) {
		delete(iOS->second);
		iOS++;
	}
	otherSpaceStats.clear();
	
	std::map<uint32, AveragePerfStats*>::iterator iOC=otherComponentStats.begin(), eOC=otherComponentStats.end();
	while (iOC != eOC) {
		delete(iOC->second);
		iOC++;
	}
	otherComponentStats.clear();
}

bool Monitoring::init(Node* parent) {
	if (!parent)
		return false;
	node = parent;
	return true;
}

bool Monitoring::startMonitoring() {
	perfType = node->getPsyType("Psyclone.Performance");

	if (!ThreadManager::CreateThread(MonitoringRun, this, threadID)) {
		LogPrint(0, 0, 0, "Could not start Monitoring thread...");
		return false;
	}
	return true;
}

bool Monitoring::run() {

	isRunning = true;

	uint32 tid;
	utils::GetCurrentThreadOSID(tid);

	uint32 counter = 0;
	while (shouldContinue) {
		// shouldContinue should be checked every 100ms or so, hence the counter implementation to do every 500ms
		utils::Sleep(100);
		if (++counter == 10) {
			gatherPerformanceStats();
			counter = 0;
		}
	}

	isRunning = false;
	return true;
}


THREAD_RET THREAD_FUNCTION_CALL MonitoringRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((Monitoring*)arg)->run() ? 0 : 1));
}


bool Monitoring::addCommunication(uint16 fromNodeID, uint16 toNodeID, uint32 from, uint32 to, uint32 size, uint32 count, uint64 time) {
	if (nodeCom->add(fromNodeID, toNodeID, (double)size, count, time))
		return compCom->add(from, to, (double)size, count, time);
	else
		return false;
}



bool Monitoring::gatherPerformanceStats() {

	uint64 now = GetTimeNow();

	std::list<PerfStats> perfStats;
	node->gatherLocalPerformanceStats(perfStats);

	uint32 osNodeProcID = node->osProcID;

	OSProcStats localTotalUsage, localNodeUsage;
	memset(&localTotalUsage, 0, sizeof(OSProcStats));

	std::map<uint32, OSProcStats> spaceIDs;
	std::list<PerfStats>::iterator i = perfStats.begin(), e = perfStats.end();

	uint32 localNodeSpaceID = 0;
	MonitoringStats* stats;
	// first calculate total ticks
	uint64 processTicks = 0, totalTicks = 0;
	while (i != e) {
		if (i->compID) {
			// This is a component
			processTicks += i->currentCPUTicks;
		}
		else if (i->spaceID) {
			// This is a space
			if (i->osID == osNodeProcID) {
				// This is a local internal space
				localNodeSpaceID = i->spaceID;
			}
			else {
				if (!(stats = localSpaceStats[i->spaceID]))
					localSpaceStats[i->spaceID] = stats = new MonitoringStats();
				spaceIDs[i->spaceID].procMem = i->currentMemoryBytes;
				spaceIDs[i->spaceID].difTicks = i->currentCPUTicks - stats->lastStats.currentCPUTicks;
				totalTicks += i->currentCPUTicks;
				// This is a space in another local process
			}
		}
		else if (i->nodeID) {
			// This is the local node
			localNodeUsage.procMem = i->currentMemoryBytes;
			localNodeUsage.difTicks = i->currentCPUTicks - localNodeStats.lastStats.currentCPUTicks;
			totalTicks += i->currentCPUTicks;
		}
		i++;
	}

	// Now we can calculate percentages...
	if (utils::GetProcAndOSCPUUsage(osNodeProcID, localNodeUsage.procCPU, localNodeUsage.osCPU, prevOSTicks, prevOSIdleTicks, prevProcTicks)) {
		localTotalUsage.add(localNodeUsage);
		computerPercentOfOSCPU.add(localNodeUsage.osCPU, 1, now);
	}

	std::map<uint32, OSProcStats>::iterator iSpace = spaceIDs.begin(), eSpace = spaceIDs.end();
	while (iSpace != eSpace) {
		if (utils::GetProcAndOSCPUUsage(osNodeProcID, iSpace->second.procCPU, iSpace->second.osCPU, prevOSTicks, prevOSIdleTicks, prevProcTicks))
			localTotalUsage.add(iSpace->second);
		iSpace++;
	}

	// Add the added up percentages of procCPU for all local processes
	totalPercentOfOSCPU.add(localTotalUsage.procCPU, 1, now);

	utils::GetSystemMemoryUsage(osMemorySize, osMemoryFree);
	computerPercentOfOSMemory.add((double)osMemoryFree/(double)osMemorySize, 1, now);

	// Add the added up percentages of proc memory for all local processes
	totalPercentOfOSMemory.add((double)localTotalUsage.procMem/(double)osMemorySize, 1, now);

	uint64 c1, c2, c3;
	computerPercentOfOSCPU.getAverageMulti(ms1, ms2, ms3, computerCPUUsage[0], c1, computerCPUUsage[1], c2, computerCPUUsage[2], c3);
	computerPercentOfOSMemory.getAverageMulti(ms1, ms2, ms3, computerMemoryUsage[0], c1, computerMemoryUsage[1], c2, computerMemoryUsage[2], c3);
	totalPercentOfOSCPU.getAverageMulti(ms1, ms2, ms3, localSystemCPUUsage[0], c1, localSystemCPUUsage[1], c2, localSystemCPUUsage[2], c3);
	totalPercentOfOSMemory.getAverageMulti(ms1, ms2, ms3, localSystemMemoryUsage[0], c1, localSystemMemoryUsage[1], c2, localSystemMemoryUsage[2], c3);

	uint32 count = (uint32)clmax(localComponentStats.size() + localSpaceStats.size() + 1, perfStats.size());
	uint32 c = 0;
	AveragePerfStats* structsData = new AveragePerfStats[count];
	i = perfStats.begin();
	e = perfStats.end();
	while (i != e) {
		if (i->compID) {
			if (!(stats = localComponentStats[i->compID]))
				localComponentStats[i->compID] = stats = new MonitoringStats();
			if (i->spaceID == localNodeSpaceID) {
				// running inside node internal space
				stats->addStats(*i, now, osMemorySize, localNodeUsage);
			}
			else {
				// running in a separate space
				stats->addStats(*i, now, osMemorySize, spaceIDs[i->spaceID]);
			}
			stats->processData(ms1, ms2, ms3, now);
			memcpy(&stats->lastPerfStruct.computerCPUUsage, computerCPUUsage, 3*sizeof(double));
			node->manager->componentMemory->setComponentPerfStats(i->compID, stats->lastPerfStruct);
			memcpy(&structsData[c++], &stats->lastPerfStruct, sizeof(AveragePerfStats));
		}
		else if (i->spaceID == localNodeSpaceID) {
			if (!(stats = localSpaceStats[i->spaceID]))
				localSpaceStats[i->spaceID] = stats = new MonitoringStats();
			// This is a local internal space
			stats->addStats(*i, now, osMemorySize, localNodeUsage);
			stats->processData(ms1, ms2, ms3, now);
			memcpy(&stats->lastPerfStruct.computerCPUUsage, computerCPUUsage, 3*sizeof(double));
			node->manager->processMemory->setProcessPerfStats(i->spaceID, stats->lastPerfStruct);
			memcpy(&structsData[c++], &stats->lastPerfStruct, sizeof(AveragePerfStats));
		}
		else if (i->spaceID) {
			if (!(stats = localSpaceStats[i->spaceID]))
				localSpaceStats[i->spaceID] = stats = new MonitoringStats();
			// This is a space in another local process
			stats->addStats(*i, now, osMemorySize, spaceIDs[i->spaceID]);
			stats->processData(ms1, ms2, ms3, now);
			memcpy(&stats->lastPerfStruct.computerCPUUsage, computerCPUUsage, 3*sizeof(double));
			node->manager->processMemory->setProcessPerfStats(i->spaceID, stats->lastPerfStruct);
			memcpy(&structsData[c++], &stats->lastPerfStruct, sizeof(AveragePerfStats));
		}
		else if (i->nodeID) {
			// This is the local node
			localNodeStats.addStats(*i, now, osMemorySize, localNodeUsage);
			localNodeStats.processData(ms1, ms2, ms3, now);
			memcpy(&localNodeStats.lastPerfStruct.computerCPUUsage, computerCPUUsage, 3*sizeof(double));
			memcpy(&localNodeStats.lastPerfStruct.computerMemoryUsage, computerMemoryUsage, 3*sizeof(double));
			//memcpy(&localTotalUsage.procCPU, computerCPUUsage, 3*sizeof(double));
			memcpy(&localNodeStats.lastPerfStruct.localSystemCPUUsage, localSystemCPUUsage, 3*sizeof(double));
			memcpy(&localNodeStats.lastPerfStruct.localSystemMemoryUsage, localSystemMemoryUsage, 3*sizeof(double));
			memcpy(&node->lastPerfStruct, &localNodeStats.lastPerfStruct, sizeof(AveragePerfStats));
			memcpy(&structsData[c++], &localNodeStats.lastPerfStruct, sizeof(AveragePerfStats));
		}
		i++;
	}

	// Send all stats to other nodes
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_PERFSYNC, node->nodeID, 0, 0, now);
	msg->setData("AveragePerfStats", (char*)structsData, c * sizeof(AveragePerfStats));
	node->networking->sendMessageToAllOtherNodes(msg, true);
	delete [] structsData;

	// Send out relevant performance messages about locals, including warnings
	sendOutPerformanceMessages(now);
	return true;
}

bool Monitoring::setPerfDataFromOtherNode(const char* data, uint32 size, uint16 remoteNodeID) {
	AveragePerfStats* structsData = (AveragePerfStats*) data;
	uint32 count = size / sizeof(AveragePerfStats);

	uint32 id;
	AveragePerfStats* perfStruct, *pstruct;
	for (uint32 c=0; c<count; c++) {
		perfStruct = &structsData[c];
		if (perfStruct->lastStats.compID) {
			if (pstruct = otherComponentStats[perfStruct->lastStats.compID])
				delete(otherComponentStats[perfStruct->lastStats.compID]);
			otherComponentStats[perfStruct->lastStats.compID] = new AveragePerfStats(*perfStruct);
		}
		else if (perfStruct->lastStats.spaceID) {
			id = (uint32)perfStruct->lastStats.compID + (((uint32)perfStruct->lastStats.nodeID) << 16);
			if (pstruct = otherSpaceStats[id])
				delete(pstruct);
			otherSpaceStats[id] = new AveragePerfStats(*perfStruct);
		}
		else if (perfStruct->lastStats.nodeID) {
			if (pstruct = otherNodeStats[perfStruct->lastStats.nodeID])
				delete(pstruct);
			otherNodeStats[perfStruct->lastStats.nodeID] = new AveragePerfStats(*perfStruct);
		}
	}

	return true;
}


char* Monitoring::getNodeMatrixBinary(uint32 &size, uint64 now) {
	return nodeCom->getThroughputBinary(size, ms1, ms2, ms3, now);
}

char* Monitoring::getCompMatrixBinary(uint32 &size, uint64 now) {
	return compCom->getThroughputBinary(size, ms1, ms2, ms3, now);
}

std::string Monitoring::getPerformanceMatrixXML() {
	std::string xml;

	uint64 now = GetTimeNow();

	std::map<uint64, PerformanceMatrixResults> localNodeCom = nodeCom->getThroughputResults(ms1, ms2, ms3, now);
	std::map<uint64, PerformanceMatrixResults> localCompCom = compCom->getThroughputResults(ms1, ms2, ms3, now);

	std::map<uint64, PerformanceMatrixResults> allNodeCom;
	std::map<uint64, PerformanceMatrixResults> allCompCom;

	node->getOtherNodeCompCom(allNodeCom, allCompCom);

	// merge them
	allNodeCom.insert(localNodeCom.begin(), localNodeCom.end());
	allCompCom.insert(localCompCom.begin(), localCompCom.end());

	xml += PerformanceMatrix::GetThroughputXMLFromResults("nodecommunication", allNodeCom);
	xml += PerformanceMatrix::GetThroughputXMLFromResults("componentcommunication", allCompCom);

	return utils::StringFormat("<performancematrix ms1=\"%u\" ms2=\"%u\" ms3=\"%u\">\n%s</performancematrix>\n",
		ms1, ms2, ms3, xml.c_str());

}


bool Monitoring::sendOutPerformanceMessages(uint64 now) {

	//MonitoringStats localNodeStats;
	//std::map<uint32, MonitoringStats> localSpaceStats;
	//std::map<uint32, MonitoringStats> localComponentStats;
	//MovingAverage percentOfOSCPU;

	DataMessage* msg = NULL;

	if (computerCPUUsage[1] > 0.95) {
		// > 95% busy over past 10 seconds
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.CPU.Overload"), 0, 0, 0, now);
	}
	else if (computerCPUUsage[1] > 0.95) {
		// > 80% busy over past 10 seconds
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.CPU.Highload"), 0, 0, 0, now);
	}
	else if (computerCPUUsage[0] > 0.9) {
		// > 90% busy over past 1 second
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.CPU.Suddenload"), 0, 0, 0, now);
	}
	if (msg) {
		msg->setInt("NodeID", node->nodeID);
		node->processMessage(msg);
		delete(msg);
		msg = NULL;
	}

	if (((double)osMemoryFree / (double)osMemorySize < 0.03) ||
		(osMemoryFree < 100000000L) ) {
		LogPrint(0, 0, 0, "Memory low: %.3f%% - %llu free of %llu total",
			((double)osMemoryFree / (double)osMemorySize)*100, osMemoryFree, osMemorySize);
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.Memory.Low"), 0, 0, 0, now);
		msg->setInt("NodeID", node->nodeID);
		node->processMessage(msg);
		delete(msg);
		msg = NULL;
	}


	if (localNodeStats.lastPerfStruct.dataQueueCountAverage[1] > 1000) {
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.Queues.Overload"), 0, 0, 0, now);
	}
	else if (localNodeStats.lastPerfStruct.dataQueueCountAverage[1] > 100) {
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.Queues.Highload"), 0, 0, 0, now);
	}
	else if (localNodeStats.lastPerfStruct.dataQueueCountAverage[0] > 100) {
		msg = new DataMessage(node->getPsyType("Psyclone.Performance.Warning.Queues.Suddenload"), 0, 0, 0, now);
	}
	if (msg) {
		msg->setInt("NodeID", node->nodeID);
		node->processMessage(msg);
		delete(msg);
		msg = NULL;
	}

	// Send out regular performance data???
	//msg = new DataMessage(perfType, 0, 0, 0, now);
	//msg->setInt("NodeID", node->nodeID);
	//node->processMessage(msg);
	//delete(msg);
	//msg = NULL;

	return true;
}



}
