

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "PsySpace.h"

namespace cmlabs {


TimeQueueSchedule::TimeQueueSchedule(uint32 interval, uint64 start, uint64 end) {
	this->interval = interval;
	this->start = start;
	this->end = end;
}

TimeQueueSchedule::~TimeQueueSchedule() {
}


TimeQueue::TimeQueue() {
	timer = new utils::Timer();
}

TimeQueue::~TimeQueue() {
	mutex.enter();
	std::map<uint32, TimeQueueSchedule*>::iterator it, itEnd;
	for (it = schedules.begin(), itEnd = schedules.end(); it!=itEnd; it++)
		delete(it->second);
	delete(timer);
	mutex.leave();
}

bool TimeQueue::addSchedule(TimeQueueSchedule* schedule) {
	mutex.enter();
	if (schedules[schedule->id]) {
		mutex.leave();
		return false;
	}
	schedules[schedule->id] = schedule;
	mutex.leave();
	return timer->addTimer(schedule->id, schedule->interval, schedule->start, schedule->end);
}

bool TimeQueue::removeSchedule(uint32 id) {
	mutex.enter();
	TimeQueueSchedule* schedule = schedules[id];
	if (!schedule) {
		mutex.leave();
		return true;
	}
	timer->removeTimer(id);
	schedules.erase(id);
	delete(schedule);
	mutex.leave();
	return true;
}

DataMessage* TimeQueue::waitForNextEvent(uint32 ms) {

	uint32 id;
	uint64 time;
	if (!timer->waitForTimer(ms, id, time))
		return NULL;

	mutex.enter();
	TimeQueueSchedule* schedule = schedules[id];
	if (!schedule) {
		mutex.leave();
		return NULL;
	}

	DataMessage* msg = new DataMessage(schedule->msgType, 0);
	if (schedule->msgTag)
		msg->setTag(schedule->msgTag);
	mutex.leave();
	return msg;
}







PsySpace::PsySpace(const char* name, bool isAdHoc, uint16 procID, bool isLocal) {
	isMaster = true;
	manager = NULL;
	masterCreatedTime = 0;
	timeQ = new TimeQueue();
	this->isAdHoc = isAdHoc;
	if (name)
		this->name = utils::TextTrimQuotes(name);
	else
		this->name = utils::StringFormat("Unnamed_Space_%u", (uint32)utils::RandomValue(MAXVALUINT32));
	this->isLocal = isLocal;
	this->procID = procID;
	finishedShuttingDown = false;

	memset(waitCounters, 0, 3*sizeof(uint32));
	memset(procCounters, 0, 3*sizeof(uint32));

	threadPoolWaitLowerThreshold = 5;
	threadPoolWaitUpperThreshold = 20;
	threadPoolWaitIncrement = 5;
	lastThreadPoolCheck = GetTimeNow();
	threadPoolCheckInterval = 5000;

	internalCranks["Simple"] = Internal_Simple;
	internalCranks["Ping"] = Internal_Ping;
	internalCranks["Pong"] = Internal_Pong;
	internalCranks["Shutdown"] = Internal_Shutdown;
	internalCranks["Print"] = Internal_Print;
	internalCranks["Time"] = Internal_Time;
	internalCranks["SignalPing"] = Internal_SignalPing;
	internalCranks["SignalPong"] = Internal_SignalPong;
	internalCranks["RetrieveTest"] = Internal_RetrieveTest;
	internalCranks["MessageScript"] = Internal_MessageScript;
	internalCranks["QueryTest"] = Internal_QueryTest;
}

PsySpace::~PsySpace() {
	reset();
	if (timeQ) delete(timeQ);
	timeQ = NULL;
}

bool PsySpace::connect(uint16 systemID, bool isMaster, const char* cmdline) {

	// LogPrint(0, LOG_SPACE, 1, "PsySpace '%s' starting up, connecting to local Node %u...", name.c_str(), systemID);

	this->isMaster = isMaster;

	// Connect to MemoryManager
	if (manager) delete(manager);
	manager = new MemoryManager();
	if (!manager->connect(systemID, isMaster)) {
		// LogPrint(0, LOG_SPACE, 0, "PsySpace couldn't connect to local Node %u...", systemID);
		delete(manager);
		manager = NULL;
		return false;
	}
	finishedShuttingDown = false;

	uint16 checkID;
	if (!isAdHoc) {
		if (!manager->processMemory->getProcessID(name.c_str(), checkID)) {
			LogPrint(0, LOG_SPACE, 0, "Could not find process '%s'...", name.c_str());
			return false;
		}
		if (!procID)
			procID = checkID;
		else if (procID != checkID) {
			LogPrint(0, LOG_SPACE, 0, "Process ID '%u' does not match shared memory id '%u'...", procID, checkID);
			return false;
		}
	}
	else {
		if (manager->processMemory->getProcessID(name.c_str(), checkID) && checkID) {
			procID = checkID;
		}
		else {
			if (!manager->processMemory->createNewProcess(name.c_str(), procID)) {
				LogPrint(0, LOG_SPACE, 0, "PsySpace couldn't connect to local Node %u...", systemID);
				return false;
			}
		}
	}

	// Send message on sysQID to inform node that process has started
	DataMessage* msg = new DataMessage(PsyAPI::CTRL_PROCESS_INITIALISE, procID);
	if (!manager->processMemory->addToCmdQ(0, msg)) {
		LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Command Queue...");
		delete(msg);
		return false;
	}
	delete(msg);

	//if (!manager->processMemory->createNewProcess(name.c_str(), procID)) {
	//	LogPrint(0, 0, 0, "Could not create Process Map\n");
	//	return false;
	//}

	if (cmdline)
		manager->processMemory->setProcessCommandLine(procID, cmdline);
	manager->processMemory->setProcessStatus(procID, PSYPROC_INIT);

	// Connect to Node
	// Send message on sysQID to inform node that process has started
	msg = new DataMessage(PsyAPI::CTRL_PROCESS_GREETING, procID);
	if (!manager->processMemory->addToCmdQ(0, msg)) {
		LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Command Queue...");
		delete(msg);
		return false;
	}
	delete(msg);

	if (isMaster) {
		// Logging will be done via LogPrint directly to the Node as we are in the same process
	}
	else {
		//LogSystem::SetLogReceiver(this);
	}

	uint64 lastSeen;
	manager->getNodeStatus(lastSeen, masterCreatedTime);

	// Setup Heartbeat
	manager->processMemory->setProcessStatus(procID, PSYPROC_READY);

	// Start SpaceThread
	if (!ThreadManager::CreateThread(PsySpaceRun, this, threadID, 0)) {
		LogPrint(procID, LOG_SPACE, 0, "Could not start PsySpace thread...");
		return false;
	}

	return true;
}

bool PsySpace::isConnected(uint32 timeoutMS) {
	if (!manager) return false;

	uint64 lastSeen, createTime;
	uint8 status = manager->getNodeStatus(lastSeen, createTime);
	if (masterCreatedTime && (createTime != masterCreatedTime)) {
		LogPrint(procID, LOG_SPACE, 0, "Psyclone Node process appears to have restarted...");
		reset();
		return false;
	}
	if (status != PSYCLONE_STATUS_READY) {
		LogPrint(procID, LOG_SPACE, 0, "Psyclone Node process doesn't appear to be running...");
		reset();
		return false;
	}
	if (GetTimeAgeMS(lastSeen) > (int32)timeoutMS) {
		LogPrint(procID, LOG_SPACE, 0, "Psyclone Node process appears to have stopped, resetting connection...");
		reset();
		return false;
	}
	return true;
}

bool PsySpace::reset() {
	if (!isMaster)
		LogSystem::SetLogReceiver(NULL);
	if (manager)
		shutdown();
	if (manager) delete(manager);
	manager = NULL;

	std::map<std::string, utils::Library*>::iterator it = libraries.begin(), itEnd = libraries.end();
	while (it != itEnd)
		delete((it++)->second);
	libraries.clear();

	std::map<uint32, PsyAPI*>::iterator ait = psyAPIs.begin(), aitEnd = psyAPIs.end();
	while (ait != aitEnd)
		delete((ait++)->second);
	psyAPIs.clear();

	std::map<PsyType, SignalStruct*>::iterator sit = signalList.begin(), sitEnd = signalList.end();
	while (sit != sitEnd) {
		delete(sit->second->lastSignalMsg);
		delete((sit++)->second);
	}
	signalList.clear();
	return true;
}

bool PsySpace::start(uint16 threadCount) {
	// Setup Threading Pool
	LogPrint(procID, LOG_SPACE, 1, "PsySpace '%s' (%u) starting up...", name.c_str(), procID);
	return setThreadPoolSize(threadCount);
}

bool PsySpace::shutdown() {
	if (finishedShuttingDown)
		return true;
	finishedShuttingDown = true;
	stop();

	//LogPrint(procID, LOG_SPACE, 1, "Space ID %u shutting down...", procID);
	LogPrint(procID, LOG_SPACE, 1, "PsySpace '%s' (%u) shutting down...", name.c_str(), procID);

	DataMessage* msg = new DataMessage(PsyAPI::CTRL_PROCESS_SHUTDOWN, procID);
	if (!manager->processMemory->addToCmdQ(0, msg)) {
		LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Command Queue...");
		delete(msg);
		return false;
	}
	delete(msg);

	// shutdown all components
	std::map<uint32, PsyAPI*>::iterator it, itEnd;
	for (it=psyAPIs.begin(), itEnd=psyAPIs.end(); it != itEnd; it++)
		it->second->stop();

	uint32 maxWaitCount = 20;
	while (threadPool.size()) {
		utils::Sleep(50);
		if (!(--maxWaitCount)) {
			LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' waited for the last %u threads to terminate, killing threads now...", name.c_str(), threadPool.size());
			std::map<uint32, uint8>::iterator threadI = threadPool.begin(), threadE = threadPool.end();
			while (threadI != threadE) {
				if (!ThreadManager::TerminateThread(threadI->first))
					LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' couldn't kill thread %u...", name.c_str(), threadI->first);
				else
					LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' killed thread %u...", name.c_str(), threadI->first);
				threadI++;
			}
			threadPool.clear();
			break;
		}
	}
	while (isRunning) {
		LogPrint(procID, LOG_SPACE,0,"PsySpace '%s' waiting for the main thread to terminate...", name.c_str());
		utils::Sleep(50);
	}

	manager->processMemory->setProcessStatus(procID, PSYPROC_TERMINATED);
	LogPrint(procID, LOG_SPACE, 0, "PsySpace '%s' has shut down (%u, %p)", name.c_str(), procID, this);
	return true;
}

bool PsySpace::hasShutdown() {
	return finishedShuttingDown;
}

// ***************** PsyProbe plugins *****************
bool PsySpace::addPsyProbeCustomView(uint32 compID, const char* name, const char* templateURL) {
	DataMessage* msg = new DataMessage(PsyAPI::CTRL_CREATECUSTOMPAGE, compID);
	msg->setString("Name", name);
	msg->setString("Template", templateURL);
	manager->processMemory->addToCmdQ(0, msg);
	delete(msg);
	return true;
}


bool PsySpace::logEntry(LogEntry* entry) {
	if (!entry)
		return false;

	DataMessage* msg = new DataMessage(CTRL_LOGPRINT, procID);
	msg->setData("LogEntry", (char*)entry, entry->size);
	manager->processMemory->addToCmdQ(0, msg);
	delete(msg);
	delete(entry);
	return true;
}

uint16 PsySpace::getID() {
	return procID;
}

bool PsySpace::setThreadPoolSize(uint16 threadCount) {

	uint32 id;
	int32 n;
	int32 c = threadCount - (uint32)threadPool.size();
	threadTarget = threadCount;
//	LogPrint(procID, LOG_SPACE,0,"Setting PsySpace thread pool size to %u...", threadTarget);
	lastThreadPoolCheck = GetTimeNow();
	if (c > 0) {
		LogPrint(procID, LOG_SPACE,2,"Creating %u more Pool Threads [%u -> %u]...",
			c, (uint32)threadPool.size(), threadCount);
		// create c more threads
		for (n=0; n<c; n++) {
			if (ThreadManager::CreateThread(PsySpacePoolRun, this, id))
				threadPool[id] = 1;
			else
				return false;
		}
	}
	return true;
}

PsyAPI* PsySpace::getCrankAPI(const char* name) {
	PsyAPI* api;
	uint16 crankID;
	if (!manager->dataMapsMemory->getCrankID(name, crankID) || !crankID)
		return NULL;

	uint32 compID = manager->dataMapsMemory->getCrankCompID(crankID);
	if (!compID)
		return NULL;

	uint16 oldCompNodeID = manager->componentMemory->getComponentNodeID(compID);
	if (!oldCompNodeID)
		return NULL;

	if (!pullRemoteComponentData(compID, oldCompNodeID))
		return NULL;

	if (!threadPoolMutex.enter(1000, __FUNCTION__))
		return NULL;
	if (!(api = psyAPIs[crankID])) {
		api = new PsyAPI(this);
		api->setCommandlineBasedir(utils::GetCommandLinePath().c_str());
		api->currentCompID = compID;
		psyAPIs[crankID] = api;
	}
	threadPoolMutex.leave();
	return api;
}

bool PsySpace::registerCrankCallback(const char* name, CrankFunction func) {
	uint16 crankID;
	if (!manager->dataMapsMemory->getCrankID(name, crankID) || !crankID)
		return false;
	if (!threadPoolMutex.enter(1000, __FUNCTION__))
		return false;
	cranks[crankID] = func;
	threadPoolMutex.leave();
	return true;
}

uint32 PsySpace::getComponentID(const char* name) {
	uint32 compID;
	if (manager->componentMemory->getComponentID(name, compID))
		return compID;
	else
		return 0;
}

bool PsySpace::pullRemoteComponentData(uint32 compID, uint16 fromNodeID) {
	DataMessage* msg = new DataMessage(PsyAPI::CTRL_PULLCOMPONENTDATA, this->procID);
	msg->setInt("CompID", compID);
	msg->setInt("ProcID", this->procID);
	msg->setInt("NodeID", fromNodeID);
	DataMessage* resultMsg = NULL;
	uint8 result = query(msg, &resultMsg, 5000);
	if (resultMsg)
		delete(resultMsg);
	delete(msg);
	return (result >= QUERY_SUCCESS);
}


//PsyAPI* PsySpace::getComponentAPI(const char* name) {
//	return getComponentAPI(getComponentID(name));
//}

//PsyAPI* PsySpace::getComponentAPI(uint32 compID) {
//	if (!compID)
//		return false;
//}

//bool PsySpace::registerComponentCallback(const char* name, CrankFunction func) {
//	return registerComponentCallback(getComponentID(name), func);
//}

//bool PsySpace::registerComponentCallback(uint32 compID, CrankFunction func) {
//	if (!compID)
//		return false;
//}


bool PsySpace::postMessage(DataMessage* msg) {
	// Add datamessage to memory
	uint64 memID = 0;
	uint64 eol = msg->getEOL();
	if (eol && (eol < msg->getCreatedTime() - 10000))
		return false;
	if (eol) {
		if (!manager->insertMessage(msg, memID)) {
			LogPrint(procID, LOG_SPACE, 0, "Could not add message from %u to shared memory...", msg->getFrom());
			return false;
		}
		msg->data->memid = memID;
	}
	if (!manager->processMemory->addToMsgQ(0, msg)) {
		LogPrint(procID, LOG_SPACE, 0, "Could not communicate with System Message Queue...");
		return false;
	}
	manager->processMemory->addToProcessStats(procID, NULL, msg);
//	printf("************ Space Post Msg %lld us old...\n", GetTimeAge(msg->getCreatedTime()));
	return true;
}


// ***************** Signals *****************
bool PsySpace::emitSignal(const PsyType& type, DataMessage* msg) {
	// Send signal to local signal struct first
	if (!signalsMutex.enter(3000, __FUNCTION__))
		return false;
	SignalStruct* signal = signalList[type];
	if (!signal) {
		signal = new SignalStruct;
		signal->lastSignalMsg = NULL;
		signalList[type] = signal;
	}
	signalsMutex.leave();

	if (signal->mutex.enter(3000, __FUNCTION__)) {
		if (signal->lastSignalMsg)
			delete(signal->lastSignalMsg);
		signal->lastSignalMsg = msg;
		signal->mutex.leave();
		signal->event.signal();
	}
	else {
		return false;
	}
	// Then to all other procs and the other nodes (via Node proc 0)
//	if (!manager->processMemory->addToAllSignalQs(msg)) {
	if (!manager->processMemory->addToAllSignalQsExcept(msg, procID)) {
		LogPrint(procID, LOG_SPACE, 0, "Could not communicate with All Signal Queues...");
		return false;
	}
	manager->processMemory->addToProcessStats(procID, NULL, msg);
	return true;
}

DataMessage* PsySpace::waitForSignal(const PsyType& type, uint32 timeout, uint64 lastReceivedTime) {
	
	if (!signalsMutex.enter(3000, __FUNCTION__))
		return NULL;
	SignalStruct* signal = signalList[type];
	if (!signal) {
		signal = new SignalStruct;
		signal->lastSignalMsg = NULL;
		signalList[type] = signal;
	}
	signalsMutex.leave();

	DataMessage* msg = NULL;

	if (lastReceivedTime) {
		if (signal->mutex.enter(3000, __FUNCTION__)) {
			if (signal->lastSignalMsg && (signal->lastSignalMsg->getCreatedTime() > lastReceivedTime)) {
				msg = new DataMessage((char*)signal->lastSignalMsg->data, true);
				// printf("************ Space NoWait Signal %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
				signal->mutex.leave();
				//LogPrint(procID, LOG_SPACE, 0, "Space signal no wait success %s...", PrintTimeString(msg->getCreatedTime()).c_str());
				manager->processMemory->addToProcessStats(procID, msg, NULL);
				return msg;
			}
			else {
				//LogPrint(procID, LOG_SPACE, 0, "Not newer %s >= %s", PrintTimeString(lastReceivedTime).c_str(),
				//	PrintTimeString(signal->lastSignalMsg->getCreatedTime()).c_str());
				//if (lastReceivedTime > signal->lastSignalMsg->getCreatedTime())
				//	LogPrint(procID, LOG_SPACE, 0, "%llu > %llu", lastReceivedTime, signal->lastSignalMsg->getCreatedTime());
				//else if (lastReceivedTime < signal->lastSignalMsg->getCreatedTime())
				//	LogPrint(procID, LOG_SPACE, 0, "%llu < %llu", lastReceivedTime, signal->lastSignalMsg->getCreatedTime());
				signal->mutex.leave();
			}
		}
	}

	//if (signal->lastSignalMsg)
	//	LogPrint(procID, LOG_SPACE, 0, "Space signal wait (%s)...", PrintTimeString(signal->lastSignalMsg->getCreatedTime()).c_str());
	//else
	//	LogPrint(procID, LOG_SPACE, 0, "Space signal wait...");

	if (signal->event.waitNext(timeout)) {
		//LogPrint(procID, LOG_SPACE, 0, "Space signal wait success...");
		if (signal->mutex.enter(3000, __FUNCTION__) && signal->lastSignalMsg) {
			if (lastReceivedTime) {
				if (signal->lastSignalMsg->getCreatedTime() > lastReceivedTime)
					msg = new DataMessage((char*)signal->lastSignalMsg->data, true);
			}
			else
				msg = new DataMessage((char*)signal->lastSignalMsg->data, true);
			//if (msg)
				//printf("************ Space Wait Signal %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
			//	LogPrint(procID, LOG_SPACE, 0, "Space signal wait success %s...", PrintTimeString(msg->getCreatedTime()).c_str());
		}
		signal->mutex.leave();
	}

	if (msg)
		manager->processMemory->addToProcessStats(procID, msg, NULL);
	return msg;
}


uint8 PsySpace::query(DataMessage* msg, DataMessage** result, uint32 timeout) {
	*result = NULL;
	if (!msg)
		return QUERY_FAILED;

	uint32 to = msg->getTo();

	uint32 reqID;
	if (!manager->dataMapsMemory->createNewRequest(msg->getFrom(), to, 0, reqID))
		return QUERY_FAILED;
//	LogPrint(procID, LOG_SPACE,0,"Created Request id %u from comp %u...", reqID, msg->getFrom());

	msg->setReference(reqID);
	manager->processMemory->addToProcessStats(procID, NULL, msg);

	uint16 otherNodeID = 0;
	uint16 otherProcID = 0;
	// If to is 0 this request is for the local Node
	if (!to) {
		manager->dataMapsMemory->setRequestStatus(reqID, REQ_PROCESSING_LOCAL);
		if (!manager->processMemory->addToReqQ(0, msg)) {
			manager->dataMapsMemory->deleteRequest(reqID);
			return QUERY_FAILED;
		}
	}
	else {
		if (!manager->componentMemory->getComponentLocation(to, otherNodeID, otherProcID))
			return QUERY_FAILED;

		if (otherNodeID != manager->getNodeID()) {
			manager->dataMapsMemory->setRequestStatus(reqID, REQ_PROCESSING_REMOTE);
			if (!manager->processMemory->addToReqQ(0, msg)) {
				manager->dataMapsMemory->deleteRequest(reqID);
				return QUERY_FAILED;
			}
		}
		else {
			manager->dataMapsMemory->setRequestStatus(reqID, REQ_PROCESSING_LOCAL);
			otherProcID = manager->componentMemory->getComponentProcessID(to);
			if (!manager->processMemory->addToReqQ(otherProcID, msg)) {
				manager->dataMapsMemory->deleteRequest(reqID);
				return QUERY_FAILED;
			}
		}
	}

	uint8 status = 0;
	uint64 time = 0;
	uint64 msgID = 0;
	uint64 msgEOL = 0;
	if (!manager->dataMapsMemory->waitForRequestReply(reqID, timeout, status, time, msgID, msgEOL)) {
		LogPrint(procID, LOG_SPACE,0,"Request reply timeout (%u) from node %u (req %u from comp %u)...",
			status, otherNodeID, reqID, msg->getFrom());
		manager->dataMapsMemory->deleteRequest(reqID);
		return QUERY_TIMEOUT;
	}
	manager->dataMapsMemory->deleteRequest(reqID);

	if (status < REQ_SUCCESS) {
		LogPrint(procID, LOG_SPACE,0,"Request reply failed (%u) from node %u (req %u from comp %u)...", status, otherNodeID, reqID, msg->getFrom());
		return QUERY_FAILED;
	}

	if (!msgID)
		return QUERY_SUCCESS;

	if (!(*result = manager->temporalMemory->getCopyOfMessage(msgID))) {
		LogPrint(procID, LOG_SPACE,0,"Couldn't getCopyOfMessage reply from node %u (req %u from comp %u)...", otherNodeID, reqID, msg->getFrom());
		return QUERY_FAILED;
	}

	if (*result)
		manager->processMemory->addToProcessStats(procID, *result, NULL);
	return QUERY_SUCCESS;
}

bool PsySpace::queryReply(uint32 id, uint8 status, DataMessage* resultMsg) {

	if (resultMsg)
		manager->processMemory->addToProcessStats(procID, resultMsg, NULL);
	// Check if this request was made locally
	uint32 remoteID;
	uint32 origin;
	uint32 source;
	if (!manager->dataMapsMemory->getRequestInfo(id, origin, source, remoteID)) {
		LogPrint(procID, LOG_SPACE,0,"Couldn't find request info for local id %u", id);
		return false;
	}
	// could this be a INTERSYSTEM_QUERY_REPLY
	if (remoteID && !origin) {
		DataMessage* netMsg = new DataMessage(PsyAPI::CTRL_INTERSYSTEM_QUERY_REPLY, source, origin, 10000);
		// keep the local request id here, needed by the InterSystemManager
		netMsg->setReference(id);
		netMsg->setStatus(status);
		if (resultMsg)
			netMsg->setAttachedMessage("DataMessage", resultMsg);
		manager->processMemory->addToReqQ(0, netMsg);
		//LogPrint(procID, LOG_SPACE,3,"Replied to remote request %u (local %u) (source %u, origin %u)...", remoteID, id, source, origin);
		delete(resultMsg);
		delete(netMsg);
		// dont delete this: manager->dataMapsMemory->deleteRequest(id);
		// needed by the InterSystemManager
		return true;
	}
	else if (remoteID) {
		uint16 otherNodeID = manager->componentMemory->getComponentNodeID(origin);
		if (otherNodeID && (otherNodeID != manager->getNodeID())) {
			DataMessage* netMsg = new DataMessage(PsyAPI::CTRL_QUERY_REPLY, source, origin, 10000);
			netMsg->setReference(remoteID);
			netMsg->setStatus(status);
			if (resultMsg)
				netMsg->setAttachedMessage("DataMessage", resultMsg);
			manager->processMemory->addToReqQ(0, netMsg);
			//LogPrint(procID, LOG_SPACE,3,"Replied to remote request %u (local %u) (source %u, origin %u)...", remoteID, id, source, origin);
			delete(resultMsg);
			delete(netMsg);
			return manager->dataMapsMemory->deleteRequest(id);
		}
		else {
			LogPrint(procID, LOG_SPACE,0,"Couldn't find otherNode %u for remote request %u (local %u) (source %u, origin %u)...",
				otherNodeID, remoteID, id, source, origin);
			return false;
		}
	}

	uint64 msgID = 0;
	uint64 msgEOL = 0;
	uint8 reqStatus = 0;

	if (resultMsg) {
		resultMsg->setTTL(10000000);
		msgEOL = resultMsg->getEOL();
		// decide on EOL ###################
		manager->temporalMemory->insertMessage(resultMsg, msgID);
		delete(resultMsg);
	}

	if (status == QUERY_SUCCESS) {
		if (msgID)
			reqStatus = REQ_SUCCESS_DATA_EOL;
		else
			reqStatus = REQ_SUCCESS;
	}
	else {
		if (msgID)
			reqStatus = REQ_FAILED_DATA_EOL;
		else
			reqStatus = REQ_FAILED;
	}
	return manager->dataMapsMemory->setRequestStatus(id, reqStatus, msgID, msgEOL);
}











bool PsySpace::threadPoolDispatch() {
	DataMessage* msg, *dmsg;
	uint32 threadID;
	ThreadManager::GetLocalThreadID(threadID);
	uint32 idleCount = 0;
	uint8 proc;
	char *crankName = new char[MAXVALUENAMELEN+1];
	crankName[0] = 0;
	char *crankFunction = new char[MAXVALUENAMELEN + 1];
	crankFunction[0] = 0;
	char *crankLanguage = new char[MAXVALUENAMELEN + 1];
	crankLanguage[0] = 0;
	char *libraryFilename = new char[MAXVALUENAMELEN + 1];
	libraryFilename[0] = 0;
	CrankFunction crank;
	uint32 compID;
	PsyAPI* api;
	uint32 s;
	TriggerSpec* trigger;

	while (shouldContinue) {
//		printf(".");
		threadPoolMutex.enter();

//		if (GetTimeAgeMS(lastThreadPoolCheck) > threadPoolCheckInterval) {
			idleCount = waitCounters[PROCMESSAGE] + waitCounters[PROCSIGNAL] + waitCounters[PROCREQUEST];
		//	idleCount = waitCounters[PROCMESSAGE] + waitCounters[PROCSIGNAL];
			if (idleCount < threadPoolWaitLowerThreshold)
				setThreadPoolSize((uint16)(threadPool.size() + threadPoolWaitIncrement));
		//	else if (idleCount > threadPoolWaitLowerThreshold)
		//		setThreadPoolSize(threadPool.size() - threadPoolWaitIncrement);
			else
				lastThreadPoolCheck = GetTimeNow();
//		}

		if (threadTarget < threadPool.size()) {
			threadPoolMutex.leave();
			break;
		}

	//	uint64 t = GetTimeNow();
		if ( (waitCounters[PROCMESSAGE] <= waitCounters[PROCSIGNAL]) && (waitCounters[PROCMESSAGE] <= waitCounters[PROCREQUEST]) ) {
	//	if (waitCounters[PROCMESSAGE] <= waitCounters[PROCSIGNAL]) {
			proc = PROCMESSAGE;
			waitCounters[proc]++;
			threadPoolMutex.leave();
			msg = manager->processMemory->waitForMsgQ(procID, 100);
		//	if (msg)
		//		printf("Wait for Msg counter: %llu [%u / %u]\n", GetTimeAge(msg->getCreatedTime()), waitCounters[PROCMESSAGE], waitCounters[PROCSIGNAL]);
		//		printf("************ Space waited %llu...\n", GetTimeAge(t));
		}
		else if ( (waitCounters[PROCREQUEST] <= waitCounters[PROCMESSAGE]) && (waitCounters[PROCREQUEST] <= waitCounters[PROCSIGNAL]) ) {
			proc = PROCREQUEST;
			waitCounters[proc]++;
			threadPoolMutex.leave();
			msg = manager->processMemory->waitForReqQ(procID, 100);
		}
		else {
			proc = PROCSIGNAL;
			waitCounters[proc]++;
			threadPoolMutex.leave();
			if (msg = manager->processMemory->waitForSigQ(procID, 100)) {
				manager->processMemory->addToProcessStats(procID, msg, NULL);
				// printf("Wait for Signal counter: %u [%u / %u]\n", GetTimeAge(msg->getCreatedTime()), waitCounters[PROCMESSAGE], waitCounters[PROCSIGNAL]);
				if (signalsMutex.enter(3000, __FUNCTION__)) {
					SignalStruct* signal = signalList[msg->getType()];
					if (!signal) {
						signal = new SignalStruct;
						signal->lastSignalMsg = NULL;
						signalList[msg->getType()] = signal;
					}
					signalsMutex.leave();
					if (signal->mutex.enter(3000, __FUNCTION__)) {
						if (signal->lastSignalMsg)
							delete(signal->lastSignalMsg);
						signal->lastSignalMsg = msg;
						signal->mutex.leave();
						signal->event.signal();
						LogPrint(procID, LOG_SPACE, 5, "Space signal signal %s...", PrintTimeString(msg->getCreatedTime()).c_str());
					}
					else
						delete(msg);
				}
				else
					delete(msg);
				msg = NULL;
			}
		}
		//else {
		//	proc = PROCTIMER;
		//	waitCounters[proc]++;
		//	threadPoolMutex.leave();
		//	msg = timeQ->waitForNextEvent(30);
		//}
		threadPoolMutex.enter();
		waitCounters[proc]--;
		procCounters[proc]++;

		uint16 crankID;
		char* compName;
		if (msg) {
			crank = NULL;
			crankID = 0;
			dmsg = NULL;

//			printf("************ Space Deliver to %u (%lld us old)     %s...\n",
//				msg->getTo(), GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
			if (!(compID = msg->getTo())) {
				delete(msg);
				procCounters[proc]--;
				threadPoolMutex.leave();
				continue;
			}

			manager->processMemory->addToProcessStats(procID, msg, NULL);

			if (msg->getType() == PsyAPI::CTRL_TRIGGER) {
				if (!(trigger = (TriggerSpec*) msg->getData("TriggerSpec", s))) {
					delete(msg);
					procCounters[proc]--;
					threadPoolMutex.leave();
					continue;
				}
				crankID = trigger->crankID;
				if (!crankID) {
					//crank = internalCranks["Simple"];
					LogPrint(procID, LOG_SPACE, 0, "Crank not found for component!");
					delete (msg);
					procCounters[proc]--;
					threadPoolMutex.leave();
					continue;
				}

				// Get actual data message
				if (!(dmsg = msg->getAttachedMessageCopy("Message"))) {
					LogPrint(procID, LOG_SPACE, 0, "Couldn't find attached input message");
					delete (msg);
					procCounters[proc]--;
					threadPoolMutex.leave();
					continue;
				}
			}
			else {
				if (!(crankID = compCrankIDs[compID])) {
					compName = new char[MAXKEYNAMELEN+1];
					if (manager->componentMemory->getComponentName(compID, compName, MAXKEYNAMELEN)) {
						utils::strcpyavail(compName+strlen(compName), "ComponentCrank", 20, true);
						if (manager->dataMapsMemory->getCrankID(compName, crankID)) {
							compCrankIDs[compID] = crankID;
						}
					}
					else
						compName[0] = 0;
					if (!crankID) {
						LogPrint(procID, LOG_SPACE, 0, "Unknown Component Crank for '%s' (%u)", compName, compID);
						delete [] compName;
						delete (msg);
						procCounters[proc]--;
						threadPoolMutex.leave();
						continue;
					}
					delete [] compName;
				}
			}

			// Load function from already loaded cranks
			if (!crank && !(crank = cranks[crankID])) {
				manager->dataMapsMemory->getCrankName(crankID, crankName, 256);
				// Check if this is a scripted module
				if (manager->dataMapsMemory->getCrankLanguage(crankID, crankLanguage, MAXVALUENAMELEN) && crankLanguage && strlen(crankLanguage)) {
					utils::StringFormatInto(crankFunction, MAXVALUENAMELEN, "%sLink", utils::TextCapitalise(crankLanguage).c_str());
					utils::StringFormatInto(libraryFilename, MAXVALUENAMELEN, "Psy%sLink", utils::TextCapitalise(crankLanguage).c_str());
					LogPrint(procID, LOG_SPACE, 0, "Scripted language (%s) crank '%s' loading binary library '%s'", crankLanguage, crankName, libraryFilename);
				}
				else {
					// Load function from library
					if (!manager->dataMapsMemory->getCrankFunction(crankID, crankFunction, MAXVALUENAMELEN) || !crankFunction) {
						LogPrint(procID, LOG_SPACE, 0, "Unknown Crank ID %u in Trigger Message", crankID);
						delete (msg);
						procCounters[proc]--;
						threadPoolMutex.leave();
						continue;
					}
					if (!manager->dataMapsMemory->getCrankLibraryFilename(crankID, libraryFilename, MAXVALUENAMELEN) || !libraryFilename) {
						LogPrint(procID, LOG_SPACE, 0, "Crank '%s' doesn't have a valid library specified", crankFunction);
						delete (msg);
						procCounters[proc]--;
						threadPoolMutex.leave();
						continue;
					}
				}

				// If crankFunction and/or libraryFilename are not yet set, accept
				// as the message will be added to the API queue for when the module
				// gets instantiated

				if (crankFunction && strlen(crankFunction)) {
					if (!(crank = loadCrankFromLibrary(crankFunction, libraryFilename))) {
						if (strlen(crankLanguage)) {
							LogPrint(procID, LOG_SPACE, 0, "The language library '%s' is unavailable - the language name '%s' may be incorrect", libraryFilename, crankLanguage);
						}
						else
							LogPrint(procID, LOG_SPACE, 0, "Couldn't find crank '%s' in library '%s'", crankFunction, libraryFilename);
						delete (msg);
						procCounters[proc]--;
						threadPoolMutex.leave();
						continue;
					}
					cranks[crankID] = crank;
				}
			}

			//printf("************ Space Trigger %llu us old, Msg %llu us old...\n", GetTimeAge(msg->getCreatedTime()), GetTimeAge(dmsg->getCreatedTime()));

			// Is the crank already running
			if (!(api = psyAPIs[crankID])) {
				api = new PsyAPI(this);
				api->setCommandlineBasedir(utils::GetCommandLinePath().c_str());
				psyAPIs[crankID] = api;
			}

			uint8 status;
			if (dmsg)
				status = api->addInputTrigger(crankID, compID, msg, dmsg);
			else
				status = api->addInputTrigger(crankID, compID, NULL, msg);
			switch(status) {
				case CRANKAPI_RUNNING:
					procCounters[proc]--;
					threadPoolMutex.leave();
					continue;
				case CRANKAPI_IDLE:
				//	threadPoolMutex.leave();
					break;
				case CRANKAPI_FAILED:
				default:
					LogPrint(procID, LOG_SPACE, 0, "Couldn't add trigger to Crank API");
					delete(msg);
					delete(dmsg);
					procCounters[proc]--;
					threadPoolMutex.leave();
					continue;
			}

			threadPoolMutex.leave();
			// Call function
			if (crank) {
				api->begin();
				do {
					try {
						if (crank(api) < 0)
							break;
					}
					catch (...) {
						LogPrint(procID, LOG_SPACE, 0, "Crank %u ('%s') caused an exception", crankID, crankName);
						break;
					}
				} while (api->shouldContinue() && api->getInputQueueSize());
				api->finish();
			}

			threadPoolMutex.enter();
			procCounters[proc]--;
		}
		threadPoolMutex.leave();

	}

	LogPrint(procID, LOG_SPACE,3,"Thread Pool %u exited, target %u, size: %u", threadID, threadTarget, threadPool.size() - 1);
	threadPoolMutex.enter();
	std::map<uint32, uint8>::iterator i = threadPool.find(threadID);
	if (i != threadPool.end())
		threadPool.erase(i);
	threadPoolMutex.leave();
	delete [] crankName;
	delete [] crankFunction;
	delete [] libraryFilename;
	return true;
}

CrankFunction PsySpace::loadCrankFromLibrary(const char* crankName, const char* libraryFilename) {
	if (!crankName || !libraryFilename)
		return NULL;

	if (!strlen(libraryFilename)) {
		// Internal cranks
		return internalCranks[crankName];
	}

	utils::Library* lib = libraries[libraryFilename];
	if (!lib) {
		if (!(lib = utils::OpenLibrary(libraryFilename))) {
			LogPrint(procID, LOG_SPACE, 0, "Could not find or load library '%s'", libraryFilename);
			return NULL;
		}
		libraries[libraryFilename] = lib;
	}

	CrankFunction crank = (CrankFunction)lib->getFunction(crankName);
	if (!crank) {
		LogPrint(procID, LOG_SPACE, 0, "Could not load Crank '%s' from library '%s'", crankName, libraryFilename);
		return NULL;
	}
	return crank;
}


bool PsySpace::startContinuousComponent(uint32 compID) {

	std::map<uint32, uint32>::iterator it = continuousComponentThreads.find(compID);
	if (it != continuousComponentThreads.end())
		return false;

	continuousComponentThreads[threadID] = 0;

	uint32 threadID;
	if (!ThreadManager::CreateThread(PsySpaceContinuousRun, this, threadID))
		return false;

	continuousComponentThreads[threadID] = compID;
	return true;
}

bool PsySpace::runContinuousComponent() {
	uint32 threadID;
	DataMessage* msg = NULL;

	ThreadManager::GetLocalThreadID(threadID);

	uint32 count = 0;
	uint32 compID;
	while (!(compID = continuousComponentThreads[threadID])) {
		if (++count > 10)
			return false;
		utils::Sleep(50);
	}

	// Find function name and library to call
	// Load function from library
	// Check stats
	// Call function
	// Record stats usage

	return true;
}


bool PsySpace::run() {
	// Space Thread
	//  Update Heartbeat
	manager->processMemory->setProcessStatus(procID, PSYPROC_ACTIVE);
	//  Update Space Stats
	uint64 currentCPUTicks;
//	utils::GetProcessCPUTicks(currentCPUTicks);

	DataMessage* cmsg;
	while (shouldContinue) {
		if (cmsg = manager->processMemory->waitForCmdQ(procID, 100)) {
			if (cmsg->getType() == PsyAPI::CTRL_SYSTEM_SHUTDOWN) {
				delete(cmsg);
				shutdown();
				break;
			}
			delete(cmsg);
		}
		//  Update Heartbeat
		utils::GetProcessCPUTicks(currentCPUTicks);
		manager->processMemory->setProcessStatus(procID, PSYPROC_ACTIVE, currentCPUTicks);
	}

	manager->processMemory->setProcessStatus(procID, PSYPROC_SHUTTING_DOWN);
	//LogPrint(procID, LOG_SPACE, 1, "PsySpace '%s' shutting down...", name.c_str());
	isRunning = false;
	return true;
}

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


THREAD_RET THREAD_FUNCTION_CALL PsySpacePoolRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((PsySpace*)arg)->threadPoolDispatch() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL PsySpaceContinuousRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((PsySpace*)arg)->runContinuousComponent() ? 0 : 1));
}


} // namespace cmlabs
