

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE version 2.1.1.260726 (26 July 2026)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2026
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  PSYCLONE 2.1 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.1 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.1 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.1.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.1 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 "Node.h"
#include "InterSystemManager.h"

namespace	cmlabs{

NodeInfoPack::NodeInfoPack(const char* data, bool copy) {
	this->data = NULL;
	if (GetObjID(data) != NODEINFOPACKID)
		return;
	if (copy) {
		this->data = (NodeInfoPackHeader*) malloc(((NodeInfoPackHeader*)data)->size);
		memcpy(this->data, data, ((NodeInfoPackHeader*)data)->size);
	}
	else
		this->data = (NodeInfoPackHeader*) data;
}

NodeInfoPack::~NodeInfoPack() {
	free(data);
	data = NULL;
}

bool NodeInfoPack::isValid() {
	return (data);
}

const NodeInfoPackHeader* NodeInfoPack::getNodeInfo() {
	return data;
}

const ProcessInfoStruct* NodeInfoPack::getProcInfo(uint16 id) {
	ProcessInfoStruct* procInfo = (ProcessInfoStruct*) (((char*)data) + sizeof(NodeInfoPackHeader));
	for (uint32 n=0; n<data->procCount; n++) {
		if (procInfo->id == id)
			return procInfo;
		procInfo++;
	}
	return NULL;
}

const ProcessInfoStruct* NodeInfoPack::getProcInfo(const char* name) {
	ProcessInfoStruct* procInfo = (ProcessInfoStruct*) (((char*)data) + sizeof(NodeInfoPackHeader));
	for (uint32 n=0; n<data->procCount; n++) {
		if (stricmp(procInfo->name, name) == 0)
			return procInfo;
		procInfo++;
	}
	return NULL;
}

const ComponentInfoStruct* NodeInfoPack::getCompInfo(uint16 id) {
	ComponentInfoStruct* compInfo = (ComponentInfoStruct*)
		(((char*)data) + sizeof(NodeInfoPackHeader) + (data->procCount*sizeof(ProcessInfoStruct)));
	for (uint32 n=0; n<data->compCount; n++) {
		if (compInfo->id == id)
			return compInfo;
		compInfo++;
	}
	return NULL;
}

const ComponentInfoStruct* NodeInfoPack::getCompInfo(const char* name) {
	ComponentInfoStruct* compInfo = (ComponentInfoStruct*)
		(((char*)data) + sizeof(NodeInfoPackHeader) + (data->procCount*sizeof(ProcessInfoStruct)));
	for (uint32 n=0; n<data->compCount; n++) {
		if (stricmp(compInfo->name, name) == 0)
			return compInfo;
		compInfo++;
	}
	return NULL;
}

std::map<uint16, std::map<uint32, ComponentInfoStruct*> > NodeInfoPack::getCatalogs() {
	std::map<uint16, std::map<uint32, ComponentInfoStruct*> > map;
	ComponentInfoStruct* compInfo = (ComponentInfoStruct*)
		(((char*)data) + sizeof(NodeInfoPackHeader) + (data->procCount*sizeof(ProcessInfoStruct)));
	for (uint32 n=0; n<data->compCount; n++) {
		if ( (compInfo->type == COMP_CATALOG) || (compInfo->type == COMP_WHITEBOARD) )
			map[compInfo->procID][compInfo->id] = compInfo;
		compInfo++;
	}
	return map;
}

std::map<uint16, std::map<uint32, ComponentInfoStruct*> > NodeInfoPack::getModules() {
	std::map<uint16, std::map<uint32, ComponentInfoStruct*> > map;
	ComponentInfoStruct* compInfo = (ComponentInfoStruct*)
		(((char*)data) + sizeof(NodeInfoPackHeader) + (data->procCount*sizeof(ProcessInfoStruct)));
	for (uint32 n=0; n<data->compCount; n++) {
		if ( (compInfo->type == COMP_INTERNAL) || (compInfo->type == COMP_EXTERNAL) )
			map[compInfo->procID][compInfo->id] = compInfo;
		compInfo++;
	}
	return map;
}

std::map<uint16, std::pair<std::map<uint32, ComponentInfoStruct*>, std::map<uint32, ComponentInfoStruct*> > > NodeInfoPack::getHierarcy() {
	std::map<uint16, std::pair<std::map<uint32, ComponentInfoStruct*>, std::map<uint32, ComponentInfoStruct*> > > map;
	ComponentInfoStruct* compInfo = (ComponentInfoStruct*)
		(((char*)data) + sizeof(NodeInfoPackHeader) + (data->procCount*sizeof(ProcessInfoStruct)));
	for (uint32 n=0; n<data->compCount; n++) {
		if ( (compInfo->type == COMP_CATALOG) || (compInfo->type == COMP_WHITEBOARD) )
			map[compInfo->procID].first[compInfo->id] = compInfo;
		else if ( (compInfo->type == COMP_INTERNAL) || (compInfo->type == COMP_EXTERNAL) )
			map[compInfo->procID].second[compInfo->id] = compInfo;
		compInfo++;
	}
	return map;
}

std::string NodeInfoPack::toXML() {
	std::string xml, moduleXML;

	ProcessInfoStruct* procInfo = (ProcessInfoStruct*) (((char*)data) + sizeof(NodeInfoPackHeader));
	for (uint32 n=0; n<data->procCount; n++) {
		if (procInfo->createdTime)
			xml += procInfo->toXML();
		procInfo++;
	}
	ComponentInfoStruct* compInfo = (ComponentInfoStruct*)
		(((char*)data) + sizeof(NodeInfoPackHeader) + (data->procCount*sizeof(ProcessInfoStruct)));
	for (uint32 n=0; n<data->compCount; n++) {
		if (compInfo->createdTime) {
			if ( (compInfo->type == COMP_INTERNAL) || (compInfo->type == COMP_EXTERNAL) )
				moduleXML += compInfo->toXML();
			else
				xml += compInfo->toXML();
		}
		compInfo++;
	}

	return utils::StringFormat("<node id=\"%u\" name=\"%s\" address=\"%s\" port=\"%u\" group=\"%s\" time=\"%llu\" created=\"%llu\">\n%s%s%s</node>\n",
		data->nodeID, data->name, data->addr, data->port, data->group, data->time, data->createdTime, data->perfStruct.toXML().c_str(), xml.c_str(), moduleXML.c_str());
}



MessageTimer* MessageTimer::Singleton = NULL;
bool MessageTimer::DrumBeat(uint32 id, uint32 count) {
	if (!Singleton)
		return false;
	TriggerSpec* spec = Singleton->triggers[id];
	if (!spec)
		return false;

	// Check if context is active
	if (!Singleton->node->subscriptionEngine->isContextActive(spec->context))
		return true;

	DataMessage* msg = new DataMessage(spec->type, 0);
	Singleton->node->deliverTriggerMessage(spec, 0, msg, GetTimeNow());
	delete(msg);

	return true;
}

MessageTimer::MessageTimer(Node* node) {
	this->node = node;
	Singleton = this;
}

MessageTimer::~MessageTimer() {
	Singleton = NULL;
	components.clear();
	std::map<uint32, TriggerSpec*>::iterator it = triggers.begin(), itEnd = triggers.end();
	while (it != itEnd) {
		free(it->second);
		triggers.erase(it++);
	}
}

uint32 MessageTimer::addTrigger(TriggerSpec* spec) {
	if (!spec || !spec->interval)
		return 0;
	uint32 id = 1;
	while (triggers[id])
		id++;

	TriggerSpec* copy = (TriggerSpec*) malloc(spec->size);
	memcpy(copy, spec, spec->size);
	triggers[id] = copy;
	components[spec->componentID].push_back(id);

	if (!utils::CreateDrumBeat(id, spec->interval, DrumBeat, true)) {
		LogPrint(0, LOG_SYSTEM, 0,"Couldn't create MessageTimer for component '%s'...", spec->name);
		return 0;
	}

	return id;
}

bool MessageTimer::removeTrigger(uint32 id) {
	TriggerSpec* spec = triggers[id];
	utils::EndDrumBeat(id);
	triggers[id] = NULL;
	free(spec);
	return true;
}

bool MessageTimer::removeComponent(uint32 compID) {
	std::list<uint32>* ids = &components[compID];
	if (!ids->size())
		return true;
	std::list<uint32>::iterator it = ids->begin(), itEnd = ids->end();
	while (it != itEnd) {
		removeTrigger(*it);
		ids->erase(it++);
	}
	return true;
}


// ============================================================================
// T1.6: JoinEngine — declarative <triggergroup> join operator.
// Accumulates member messages Node-side and activates the crank once per set.
// ============================================================================
JoinEngine* JoinEngine::Singleton = NULL;

JoinEngine::JoinEngine(Node* node) {
	this->node = node;
	this->sweepBeatID = 0;
	this->sweepRunning = false;
	Singleton = this;
}

JoinEngine::~JoinEngine() {
	if (sweepRunning) { utils::EndDrumBeat(sweepBeatID); sweepRunning = false; }
	mutex.enter();
	std::map<uint32, GroupState>::iterator it, itEnd;
	for (it = groups.begin(), itEnd = groups.end(); it != itEnd; it++) {
		std::map<uint32, PartialSet>::iterator pit, pitEnd;
		for (pit = it->second.partials.begin(), pitEnd = it->second.partials.end(); pit != pitEnd; pit++)
			freePartial(pit->second);
		if (it->second.spec) free(it->second.spec);
		if (it->second.memberTemplate) free(it->second.memberTemplate);
	}
	groups.clear();
	mutex.leave();
	Singleton = NULL;
}

bool JoinEngine::SweepBeat(uint32 id, uint32 count) {
	if (Singleton) Singleton->sweep();
	return true;
}

void JoinEngine::stopSweep() {
	if (sweepRunning) {
		Singleton = NULL;				// racing SweepBeat callbacks no-op immediately
		utils::EndDrumBeat(sweepBeatID);
		sweepRunning = false;
		utils::Sleep(80);				// let any in-flight 50ms sweep callback drain before manager teardown
	}
}

void JoinEngine::ensureSweep() {
	// Lazily start a 50ms sweep once the first group is registered.
	if (!sweepRunning) {
		// Use a high, unique DrumBeat id: MessageTimer allocates ids counting up
		// from 1 (shared global DrumBeatMap), so a low id like 1 collides and the
		// two teardowns race on the same timer. Pick an id well above any plausible
		// MessageTimer trigger count.
		sweepBeatID = 0x7F000001;
		if (utils::CreateDrumBeat(sweepBeatID, 50, SweepBeat, true))
			sweepRunning = true;
		else
			LogPrint(node->networking->localNodeID, LOG_NODE, 0, "T1.6: could not create JoinEngine sweep timer...");
	}
}

bool JoinEngine::registerGroup(TriggerGroupSpec* group, uint16 crankID) {
	if (!group) return false;
	mutex.enter();
	GroupState& g = groups[group->id];
	if (g.spec) free(g.spec);
	g.spec = (TriggerGroupSpec*) malloc(group->size);
	memcpy(g.spec, group, group->size);
	g.crankID = crankID;
	g.memberTemplate = NULL;   // T1.6: captured lazily from the first routed member
	ensureSweep();
	mutex.leave();
	LogPrint(node->networking->localNodeID, LOG_NODE, 2, "T1.6: registered join group '%s' (id %u, %u members)...", group->name, group->id, group->memberCount);
	return true;
}

bool JoinEngine::removeComponent(uint32 compID) {
	mutex.enter();
	std::map<uint32, GroupState>::iterator it = groups.begin();
	while (it != groups.end()) {
		if (it->second.spec && it->second.spec->componentID == compID) {
			std::map<uint32, PartialSet>::iterator pit, pitEnd;
			for (pit = it->second.partials.begin(), pitEnd = it->second.partials.end(); pit != pitEnd; pit++)
				freePartial(pit->second);
			if (it->second.spec) free(it->second.spec);
			if (it->second.memberTemplate) free(it->second.memberTemplate);
			groups.erase(it++);
		}
		else it++;
	}
	mutex.leave();
	return true;
}

bool JoinEngine::isJoinGroup(uint32 groupID) {
	if (!groupID) return false;
	mutex.enter();
	bool found = (groups.find(groupID) != groups.end());
	mutex.leave();
	return found;
}

void JoinEngine::freePartial(PartialSet& ps) {
	std::map<std::string, std::list<MemberValue> >::iterator mit, mitEnd;
	for (mit = ps.members.begin(), mitEnd = ps.members.end(); mit != mitEnd; mit++) {
		std::list<MemberValue>::iterator vit, vitEnd;
		for (vit = mit->second.begin(), vitEnd = mit->second.end(); vit != vitEnd; vit++)
			if (vit->msg) delete(vit->msg);
	}
	ps.members.clear();
	ps.order.clear();
}

bool JoinEngine::routeMessage(TriggerSpec* memberSpec, DataMessage* msg, uint64 recvtime) {
	if (!memberSpec || !msg) return false;
	uint32 groupID = memberSpec->group;
	mutex.enter();
	std::map<uint32, GroupState>::iterator git = groups.find(groupID);
	if (git == groups.end()) { mutex.leave(); return false; }
	GroupState& g = git->second;
	TriggerGroupSpec* gs = g.spec;

	// T1.6: capture a clone of a real member TriggerSpec once; it carries the
	// owning module's post specs (packed in its tail), which the synthetic group
	// activation needs so the consuming crank can postOutputMessage().
	if (!g.memberTemplate && memberSpec->size) {
		g.memberTemplate = (TriggerSpec*) malloc(memberSpec->size);
		memcpy(g.memberTemplate, memberSpec, memberSpec->size);
	}

	// Grouping key = message tag id; untagged (0) maps to the "" tag key so it
	// groups on that and never mixes with real tags.
	uint32 tagKey = msg->getTag();
	if (!tagKey) tagKey = node->getTagID("");

	// mode=snapshot: fire on the primary member, attach latest of others, no wait.
	if (gs->mode == TRIGGERGROUP_MODE_SNAPSHOT) {
		GroupMemberSpec* mm = gs->getMemberSpec(memberSpec->name);
		bool isPrimary = (mm && mm->primary);
		PartialSet& ps = g.partials[tagKey];
		// Always record latest value of this member.
		MemberValue mv; mv.msg = new DataMessage(*msg); mv.arrival = recvtime;
		std::list<MemberValue>& slot = ps.members[memberSpec->name];
		for (std::list<MemberValue>::iterator it = slot.begin(); it != slot.end(); it++) if (it->msg) delete(it->msg);
		slot.clear(); slot.push_back(mv);
		if (isPrimary) {
			fire(g, tagKey, ps, false);
			// snapshot keeps other members' latest values; only clear primary marker by leaving them.
		}
		mutex.leave();
		return true;
	}

	PartialSet& ps = g.partials[tagKey];
	if (ps.members.empty()) {
		ps.firstArrival = recvtime; ps.lastFire = 0; ps.fireCount = 0;
		ps.minSend = ps.maxSend = msg->getSendTime();
		ps.minCreated = ps.maxCreated = msg->getCreatedTime();
	}
	else {
		uint64 st = msg->getSendTime(), ct = msg->getCreatedTime();
		if (st < ps.minSend) ps.minSend = st; if (st > ps.maxSend) ps.maxSend = st;
		if (ct < ps.minCreated) ps.minCreated = ct; if (ct > ps.maxCreated) ps.maxCreated = ct;
	}
	// dup handling: latest-wins (default) overwrites; keep queues all.
	std::list<MemberValue>& slot = ps.members[memberSpec->name];
	MemberValue mv; mv.msg = new DataMessage(*msg); mv.arrival = recvtime;
	if (gs->dup == TRIGGERGROUP_DUP_KEEP) {
		slot.push_back(mv);
	}
	else {
		for (std::list<MemberValue>::iterator it = slot.begin(); it != slot.end(); it++) if (it->msg) delete(it->msg);
		slot.clear(); slot.push_back(mv);
	}
	ps.order.push_back(memberSpec->name);

	if (isComplete(g, ps, false)) {
		fire(g, tagKey, ps, false);
		// min= : keep the set open to re-fire as more of the tag-set arrive.
		// slide= : keep a rolling window (re-emit on update while window holds).
		// Otherwise close the set.
		if (!gs->minCount && !gs->slide) {
			freePartial(ps);
			g.partials.erase(tagKey);
		}
	}
	mutex.leave();
	return true;
}

bool JoinEngine::isComplete(GroupState& g, PartialSet& ps, bool timedOut) {
	TriggerGroupSpec* gs = g.spec;
	uint64 now = GetTimeNow();
	// Count present REQUIRED members (respecting per-member maxage), and total present.
	uint32 requiredPresent = 0, requiredTotal = 0, anyPresent = 0;
	for (uint32 i = 0; i < gs->memberCount; i++) {
		GroupMemberSpec* m = gs->getMemberSpec(i);
		if (!m) continue;
		std::map<std::string, std::list<MemberValue> >::iterator mit = ps.members.find(m->name);
		bool present = false;
		if (mit != ps.members.end() && !mit->second.empty()) {
			present = true;
			// per-member maxage: a stale member does not count.
			if (m->maxage) {
				uint64 age = (now - mit->second.back().arrival) / 1000; // us->ms
				if (age > m->maxage) present = false;
			}
		}
		if (present) anyPresent++;
		if (!m->optional) { requiredTotal++; if (present) requiredPresent++; }
	}
	// order=strict: members must have arrived in declared order.
	if (gs->order == TRIGGERGROUP_ORDER_STRICT && !timedOut) {
		uint32 idx = 0;
		for (size_t k = 0; k < ps.order.size() && idx < gs->memberCount; k++) {
			GroupMemberSpec* m = gs->getMemberSpec(idx);
			if (m && ps.order[k] == m->name) idx++;
		}
		if (idx < requiredTotal) return false;
	}
	// within= : post-time spread bound.
	if (gs->within && (ps.maxSend - ps.minSend) / 1000 > gs->within) return false;
	// time= : content-time spread bound (time=0 => identical content time).
	if (gs->hasTimeWindow && (ps.maxCreated - ps.minCreated) / 1000 > gs->timeWindow) return false;
	// count/min policy.
	if (gs->minCount) return (anyPresent >= gs->minCount);
	if (gs->count) return (anyPresent >= gs->count); // any-N-of-M
	return (requiredPresent >= requiredTotal);       // all required present
}

void JoinEngine::fire(GroupState& g, uint32 tagKey, PartialSet& ps, bool partial) {
	TriggerGroupSpec* gs = g.spec;
	uint64 now = GetTimeNow();
	// debounce= : fire at most once per N ms.
	if (gs->debounce && ps.lastFire && (now - ps.lastFire) / 1000 < gs->debounce) return;
	ps.lastFire = now; ps.fireCount++;
	if (partial && gs->ontimeout == TRIGGERGROUP_ONTIMEOUT_FIRE)
		LogPrint(node->networking->localNodeID, LOG_NODE, 1, "T1.6: join group '%s' partial-fire on timeout (tag %u)...", gs->name, tagKey);

	// T1.6 Step 4: emit ONE CTRL_TRIGGER_GROUP carrying the full member map (trigger
	// name -> latest message) so the SDK waitForNewMessageGroup() delivers them together.
	// A primary member message is also attached as "Message" for back-compat.
	std::map<std::string, DataMessage*> memberMap;
	DataMessage* primaryMsg = NULL;
	GroupMemberSpec* primaryMember = NULL;
	for (uint32 i = 0; i < gs->memberCount; i++) {
		GroupMemberSpec* m = gs->getMemberSpec(i);
		if (!m) continue;
		std::map<std::string, std::list<MemberValue> >::iterator mit = ps.members.find(m->name);
		if (mit == ps.members.end() || mit->second.empty()) continue;
		DataMessage* latest = mit->second.back().msg;
		memberMap[m->name] = latest;
		if (m->primary) { primaryMsg = latest; primaryMember = m; }
	}
	if (!primaryMsg && !memberMap.empty()) primaryMsg = memberMap.begin()->second;
	if (memberMap.empty()) return;
	(void)primaryMember;
	node->deliverTriggerGroup(gs->componentID, g.crankID, gs->name, memberMap, primaryMsg, now, g.memberTemplate);
}

void JoinEngine::sweep() {
	if (!node || !node->shouldContinue) return;	// T1.6: never sweep during/after teardown
	uint64 now = GetTimeNow();
	mutex.enter();
	std::map<uint32, GroupState>::iterator git, gitEnd;
	for (git = groups.begin(), gitEnd = groups.end(); git != gitEnd; git++) {
		GroupState& g = git->second;
		TriggerGroupSpec* gs = g.spec;
		if (!gs) continue;
		uint32 evictMs = gs->timeout ? gs->timeout : gs->maxage;
		std::map<uint32, PartialSet>::iterator pit = g.partials.begin();
		while (pit != g.partials.end()) {
			uint64 age = (now - pit->second.firstArrival) / 1000;
			bool expired = (age > evictMs);
			if (expired) {
				if (gs->ontimeout == TRIGGERGROUP_ONTIMEOUT_FIRE && isComplete(g, pit->second, true))
					fire(g, pit->first, pit->second, true);
				else if (gs->ontimeout == TRIGGERGROUP_ONTIMEOUT_FIRE)
					fire(g, pit->first, pit->second, true); // partial-fire whatever is present
				freePartial(pit->second);
				g.partials.erase(pit++);
			}
			else pit++;
		}
	}
	mutex.leave();
}


Node::Node(const char* logdir, const char* procname) {
	logIntersystemTraffic = false;
	createTime = GetTimeNow();
	currentCPUTicks = 0;
	procCount = 0;
	procBytes = 0;
	toCount = 0;
	toBytes = 0;
	deliverCount = 0;
	deliverBytes = 0;
	defaultContext = NOCONTEXT;
	shouldContinue = true;
	shouldRestart = false;
	isNodeRunning = false;
	isMsgQRunning = false;
	isSigQRunning = false;
	isReqQRunning = false;
	msgThreadID = 0;
	sigThreadID = 0;
	reqThreadID = 0;
	configNodeMessage = NULL;
	nodeID = 0;
	osProcID = utils::GetLocalProcessID();
	manager = NULL;
	idManager = new IDManager();
	messageTimer = new MessageTimer(this);
	joinEngine = new JoinEngine(this);	// T1.6
	interSystemManager = NULL;
	networkThreadID = 0;
	networkPort = 0;
	listenSocket = 0;
	networking = NULL;
	psyProbe = NULL;
	monitoring = NULL;
	logging = NULL;
	console = NULL;
	serviceManager = NULL;
	subscriptionEngine = NULL;
	internalSpace = NULL;
	externalExec = NULL;
	if (logdir)
		this->logdir = logdir;
	else
		this->logdir = ".";
	if (procname)
		this->procname = procname;
}

Node::~Node() {
	shouldContinue = false;
	while (isNodeRunning)
		utils::Sleep(10);
	delete(externalExec);	// stops managed external executables (T1.4)
	externalExec = NULL;
	delete(internalSpace);
	internalSpace = NULL;
	delete(idManager);
	idManager = NULL;
	// T1.6: null joinEngine BEFORE stopping/deleting messageTimer. A MessageTimer
	// DrumBeat callback can still fire here and calls deliverTriggerMessage(), which
	// dereferences joinEngine (isJoinGroup). Its own `joinEngine &&` guard then makes
	// a late beat a safe no-op instead of a use-after-free (segfault at offset 8).
	JoinEngine* je = joinEngine;
	joinEngine = NULL;
	if (je) je->stopSweep();					// stop our own sweep beat + drain
	delete(messageTimer);					// stops MessageTimer beats (may have one in flight)
	messageTimer = NULL;
	utils::Sleep(80);						// let any in-flight beat callback finish before we free the engine
	delete(je);								// now safe to free the engine object
	delete(serviceManager);
	serviceManager = NULL;
	delete(subscriptionEngine);
	subscriptionEngine = NULL;
	if (logging)
		logging->shutdown();
	delete(console);
	console = NULL;
	delete(interSystemManager);
	interSystemManager = NULL;
	delete(psyProbe);
	psyProbe = NULL;
	delete(monitoring);
	monitoring = NULL;
	delete(networking);
	networking = NULL;
	delete(logging);
	logging = NULL;
	delete(manager);
	manager = NULL;
}

bool Node::checkPsyclone(const char* addr, uint16 port) {
	uint64 location;
	TCPConnection* con = new TCPConnection();
	if (!con->connect(addr, port, location, NODESYNC_TIMEOUT_MS)) {
		delete(con);
		return false;
	}
	DataMessage* msg = new DataMessage(CTRL_PING, 0);
	if (!MessageProtocol::SendMessage(con, msg)) {
		delete(msg);
		delete(con);
		return false;
	}
	delete(msg);
	if (!(msg = MessageProtocol::ReceiveMessage(con, NODESYNC_TIMEOUT_MS))) {
		delete(con);
		return false;
	}

	PsyType res = msg->getType();
	delete(con);
	delete(msg);
	return (res == CTRL_PING_REPLY);
}


bool Node::initExternal(uint16 port, const char* otherNodeAddress, uint16 otherNodePort) {
	if (!otherNodeAddress) {
		LogPrint(0, LOG_NODE, 0, "No existing node specified...");
		return false;
	}
	else if (!strlen(otherNodeAddress) || otherNodeAddress) {
		LogPrint(0, LOG_NODE, 0, "Error in existing node specification (address: '%s', port: %u)...", otherNodeAddress, otherNodePort);
		return false;
	}

	if (!initialConfig(port))
		return false;

	if (!networking->connectToNodeNetwork(otherNodeAddress, otherNodePort)) {
		LogPrint(0, LOG_NODE, 0, "Couldn't connect to existing node network on %s:%u...", otherNodeAddress, otherNodePort);
		return false;
	}

	LogPrint(networking->localNodeID, LOG_NODE, 1, "Connecting to existing node network on %s:%u...", otherNodeAddress, otherNodePort);

	// Join existing node network
	// Connect with other nodes
	// Accept and buffer all changes
	// Pull and apply Node Dump
	// Apply all changes dates post dump, discard older ones

	return true;
}

// Init with or without config file
bool Node::init(uint16 port, const char* configXML, const char* htmldir) {
	if (!initialConfig(port))
		return false;

	interSystemManager = new InterSystemManager(this);
	interSystemManager->init(logIntersystemTraffic);

	if (htmldir) {
		psyProbe->setRootDir(htmldir);
	}

	if (configXML == NULL) {
		// If we don't have a config file we just
		// wait for a connection from the config Node
		// idManager->init(networking);
		LogPrint(networking->localNodeID, LOG_NODE, 1, "No configuration provided, waiting for Config Node to connect...", networkPort);
		networking->waitForNodeNetwork();
	}
	else {
		LogPrint(networking->localNodeID, LOG_NODE, 1, "Parsing configuration and setting up system...");
		XMLNode	mainNode = parseAndExpandConfig(configXML);
		if (mainNode.isEmpty()){
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Error parsing PsySpec...");
			return false;
		}

		// Check LicenceManager for permissions
		if (!checkRuntimeLicence(mainNode)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Runtime licence check failed...");
			return false;
		}

		if (!networking->setupNodeNetwork(networkPort, mainNode)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Error setting up Node Network...");
			return false;
		}

		// ##############################

		nodeID = networking->localNodeID;
		manager->setNodeID(nodeID);

		//networking->localNodeEntry->status = NODESTATUS_IDSYNC;
		networking->setLocalNodeStatus(NODESTATUS_IDSYNC);

		LogPrint(networking->localNodeID, LOG_NODE, 3, "Synchronising ID Manager...");
		if (!idManager->setup(networking, networking->getNodeCount())) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed synchronising ID Manager...");
			return false;
		}
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Success synchronising ID Manager...");

		LogPrint(networking->localNodeID, LOG_NODE, 0, "Configuring local node...");

		// Configure local node
		if (!configNode(mainNode, true)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed configuring local Node...");
			return false;
		}

		int s = 0;
		XMLSTR xml = mainNode.createXMLString(1, &s);
		DataMessage* msg = new DataMessage(Networking::CTRL_NODE_CONFIG, networking->localNodeID);
		msg->setString("Config", xml);
		origConfigXML = xml;
		free(xml);

		LogPrint(networking->localNodeID, LOG_NODE, 0, "Sending config to all other nodes...");
		networking->sendMessageToAllOtherNodes(msg, true);

		uint64 start = GetTimeNow();
		uint64 timeout = 20000;
		int64 dif;
		bool allReady;
		// Now cycle until all nodes have responded back or until the timeout
		std::map<uint16, NodeEntry*>::iterator it, itEnd;
		while (true) {
			if (networking->getLocalNodeStatus() == NODESTATUS_ERROR) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Error configuring remote nodes...");
				return false;
			}
			dif = GetTimeAgeMS(start);
			if (networking->nodeEntriesMutex.enter()) {
				allReady = true;
				for (it = networking->nodeEntries.begin(), itEnd = networking->nodeEntries.end(); it != itEnd; it++) {
					if ((it->first != this->nodeID) && it->second) {
						if (it->second->status < NODESTATUS_READY) {
							allReady = false;
							if (dif > (int32)timeout) {
								LogPrint(networking->localNodeID, LOG_NODE, 0, "Error configuring node '%s', response timeout...", networking->nodeEntries[it->second->nodeID]->name);
								networking->nodeEntriesMutex.leave();
								return false;
							}
							break;
						}
					}
				}
				networking->nodeEntriesMutex.leave();
				if (allReady)
					break;
			}
			utils::Sleep(20);
		}

		networking->setLocalNodeStatus(NODESTATUS_READY);

		// Send out SystemReady message
		msg = new DataMessage(PsyAPI::CTRL_SYSTEM_READY, networking->localNodeID, 0, HOUR_US);
		networking->sendMessageToAllOtherNodes(msg, true);
		LogPrint(networking->localNodeID, LOG_NODE, 1, "**********************************");
		LogPrint(networking->localNodeID, LOG_NODE, 1, "********** SYSTEM READY **********");
		LogPrint(networking->localNodeID, LOG_NODE, 1, "**********************************");

		psyProbe->setSystemReady();

		LogPrint(nodeID, LOG_NODE, 2, "Current full registration:\n%s\n", subscriptionEngine->printFriendly().c_str());

		LogPrint(nodeID, LOG_NODE, 2, "Component list\n%s...", manager->componentMemory->printFriendly().c_str());

		LogPrint(nodeID, LOG_NODE, 2, "Types list\n%s...", manager->dataMapsMemory->printAllTypes().c_str());

		LogPrint(nodeID, LOG_NODE, 2, "Contexts list\n%s...", manager->dataMapsMemory->printAllContexts().c_str());

		// Snippit for testing speed of subscription matching
		//DataMessage* msg2 = new DataMessage();
		//msg2->setType(this->getPsyType("Psyclone.Ready"));
		//std::list<TriggerSpec*>* triggers;
		//triggers = subscriptionEngine->getTriggersForMsg(msg2);
		//delete triggers;
		//uint32 count = 10000;
		//uint64 now = GetTimeNow();
		//for (int n = 0; n < count; n++) {
		//	triggers = subscriptionEngine->getTriggersForMsg(msg2);
		//	delete triggers;
		//}
		//uint64 after = GetTimeNow();
		//printf("GetTriggers took %s, %.3fus per msg...\n", PrintTimeDifString(after-now).c_str(), (double)(after - now) / count);
		//utils::Sleep(20000);


		msg = new DataMessage(systemReadyType, 0, 0, HOUR_US);

		uint64 memID = 0;
		if (!manager->insertMessage(msg, memID))
			LogPrint(nodeID, LOG_NODE, 0, "Could not add startup message to shared memory...");
		else
			msg->data->memid = memID;
		processMessage(msg);
		delete(msg);

		PsyType type;
		// Now list all nodes as connected
		for (it = networking->nodeEntries.begin(), itEnd = networking->nodeEntries.end(); it != itEnd; it++) {
			if (it->second) {
				type = getPsyType(utils::StringFormat("Psyclone.%s.Connect", it->second->name).c_str());
				if (type.isValid()) {
					msg = new DataMessage();
					msg->setType(type);
					this->processMessage(msg);
					delete(msg);
				}
			}
		}

	}
	return true;
}

bool Node::initialConfig(uint16 port) {
	networkPort = port;

	uint32 slotCount = 100000;
	uint16 binCount = 2;
	uint32 minBlockSize = 1024; // 1kb
	uint32 maxBlockSize = 64 * 1024; // 256kb
	uint64 initSize = 50000000; // 50MB
	uint64 maxSize = 5000000000L; // 5GB

	// Create Logging
	if (logging) delete(logging);
	logging = new Logging();
	if (!logging->init(logdir.c_str())) {
		LogPrint(0, LOG_NODE, 0, "Could not create Logging...");
		return false;
	}

	LogPrint(0, LOG_NODE, 1, "Psyclone Node starting on port %u...", networkPort);

	// Initialise MemoryManager
	if (manager) delete(manager);
	manager = new MemoryManager();
	if (!manager->create(networkPort, slotCount, binCount, minBlockSize, maxBlockSize, initSize, maxSize, false)) {
		// Find out if another Psyclone is running
		if (checkPsyclone("localhost", networkPort)) {
			LogPrint(0, LOG_NODE, 0, "Another instance is already running on port %u...", networkPort);
			return false;
		}
		else {
			LogPrint(0, LOG_NODE, 0, "Couldn't create memory manager on port %u...", networkPort);
			return false;
		}
		//// or is this just dead memory to be forcibly replaced
		//if (!manager->create(networkPort, slotCount, slotSize, slotDuration, true)) {
		//	LogPrint(0, 0, 0, "Node cannot start on port %u...", networkPort);
		//	return false;
		//}
		// LogPrint(0, LOG_SYSTEM, 1, "Cleared old memory on port %u...", networkPort);
	}

	initialiseControlTypes();

	if (subscriptionEngine) delete(subscriptionEngine);
	subscriptionEngine = new SubscriptionEngine(this);
	if (serviceManager) delete(serviceManager);
	serviceManager = new ServiceManager();
	if (!serviceManager->init()) {
		LogPrint(0, LOG_NODE, 0, "Could not create Service Manager...");
		return false;
	}

	// Start up networking
	if (networking) delete(networking);
	networking = new Networking(this);
	if (!networking->init(networkPort, serviceManager)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Error starting local network on port %u...", networkPort);
		return false;
	}

	// Create Console
	if (console) delete(console);
	console = new Console();
	if (!console->init(this, networking->manager, "Welcome to the system")) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create Console...");
		delete(manager);
		manager = NULL;
		return false;
	}
	logging->addLogEntryReceiver(console);

	// Create Monitoring
	if (monitoring) delete(monitoring);
	monitoring = new Monitoring(1000, 10000, 30000);
	if (!monitoring->init(this)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create Monitoring...");
		delete(manager);
		manager = NULL;
		return false;
	}

	// Create PsyProbe
	if (psyProbe) delete(psyProbe);
	psyProbe = new PsyProbe();
	if (!psyProbe->init(this)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create PsyProbe...");
		delete(manager);
		manager = NULL;
		return false;
	}
	
	// Add default services
	uint32 psyProbeID = serviceManager->addService("<service name=\"PsyProbe\" rootdir=\"../html\" />", psyProbe);
	uint32 consoleID = serviceManager->addService("<service name=\"Console\" />", console);

	// Add default interfaces
	if (!networking->addInterface("PsyProbe", networkPort, INTERFACE_PROTO_TCP, NOENC, INTERFACE_HTTP, NODESYNC_TIMEOUT_MS, false, "", "PsyProbe")) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start PsyProbe...");
		delete(manager);
		manager = NULL;
		return false;
	}
	if (!networking->addInterface("Console", networkPort, INTERFACE_PROTO_TCP, NOENC, INTERFACE_TELNET, NODESYNC_TIMEOUT_MS, false, "", "Console")) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start Console...");
		delete(manager);
		manager = NULL;
		return false;
	}

	// start other threads...
	if (!ThreadManager::CreateThread(NodeMessageThread, this, msgThreadID)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start MessageQ thread...");
		delete(manager);
		manager = NULL;
		return false;
	}
	if (!ThreadManager::CreateThread(NodeSignalThread, this, sigThreadID)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start SignalQ thread...");
		delete(manager);
		manager = NULL;
		return false;
	}
	if (!ThreadManager::CreateThread(NodeRequestThread, this, reqThreadID)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start RequestQ thread...");
		delete(manager);
		manager = NULL;
		return false;
	}

	return true;
}

bool Node::setConfigFile(const char* configfile) {
	this->configfile = configfile;
	return true;
}


bool Node::checkRuntimeLicence(XMLNode &mainNode) {
	if (mainNode.isEmpty())
		return true;

	// Check for max number of Nodes
	// Check for xxx
	// #######################

	return true;
}

bool Node::rejoinNode(uint16 nodeID) {

	LogPrint(networking->localNodeID, LOG_NODE, 3, "Resynchronising ID Manager...");
	if (!idManager->resyncAllReservations()) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed resynchronising ID Manager...");
		return false;
	}
	LogPrint(networking->localNodeID, LOG_NODE, 1, "Success resynchronising ID Manager...");

	if (!bulkIDSyncNode(nodeID)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed bulk synchronising IDs...");
		return false;
	}
	LogPrint(networking->localNodeID, LOG_NODE, 1, "Success bulk synchronising IDs...");

	if (!bulkContextSyncNode(nodeID)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed bulk synchronising contexts...");
		return false;
	}
	LogPrint(networking->localNodeID, LOG_NODE, 1, "Success bulk synchronising IDs...");

	if (origConfigXML.length()) {
		int s = 0;
		DataMessage* msg = new DataMessage(Networking::CTRL_NODE_CONFIG, networking->localNodeID);
		msg->setString("Config", origConfigXML.c_str());
		LogPrint(networking->localNodeID, LOG_NODE, 1, "Sending config to all other nodes...");
		networking->sendMessageToNode(nodeID, msg, true);
	}

	uint64 start = GetTimeNow();
	uint64 timeout = 20000;
	int64 dif;
	// Now cycle until rejoining nodes have responded back or until the timeout
	std::map<uint16, NodeEntry*>::iterator it, itEnd;
	while (true) {
		dif = GetTimeAgeMS(start);
		if (networking->nodeEntriesMutex.enter()) {
			if (networking->nodeEntries[nodeID]->status < NODESTATUS_READY) {
				if (dif > (int32)timeout) {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Error configuring node '%s', response timeout...", networking->nodeEntries[nodeID]->name);
					networking->nodeEntriesMutex.leave();
					return false;
				}
				networking->nodeEntriesMutex.leave();
			}
			else {
				networking->nodeEntriesMutex.leave();
				break;
			}
		}
		utils::Sleep(20);
	}

	DataMessage* msg = new DataMessage(PsyAPI::CTRL_SYSTEM_READY, networking->localNodeID, 0, HOUR_US);
	networking->sendMessageToAllOtherNodes(msg, true);

	LogPrint(nodeID, LOG_NODE, 0, "Node '%s' (%u) has rejoined successfully...", networking->nodeEntries[nodeID]->name, nodeID);
	PsyType type = getPsyType(utils::StringFormat("Psyclone.%s.Connect", networking->nodeEntries[nodeID]->name).c_str());
	if (type.isValid()) {
		msg = new DataMessage();
		msg->setType(type);
		this->processMessage(msg);
		delete(msg);
	}

	LogPrint(nodeID, LOG_NODE, 2, "Current full registration:\n%s\n", subscriptionEngine->printFriendly().c_str());

	LogPrint(nodeID, LOG_NODE, 2, "Component list\n%s...", manager->componentMemory->printFriendly().c_str());

	LogPrint(nodeID, LOG_NODE, 2, "Types list\n%s...", manager->dataMapsMemory->printAllTypes().c_str());

	LogPrint(nodeID, LOG_NODE, 2, "Contexts list\n%s...", manager->dataMapsMemory->printAllContexts().c_str());

	return true;
}


bool Node::join(uint16 nodeID) {
	this->nodeID = nodeID;
	manager->setNodeID(nodeID);
	idManager->init(networking);
	// #######################
	return true;
}

XMLNode Node::parseAndExpandConfig(const char* configXML) {
	if (!configXML)
		return XMLNode::emptyXMLNode;

	// First parse the XML to get errors, variables and includes

	XMLResults xmlResults;
	XMLNode mainNode=XMLNode::parseString(configXML,"psyspec", &xmlResults);
	
	// display error message (if any)
	if (xmlResults.error != eXMLErrorNone)
	{
		if (xmlResults.error == eXMLErrorFirstTagNotFound) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "PsySpec format error, main tag '<psyspec>' not found");
			return XMLNode::emptyXMLNode;
		}
		else {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "PsySpec parsing error at line %i, column %i:\n   %s\n",
				xmlResults.nLine,xmlResults.nColumn, XMLNode::getError(xmlResults.error));
			return XMLNode::emptyXMLNode;
		}
	}
	if(mainNode.isEmpty())
		return mainNode;

	// Check for XML includes
	std::string includeXML;
	int i = 0;
	XMLNode node;
	while (!(node = mainNode.getChildNode("include", i++)).isEmpty())
		includeXML += utils::ReadAFileString(node.getAttribute("file"));

	char* xmlMainNodeText;
	std::string xml;
	char* str = NULL;
	if (includeXML.size()) {
		// Then recreate XML without <psyspec> main tag
		const char* firstNode = strchr(configXML, '>');
		if (!firstNode) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "PsySpec format error, main tag '<psyspec>' not found");
			return XMLNode::emptyXMLNode;
		}
		uint32 pStart = (uint32)(firstNode-configXML+1);
		xmlMainNodeText = new char[pStart+1];
		memcpy(xmlMainNodeText, configXML, pStart);
		xmlMainNodeText[pStart] = 0;
		
		uint32 pEnd = (uint32)strlen(configXML)-1;
		while ( pEnd && (configXML[pEnd] != '<') ) {
			pEnd--;
		}
		if (!pEnd) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "PsySpec format error, main tag '<psyspec>' not found");
			return XMLNode::emptyXMLNode;
		}
		str = new char[pEnd - pStart + 1];
		memcpy(str, configXML+pStart, pEnd - pStart);
		str[pEnd - pStart] = 0;

		xml = utils::StringFormat("%s%s%s\n</psyspec>",
			xmlMainNodeText, includeXML.c_str(), str);
		delete [] str;
		delete [] xmlMainNodeText;

		mainNode = XMLNode::parseString(xml.c_str(), "psyspec", &xmlResults);

		// display error message (if any)
		if (xmlResults.error != eXMLErrorNone)
		{
			if (xmlResults.error == eXMLErrorFirstTagNotFound) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Expanded PsySpec format error, main tag '<psyspec>' not found");
				return XMLNode::emptyXMLNode;
			}
			else {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Expanded PsySpec parsing error at line %i, column %i:\n   %s\n",
					xmlResults.nLine, xmlResults.nColumn, XMLNode::getError(xmlResults.error));
				return XMLNode::emptyXMLNode;
			}
		}
		if (mainNode.isEmpty()) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Error parsing Expanded PsySpec file...");
			return XMLNode::emptyXMLNode;
		}
	}

	i = 0;
	ExpandConfigVariables(mainNode, configXML, xml);
	
	// if XML has been altered, reparse...
	if (xml.size()) {
		mainNode=XMLNode::parseString(xml.c_str(),"psyspec", &xmlResults);
		
		// display error message (if any)
		if (xmlResults.error != eXMLErrorNone)
		{
			if (xmlResults.error == eXMLErrorFirstTagNotFound) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Expanded PsySpec format error, main tag '<psyspec>' not found");
				return XMLNode::emptyXMLNode;
			}
			else {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Expanded PsySpec parsing error at line %i, column %i:\n   %s\n",
					xmlResults.nLine,xmlResults.nColumn, XMLNode::getError(xmlResults.error));
				return XMLNode::emptyXMLNode;
			}
		}
		if(mainNode.isEmpty()){
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Error parsing Expanded PsySpec file...");
			return XMLNode::emptyXMLNode;
		}
		// printf("%s", xml.c_str());
	}

	// We now have a complete XML Config File...
	return mainNode;

}

// Perform %variable% substitution on the PsySpec. Values come from the spec's
// <variable name=".." value=".."/> defaults, but a matching 'name=value'
// argument given on the Psyclone command line overrides the spec default.
bool Node::ExpandConfigVariables(XMLNode& mainNode, const char* configXML, std::string& xml) {
	if (!configXML)
		return false;
	int i = 0;
	XMLNode node;
	const char *key, *value;
	std::string cmdlineValue;
	while (!(node = mainNode.getChildNode("variable", i++)).isEmpty()) {
		if ( (key = node.getAttribute("name")) && (value = node.getAttribute("value")) ) {
			// Command line 'key=value' wins over the spec default
			if (utils::HasCommandLineArg(key, cmdlineValue))
				value = cmdlineValue.c_str();
			if (!xml.size()) xml = configXML;
			utils::StringSingleReplace(xml, utils::StringFormat("%%%s%%", key).c_str(), value, false);
		}
	}
	return true;
}


bool Node::shutdown(uint32 ms) {
	if (ms > 0) {
		// Send shutdown to all other nodes
		networking->sendMessageToAllOtherNodes(new DataMessage(PsyAPI::CTRL_SYSTEM_SHUTDOWN, nodeID));
	}
	shouldContinue = false;
	return true;
}

bool Node::terminate() {
	shouldContinue = false;
	if (joinEngine) joinEngine->stopSweep();

	uint32 n;
	// Notify every space (internal AND external) of the shutdown FIRST, so
	// externally managed executables can see their space shut down and exit
	// on their own while stopAll() below waits for them gracefully. Posting
	// after stopAll() meant the external helper never got the notice within
	// the stop timeout and was force-terminated (noisy on Windows, where a
	// console CTRL_BREAK cannot reach a CREATE_NEW_CONSOLE child).
	DataMessage* msg = new DataMessage(PsyAPI::CTRL_SYSTEM_SHUTDOWN, nodeID);
	std::vector<ProcessInfoStruct>* processes = manager->processMemory->getAllProcesses();
	for (n=0; n<processes->size(); n++)
		manager->processMemory->addToCmdQ(processes->at(n).id, msg);
	delete(msg);

	// T1.4: stop managed external executables (graceful stopsignal, then SIGKILL)
	if (externalExec)
		externalExec->stopAll();

	utils::Sleep(1000);

	for (n=0; n<processes->size(); n++)
		// Never EndProcess ourselves: the internal Root space is registered with
		// this node's own OS pid, and killing it here SIGKILLed the node mid-
		// shutdown (Linux) / silently failed (Windows), leaving children behind.
		if (processes->at(n).osID && (processes->at(n).osID != osProcID))
			utils::EndProcess(processes->at(n).osID);
	delete(processes);

	uint32 timeout = 100;
	while (timeout > 0) {
		if (!isMsgQRunning)
			break;
		utils::Sleep(5);
		timeout -= 5;
	}
	if (isMsgQRunning)
		ThreadManager::TerminateThread(msgThreadID);

	timeout = 100;
	while (timeout > 0) {
		if (!isSigQRunning)
			break;
		utils::Sleep(5);
		timeout -= 5;
	}
	if (isSigQRunning)
		ThreadManager::TerminateThread(sigThreadID);

	timeout = 100;
	while (timeout > 0) {
		if (!isReqQRunning)
			break;
		utils::Sleep(5);
		timeout -= 5;
	}
	if (isReqQRunning)
		ThreadManager::TerminateThread(reqThreadID);

	timeout = 100;
	while (timeout > 0) {
		if (!isNodeRunning)
			break;
		utils::Sleep(5);
		timeout -= 5;
	}

	// shutdown local spaces
	if (internalSpace) {
		delete(internalSpace);
		internalSpace = NULL;
	}

	//printf("[1]"); fflush(stdout);
	// disconnect local networking
	networking->shutdown();
	//printf("[2]"); fflush(stdout);


	// shutdown local IDManager
	delete(idManager);
	idManager = NULL;

	//printf("[2.0]"); fflush(stdout);
	// shutdown PsyProbe
	delete(psyProbe);
	psyProbe = NULL;
	//printf("[2.1]"); fflush(stdout);

	// shutdown Monitoring
	delete(monitoring);
	monitoring = NULL;
	//printf("[2.2]"); fflush(stdout);

	// shutdown Console and Logging
	logging->shutdown();
	delete(console);
	console = NULL;
	//delete(logging);
	logging = NULL;
	//printf("[2.3]"); fflush(stdout);

	// shutdown local networking
	delete(networking);
	networking = NULL;
	//printf("[2.4]"); fflush(stdout);

	// shutdown ServiceManager and SubscriptionEngine
	delete(serviceManager);
	serviceManager = NULL;
	delete(subscriptionEngine);
	subscriptionEngine = NULL;
	//printf("[2.5]"); fflush(stdout);

	//LogPrint(0, LOG_NODE, 0, "Psyclone Node on port %u has shutdown successfully...", networkPort);

	//printf("3"); fflush(stdout);
	// shutdown MemoryManager
	delete(manager);
	manager = NULL;
	//printf("4"); fflush(stdout);

	LogPrint(0, LOG_NODE, 0, "Psyclone Node on port %u has shutdown successfully...", networkPort);

	return true;
}

bool Node::restart() {
	return false;
}

bool Node::reinit(const char* filename) {
	return false;
}

bool Node::reinit(uint16 port, const char* filename) {
	return false;
}

uint16 Node::getNodeID(uint32 compID) {
	return manager->componentMemory->getComponentNodeID(compID);
}


bool Node::configNode(const char* configXML, bool isConfigNode) {
	if (!configXML)
		return false;

	// First parse the XML to get errors, variables and includes

	XMLResults xmlResults;
	XMLNode mainNode=XMLNode::parseString(configXML,"psyspec", &xmlResults);

	// display error message (if any)
	if (xmlResults.error != eXMLErrorNone)
	{
		if (xmlResults.error == eXMLErrorFirstTagNotFound) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "PsySpec format error, main tag '<psyspec>' not found");
			return false;
		}
		else {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "PsySpec parsing error at line %i, column %i:\n   %s\n",
				xmlResults.nLine,xmlResults.nColumn, XMLNode::getError(xmlResults.error));
			return false;
		}
	}
	if(mainNode.isEmpty())
		return false;
	return configNode(mainNode, isConfigNode);
}

//bool Node::processIDSync(uint8 idType, const char* entryData, uint16 remoteNodeID, uint32 syncID, uint32& existingID) {
bool Node::processIDSync(DataMessage* msg, uint16 remoteNodeID, uint32 syncID, uint32& existingID) {
	if (!msg)
		return false;

	uint8 idType = (uint8)msg->getReference();
	uint32 s;
	const char* entryData = msg->getData("IDEntry", s);

	uint16 id16;
	uint32 id32;

	existingID = 0;
	bool result = false;
	switch(idType) {
		case SYNC_RESERVE_ID:
			if (idManager->enterIDReservation(*(IDReservation*)entryData))
				result = true;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync reserve ID range from Node %u {%u}", remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync reserve ID range from Node %u {%u}", remoteNodeID, syncID);
			break;
		case SYNC_RESERVE_IDS:
			if (idManager->enterIDReservations(msg))
				result = true;
			if (result)
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS sync reserve ID bulk from Node %u {%u}", remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync reserve ID bulk from Node %u {%u}", remoteNodeID, syncID);
			break;
		case SYNC_TYPELEVEL_ID:
			if (manager->dataMapsMemory->createNewTypeLevel(((TypeLevelID*)entryData)->id, ((TypeLevelID*)entryData)->name, ((TypeLevelID*)entryData)->createdTime, id16))
				result = true;
			else
				existingID = id16;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync type ID %u '%s' (%u) from Node %u {%u}", ((TypeLevelID*)entryData)->id, ((TypeLevelID*)entryData)->name, existingID, remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync type ID %u '%s' (%u) from Node %u {%u}", ((TypeLevelID*)entryData)->id, ((TypeLevelID*)entryData)->name, existingID, remoteNodeID, syncID);
			break;
		case SYNC_CONTEXTLEVEL_ID:
			if (manager->dataMapsMemory->createNewContextLevel(((ContextLevelID*)entryData)->id, ((ContextLevelID*)entryData)->name, ((ContextLevelID*)entryData)->createdTime, id16))
				result = true;
			else
				existingID = id16;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync context ID %u '%s' (%u) from Node %u {%u}", ((ContextLevelID*)entryData)->id, ((ContextLevelID*)entryData)->name, existingID, remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync context ID %u '%s' (%u) from Node %u {%u}", ((ContextLevelID*)entryData)->id, ((ContextLevelID*)entryData)->name, existingID, remoteNodeID, syncID);
			break;
		case SYNC_TAG_ID:
			if (manager->dataMapsMemory->createNewTag(((TagID*)entryData)->id, ((TagID*)entryData)->name, ((TagID*)entryData)->createdTime, id32))
				result = true;
			else
				existingID = id32;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync tag ID %u '%s' (%u) from Node %u {%u}", ((TagID*)entryData)->id, ((TagID*)entryData)->name, existingID, remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync tag ID %u '%s' (%u) from Node %u {%u}", ((TagID*)entryData)->id, ((TagID*)entryData)->name, existingID, remoteNodeID, syncID);
			break;
		case SYNC_CRANK_ID:
			if (manager->dataMapsMemory->createNewCrank(((CrankID*)entryData)->id, ((CrankID*)entryData)->compID, ((CrankID*)entryData)->name, ((CrankID*)entryData)->function, ((CrankID*)entryData)->libraryFilename, ((CrankID*)entryData)->language, ((CrankID*)entryData)->script, ((CrankID*)entryData)->createdTime, id16))
				result = true;
			else
				existingID = id16;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync crank ID %u/%u '%s' (%u) from Node %u {%u}", ((CrankID*)entryData)->id, ((CrankID*)entryData)->compID, ((CrankID*)entryData)->name, existingID, remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync crank ID %u/%u '%s' (%u) from Node %u {%u}", ((CrankID*)entryData)->id, ((CrankID*)entryData)->compID, ((CrankID*)entryData)->name, existingID, remoteNodeID, syncID);
			break;
		case SYNC_COMPONENT_INFO:
			if (manager->componentMemory->updateComponentInformation(((ComponentID*)entryData)->id, ((ComponentID*)entryData)->type, ((ComponentID*)entryData)->policy, ((ComponentID*)entryData)->selfTrigger, ((ComponentID*)entryData)->nodeID, ((ComponentID*)entryData)->procID))
				result = true;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync component type for ID %u '%s' (%u) from Node %u {%u}", ((ComponentID*)entryData)->id, ((ComponentID*)entryData)->name, existingID, remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync component type for ID %u '%s' (%u) from Node %u {%u}", ((ComponentID*)entryData)->id, ((ComponentID*)entryData)->name, existingID, remoteNodeID, syncID);
			break;
		case SYNC_COMPONENT_ID:
			if (manager->componentMemory->createComponent(((ComponentID*)entryData)->id, ((ComponentID*)entryData)->type, ((ComponentID*)entryData)->policy, ((ComponentID*)entryData)->selfTrigger, ((ComponentID*)entryData)->name, 1024, ((ComponentID*)entryData)->nodeID, ((ComponentID*)entryData)->procID, ((ComponentID*)entryData)->createdTime, id32))
				result = true;
			else
				existingID = id32;
			if (result)
				LogPrint(nodeID, LOG_NODE, 3, "SUCCESS sync component ID %u '%s' (%u) from Node %u {%u}", ((ComponentID*)entryData)->id, ((ComponentID*)entryData)->name, existingID, remoteNodeID, syncID);
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED sync component ID %u '%s' (%u) from Node %u {%u}", ((ComponentID*)entryData)->id, ((ComponentID*)entryData)->name, existingID, remoteNodeID, syncID);
			break;
		case BULK_SYNC_TYPELEVEL_ID:
			if (manager->dataMapsMemory->syncTypeLevels(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync type IDs from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync type IDs from Node %u {%u}", remoteNodeID, syncID);
			break;
		case BULK_SYNC_CONTEXTLEVEL_ID:
			if (manager->dataMapsMemory->syncContextLevels(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync context IDs from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync context IDs from Node %u {%u}", remoteNodeID, syncID);
			break;
		case BULK_SYNC_TAG_ID:
			if (manager->dataMapsMemory->syncTags(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync tag IDs from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync tag IDs from Node %u {%u}", remoteNodeID, syncID);
			break;
		case BULK_SYNC_CRANK_ID:
			if (manager->dataMapsMemory->syncCranks(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync crank IDs from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync crank IDs from Node %u {%u}", remoteNodeID, syncID);
			break;
		case BULK_SYNC_COMPONENT_ID:
			if (manager->componentMemory->syncComponents(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync component IDs from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync component IDs from Node %u {%u}", remoteNodeID, syncID);
			break;
		case BULK_SYNC_SUBSCRIPTION:
			if (subscriptionEngine->syncSubscriptions(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync subscriptions IDs from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync subscriptions IDs from Node %u {%u}", remoteNodeID, syncID);
			break;
		case BULK_SYNC_CONTEXTS:
			if (subscriptionEngine->syncContexts(msg)) {
				result = true;
				LogPrint(nodeID, LOG_NODE, 1, "SUCCESS bulk sync contexts from Node %u {%u}", remoteNodeID, syncID);
			}
			else
				LogPrint(nodeID, LOG_NODE, 0, "FAILED bulk sync contexts from Node %u {%u}", remoteNodeID, syncID);
			break;
		default:
			LogPrint(nodeID, LOG_NODE, 0, "UNKNOWN sync from Node %u {%u}", remoteNodeID, syncID);
			break;
	}
	return result;
}

bool Node::processConfirmIDSync(uint8 idType, uint32 id, uint16 remoteNodeID) {
	if (!id)
		return false;

	bool result = false;
	switch(idType) {
		case SYNC_TYPELEVEL_ID:
			LogPrint(nodeID, LOG_NODE, 5, "Received CONFIRM sync type ID %u from Node %u", id, remoteNodeID);
			if (manager->dataMapsMemory->confirmTypeLevelID((uint16)id))
				result = true;
			break;
		case SYNC_CONTEXTLEVEL_ID:
			LogPrint(nodeID, LOG_NODE, 5, "Received CONFIRM sync context ID from Node %u", remoteNodeID);
			if (manager->dataMapsMemory->confirmContextLevelID((uint16)id))
				result = true;
			break;
		case SYNC_TAG_ID:
			LogPrint(nodeID, LOG_NODE, 5, "Received CONFIRM sync tag ID from Node %u", remoteNodeID);
			if (manager->dataMapsMemory->confirmTagID(id))
				result = true;
			break;
		case SYNC_CRANK_ID:
			LogPrint(nodeID, LOG_NODE, 5, "Received CONFIRM sync crank ID from Node %u", remoteNodeID);
			if (manager->dataMapsMemory->confirmCrankID((uint16)id))
				result = true;
			break;
		case SYNC_COMPONENT_ID:
			LogPrint(nodeID, LOG_NODE, 5, "Received CONFIRM sync component ID from Node %u", remoteNodeID);
			if (manager->componentMemory->confirmComponentID(id))
				result = true;
			break;
		default:
			break;
	}
	return result;
}

bool Node::processCancelIDSync(uint8 idType, uint32 id, uint16 remoteNodeID) {
	if (!id)
		return false;

	bool result = false;
	switch(idType) {
		case SYNC_TYPELEVEL_ID:
			LogPrint(nodeID, LOG_NODE, 0, "Received CANCEL sync type ID %u from Node %u", id, remoteNodeID);
			if (manager->dataMapsMemory->cancelTypeLevelID((uint16)id))
				result = true;
			break;
		case SYNC_CONTEXTLEVEL_ID:
			LogPrint(nodeID, LOG_NODE, 0, "Received CANCEL sync context ID from Node %u", remoteNodeID);
			if (manager->dataMapsMemory->cancelContextLevelID((uint16)id))
				result = true;
			break;
		case SYNC_TAG_ID:
			LogPrint(nodeID, LOG_NODE, 0, "Received CANCEL sync tag ID from Node %u", remoteNodeID);
			if (manager->dataMapsMemory->cancelTagID(id))
				result = true;
			break;
		case SYNC_CRANK_ID:
			LogPrint(nodeID, LOG_NODE, 0, "Received CANCEL sync crank ID from Node %u", remoteNodeID);
			if (manager->dataMapsMemory->cancelCrankID((uint16)id))
				result = true;
			break;
		case SYNC_COMPONENT_ID:
			LogPrint(nodeID, LOG_NODE, 0, "Received CANCEL sync component ID from Node %u", remoteNodeID);
			if (manager->componentMemory->cancelComponentID(id))
				result = true;
			break;
		default:
			break;
	}
	return result;
}

uint8 Node::confirmIDSync(uint8 idType, uint32 id) {
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_CONFIRM_ID, networking->localNodeID);
	msg->setReference(idType);
	msg->setInt("ID", id);
	return networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true);
}

uint8 Node::cancelIDSync(uint8 idType, uint32 id) {
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_CANCEL_ID, networking->localNodeID);
	msg->setReference(idType);
	msg->setInt("ID", id);
	return networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true);
}

bool Node::bulkContextSyncNode(uint16 nodeID) {
	//TypeLevelID
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_CONTEXTS);
	subscriptionEngine->writeContextsToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising active contexts to rejoining Node %u", nodeID);
		return false;
	}
	return true;
}

bool Node::bulkIDSyncNode(uint16 nodeID) {

	//TypeLevelID
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_TYPELEVEL_ID);
	manager->dataMapsMemory->writeTypeLevelsToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising type levels to rejoining Node %u", nodeID);
		return false;
	}

	//ContextLevelID
	msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_CONTEXTLEVEL_ID);
	manager->dataMapsMemory->writeContextLevelsToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising context levels to rejoining Node %u", nodeID);
		return false;
	}

	//TagID
	msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_TAG_ID);
	manager->dataMapsMemory->writeTagsToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising cranks to rejoining Node %u", nodeID);
		return false;
	}

	//CrankID(struct)
	msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_CRANK_ID);
	manager->dataMapsMemory->writeCranksToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising components to rejoining Node %u", nodeID);
		return false;
	}

	//ComponentID(struct)
	msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_COMPONENT_ID);
	manager->componentMemory->writeComponentsToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising tags to rejoining Node %u", nodeID);
		return false;
	}

	//ComponentTriggers (complex struct)
	msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(BULK_SYNC_SUBSCRIPTION);
	subscriptionEngine->writeTriggersToMsg(msg);
	if (!networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true)) {
		LogPrint(nodeID, LOG_NODE, 0, "Failed bulk synchronising subscriptions to rejoining Node %u", nodeID);
		return false;
	}

	return true;
}

uint8 Node::negotiateIDSync(uint8 type, uint32 id, const char* name, const void* idEntry, uint32 size) {

	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_SYNC_ID, networking->localNodeID);
	msg->setReference(type);
	msg->setData("IDEntry", (char*)idEntry, size);

	std::map<uint16, DataMessage*> replyMap;
	std::map<uint16, DataMessage*>::iterator i, e;
	uint8 status = networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true, &replyMap);
	if (status == SYNC_SUCCESS) {
		for (i = replyMap.begin(), e = replyMap.end(); i!=e; i++)
			delete(i->second);
		if (confirmIDSync(type, id) == SYNC_SUCCESS)
			return SYNC_SUCCESS;
		else {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed confirming %s (%u) for '%s'...", SYNCTYPES[type], id, name);
			return SYNC_FAILED;
		}
	}
	else if (status == SYNC_FAILED) {
		if (!replyMap.size()) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing %s (%u) for '%s', no replies...", SYNCTYPES[type], id, name);
			return SYNC_FAILED;
		}
		// Did I get the lowest number node to confirm my entry
		uint16 nID = 0;
		while (!replyMap[nID] && (nID < 100)) nID++;
		if (!replyMap[nID]) {
			for (i = replyMap.begin(), e = replyMap.end(); i!=e; i++)
				delete(i->second);
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing %s (%u) for '%s', no replies under ID 100...", SYNCTYPES[type], id, name);
			return SYNC_FAILED;
		}
		if (replyMap[nID]->getType() == Networking::CTRL_NODE_SYNC_ID_SUCCESS) {
			// Our ID wins, wait for a bit and send out request again...
			utils::Sleep(10);
			for (i = replyMap.begin(), e = replyMap.end(); i!=e; i++)
				delete(i->second);
			if (negotiateIDSync(type, id, name, idEntry, size) == SYNC_SUCCESS)
				return SYNC_SUCCESS;
			else {
				LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed confirming %s (%u) for '%s' after negotiation...", SYNCTYPES[type], id, name);
				return SYNC_FAILED;
			}
		}
		else {
			for (i = replyMap.begin(), e = replyMap.end(); i!=e; i++)
				delete(i->second);
			// Nope, someone else got there before us, cancel our request and wait a bit
			if (cancelIDSync(type, id) != SYNC_SUCCESS) {
				LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed cancelling failed sync %s (%u) for '%s'...", SYNCTYPES[type], id, name);
				return SYNC_FAILED;
			}
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "ID clash sync %s (%u) for '%s', waiting for other node to succeed...", SYNCTYPES[type], id, name);
			utils::Sleep(100); // wait for the other node to succeed in setting up this ID
			return SYNC_RETRY;
		}
	}
	else if (status == SYNC_TIMEOUT) {
		for (i = replyMap.begin(), e = replyMap.end(); i!=e; i++)
			delete(i->second);
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "TIMEOUT syncing %s (%u) for '%s', cancelling and retrying...", SYNCTYPES[type], id, name);
		if (cancelIDSync(type, id) != SYNC_SUCCESS) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed cancelling timed out sync %s (%u) for '%s'...", SYNCTYPES[type], id, name);
			return SYNC_FAILED;
		}
		utils::Sleep(100); // wait a bit and try again
		return SYNC_RETRY;
	}
	else {
		for (i = replyMap.begin(), e = replyMap.end(); i!=e; i++)
			delete(i->second);
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "Error syncing %s (%u) for '%s'...", SYNCTYPES[type], id, name);
		return SYNC_FAILED;
	}
}


uint16 Node::syncTypeLevelID(const char* name) {

	uint32 usTimeout = IDSYNC_RESERVE_TIMEOUT_MS*1000;
	uint64 start;
	uint16 id;

	// Check to see if id already exists
	uint8 status = manager->dataMapsMemory->lookupTypeLevelID(name, id);
	if (status == 2)
		return id;
	else if (status == 1) {
		start = GetTimeNow();
		do {
			utils::Sleep(5);
			if ( (status = manager->dataMapsMemory->lookupTypeLevelID(name, id)) > 1)
				return id;
			else if (status < 1)
				break;
		} while (GetTimeAge(start) < usTimeout);
		if (status == 1) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing type id for '%s', stuck on reserved...", name);
			return 0;
		}
	}

	// This id has never been created, create it
	id = idManager->getNextTypeID();
	if (!id) {
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "No more IDs available for type '%s'...", name);
		return 0;
	}

	// See if everybody else can agree to this id
	TypeLevelID type;
	type.size = sizeof(TypeLevelID);
	type.cid = TYPELEVELID;
	type.id = id;
	type.createdTime = GetTimeNow();
	utils::strcpyavail(type.name, name, MAXKEYNAMELEN, true);
	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing type id [%u] for '%s'...", id, name);

	uint16 existingID;
	if (!manager->dataMapsMemory->createNewTypeLevel(id, name, type.createdTime, existingID)) {
		return syncTypeLevelID(name);
	}
	if ( (status = negotiateIDSync(SYNC_TYPELEVEL_ID, id, name, &type, type.size)) == SYNC_SUCCESS) {
		manager->dataMapsMemory->confirmTypeLevelID(id);
		return id;
	}
	else
		manager->dataMapsMemory->cancelTypeLevelID(id);

	if (status == SYNC_RETRY) {
		return syncTypeLevelID(name);
	}
	else
		return 0;
}

uint16 Node::syncContextLevelID(const char* name) {

	uint32 usTimeout = IDSYNC_RESERVE_TIMEOUT_MS*1000;
	uint64 start;
	uint16 id;

	// Check to see if id already exists
	uint8 status = manager->dataMapsMemory->lookupContextLevelID(name, id);
	if (status == 2)
		return id;
	else if (status == 1) {
		start = GetTimeNow();
		do {
			utils::Sleep(5);
			if ( (status = manager->dataMapsMemory->lookupContextLevelID(name, id)) > 1)
				return id;
			else if (status < 1)
				break;
		} while (GetTimeAge(start) < usTimeout);
		if (status == 1) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing context id for '%s', stuck on reserved...", name);
			return 0;
		}
	}

	// This id has never been created, create it
	id = idManager->getNextContextID();
	if (!id) {
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "No more IDs available for context '%s'...", name);
		return 0;
	}

	// See if everybody else can agree to this id
	ContextLevelID context;
	context.size = sizeof(ContextLevelID);
	context.cid = CONTEXTLEVELID;
	context.id = id;
	context.createdTime = GetTimeNow();
	utils::strcpyavail(context.name, name, MAXKEYNAMELEN, true);
	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing context id [%u] for '%s'...", id, name);

	uint16 existingID;
	if (!manager->dataMapsMemory->createNewContextLevel(id, name, context.createdTime, existingID))
		return syncContextLevelID(name);
	if ( (status = negotiateIDSync(SYNC_CONTEXTLEVEL_ID, id, name, &context, context.size)) == SYNC_SUCCESS) {
		manager->dataMapsMemory->confirmContextLevelID(id);
		return id;
	}
	else
		manager->dataMapsMemory->cancelContextLevelID(id);

	if (status == SYNC_RETRY)
		return syncContextLevelID(name);
	else
		return 0;
}

uint32 Node::syncTagID(const char* name) {

	uint32 usTimeout = IDSYNC_RESERVE_TIMEOUT_MS*1000;
	uint64 start;
	uint32 id;

	// Check to see if id already exists
	uint8 status = manager->dataMapsMemory->lookupTagID(name, id);
	if (status == 2)
		return id;
	else if (status == 1) {
		start = GetTimeNow();
		do {
			utils::Sleep(5);
			if ( (status = manager->dataMapsMemory->lookupTagID(name, id)) > 1)
				return id;
			else if (status < 1)
				break;
		} while (GetTimeAge(start) < usTimeout);
		if (status == 1) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing tag id for '%s', stuck on reserved...", name);
			return 0;
		}
	}

	// This id has never been created, create it
	id = idManager->getNextTagID();
	if (!id) {
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "No more IDs available for tag '%s'...", name);
		return 0;
	}

	// See if everybody else can agree to this id
	TagID tag;
	tag.size = sizeof(TagID);
	tag.cid = TAGID;
	tag.id = id;
	tag.createdTime = GetTimeNow();
	utils::strcpyavail(tag.name, name, MAXKEYNAMELEN, true);
	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing tag id [%u] for '%s'...", id, name);

	uint32 existingID;
	if (!manager->dataMapsMemory->createNewTag(id, name, tag.createdTime, existingID))
		return syncTagID(name);
	if ( (status = negotiateIDSync(SYNC_TAG_ID, id, name, &tag, tag.size)) == SYNC_SUCCESS) {
		manager->dataMapsMemory->confirmTagID(id);
		return id;
	}
	else
		manager->dataMapsMemory->cancelTagID(id);

	if (status == SYNC_RETRY)
		return syncTagID(name);
	else
		return 0;
}

uint16 Node::syncCrankID(uint32 compID, const char* name, const char* function, const char* libraryFilename, const char* language, const char* script) {

	uint32 usTimeout = IDSYNC_RESERVE_TIMEOUT_MS*1000;
	uint64 start;
	uint16 id;

	// Check to see if id already exists
	uint8 status = manager->dataMapsMemory->lookupCrankID(name, id);
	if (status == 2)
		return id;
	else if (status == 1) {
		start = GetTimeNow();
		do {
			utils::Sleep(5);
			if ( (status = manager->dataMapsMemory->lookupCrankID(name, id)) > 1)
				return id;
			else if (status < 1)
				break;
		} while (GetTimeAge(start) < usTimeout);
		if (status == 1) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing crank id for '%s', stuck on reserved...", name);
			return 0;
		}
	}

	// This id has never been created, create it
	id = idManager->getNextCrankID();
	if (!id) {
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "No more IDs available for crank '%s'...", name);
		return 0;
	}

	// See if everybody else can agree to this id
	CrankID crank;
	crank.size = sizeof(CrankID);
	crank.cid = CRANKID;
	crank.id = id;
	crank.compID = compID;
	crank.createdTime = GetTimeNow();
	utils::strcpyavail(crank.name, name, MAXKEYNAMELEN, true);
	utils::strcpyavail(crank.language, language, MAXKEYNAMELEN, true);
	utils::strcpyavail(crank.function, function, MAXKEYNAMELEN, true);
	if (libraryPaths.find(libraryFilename) != libraryPaths.end())
		utils::strcpyavail(crank.libraryFilename, libraryPaths[libraryFilename].c_str(), MAXKEYNAMELEN, true);
	else
		utils::strcpyavail(crank.libraryFilename, libraryFilename, MAXKEYNAMELEN, true);
	utils::strcpyavail(crank.script, script, MAXSCRIPTLEN, true);
	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing crank id [%u] for '%s'...", id, name);

	uint16 existingID;
	if (!manager->dataMapsMemory->createNewCrank(id, compID, name, function, crank.libraryFilename, language, script, crank.createdTime, existingID))
		return syncCrankID(compID, name, function, crank.libraryFilename, language, script);

	if ( (status = negotiateIDSync(SYNC_CRANK_ID, id, name, &crank, crank.size)) == SYNC_SUCCESS) {
		manager->dataMapsMemory->confirmCrankID(id);
		return id;
	}
	else
		manager->dataMapsMemory->cancelCrankID(id);

	if (status == SYNC_RETRY)
		return syncCrankID(compID, name, function, crank.libraryFilename, language, script);
	else
		return 0;
}

uint32 Node::syncComponentID(uint8 type, uint8 policy, uint8 selfTrigger, const char* name, uint16 nodeID, uint16 procID) {

	uint32 usTimeout = IDSYNC_RESERVE_TIMEOUT_MS*1000;
	uint64 start;
	uint32 id;
	ComponentID component;

	// Check to see if id already exists
	uint8 status = manager->componentMemory->lookupComponentID(name, id); // dataMapsMemory->lookupTypeLevelID(name, id);
	// if completely synchronised
	if (status == 3)
		return id;
	else if (status == 2) {
		// name and id synchronised, but type not yet specified, we may know that now
		if (type) {
			component.size = sizeof(ComponentID);
			component.cid = COMPONENTID;
			component.id = id;
			component.type = type;
			component.selfTrigger = selfTrigger;
			component.policy = policy;
			component.nodeID = nodeID;
			component.procID = procID;
			LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing component type for ID [%u] '%s'...", id, name);
			if ((status = negotiateIDSync(SYNC_COMPONENT_INFO, id, name, &component, component.size)) == SYNC_SUCCESS)
				manager->componentMemory->updateComponentInformation(id, type, policy, selfTrigger, nodeID, procID);
			return id;
		}
		else {
			// we still don't know so just go with it
			return id;
		}
	}
	else if (status == 1) {
		start = GetTimeNow();
		do {
			utils::Sleep(5);
			status = manager->componentMemory->lookupComponentID(name, id);
			if (status > 2)
				return id;
			else if (!type && (status > 1))
				return id;
			else if (status < 1)
				break;
		} while (GetTimeAge(start) < usTimeout);
		if (status == 2) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing component type for '%s', stuck on no type...", name);
			return id;
		}
		else if (status == 1) {
			LogPrint(networking->localNodeID, LOG_SYNC, 0, "Failed syncing component id for '%s', stuck on reserved...", name);
			return 0;
		}
	}

	// This id has never been created, create it
	id = idManager->getNextComponentID();
	if (!id) {
		LogPrint(networking->localNodeID, LOG_SYNC, 0, "No more IDs available for component '%s'...", name);
		return 0;
	}

	// See if everybody else can agree to this id
	component.size = sizeof(ComponentID);
	component.cid = COMPONENTID;
	component.id = id;
	component.createdTime = GetTimeNow();
	component.type = type;
	component.selfTrigger = selfTrigger;
	component.policy = policy;
	component.nodeID = nodeID;
	component.procID = procID;
	utils::strcpyavail(component.name, name, MAXKEYNAMELEN, true);
	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing component id [%u] for '%s'...", id, name);

	uint32 existingID;
	if (!manager->componentMemory->createComponent(id, type, policy, selfTrigger, name, 1024, nodeID, procID, component.createdTime, existingID))
		return syncComponentID(type, policy, selfTrigger, name, nodeID, procID);

	if ( (status = negotiateIDSync(SYNC_COMPONENT_ID, id, name, &component, component.size)) == SYNC_SUCCESS) {
		manager->componentMemory->confirmComponentID(id);
		return id;
	}
	else
		manager->componentMemory->cancelComponentID(id);

	if (status == SYNC_RETRY)
		return syncComponentID(type, policy, selfTrigger, name, nodeID, procID);
	else
		return 0;
}

bool Node::syncComponentLocation(uint32 id, uint16 nodeID, uint16 procID) {

	if (!manager->componentMemory->updateComponentLocation(id, nodeID, procID)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not update local location for component '%u'...", id);
		return false;
	}

	ComponentID component;
	memset(&component, 0, sizeof(ComponentID));
	component.size = sizeof(ComponentID);
	component.cid = COMPONENTID;
	component.id = id;
	component.createdTime = GetTimeNow();
	component.nodeID = nodeID;
	component.procID = procID;
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_SYNC_COMPONENT, networking->localNodeID);
	msg->setData("ComponentLocation", (char*)&component, sizeof(ComponentID));
	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing component location [%u] for node %u space %u...", id, nodeID, procID);
	return networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true);
}

bool Node::syncComponentTriggers(uint32 compID, std::list<TriggerSpec*> &triggers) {

	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_SYNC_SUBSCRIPTION, networking->localNodeID);
	msg->setInt("CompID", (int64)compID);

	uint32 n = 0;
	char* label = new char[MAXKEYNAMELEN+1];
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = triggers.begin(), itEnd = triggers.end(); it != itEnd; it++) {
		snprintf(label, MAXKEYNAMELEN+1, "Trigger%u", n++);
		msg->setData(label, (char*)(*it), (*it)->size);
	}
	delete [] label;

	LogPrint(networking->localNodeID, LOG_SYNC, 3, "Syncing %u triggers for component id [%u]...", triggers.size(), compID);
	return networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true);
}

bool Node::startConfigNode(DataMessage* msg) {
	configNodeMessage = msg;
	return true;
}

bool Node::configNode(XMLNode &n, bool isConfigNode) {
	
	// All nodes will go through the config file and do what is relevant to them
	// if isConfigNode we will deal with 'local' options, otherwise ignore

	XMLNode node, subNode;
	uint32 i;
	int j;
	const char *tag, *name, *nodeName, *subTag, *str, *loc, *defaultFile;

	// Global SSL client verification policy: <psyspec allowselfsigned="yes">
	// loosens verification for all SSL client connections; default is secure
	// (verify peer). Per-interface allowselfsigned overrides this either way.
	if (str = n.getAttribute("allowselfsigned"))
		SSLConnection::SetDefaultAllowSelfSigned((stricmp(str, "yes") == 0) || (stricmp(str, "true") == 0) || (strcmp(str, "1") == 0));

	// Global custom CA location for SSL client verification:
	// <psyspec cafile="/path/ca.pem" capath="/path/cadir"> verifies peers
	// against this CA instead of the OS trust store. Per-interface
	// cafile/capath overrides this.
	{
		const char* caFileStr = n.getAttribute("cafile");
		const char* caPathStr = n.getAttribute("capath");
		if (caFileStr || caPathStr)
			SSLConnection::SetDefaultCALocation(caFileStr, caPathStr);
	}

	// Register basic types and contexts
	PsyType type = getPsyType("PsyControl"); // gets subtype number 1, reserved for system use
	if (type == NOTYPE) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic type PsyControl...");
		return false;
	}
	systemReadyType = getPsyType("Psyclone.Ready"); // gets subtype number 2 and 3
	if (systemReadyType == NOTYPE) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic type Psyclone.Ready...");
		return false;
	}
	defaultContext = getPsyContext("Psyclone.Ready");
	if (defaultContext == NOCONTEXT) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic context Psyclone.Ready...");
		return false;
	}
	shutdownType = getPsyType("Psyclone.Shutdown");
	if (shutdownType == NOTYPE) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic type Psyclone.Shutdown...");
		return false;
	}
	
	// some additional type levels used frequently
	type = getPsyType("Connect");
	if (type == NOTYPE) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic type Connect...");
		return false;
	}
	type = getPsyType("Disconnect");
	if (type == NOTYPE) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic type Connect...");
		return false;
	}
	type = getPsyType("Root");
	if (type == NOTYPE) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to synchronise basic type Connect...");
		return false;
	}

	subscriptionEngine->setContextActive(defaultContext, NOCONTEXT);

	if (!monitoring->startMonitoring()) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed to start Monitoring...");
		return false;
	}

//	testTypesContexts();

	uint32 count = n.nChildNode("service");
	// We need to add services first
	for(i=0; i<count; i++) {
		if (!(node=n.getChildNode("service", i)).isEmpty()) {
			name = node.getAttribute("name");
			nodeName = node.getAttribute("node");
			if ( !nodeName || (stricmp(nodeName, networking->localNodeName) == 0) ) {
				LogPrint(networking->localNodeID, LOG_NODE, 1, "Configuring Service '%s'...", name);
				serviceManager->addService(node);
			}
		}
	}

	uint32 id;
	count = n.nChildNode("psyprobe");
	for (i = 0; i<count; i++) {
		if (!(node = n.getChildNode("psyprobe", i)).isEmpty()) {
			if ((str = node.getAttribute("defaultport")) && (stricmp(str, "no"))) {
				if (serviceManager->getServiceID("PsyProbe", id)) {
					if (!networking->removeInterface(networkPort, INTERFACE_HTTP))
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Couldn't remove default PsyProbe port");
					else
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Removed PsyProbe interface from Psyclone port %u", networkPort);
				}
			}
			uint16 port2;
			if ((str = node.getAttribute("port")) && (port2 = (uint16)utils::Ascii2Uint32(str))) {
				if (!networking->addInterface("PsyProbe", port2, INTERFACE_PROTO_TCP, NOENC, INTERFACE_HTTP, NODESYNC_TIMEOUT_MS, false, "", "PsyProbe"))
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Couldn't add PsyProbe port %u", port2);
				else
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Added PsyProbe interface on port %u", port2);
			}
			if ((loc = node.getAttribute("location")) && (strlen(loc))) {
				psyProbe->setRootDir(loc);
			}
			uint32 l;
			uint32 subCount = node.nChildNode("alias");
			for (l = 0; l < subCount; l++) {
				if (!(subNode = node.getChildNode("alias", i)).isEmpty()) {
					name = subNode.getAttribute("name");
					loc = subNode.getAttribute("location");
					defaultFile = subNode.getAttribute("default");
					if (name && loc) {
						if (psyProbe->addAlias(name, loc, defaultFile))
							LogPrint(networking->localNodeID, LOG_NODE, 1, "Added PsyProbe alias '%s'", name);
						else
							LogPrint(networking->localNodeID, LOG_NODE, 1, "Couldn't add PsyProbe alias '%s'", name);
					}
				}
			}

			subCount = node.nChildNode("post");
			for (l = 0; l < subCount; l++) {
				if (!(subNode = node.getChildNode("post", l)).isEmpty()) {
					PostSpec* postSpec = new PostSpec;
					memset(postSpec, 0, sizeof(PostSpec));
					postSpec->context = defaultContext;
					if (str = subNode.getAttribute("name"))
						utils::strcpyavail(postSpec->name, str, MAXKEYNAMELEN, true);
					if (str = subNode.getAttribute("type"))
						postSpec->type = getPsyType(str);
					if (str = subNode.getAttribute("context"))
						postSpec->contextchange = getPsyContext(str);
					if (str = subNode.getAttribute("to"))
						postSpec->to = getComponentID(str);
					if (str = subNode.getAttribute("ttl"))
						postSpec->ttl = utils::Ascii2Uint64(str) * 1000;
					else
						postSpec->ttl = 0;
					if (str = subNode.getAttribute("guaranteed"))
						if (stricmp(str, "no") == 0)
							postSpec->policy |= MESSAGE_NON_GUARANTEED;
					//if (str = subNode.getAttribute("contentkey")) {
					//	utils::strcpyavail(postSpec->contentKey, str, MAXKEYNAMELEN, true);
					//	if (str = subNode.getAttribute("content"))
					//		utils::strcpyavail(postSpec->content, html::DecodeHTML(str).c_str(), MAXVALUENAMELEN, true);
					//}
					postSpec->contentFromXML(subNode);
					//if (str = subNode.getAttribute("tag")) {
					//	if (strcmp(str, "*") == 0)
					//		postSpec->tag = getTagID(spec->setup->name);
					//	else
					//		postSpec->tag = getTagID(str);
					//}
					if (postSpec->name[0] && (postSpec->type.isValid() || postSpec->contextchange.isValid())) {
						psyProbe->addPost(postSpec->name, postSpec);
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Added PsyProbe post '%s'", name);
					}
					else {
						delete(postSpec);
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Couldn't add PsyProbe post '%s'", name);
					}
				}
			}
		}
	}

	// Add main PsySystem library
	addLibrary("PsySystem", "PsySystem", NULL);

	uint16 spaceID;
	// First setup non-Component elements
	count = n.nChildNode();
	for(i=0; i<count; i++) {
		if (!(node=n.getChildNode(i)).isEmpty()) {
			tag = node.getName();
			name = node.getAttribute("name");
			nodeName = node.getAttribute("node");

			if (stricmp(tag, "node") == 0) {
				if ( (!name && isConfigNode) || (name && stricmp(name, networking->localNodeName) == 0) ) {
					for (j=0; j<node.nChildNode(); j++) {
						if (!(subNode=node.getChildNode(j)).isEmpty()) {
							subTag = subNode.getName();
							if (stricmp(subTag, "library") == 0)
								addLibrary(subNode.getAttribute("name"), subNode.getAttribute("library"), subNode.getAttribute("path"));
						}
					}
				}
			}
			else if (stricmp(tag, "space") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					// Create Space
					str = node.getAttribute("type");
					if (str && (stricmp(str, "external") == 0)) {
						createExternalSpace(name, spaceID);
					}
				}
			}
			else if (stricmp(tag, "library") == 0) {
				if ( !nodeName || (stricmp(nodeName, networking->localNodeName) == 0) ) {
					// Load library
					addLibrary(name, node.getAttribute("library"), node.getAttribute("path"));
				}
			}
			else if (stricmp(tag, "interface") == 0) {
				if ( !nodeName || (stricmp(nodeName, networking->localNodeName) == 0) ) {
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Configuring Interface '%s'...", name);
					networking->addInterface(node);
				}
			}
			else if (stricmp(tag, "supervisor") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					// ##################################
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Configuring Supervisor '%s'...", name);
				}
			}
			else if (stricmp(tag, "monitoring") == 0) {
				if ( !nodeName || (stricmp(nodeName, networking->localNodeName) == 0) ) {
					// ##################################
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Monitoring '%s' not yet implemented...", name);
				}
			}
			else if (stricmp(tag, "performance") == 0) {
				if ( !nodeName || (stricmp(nodeName, networking->localNodeName) == 0) ) {
					// ##################################
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Performance '%s' not yet implemented...", name);
				}
			}
		}
	}

	// Then setup Components
	count = n.nChildNode();
	for(i=0; i<count; i++) {
		if (!(node=n.getChildNode(i)).isEmpty()) {
			tag = node.getName();
			name = node.getAttribute("name");
			nodeName = node.getAttribute("node");

			if (stricmp(tag, "feed") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					// ##################################
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Configuring Feed '%s'...", name);
				}
			}
			else if (stricmp(tag, "whiteboard") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					if (createComponent(name, node))
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Configured Whiteboard '%s' (%u) successfully", name, getComponentID(name));
					else {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed creating Whiteboard '%s'...", name);
						return false;
					}
				}
			}
			else if (stricmp(tag, "stream") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					if (createComponent(name, node))
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Configured Stream '%s' (%u) successfully", name, getComponentID(name));
					else {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed creating Stream '%s'...", name);
						return false;
					}
				}
			}
			else if (stricmp(tag, "catalog") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					if (createComponent(name, node))
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Configured Catalog '%s' (%u) successfully", name, getComponentID(name));
					else {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed creating Catalog '%s'...", name);
						return false;
					}
				}
			}
			else if (stricmp(tag, "service") == 0) { // This time for the subscription inside
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					if (createComponent(name, node))
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Configured Service '%s' successfully", name);
					else {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed creating Service '%s'...", name);
						return false;
					}
				}
			}
			else if (stricmp(tag, "module") == 0) {
				if ( (!nodeName && isConfigNode)
					|| (nodeName && stricmp(nodeName, networking->localNodeName) == 0)
					|| (isConfigNode && (strcmp(nodeName, "*") == 0) ) ) {
					if (createComponent(name, node))
						LogPrint(networking->localNodeID, LOG_NODE, 1, "Configured Module '%s' (%u) successfully", name, getComponentID(name));
					else {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Failed creating Module '%s'...", name);
						return false;
					}
				}
			}
			else if (stricmp(tag, "group") == 0) {
				if ( (!nodeName && isConfigNode) || (nodeName && stricmp(nodeName, networking->localNodeName) == 0) ) {
					// ##################################
					LogPrint(networking->localNodeID, LOG_NODE, 1, "Configuring Group '%s'...", name);
				}
			}
		}
	}

	return true;
}

bool Node::addLibrary(const char* name, const char* library, const char* path) {
	LogPrint(networking->localNodeID, LOG_NODE, 2, "Configuring Library '%s'...", name);
	if (!path || !strlen(path))
		libraryPaths[name] = library;
	else if (utils::TextEndsWith(path, "\\") || utils::TextEndsWith(path, "/"))
		libraryPaths[name] = utils::StringFormat("%s%s", path, library);
	else
		libraryPaths[name] = utils::StringFormat("%s/%s", path, library);
	return true;
}

bool Node::createComponent(const char* name, XMLNode &n) {
	if (!name)
		return false;

	uint32 id;
	if (manager->componentMemory->getComponentID(name, id)) {
		if (manager->componentMemory->getComponentType(id)) {
			LogPrint(networking->localNodeID, LOG_NODE, 1, "Component name '%s' is already in use for id %u...", name, id);
			return false;
		}
	}

	LogPrint(networking->localNodeID, LOG_NODE, 1, "Creating Module '%s'...", name);

	ComponentSpec* spec = componentSpecFromXML(n);
	if (!spec) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not register Module subscription '%s'...", name);
		return false;
	}

//	printf("*** Adding Subscriptions for '%s'...\n", name);

	// Send to local Subscription Engine
	if (!subscriptionEngine->addSubscription(spec->setup->componentID, spec->triggers)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not add Module subscription '%s'...", name);
		return false;
	}

//	printf("*** Adding Timers for '%s'...\n", name);

	// Check for any interval triggers
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = spec->triggers.begin(), itEnd = spec->triggers.end(); it != itEnd; it++) {
		if ((*it)->interval) {
			messageTimer->addTrigger(*it);
		}
	}

	// T1.6: register any <triggergroup> join groups with the JoinEngine. The group's
	// crank is the same crank stamped onto its member triggers (matched by group id).
	std::list<TriggerGroupSpec*>::iterator gg, ggEnd;
	for (gg = spec->triggerGroups.begin(), ggEnd = spec->triggerGroups.end(); gg != ggEnd; gg++) {
		uint16 gcrank = 0;
		for (it = spec->triggers.begin(), itEnd = spec->triggers.end(); it != itEnd; it++) {
			if ((*it)->group == (*gg)->id) { gcrank = (*it)->crankID; break; }
		}
		(*gg)->crankID = gcrank;
		joinEngine->registerGroup(*gg, gcrank);
	}

	// Add PsyProbe custom view to all PsyProbes
	std::list<CustomView>::iterator ic = spec->customViews.begin(), ec = spec->customViews.end();
	while (ic != ec) {
		psyProbe->addCustomView(spec->setup->componentID, ic->name, ic->templateURL);
		ic++;
	}

//	printf("*** Subscriptions are now ***\n%s*****************************\n", subscriptionEngine->toString().c_str());
//	printf("*** Registering Triggers for '%s'...\n", name);

	// Send to all other nodes
	if (!syncComponentTriggers(spec->setup->componentID, spec->triggers)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not sync Module subscription '%s'...", name);
		return false;	
	}

	// Maybe we should save the spec???
	delete(spec);

	// printf("*** Done for '%s'...\n", subscriptionEngine->toString().c_str());

	return true;
}

bool Node::createExternalSpace(const char* name, uint16 &id) {
	// Does it already exist?
	if (manager->processMemory->getProcessID(name, id))
		return true;
	if (stricmp(name, "Root") == 0) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "External Spaces cannot be named Root...");
		return false;
	}
	if (!manager->processMemory->createNewProcess(name, id)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create ProcessMap entry for space '%s'...", name);
		return false;
	}
	LogPrint(networking->localNodeID, LOG_NODE, 0, "External PsySpace '%s' created, waiting for connection...", name);
	return true;
}

bool Node::createSpace(const char* name, uint16 &id) {
	
	// Does it already exist?
	if (manager->processMemory->getProcessID(name, id)) {
		// Reusing an existing ProcessMap entry. If it is this node's own Root
		// space the entry may be left over from a previously killed run (stale
		// shared memory) and could carry a stale OS pid. Root always runs in
		// this very process, so refresh its type/osID to the current process.
		// Other (spawned) spaces keep the untouched fast-path so genuinely
		// live spaces and internal-space auto-restart are unaffected.
		if (stricmp(name, "Root") == 0) {
			uint32 oldOSID = manager->processMemory->getProcessOSID(id);
			if (oldOSID != osProcID)
				LogPrint(networking->localNodeID, LOG_NODE, 1, "Reusing existing Root ProcessMap entry (%u), refreshing stale osID %u -> %u...", id, oldOSID, osProcID);
			manager->processMemory->setProcessType(id, 1);
			manager->processMemory->setProcessOSID(id, osProcID);
		}
		return true;
	}

	if (!manager->processMemory->createNewProcess(name, id)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create ProcessMap entry for space '%s'...", name);
		return false;
	}

	if (stricmp(name, "Root") == 0) {
		internalSpace = new PsySpace(name, false, id, true);
		if (!internalSpace->connect(networkPort, false)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create internal Root space...");
			delete (internalSpace);
			internalSpace = NULL;
			return false;
		}
		if (!internalSpace->start()) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start internal Root space...");
			delete (internalSpace);
			internalSpace = NULL;
			return false;
		}
		manager->processMemory->setProcessType(id, 1);
		manager->processMemory->setProcessOSID(id, osProcID);
	}
	else {
		std::string cmdline = utils::StringFormat("%s space=%u name=\"%s\" port=%u", procname.c_str(), id, name, networkPort);

		LogPrint(networking->localNodeID, LOG_NODE, 1,"Starting new space: %s", cmdline.c_str());

		uint32 procid = utils::NewProcess(cmdline.c_str());

		if (procid == 0) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not start process for space '%s' (%s)...", name, cmdline.c_str());
			return false;
		}
		manager->processMemory->setProcessCommandLine(id, cmdline.c_str());
		manager->processMemory->setProcessOSID(id, procid);
		internallySpawnedSpaces.insert(id);	// eligible for auto-restart if it crashes
	}

	utils::Sleep(100);
	if (!waitForSpaceToBeReady(id, 10000)) {
		uint64 time;
		uint8 status = manager->processMemory->getProcessStatus(id, time);
		if (status < PSYPROC_INIT) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not connect to space '%s' (%u)...", name, id);
			return false;
		}
		else if (status < PSYPROC_ACTIVE) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not communicate with space '%s' (%u)...", name, id);
			return false;
		}
	}

	return true;
}

bool Node::waitForSpaceToBeReady(uint16 id, uint32 timeout) {
	uint64 time, start = GetTimeNow();
	while(manager->processMemory->getProcessStatus(id, time) < PSYPROC_ACTIVE) {
		utils::Sleep(10);
		if (GetTimeAgeMS(start) > (int32)timeout)
			return false;
	}
	return true;
}

bool Node::processSpaceConnect(uint16 procID) {
	// post message with type Psyclone.<spacename>.Connect
	char* procName = new char[MAXKEYNAMELEN];
	if (manager->processMemory->getProcessName(procID, procName, MAXKEYNAMELEN)) {
		LogPrint(networking->localNodeID, LOG_NODE, 1, "Space '%s' connected", procName);
		PsyType type = getPsyType(utils::StringFormat("Psyclone.%s.%s.Connect", networking->localNodeName, procName).c_str());
		if (type.isValid()) {
			DataMessage* msg = new DataMessage();
			msg->setType(type);
			this->processMessage(msg);
			delete(msg);
		}
		delete[] procName;
		return true;
	}
	else {
		delete[] procName;
		return false;
	}
}

bool Node::processSpaceDisconnect(uint16 procID) {
	// post message with type Psyclone.<spacename>.Disconnect
	char* procName = new char[MAXKEYNAMELEN];
	if (manager->processMemory->getProcessName(procID, procName, MAXKEYNAMELEN)) {
		LogPrint(networking->localNodeID, LOG_NODE, 1, "Space '%s' disconnected", procName);
		PsyType type = getPsyType(utils::StringFormat("Psyclone.%s.%s.Disconnect", networking->localNodeName, procName).c_str());
		if (type.isValid()) {
			DataMessage* msg = new DataMessage();
			msg->setType(type);
			this->processMessage(msg);
			delete(msg);
		}
		delete[] procName;
		return true;
	}
	else {
		delete[] procName;
		return false;
	}
}

bool Node::processSpaceShutdown(uint16 procID) {
	processSpaceDisconnect(procID);
	manager->processMemory->setProcessStatus(procID, PSYPROC_IDLE);
	//std::list<uint16>* comps = manager->componentMemory->findProcessComponents(procID);
	//if (comps) {
	//	std::list<uint16>::iterator i = comps->begin(), e = comps->end();
	//	while (i != e) {
	//		syncComponentLocation(*i, nodeID, 0);
	//		i++;
	//	}
	//	delete(comps);
	//}
	return true;
}

// T-space: respawn a crashed internally-spawned (plain, non-external) space.
// External spaces (clients connect on their own) and executable-managed spaces
// (T1.4 ExternalExec has its own restart policy) are not touched: only ids we
// recorded in internallySpawnedSpaces at createSpace() time are eligible.
// Rate-limited to at most 5 restarts per space within a rolling 60s window.
bool Node::restartInternalSpace(uint16 procID) {
	if (internallySpawnedSpaces.find(procID) == internallySpawnedSpaces.end())
		return false;

	char cmdline[1024];
	if (!manager->processMemory->getProcessCommandLine(procID, cmdline, sizeof(cmdline)) || !strlen(cmdline)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Cannot restart space %u: no stored command line...", procID);
		return false;
	}

	// Rate limiting: max 5 restarts per rolling 60s
	uint64 now = GetTimeNow();
	std::list<uint64>& hist = spaceRestartHistory[procID];
	while (hist.size() && (now - hist.front()) > 60000000ULL)
		hist.pop_front();
	if (hist.size() >= 5) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Space %u crashed too often (%u restarts in 60s), giving up on it...", procID, (uint32)hist.size());
		return false;
	}
	hist.push_back(now);

	LogPrint(networking->localNodeID, LOG_NODE, 1, "Restarting crashed space %u: %s", procID, cmdline);

	// Reset the ProcessMap entry so the child can rendezvous afresh
	manager->processMemory->setProcessStatus(procID, PSYPROC_CREATED);

	uint32 osid = utils::NewProcess(cmdline);
	if (osid == 0) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not restart process for space %u (%s)...", procID, cmdline);
		return false;
	}
	manager->processMemory->setProcessOSID(procID, osid);

	if (!waitForSpaceToBeReady(procID, 10000)) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Restarted space %u did not become ready in time...", procID);
		return false;
	}
	LogPrint(networking->localNodeID, LOG_NODE, 1, "Space %u restarted successfully", procID);
	return true;
}

bool Node::addPsyProbeCustomView(uint32 compID, const char* name, const char* templateURL) {
	if (!psyProbe->addCustomView(compID, name, templateURL))
		return false;
	DataMessage* msg = new DataMessage(Networking::CTRL_PSYPROBE_CREATECUSTOMPAGE, networking->localNodeID);
	msg->setInt("ComponentID", compID);
	msg->setString("Name", name);
	msg->setString("Template", templateURL);
	return networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true);
}


NodeInfoPackHeader* Node::getLocalInfoPack() {
	if (!nodeID || !networking->nodeEntries[nodeID])
		return NULL;

	uint32 size;
	std::vector<ProcessInfoStruct>* procInfos = manager->processMemory->getAllProcesses();
	std::vector<ComponentInfoStruct>* compInfos = manager->componentMemory->getAllLocalComponents();

	size = (uint32)((sizeof(NodeInfoPackHeader)
		+ (procInfos->size() * sizeof(ProcessInfoStruct))
		+ (compInfos->size() * sizeof(ComponentInfoStruct))));

	NodeInfoPackHeader* header = (NodeInfoPackHeader*) malloc(size);
	memset(header, 0, size);
	header->size = size;
	header->cid = NODEINFOPACKID;
	header->time = GetTimeNow();
	header->createdTime = createTime;
	header->nodeID = nodeID;
	header->port = networkPort;
	utils::strcpyavail(header->name, networking->nodeEntries[nodeID]->name, 256, true);
	utils::strcpyavail(header->addr, networking->nodeEntries[nodeID]->addr, 256, true);
	header->procCount = (uint16)procInfos->size();
	header->compCount = (uint16)compInfos->size();
	memcpy(&header->perfStruct, &lastPerfStruct, sizeof(AveragePerfStats));

	char* dst = ((char*)header) + sizeof(NodeInfoPackHeader);
	std::vector<ProcessInfoStruct>::iterator ip = procInfos->begin(), ep = procInfos->end();
	while (ip != ep) {
		memcpy(dst, &(*ip), sizeof(ProcessInfoStruct));
		dst += sizeof(ProcessInfoStruct);
		ip++;
	}
	std::vector<ComponentInfoStruct>::iterator ic = compInfos->begin(), ec = compInfos->end();
	while (ic != ec) {
		memcpy(dst, &(*ic), sizeof(ComponentInfoStruct));
		dst += sizeof(ComponentInfoStruct);
		ic++;
	}
	delete(compInfos);
	delete(procInfos);
	return header;
}

bool Node::getAllInfoPacks(std::map<uint16, NodeInfoPack*>& infopacks) {
	NodeInfoPackHeader* localInfoPack = getLocalInfoPack();
	if (!localInfoPack)
		return false;

	infopacks[nodeID] = new NodeInfoPack((char*)localInfoPack, false);

	std::map<uint16, DataMessage*> replyMap;
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
	msg->setReference(QUERYNODE_GETINFOPACK);
	if (networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true, &replyMap) != SYNC_SUCCESS)
		LogPrint(networking->localNodeID, LOG_NODE, 0,"Only got %u NodeInfoPacks back from other nodes...", replyMap.size());

	const char* data;
	uint32 size;
	std::map<uint16, DataMessage*>::iterator i = replyMap.begin(), e = replyMap.end();
	while (i != e) {
		if (i->second) {
			if (data = i->second->getData("NodeInfoPackStruct", size))
				infopacks[i->first] = new NodeInfoPack(data, true);
			delete(i->second);
		}
		i++;
	}
	replyMap.clear();
	return true;
}

bool Node::getOtherNodeCompCom(std::map<uint64, PerformanceMatrixResults>& nodeCom, std::map<uint64, PerformanceMatrixResults>& compCom) {

	std::map<uint16, DataMessage*> replyMap;
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
	msg->setReference(QUERYNODE_GETPERFMATRICES);
	if (networking->syncNodes(msg, NODESYNC_TIMEOUT_MS, true, &replyMap) != SYNC_SUCCESS)
		LogPrint(networking->localNodeID, LOG_NODE, 0,"Only got %u Performance Matrices back from other nodes...", replyMap.size());

	std::map<uint64, PerformanceMatrixResults> results;

	const char* data;
	uint32 size;
	std::map<uint16, DataMessage*>::iterator i = replyMap.begin(), e = replyMap.end();
	while (i != e) {
		if (i->second) {
			if (data = i->second->getData("PerformanceMatrixResultsNode", size)) {
				 results = PerformanceMatrix::GetThroughputResultsFromBinary(data, size);
				 nodeCom.insert(results.begin(), results.end());
				 results.clear();
			}
			if (data = i->second->getData("PerformanceMatrixResultsComponent", size)) {
				 results = PerformanceMatrix::GetThroughputResultsFromBinary(data, size);
				 compCom.insert(results.begin(), results.end());
				 results.clear();
			}
			delete(i->second);
		}
		i++;
	}
	replyMap.clear();
	return true;

}


bool Node::processMessage(DataMessage* msg) {
	if (!msg || !msg->isValid())
		return false;

	procCount++;
	procBytes += msg->data->size;

	uint64 now = GetTimeNow();
	uint64 eol = msg->getEOL();
	if (eol && (eol < now - 10000))
		return false;

	uint32 to;
	if (to = msg->getTo()) {
		toCount++;
		toBytes += msg->data->size;
		uint16 otherNodeID = 0;
		uint16 otherProcID = 0;
		if (manager->componentMemory->getComponentLocation(to, otherNodeID, otherProcID)) {
			if (otherNodeID != manager->getNodeID()) {
				if (networking->sendMessageToNode(otherNodeID, msg, false)) {
					monitoring->addCommunication(nodeID, otherNodeID, msg->data->from, to, msg->data->size, 1, now);
					//LogPrint(nodeID, LOG_NETWORK, 3, "Posted msg directly to Node %u...", otherNodeID);
				//	printf("************ Node Network Direct from %u to %u on node %u (%lld us old)   %s...\n",
				//		msg->getFrom(), to, otherNodeID, GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
				}
				else {
					LogPrint(nodeID, LOG_NETWORK, 0, "Failed to post msg directly to Node %u...", otherNodeID);
				//	printf("************ Node Network Direct FAILED from %u to %u on node %u (%lld us old)   %s...\n",
				//		msg->getFrom(), to, otherNodeID, GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
				}
			}
			else {
				msg->setRecvTime(now);
				manager->processMemory->addToMsgQ(otherProcID, msg);
				monitoring->addCommunication(nodeID, otherNodeID, msg->data->from, to, msg->data->size, 1, now);
				//LogPrint(nodeID, LOG_NETWORK, 3, "Posted msg directly to Proc %u...", otherProcID);
				//printf("************ Node Direct from %u to %u (%lld us old)   %s...\n",
				//	msg->getFrom(), to, GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
			}
		}
		else
			LogPrint(nodeID, LOG_NETWORK, 0, "Failed to find destination component %u location...", to);
	}
	else {
		//printf("************ Node Indirect from %u to %u (%lld us old)   %s...\n",
		//	msg->getFrom(), to, GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
	}

//	int64 t = GetTimeAge(msg->getCreatedTime());
//	if ((t < 0) || (t > 100))
//		printf("************ Node Process %lld us old...\n", t);
//	printf("************ Node Process %lld us old...\n", GetTimeAge(msg->getCreatedTime()));

	PsyType type = msg->getType();
	//if (type[0] == 1) {
	//	LogPrint(0,0,0,"Processing control message type '%s' from %u to %u...", manager->typeToText(msg->getType()).c_str(), msg->getFrom(), msg->getTo());
	//}
	if (type == PsyAPI::CTRL_CONTEXT_CHANGE) {
		PsyContext newContext = msg->getContextChange();
		if (newContext.isValid())
			subscriptionEngine->processContext(newContext);
	}
	else if (type == shutdownType) {
		LogPrint(networking->localNodeID, LOG_NODE, 1, "*************************************");
		LogPrint(networking->localNodeID, LOG_NODE, 1, "********** SYSTEM SHUTDOWN **********");
		LogPrint(networking->localNodeID, LOG_NODE, 1, "*************************************");
		char* str = new char[1024];
		if (manager->componentMemory->getComponentName(msg->getFrom(), str, 1024))
			LogPrint(networking->localNodeID, LOG_NODE, 1,"Module %s requested system shutdown, shutting down...", str);
		else
			LogPrint(networking->localNodeID, LOG_NODE, 1,"Module %u requested system shutdown, shutting down...", msg->getFrom());
		delete [] str;
		return shutdown(3000);
	}

//	LogPrint(0,0,0,"Processing message type '%s' from %u to %u...", manager->typeToText(msg->getType()).c_str(), msg->getFrom(), msg->getTo());

	// ** moved to PsySpace instead
	// Add datamessage to memory
	//uint64 memID = 0;
	//if (eol) {
	//	if (!manager->insertMessage(msg, memID)) {
	//		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not add message from %u to shared memory...", msg->getFrom());
	//		return false;
	//	}
	//}

	// Push message through subscription engine. Use a stack-local list filled
	// in place to avoid a per-message heap allocation on this hot path (and the
	// leak the heap-returning variant had when no triggers matched).
	std::list<TriggerSpec*> triggersStorage;
	subscriptionEngine->getTriggersForMsg(msg, triggersStorage);
	std::list<TriggerSpec*>* triggers = &triggersStorage;

	int8 selfTrigger = -1;
	uint32 from = msg->getFrom();
	if (triggers && triggers->size()) {
		std::list<TriggerSpec*>::iterator it, itEnd;
		for (it = triggers->begin(), itEnd = triggers->end(); it != itEnd; it++) {
			if (from == (*it)->componentID) {
				selfTrigger = (int8)manager->componentMemory->getComponentSelfTrigger(from);
				if (selfTrigger != SELFTRIGGER_ALLOW) {
					char* name = new char[1024];
					if (selfTrigger == SELFTRIGGER_NONE) {
						if (manager->componentMemory->getComponentName(from, name, 1024))
							LogPrint(networking->localNodeID, LOG_NODE, 1, "Module %s trigger %s on own post is not allowed...", name, (*it)->name);
						else
							LogPrint(networking->localNodeID, LOG_NODE, 1, "Module ID %u trigger %s on own post is not allowed...", from, (*it)->name);
					}
					else {
						if (manager->componentMemory->getComponentName(from, name, 1024))
							LogPrint(networking->localNodeID, LOG_NODE, 1, "Warning: Module %s trigger %s on own post, use with care...", name, (*it)->name);
						else
							LogPrint(networking->localNodeID, LOG_NODE, 1, "Warning: Module ID %u trigger %s on own post, use with care...", from, (*it)->name);
					}
					delete[] name;
				}
			}
			if ( (selfTrigger < 0) || (selfTrigger > SELFTRIGGER_NONE)) {
				deliverTriggerMessage(*it, eol ? (uint32)(eol - now) : 0, msg, now);
				deliverCount++;
				deliverBytes += msg->data->size;
			}
		}
	}
	else {
		LogPrint(nodeID, LOG_NODE, 3, "Found no triggers for msg from %u...\n%s\n---\n%s\n",
			msg->getFrom(),
			//subscriptionEngine->printFriendly().c_str(),""
			subscriptionEngine->printFriendlyTriggersForMsg(msg).c_str(), ""
			);
	}
	return true;
}

// T1.6: deliver one joined <triggergroup> activation. Carries the full member map
// (trigger name -> message) via setAttachedMessageMap, plus a "Message" primary for
// back-compat with cranks that still call waitForNewMessage(). Node-local delivery:
// the join completed on this node, so the owning component's PsySpace is local.
bool Node::deliverTriggerGroup(uint32 componentID, uint16 crankID, const char* groupName,
		std::map<std::string, DataMessage*>& members, DataMessage* primary, uint64 recvtime, TriggerSpec* templateSpec) {
	uint16 nID = 0, procID = 0;
	if (!manager->componentMemory->getComponentLocation(componentID, nID, procID))
		return false;
	if (nID != nodeID || !procID) {
		LogPrint(networking->localNodeID, LOG_NODE, 1, "T1.6: join group '%s' component not local; cannot deliver grouped activation...", groupName ? groupName : "");
		return false;
	}
	// Synthetic TriggerSpec so PsySpace dispatch resolves crankID + group name uniformly.
	// Clone a real member TriggerSpec when available so the module's post specs (packed
	// in the spec tail) travel with the activation — otherwise a crank consuming the join
	// cannot postOutputMessage() (getPostSpecs reads them off the current trigger spec).
	TriggerSpec* ts = NULL;
	uint32 tsize;
	if (templateSpec && templateSpec->size >= sizeof(TriggerSpec)) {
		tsize = templateSpec->size;
		ts = (TriggerSpec*) malloc(tsize);
		memcpy(ts, templateSpec, tsize);
	} else {
		tsize = sizeof(TriggerSpec);
		ts = (TriggerSpec*) malloc(tsize);
		ts->reset(tsize);
	}
	ts->componentID = componentID;
	ts->crankID = crankID;
	if (primary) ts->type = primary->getType();
	utils::strcpyavail(ts->name, groupName ? groupName : "", MAXKEYNAMELEN, true);

	uint16 policy = primary ? primary->getPolicy() : 0;
	DataMessage* dmsg = new DataMessage(PsyAPI::CTRL_TRIGGER_GROUP, 0, componentID, 0, recvtime, 0, policy);
	dmsg->setData("TriggerSpec", (char*)ts, ts->size);
	dmsg->setAttachedMessageMap("Members", members);
	if (primary) dmsg->setAttachedMessage("Message", primary);
	if (primary) {
		dmsg->setSendTime(primary->getSendTime());
		dmsg->setRecvTime(recvtime);
	}
	bool ok = manager->processMemory->addToMsgQ(procID, dmsg);
	if (ok)
		monitoring->addCommunication(nodeID, nID, primary ? primary->data->from : 0, componentID, primary ? primary->data->size : 0, 1, recvtime);
	else {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "T1.6: could not deliver grouped activation for '%s' to component queue (PsySpace %u)...", groupName ? groupName : "", procID);
		delete(dmsg);
	}
	free(ts);
	return ok;
}

bool Node::deliverTriggerMessage(TriggerSpec* spec, uint64 ttl, DataMessage* msg, uint64 recvtime) {
	// T1.6: if this trigger belongs to a registered <triggergroup>, route the message
	// into the JoinEngine buffer instead of enqueuing it now. The crank is activated
	// once per completed set (the JoinEngine calls back here with a synthetic non-group
	// spec, so there is no re-entrancy). Non-group triggers fall through unchanged.
	if (spec->group && joinEngine && joinEngine->isJoinGroup(spec->group)) {
		return joinEngine->routeMessage(spec, msg, recvtime);
	}
	uint16 nID = 0, procID = 0;
	DataMessage* dmsg = new DataMessage(PsyAPI::CTRL_TRIGGER, 0, spec->componentID, 0, recvtime, ttl, msg->getPolicy());
	dmsg->setData("TriggerSpec", (char*)spec, spec->size);
	dmsg->setAttachedMessage("Message", msg);
	if (!manager->componentMemory->getComponentLocation(spec->componentID, nID, procID))
		return false;
	if (nID == nodeID) {
		dmsg->setReference(msg->data->memid);
		dmsg->setSendTime(msg->getSendTime());
		dmsg->setRecvTime(recvtime);
		if (procID) { 
			if (spec->delay) {
				dmsg->setDestination(procID);
				delayedLocalMsgs.add(recvtime + (spec->delay * 1000), dmsg);
				monitoring->addCommunication(nodeID, nID, msg->data->from, spec->componentID, msg->data->size, 1, recvtime);
				//LogPrint(networking->localNodeID, LOG_NODE, 0, "Delaying message delivery by %ums...", spec->delay);
				return true;
			}
			else {
				//printf("************ Node Process adding %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
				if (manager->processMemory->addToMsgQ(procID, dmsg))
					monitoring->addCommunication(nodeID, nID, msg->data->from, spec->componentID, msg->data->size, 1, recvtime);
				else
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not deliver local message to component '%s' queue (PsySpace %u) from '%s'...",
						manager->componentMemory->getComponentNameString(spec->componentID).c_str(), procID,
						manager->componentMemory->getComponentNameString(msg->getFrom()).c_str());
				//else
				//	LogPrint(networking->localNodeID, LOG_NODE, 3, "Delivered local msg from %u to component %u queue...", msg->getFrom(), spec->componentID);
				//	printf("************ Node Deliver from %u to %u (%lld us old)   %s...\n",
				//		msg->getFrom(), spec->componentID, GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
			}
		}
		else
			LogPrint(networking->localNodeID, LOG_NODE, 3, "Could not find the PsySpace for component '%s' for incoming local message from '%s'...",
				manager->componentMemory->getComponentNameString(spec->componentID).c_str(),
				manager->componentMemory->getComponentNameString(msg->getFrom()).c_str());
		delete(dmsg);
	}
	else {
	//	uint64 t = msg->getCreatedTime();
	//	uint64 start = GetTimeNow();
		if (spec->delay) {
			dmsg->setDestination(nID);
			delayedNetworkMsgs.add(recvtime + (spec->delay * 1000), dmsg);
			monitoring->addCommunication(nodeID, nID, msg->data->from, spec->to, msg->data->size, 1, recvtime);
			return true;
		}
		else {
			if (networking->sendMessageToNode(nID, dmsg, true))
				monitoring->addCommunication(nodeID, nID, msg->data->from, spec->to, msg->data->size, 1, recvtime);
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not send message from '%s' to '%s' on PsyNode %u...",
					manager->componentMemory->getComponentNameString(spec->componentID).c_str(),
					manager->componentMemory->getComponentNameString(msg->getFrom()).c_str(),
					nID);
			//else
			//	LogPrint(networking->localNodeID, LOG_NODE, 3, "Sent message from %u to Node %u...", msg->getFrom(), nID);
			//else
			//	printf("************ Node Network Deliver from %u to %u (%lld us old)...\n",
			//		msg->getFrom(), spec->componentID, GetTimeAge(msg->getCreatedTime()));
		//	printf("*** Net Msg Send    start %lld     send %lld...\n", start-t, GetTimeNow()-start);
	//		printf("************ Node Net Send %lld us old...\n", GetTimeAge(msg->getCreatedTime()));
		}
	}
	return true;
}

DataMessage* Node::getCopyOfMessage(uint64 id) {
	DataMessage* msg;
	uint16 msgNode = GETMSGNODE(id);
	if (msgNode != nodeID) {
		DataMessage* queryMsg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
		queryMsg->setReference(QUERYNODE_GETMESSAGECOPY);
		queryMsg->setMemID(id);
		DataMessage* replyMsg = NULL;
		if (networking->queryNode(msgNode, queryMsg, NODEQUERY_TIMEOUT_MS, true, &replyMsg) != SYNC_SUCCESS) {
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Didn't get Message Copy back from Node %u...", msgNode);
			delete(replyMsg);
			return NULL;
		}

		if (!(msg = replyMsg->getAttachedMessageCopy("Message"))) {
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Message Copy reply back from Node %u didn't contain any data...", msgNode);
			delete(replyMsg);
			return NULL;
		}
		delete(replyMsg);
	}
	else
		msg = manager->getCopyOfMessage(id);
	return msg;
}

bool Node::runMessageQ() {
	DataMessage* msg;
	isMsgQRunning = true;
	uint32 c = 0;
	while (shouldContinue) {
		if (msg = manager->processMemory->waitForMsgQ(0, 200)) {
			if ( (manager->processMemory->getMsgQCount(0) > 100) && (msg->getPolicy() & MESSAGE_NON_GUARANTEED) ) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Non guaranteed msg from %u deleted due to large queue size...", msg->getFrom());
				delete(msg);
			}
			else {
			//	printf("[%u] ************ Node Run Msg %lld us old   %s\n", c++, GetTimeAge(msg->getCreatedTime()), PrintTimeString(msg->getCreatedTime()).c_str());
				processMessage(msg);
				//printf("[%llu us]\n", GetTimeAge(time));
				delete(msg);
			}
		}
	}
	isMsgQRunning = false;
	return true;
}

bool Node::runSignalQ() {
	DataMessage* msg;
	isSigQRunning = true;
	while (shouldContinue) {
		// Deliver to local processes
		if (msg = manager->processMemory->waitForSigQ(0, 200)) {
		//	printf("************ Node before alltoall Signal %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
		// Don't send to local procs, only other nodes
		//	manager->processMemory->addToAllSignalQs(msg);
		//	printf("************ Node after alltoall Signal %llu us old...\n", GetTimeAge(msg->getCreatedTime()));
			// Send to other nodes
			DataMessage* smsg = new DataMessage(Networking::CTRL_NODE_SIGNAL, networking->localNodeID);
			smsg->setAttachedMessage("Message", msg);
			networking->sendMessageToAllOtherNodes(smsg, true);
			delete(msg);
		}
	}
	isSigQRunning = false;
	return true;
}

DataMessage* Node::processQueryMsg(DataMessage* msg) {
	DataMessage* replyMsg = new DataMessage(Networking::CTRL_NODE_QUERY_FAILED, nodeID);
	if (!msg)
		return replyMsg;

	uint16 remoteNodeID = msg->getFrom();

	const char* data;
	uint64 size;
	uint32 size32;
	int64 intVal;

	NodeInfoPackHeader* infopack;
	char* nodeMatrixResults;
	char* compMatrixResults;
	std::map<std::string, std::string> map;
	std::string str;

	switch(msg->getReference()) {
		case QUERYNODE_GETINFOPACK:
			infopack = getLocalInfoPack();
			if (infopack) {
				replyMsg->setType(Networking::CTRL_NODE_QUERY_SUCCESS);
				replyMsg->setData("NodeInfoPackStruct", (char*)infopack, infopack->size);
				//LogPrint(0, 0, 0, "Received ID SYNC %u, success to Node %u...", msg->getTag(), remoteNodeID);
			}
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETINFOPACK request %u, failed to create for Node %u...", msg->getTag(), remoteNodeID);
			break;
		case QUERYNODE_GETCOMPDATA:
			if ((msg->getInt("CompID", intVal)) &&
				(data = manager->componentMemory->getComponentData((uint32)intVal, size)) && (size <= MAXVALUINT32)) {
				replyMsg->setType(Networking::CTRL_NODE_QUERY_SUCCESS);
				replyMsg->setData("ComponentData", data, (uint32)size);
				delete [] data;
			}
			else if (size > MAXVALUINT32)
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETCOMPDATA request %u, component data too big %s for Node %u...", msg->getTag(), utils::BytifySize((double)size).c_str(), remoteNodeID);
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETCOMPDATA request %u, failed to create for Node %u...", msg->getTag(), remoteNodeID);
			break;
		case QUERYNODE_SETCOMPDATA:
			if (msg->getInt("CompID", intVal) && (data = msg->getData("ComponentData", size32)) &&
				(manager->componentMemory->setComponentData((uint32)intVal, data, size32, true)) ) {
				replyMsg->setType(Networking::CTRL_NODE_QUERY_SUCCESS);
			}
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETCOMPDATA request %u, failed to create for Node %u...", msg->getTag(), remoteNodeID);
			break;
		case QUERYNODE_GETCOMPPRIVATEDATA:
			const char* dataName;
			if (msg->getInt("CompID", intVal) && (dataName = msg->getString("DataName")) &&
				(data = manager->componentMemory->getPrivateDataCopy((uint32)intVal, dataName, size)) ) {
				str = manager->componentMemory->getPrivateDataMimetype((uint32)intVal, dataName);
				replyMsg->setString("Mimetype", str.c_str());
				replyMsg->setType(Networking::CTRL_NODE_QUERY_SUCCESS);
				replyMsg->setData("ComponentPrivateData", data, (uint32)size);
				delete [] data;
			}
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETCOMPPRIVATEDATA request %u, failed to create for Node %u...", msg->getTag(), remoteNodeID);
			break;
		case QUERYNODE_GETCOMPPRIVATEKEYSTYPES:
			if (msg->getInt("CompID", intVal)) {
				map = manager->componentMemory->getPrivateDataKeysAndTypes((uint32)intVal);
				replyMsg->setType(Networking::CTRL_NODE_QUERY_SUCCESS);
				replyMsg->setStringMap("ComponentPrivateDataKeysAndTypes", map);
			}
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETCOMPPRIVATEKEYSTYPES request %u, failed to create for Node %u...", msg->getTag(), remoteNodeID);
			break;
		case QUERYNODE_GETMESSAGECOPY:
			DataMessage* msgCopy;
			if (msgCopy = manager->getCopyOfMessage(msg->getMemID())) {
				replyMsg->setType(Networking::CTRL_NODE_QUERY_SUCCESS);
				replyMsg->setAttachedMessage("Message", msgCopy);
				delete(msgCopy);
			}
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Received QUERYNODE_GETCOMPPRIVATEDATA request %u, failed to create for Node %u...", msg->getTag(), remoteNodeID);
			break;
		case QUERYNODE_GETPERFMATRICES:
			uint32 size1, size2;
			nodeMatrixResults = monitoring->getNodeMatrixBinary(size1, msg->data->time);
			compMatrixResults = monitoring->getCompMatrixBinary(size2, msg->data->time);

			if (nodeMatrixResults && size1)
				replyMsg->setData("PerformanceMatrixResultsNode", nodeMatrixResults, size1);
			if (compMatrixResults && size2)
				replyMsg->setData("PerformanceMatrixResultsComponent", compMatrixResults, size2);

			delete [] nodeMatrixResults;
			delete [] compMatrixResults;
			break;
		default:	
			break;
	}

	return replyMsg;
}

bool Node::processLocalRequest(DataMessage* msg, DataMessage** resultMsg) {
	if (!msg || !resultMsg)
		return false;
	*resultMsg = NULL;

	int64 otherNodeID, compID, localProcID;
	const char* data;

	PsyType type = msg->getType();

	if (type == PsyAPI::CTRL_PULLCOMPONENTDATA) {
		if ( msg->getInt("CompID", compID) && msg->getInt("NodeID", otherNodeID) && msg->getInt("ProcID", localProcID) ) {
			// Move CompData from other node to local node
			return migrateComponentData((uint32)compID, (uint16)otherNodeID, this->nodeID, (uint16)localProcID);
		}
	}
	else if (type == PsyAPI::CTRL_ADDSUBSCRIPTION) {
		if (data = msg->getString("Subscription")) {

			std::list<TriggerSpec*> triggers;
			std::list<RetrieveSpec*> retrieves;
			std::list<QuerySpec*> queries;
			std::list<PostSpec*> posts;
			std::list<SignalSpec*> signals;
			std::list<uint16> cranks;

			XMLResults xmlResults;
			XMLNode mainNode = XMLNode::parseString(data, "subscription", &xmlResults);

			// display error message (if any)
			if (xmlResults.error != eXMLErrorNone) {
				if (xmlResults.error == eXMLErrorFirstTagNotFound) {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Subscription format error, main tag '<subscription>' not found");
					return false;
				}
				else {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Subscription parsing error at line %i, column %i:\n   %s\n",
						xmlResults.nLine, xmlResults.nColumn, XMLNode::getError(xmlResults.error));
					return false;
				}
			}
			if (mainNode.isEmpty())
				return false;

			if (!(data = mainNode.getAttribute("name"))) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Subscription format error, name not found");
				return false;
			}

			ComponentSpec* spec = new ComponentSpec();
			//utils::strcpyavail(spec->setup->name, data, MAXKEYNAMELEN, true);

			uint32 compID = 0;
			if (!manager->componentMemory->getComponentID(data, compID) || !compID) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Subscription error, unknown component '%s'", data);
				return false;
			}

			spec->setup->componentID = compID;

			XMLNode node;
			uint32 n, count = mainNode.nChildNode();
			for (n = 0; n < count; n++) {
				node = mainNode.getChildNode(n);
				if (!getTriggersFromXML(node, spec, defaultContext, triggers, retrieves, queries, posts, signals, cranks)) {
					delete(spec);
					return false;
				}
			}

			if (triggers.size() > 0) {
				// Add remaining triggers, cranks, etc. to default context
				addTriggersToComponent(spec, triggers, retrieves, queries, posts, signals, cranks);

				// Send to local Subscription Engine
				if (!subscriptionEngine->addSubscription(compID, spec->triggers)) {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not add Module subscription '%s'...", data);
					return false;
				}

				//	printf("*** Adding Timers for '%s'...\n", name);

				// Check for any interval triggers
				std::list<TriggerSpec*>::iterator it, itEnd;
				for (it = spec->triggers.begin(), itEnd = spec->triggers.end(); it != itEnd; it++) {
					if ((*it)->interval) {
						messageTimer->addTrigger(*it);
					}
				}

				// Add PsyProbe custom view to all PsyProbes
				std::list<CustomView>::iterator ic = spec->customViews.begin(), ec = spec->customViews.end();
				while (ic != ec) {
					psyProbe->addCustomView(spec->setup->componentID, ic->name, ic->templateURL);
					ic++;
				}

				// Send to all other nodes
				if (!syncComponentTriggers(compID, spec->triggers)) {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not sync Module subscription '%s'...", data);
					return false;
				}
				delete spec;
				return true;
			}
			// Subscription parsed cleanly but queued no new triggers
			// (e.g. context/post-only or idempotent re-subscription).
			// This is a success, not a failure: return true so the client
			// PsyAPI::addSubscription() call does not spuriously report false.
			delete spec;
			return true;
		}
	}

	return false;
}

bool Node::migrateComponentData(uint32 compID, uint16 fromNode, uint16 toNode, uint16 toProc) {
	
	const char* data = NULL;
	uint64 size = 0;
	uint32 size32 = 0;
	DataMessage* replyMsg, *msg;
	
	// Update component location on all nodes
	this->syncComponentLocation(compID, toNode, toProc);
	
	if (fromNode == toNode)
		return true;
	
	// request a copy of the component data
	if (fromNode == this->nodeID) {
		if (!(data = manager->componentMemory->getComponentData(compID, size))) {
			return false;
		}
	}
	else {
		msg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
		msg->setReference(QUERYNODE_GETCOMPDATA);
		msg->setInt("CompID", compID);
		if (networking->queryNode(fromNode, msg, NODEQUERY_TIMEOUT_MS, true, &replyMsg) != SYNC_SUCCESS) {
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Didn't get Component Data Pack back from Node %u...", fromNode);
			delete(replyMsg);
			return false;
		}
		if (!(data = replyMsg->getDataCopy("ComponentData", size32))) {
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Component Data Pack reply back from Node %u didn't contain any data...", fromNode);
			delete(replyMsg);
			return false;
		}
		size = size32;
	}
	delete(replyMsg);

	if (toNode == this->nodeID) {
		// update local copy of component data
		if (!manager->componentMemory->setComponentData(compID, data, size, true)) {
			delete [] data;
			return false;
		}
	}
	else {
		size32 = size ? (uint32)size : size32;
		// update remote copy of component data
		msg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
		msg->setReference(QUERYNODE_SETCOMPDATA);
		msg->setInt("CompID", compID);
		msg->setData("ComponentData", data, size32);
		if (networking->queryNode(toNode, msg, NODEQUERY_TIMEOUT_MS, true, &replyMsg) != SYNC_SUCCESS) {
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Didn't get Component Data Pack back from Node %u...", fromNode);
			delete [] data;
			delete(replyMsg);
			return false;
		}
		delete(replyMsg);
	}
	delete [] data;
	return true;
}

bool Node::runRequestQ() {
	DataMessage* msg, *resultMsg;
	uint16 otherNode;
	PsyType type;
	uint32 to, reqID;
	uint64 msgID, msgEOL;
	uint8 reqStatus;
	uint32 remoteID;
	uint32 origin;
	uint32 source;
	bool result;
	isReqQRunning = true;
	while (shouldContinue) {
		if (msg = manager->processMemory->waitForReqQ(0, 200)) {
			type = msg->getType();
			if (type == PsyAPI::CTRL_INTERSYSTEM_QUERY) {
				reqID = (uint32)msg->getReference();
				if (!interSystemManager->sendQuery(msg)) {
					manager->dataMapsMemory->setRequestStatus(reqID, REQ_FAILED_TO_SEND, 0, 0);
				}
			}
			else if (type == PsyAPI::CTRL_INTERSYSTEM_QUERY_REPLY) {
				if (!interSystemManager->sendQueryReply(msg)) {
					// Not much we can do about this as we cannot talk to the other end
				}
			}

			else if (!(to = msg->getTo())) {
				// This is a query from a local PsySpace
				reqID = (uint32) msg->getReference();
				if (manager->dataMapsMemory->getRequestInfo(reqID, origin, source, remoteID)) {
					result = processLocalRequest(msg, &resultMsg);
					msgID = 0;
					msgEOL = 0;
					if (resultMsg) {
						resultMsg->setTTL(10000000);
						msgEOL = resultMsg->getEOL();
						// decide on EOL ###################
						manager->temporalMemory->insertMessage(resultMsg, msgID);
					}
					if (result) {
						if (msgID)
							reqStatus = REQ_SUCCESS_DATA_EOL;
						else
							reqStatus = REQ_SUCCESS;
					}
					else {
						if (msgID)
							reqStatus = REQ_FAILED_DATA_EOL;
						else
							reqStatus = REQ_FAILED;
					}
					manager->dataMapsMemory->setRequestStatus(reqID, reqStatus, msgID, msgEOL);
				}
			}
			else if (otherNode = manager->componentMemory->getComponentNodeID(to)) {
				if (type == PsyAPI::CTRL_QUERY)
					networking->sendMessageToNode(otherNode, msg, false);
				else if (type == PsyAPI::CTRL_QUERY_REPLY)
					networking->sendMessageToNode(otherNode, msg, false);
			}
			// Process Request Message
			// #######################
			delete(msg);
		}
	}
	isReqQRunning = false;
	return true;
}




uint8 Node::retrieve(std::list<DataMessage*> &result, RetrieveSpec* spec, uint32 timeout) {
	if (!spec)
		return QUERY_FAILED;
	spec->origin = 0;

	DataMessage* msg = new DataMessage(PsyAPI::CTRL_QUERY, 0, spec->source, timeout);
	msg->setData("RetrieveSpec", (char*)spec, sizeof(RetrieveSpec));

	DataMessage* resultMsg = NULL;
	uint8 status = query(msg, &resultMsg, timeout);
	delete(msg);

	if (!resultMsg) {
		if (status == QUERY_FAILED)
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Query failed and did not return a message...");
		return status;
	}

	uint32 size;
	int64 count;
	const char* data = resultMsg->getData("ReplyData", size);
	if (!data || !resultMsg->getInt("ReplyCount", count)) {
		if (!data)
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Query did not return any ReplyData...");
		else
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Query did not return any ReplyCount...");
		delete(resultMsg);
		return QUERY_FAILED;
	}

	DataMessage* subMsg;
	const char* src = data;
	for (uint32 n=0; n<(uint32)count; n++) {
		if (GetObjID(src) != DATAMESSAGEID)
			break;
		subMsg = new DataMessage(src);
		result.push_back(subMsg);
		src += subMsg->getSize();
	}

	delete(spec);
	delete(resultMsg);
	return QUERY_SUCCESS;
}

uint8 Node::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) {
		// ###### Should just call the function directly without adding to the queue
		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()) {
			// ###### Should just send out via network directly without adding to the queue
			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(networking->localNodeID, LOG_NODE, 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(networking->localNodeID, LOG_NODE, 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(networking->localNodeID, LOG_NODE, 0,"Couldn't getCopyOfMessage reply from node %u (req %u from comp %u)...", otherNodeID, reqID, msg->getFrom());
		return QUERY_FAILED;
	}

	return QUERY_SUCCESS;
}




int8 Node::run() {
	isNodeRunning = true;
	uint32 s;
	DataMessage* msg;
	int32 idCheckInterval = 30000;
	uint64 lastIDCheck = GetTimeNow() - idCheckInterval;
	uint64 time = 0;
	std::list<ProcessInfoStruct> procIssues;
	std::list<ProcessInfoStruct>::iterator i, e = procIssues.end();

	utils::GetProcessCPUTicks(currentCPUTicks);

	// Check queues:
	// CmdQ: commands, logging
	// MsgQ:
	// SigQ:
	// RegQ:

	std::list<NodeEntry> nodeIssues;
	std::list<NodeEntry>::iterator ni, ne;
	uint16 minStatus;

	uint16 localNodeStatus;
	while (shouldContinue && !shouldRestart) {

		localNodeStatus = networking->getLocalNodeStatus();
		if (networking && (localNodeStatus < NODESTATUS_READY)) {
			if (configNodeMessage) {
				LogPrint(nodeID, LOG_NODE, 3, "Starting node configuration...");
				if (configNode(configNodeMessage->getString("Config"), false)) {
					LogPrint(nodeID, LOG_NODE, 3, "Node configuration success...");
					msg = new DataMessage(Networking::CTRL_NODE_CONFIG_SUCCESS, nodeID);
					networking->setLocalNodeStatus(NODESTATUS_READY);
					networking->sendHeartbeat();
				}
				else {
					LogPrint(nodeID, LOG_NODE, 0, "Node configuration failed...");
					msg = new DataMessage(Networking::CTRL_NODE_CONFIG_FAILED, nodeID);
					msg->setString("ErrorText", "Please see node console/log output");
					networking->setLocalNodeStatus(NODESTATUS_ERROR);
				}
				networking->sendMessageToNode(configNodeMessage->getOrigin(), msg, true);
				delete(configNodeMessage);
				configNodeMessage = NULL;
			}
		}

		if (networking && (localNodeStatus == NODESTATUS_READY)) {
			// Check if any nodes have reconnected and need rejoining...
			if (networking->localNodeEntry->type == NODE_TIMENODE) {
				nodeIssues.clear();
				minStatus = networking->getNodeIssues(nodeIssues);
				if (minStatus < NODESTATUS_NETREADY) {
					if (!networking->waitForAllNodesStatus(NODESTATUS_NETREADY, 50)) {
						ni = nodeIssues.begin();
						ne = nodeIssues.end();
						while (ni != ne) {
							if ((*ni).status < NODESTATUS_NETREADY)
								LogPrint(nodeID, LOG_NODE, 0, "Node '%s' (%u) has reconnected, but not ready to rejoin yet (status %u)...", (*ni).name, (*ni).nodeID, (*ni).status);
							ni++;
						}
					}
				}
				else if (minStatus < NODESTATUS_READY) {
					ni = nodeIssues.begin();
					ne = nodeIssues.end();
					while (ni != ne) {
						if ((*ni).status == NODESTATUS_NETREADY) {
							LogPrint(nodeID, LOG_NODE, 0, "Node '%s' (%u) has reconnected, network ready, continuing configuration...", (*ni).name, (*ni).nodeID);
							rejoinNode((*ni).nodeID);
						}
						else if ((*ni).status < NODESTATUS_READY) {
							LogPrint(nodeID, LOG_NODE, 0, "Node '%s' (%u) has reconnected, still being configured...", (*ni).name, (*ni).nodeID);
						}
						ni++;
					}
				}
			}

			if (GetTimeAgeMS(lastIDCheck) > idCheckInterval) {
				if (idManager)
					idManager->checkSync();
				else
					break;
				lastIDCheck = GetTimeNow();
			}
		}

		while (msg = delayedLocalMsgs.getNextEntryDue()) {
			if (!manager->processMemory->addToMsgQ(msg->getDestination(), msg))
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not deliver local delayed msg to proc %u queue...", msg->getDestination());
			delete(msg);
		}

		while (msg = delayedNetworkMsgs.getNextEntryDue()) {
			if (networking->sendMessageToNode(msg->getDestination(), msg, true))
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not deliver network delayed msg to Node %u...", msg->getDestination());
			delete(msg);
		}

		// time = GetTimeNow();
		if (msg = manager->processMemory->waitForCmdQ(0, 50)) {
			// Process Command Message
			if (msg->getType() == CTRL_LOGPRINT) {
				// ### should send to PsyProbe node? ###
				logging->logEntry((LogEntry*)msg->getDataCopy("LogEntry", s));
			}
			else if (msg->getType() == PsyAPI::CTRL_PROCESS_GREETING)
				processSpaceConnect((uint16)msg->getFrom());
			else if (msg->getType() == PsyAPI::CTRL_PROCESS_SHUTDOWN)
				processSpaceShutdown((uint16)msg->getFrom());
			else if (msg->getType() == PsyAPI::CTRL_CREATECUSTOMPAGE)
				addPsyProbeCustomView(msg->getFrom(), msg->getString("Name"), msg->getString("Template"));
			//printf("[%llu us]\n", GetTimeAge(time));
			delete(msg);
		}
	//	printf("***%llu us***\n", GetTimeAge(time));

		if (!manager->processMemory->checkProcessHeartbeats(5000, procIssues) && procIssues.size()) {
			i = procIssues.begin();
			while (i != e) {
				LogPrint(0,0,0,"Issues with space %u '%s'...", (*i).id, (*i).name);
				if ((*i).status == PSYPROC_TERMINATED)
					processSpaceShutdown((*i).id);
				else // heartbeat timeout reached
					processSpaceShutdown((*i).id);
					//processSpaceDisconnect((*i).id);
				// If this was a plain internal space that we spawned ourselves,
				// bring it back up: the whole point of a separate space process is
				// crash isolation, so a crashed space should not stay down.
				restartInternalSpace((*i).id);
				i++;
			}
			procIssues.clear();
		}
		
		utils::GetProcessCPUTicks(currentCPUTicks);
	}

	if (shouldRestart) {
		LogPrint(networking->localNodeID, LOG_NODE, 1, "Local Node is preparing to restart...");
		isNodeRunning = false;
		return NODE_EXIT_RESTART;
	}
	
	LogPrint(networking->localNodeID, LOG_NODE, 1, "Local Node is preparing to shutdown...");
	isNodeRunning = false;
	return NODE_EXIT_DONE;
}

// Subscriptions
ComponentSpec* Node::componentSpecFromXML(XMLNode &specNode) {
	ComponentSpec* spec;
	ComponentSetup* setup;
	uint8 type;

	if (specNode.isEmpty())
		return NULL;

	const char *str, *str2, *sname, *snode, *sspace, *stag, *stagname = specNode.getName();
	const char *pname, *svalue, *sinterval;
	int64 ivalue;
	float64 fvalue;
	uint8 ptype;
	uint8 selfTrigger = SELFTRIGGER_WARN;
	std::string strString;
	std::vector<std::string> values;
	const char *stype = specNode.getAttribute("type");
	XMLNode node, subNode;

	char* compName = new char[MAXKEYNAMELEN+1];

	uint8 policy = 0;

	if (stricmp(stagname, "whiteboard") == 0)
		type = COMP_WHITEBOARD;
	else if (stricmp(stagname, "catalog") == 0)
		type = COMP_CATALOG;
	else if (stricmp(stagname, "stream") == 0)
		type = COMP_STREAM;
	else if (stricmp(stagname, "service") == 0)
		type = COMP_SERVICE;
	else if (stricmp(stagname, "feed") == 0)
		type = COMP_FEED;
	else if (stricmp(stagname, "module") == 0) {
		type = COMP_INTERNAL;
		if (stype && (stricmp(stype, "external") == 0)) {
			type = COMP_EXTERNAL;
			policy = COMPONENT_CAN_MIGRATE;
		}
	}
	else {
		delete [] compName;
		return NULL;
	}

	if (!(sname = specNode.getAttribute("name"))) {
		delete [] compName;
		return NULL;
	}

	stag = specNode.getAttribute("tag");
	
	if (stag)
		snprintf(compName, MAXKEYNAMELEN+1, "%s_%s", sname, stag);
	else
		snprintf(compName, MAXKEYNAMELEN+1, "%s", sname);

	uint16 compNodeID = nodeID;
	snode = specNode.getAttribute("node");

	sspace = specNode.getAttribute("space");

	if (!sspace) {
		// Check if this module contains a scripted crank which needs its own space
		if (!(node = specNode.getChildNodeWithAttribute("crank", "language")).isEmpty()) {
			if (utils::stristr(node.getAttribute("language"), "python")) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Component '%s' should have its own space, creating space with the same name...", compName);
				sspace = compName;
			}
		}
	}

	if (str = specNode.getAttribute("selftrigger")) {
		if (stricmp(str, "allow") == 0)
			selfTrigger = SELFTRIGGER_ALLOW;
		else if (stricmp(str, "warn") == 0)
			selfTrigger = SELFTRIGGER_WARN;
		else
			selfTrigger = SELFTRIGGER_NONE;
	}

	uint16 spaceID = 0;
	if (type != COMP_EXTERNAL) {
		if (!sspace || strlen(sspace) == 0) {
			if (!createSpace("Root", spaceID)) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create root space for component '%s'...", compName);
				delete [] compName;
				return NULL;
			}
		}
		else if (!createSpace(sspace, spaceID)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create space '%s' for component '%s'...", sspace, compName);
			delete [] compName;
			return NULL;
		}
	}


//	uint32 existingID;
	uint32 compID = syncComponentID(type, policy, selfTrigger, compName, compNodeID, spaceID);
	if (!compID) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not register component '%s' with node network...", compName);
		delete [] compName;
		return NULL;
	}

	//if (!manager->componentMemory->createComponent(compID, type, policy, compName, 1024, compNodeID, spaceID, 0, existingID)) {
	//	if (existingID)
	//		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create shared memory component data for '%s' (ID %u already exists as ID %u)...", compName, compID, existingID);
	//	else
	//		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create shared memory component data for '%s'...", compName);
	//	delete [] compName;
	//	return NULL;
	//}

			//uint32 test;
			//if ( (test = getComponentID(compName)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "1 Component '%s' not created correctly (%u != %u)\n%s...", compName, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return false;
			//}
	
			//if ( (test = getComponentID(compName)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "2 Component '%s' not created correctly (%u != %u)\n%s...", compName, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return false;
			//}

	spec = new ComponentSpec();
	setup = spec->setup;
	utils::strcpyavail(setup->name, compName, MAXKEYNAMELEN, true);
	setup->componentID = compID;
	delete [] compName;

	setup->spaceID = spaceID;

			//if ( (test = getComponentID(setup->name)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "3 Component '%s' not created correctly (%u != %u)\n%s...", setup->name, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return NULL;
			//}
	if (stag) {
		if (strcmp(stag, "*") == 0)
			setup->tag = getTagID(setup->name);
		else
			setup->tag = getTagID(stag);
	}

	// All component types
	int16 moduleVerbose = -1, moduleDebug = -1;
	if (str = specNode.getAttribute("logfile"))
		utils::strcpyavail(setup->logfile, str, MAXKEYNAMELEN, true);
	if (str = specNode.getAttribute("verbose"))
		setup->verbose = (uint8) utils::Ascii2Uint32(str), moduleVerbose = (int16)setup->verbose;
	if (str = specNode.getAttribute("debug"))
		setup->debug = (uint8) utils::Ascii2Uint32(str), moduleDebug = (int16)setup->debug;
	// Per-module logging: wire the module's own verbose/debug level and/or
	// separate logfile into the Node's central logging
	if (logging && ((moduleVerbose >= 0) || (moduleDebug >= 0) || strlen(setup->logfile)))
		logging->setModuleLogging(setup->componentID, setup->name, moduleVerbose, moduleDebug, setup->logfile);
	if (str = specNode.getAttribute("migration"))
		setup->migration = (stricmp(str, "yes") == 0) ? COMP_MIGRATE_ALLOW : COMP_MIGRATE_NO;
	if (str = specNode.getAttribute("priority"))
		setup->priority = (uint8) utils::Ascii2Uint32(str);

			//if ( (test = getComponentID(setup->name)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "4 Component '%s' not created correctly (%u != %u)\n%s...", setup->name, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return NULL;
			//}
	switch(type) {
	case COMP_WHITEBOARD:
	case COMP_STREAM:
	case COMP_CATALOG:
	case COMP_SERVICE:
	case COMP_FEED:
		if ( (str = specNode.getAttribute("function")) ||
			 (str = specNode.getAttribute("type")) ) {
			if (!strchr(str, ':'))
				snprintf(setup->function, MAXKEYNAMELEN, "PsySystem::%s", str);
			else
				utils::strcpyavail(setup->function, str, MAXKEYNAMELEN, true);
		}
		else if (type == COMP_WHITEBOARD) {
			utils::strcpyavail(setup->function, "PsySystem::Whiteboard", MAXKEYNAMELEN, true);
		}
		else if (type == COMP_STREAM) {
			utils::strcpyavail(setup->function, "PsySystem::Stream", MAXKEYNAMELEN, true);
		}
		else {
			LogPrint(networking->localNodeID, LOG_NODE, 0,"Component '%s' didn't specify a valid function", setup->name);
			delete(spec);
			return NULL;
		}
		if (str = specNode.getAttribute("root")) {
			utils::strcpyavail(setup->root, str, MAXKEYNAMELEN, true);
			manager->componentMemory->createParameter(compID, "root", str);
		}
		if (str = specNode.getAttribute("subdir")) {
			utils::strcpyavail(setup->subdir, str, MAXKEYNAMELEN, true);
			manager->componentMemory->createParameter(compID, "subdir", str);
		}
		if (str = specNode.getAttribute("ext")) {
			utils::strcpyavail(setup->extension, str, MAXKEYNAMELEN, true);
			manager->componentMemory->createParameter(compID, "extension", str);
		}
		if (str = specNode.getAttribute("operation")) {
			utils::strcpyavail(setup->operation, str, MAXKEYNAMELEN, true);
			manager->componentMemory->createParameter(compID, "operation", str);
		}
		if ((str = specNode.getAttribute("key")) && (str2 = specNode.getAttribute("keytype"))) {
			if (!strlen(setup->keys))
				snprintf(setup->keys, MAXKEYNAMELEN, "%s=%s", str, str2);
			else
				snprintf(setup->keys + strlen(setup->keys), MAXKEYNAMELEN-strlen(setup->keys), ";%s=%s", str, str2);
			manager->componentMemory->createParameter(compID, "keys", setup->keys);
		}
		if (str = specNode.getAttribute("timeout")) {
			setup->timeout = utils::Ascii2Uint32(str);
			manager->componentMemory->createParameter(compID, "timeout", (int64)setup->timeout);
		}
		if (str = specNode.getAttribute("interval")) {
			setup->interval = utils::Ascii2Uint32(str);
			manager->componentMemory->createParameter(compID, "interval", (int64)setup->interval);
		}
		if (str = specNode.getAttribute("maxcount")) {
			setup->maxcount = utils::Ascii2Uint32(str);
			manager->componentMemory->createParameter(compID, "maxcount", (int64)setup->maxcount);
		}
		if (str = specNode.getAttribute("maxsize")) {
			setup->maxsize = utils::Ascii2Uint64(str);
			manager->componentMemory->createParameter(compID, "maxsize", (int64)setup->maxsize);
		}
		if (str = specNode.getAttribute("rotate")) {
			setup->rotate = (stricmp(str, "yes") == 0);
			manager->componentMemory->createParameter(compID, "rotate", str);
		}
		break;
	case COMP_INTERNAL:
	case COMP_EXTERNAL:
	//	if (str = specNode.getAttribute("group"))
	//		group = str;
		break;
	default:
		delete(spec);
		return NULL;
	}

	PsyContext context;
	std::list<TriggerSpec*> triggers;
	std::list<RetrieveSpec*> retrieves;
	std::list<QuerySpec*> queries;
	std::list<PostSpec*> posts;
	std::list<SignalSpec*> signals;
	std::list<uint16> cranks;

	if (strlen(setup->function)) {
		TriggerSpec* triggerSpec = (TriggerSpec*) malloc(sizeof(TriggerSpec));
		triggerSpec->reset(sizeof(TriggerSpec));
		triggerSpec->componentID = spec->setup->componentID;
		triggerSpec->context = defaultContext;
		utils::strcpyavail(triggerSpec->name, "DefaultComponentTrigger", MAXKEYNAMELEN, true);
		triggerSpec->type = systemReadyType;
		triggers.push_back(triggerSpec);
	}

			//if ( (test = getComponentID(setup->name)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "5 Component '%s' not created correctly (%u != %u)\n%s...", setup->name, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return NULL;
			//}

	CustomView customView;
	ExternalExecSpec execSpec;		// T1.4: external executable auto-start
	bool hasExecutable = false;
	uint32 n, count = specNode.nChildNode();
	for(n=0; n<count; n++){
		node=specNode.getChildNode(n);
		if (stricmp("recording", node.getName()) == 0) {
			if (str = node.getAttribute("root"))
				utils::strcpyavail(setup->root, str, MAXKEYNAMELEN, true);
			if (str = node.getAttribute("filesize"))
				setup->maxsize = utils::Ascii2Uint32(str);
			else
				setup->maxsize= 0;
			if (str = node.getAttribute("overwrite"))
				setup->rotate = (stricmp(str, "yes") == 0);
			else
				setup->rotate = true;
			manager->componentMemory->createParameter(compID, "componentrecordingroot", setup->root);
			manager->componentMemory->createParameter(compID, "componentrecordingsize", (int64)setup->maxsize);
			if (setup->rotate)
				manager->componentMemory->createParameter(compID, "componentrecordingoverwrite", "yes");
		}
		else if (stricmp("playback", node.getName()) == 0) {
			if (str = node.getAttribute("root"))
				utils::strcpyavail(setup->root, str, MAXKEYNAMELEN, true);
			if (str = node.getAttribute("interval"))
				setup->interval = utils::Ascii2Uint32(str);
			else
				setup->interval = 0;
			manager->componentMemory->createParameter(compID, "componentplaybackroot", setup->root);
			manager->componentMemory->createParameter(compID, "componentplaybackinterval", (int64)setup->interval);
		}
		else if (stricmp("executable", node.getName()) == 0) {
			// T1.4: full parse of <executable> options; see Docs/T1.4-external-autostart-spec.md
			hasExecutable = true;
			if (str = node.getText()) {
				utils::strcpyavail(setup->cmdline, str, MAXKEYNAMELEN, true);
				execSpec.cmdline = str;
			}
			if (str = node.getAttribute("consoleoutput"))
				setup->consoleOutput = (stricmp(str, "yes") == 0);
			else
				setup->consoleOutput = 0;
			if (setup->consoleOutput)
				execSpec.outputMode = EXTEXEC_OUTPUT_CONSOLE;	// legacy shorthand
			if (str = node.getAttribute("output")) {
				int mode = ExternalExecManager::ParseOutputMode(str);
				if (mode < 0)
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Unknown <executable> output mode '%s' for '%s', using 'ignore'...", str, setup->name);
				else
					execSpec.outputMode = (uint8)mode;
			}
			if (str = node.getAttribute("logfile"))
				execSpec.logfile = str;
			// Default flipped to YES per T1.4
			if (str = node.getAttribute("autorestart"))
				setup->autoRestart = (stricmp(str, "yes") == 0);
			else
				setup->autoRestart = 1;
			execSpec.autoRestart = (setup->autoRestart != 0);
			if (str = node.getAttribute("shutdownondelete"))
				execSpec.shutdownOnDelete = (stricmp(str, "yes") == 0);
			if (str = node.getAttribute("enabled"))
				execSpec.enabled = (stricmp(str, "yes") == 0);
			else if (str = node.getAttribute("autostart"))
				execSpec.enabled = (stricmp(str, "yes") == 0);
			if (str = node.getAttribute("restartdelay"))
				execSpec.restartDelay = utils::Ascii2Uint32(str);
			if (str = node.getAttribute("restartlimit"))
				execSpec.restartLimit = utils::Ascii2Uint32(str);
			if (str = node.getAttribute("restartwindow"))
				execSpec.restartWindow = utils::Ascii2Uint32(str);
			if (str = node.getAttribute("maxrestartdelay"))
				execSpec.maxRestartDelay = utils::Ascii2Uint32(str);
			if (str = node.getAttribute("startuptimeout"))
				execSpec.startupTimeout = utils::Ascii2Uint32(str);
			if (str = node.getAttribute("stoptimeout"))
				execSpec.stopTimeout = utils::Ascii2Uint32(str);
			if (str = node.getAttribute("stopsignal")) {
				int sig = ExternalExecManager::ParseSignal(str);
				if (sig < 0)
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Unknown <executable> stopsignal '%s' for '%s', using SIGTERM...", str, setup->name);
				else
					execSpec.stopSignal = sig;
			}
			if (str = node.getAttribute("workdir"))
				execSpec.workdir = str;
			// <env name=".." value=".."/> children
			uint32 m, subCount = node.nChildNode();
			for (m=0; m<subCount; m++) {
				subNode = node.getChildNode(m);
				if (stricmp("env", subNode.getName()) == 0) {
					if ((str = subNode.getAttribute("name")) && (str2 = subNode.getAttribute("value")))
						execSpec.env[str] = str2;
				}
			}
		}
		else if (stricmp("customview", node.getName()) == 0) {
			if (str = node.getAttribute("name")) {
				utils::strcpyavail(customView.name, str, MAXKEYNAMELEN, true);
				if (str = node.getAttribute("template")) {
					customView.compID = compID;
					utils::strcpyavail(customView.templateURL, str, MAXKEYNAMELEN, true);
					spec->customViews.push_back(customView);
				}
			}
			else if (node.nChildNode()) {
				char* xml = node.createXMLString();			
				utils::strcpyavail(setup->setup, xml, 1024*1024, true);
				delete [] xml;
			}
			else
				utils::strcpyavail(setup->setup, node.getText(), 1024*1024, true);
			manager->componentMemory->createParameter(compID, "componentsetup", setup->setup);
		}
		else if (stricmp("setup", node.getName()) == 0) {
			if (str = node.getAttribute("file")) {
				strString = utils::ReadAFileString(str);
				if (strString.size())
					utils::strcpyavail(setup->setup, strString.c_str(), MAXKEYNAMELEN, true);
			}
			else if (node.nChildNode()) {
				char* xml = node.createXMLString();			
				utils::strcpyavail(setup->setup, xml, 1024*1024, true);
				delete [] xml;
			}
			else
				utils::strcpyavail(setup->setup, node.getText(), 1024*1024, true);
			manager->componentMemory->createParameter(compID, "componentsetup", setup->setup);
		}
		else if (stricmp("key", node.getName()) == 0) {
			if ((str = node.getAttribute("name")) && (str2 = node.getAttribute("type"))) {
				if (!strlen(setup->keys))
					snprintf(setup->keys, MAXKEYNAMELEN, "%s=%s", str, str2);
				else
					snprintf(setup->keys + strlen(setup->keys), MAXKEYNAMELEN-strlen(setup->keys), ";%s=%s", str, str2);
				manager->componentMemory->createParameter(compID, "keys", setup->keys);
			}
		}
		else if (stricmp("parameter", node.getName()) == 0) {
			if (pname = node.getAttribute("name")) {
				//<parameter name="MaxCount" type="Integer" value="0"/>
				//<parameter name="BitmapFiles" type="String" value="%videofiles%" />
				//<parameter name="MaxCount" type="Integer" value="[0:110]"/> <!-- any integer in range, initial value lowest value -->
				//<parameter name="MaxCount" type="Integer" value="[0:110]=55"/> <!-- any integer in range, initial value 55 -->
				//<parameter name="MaxCount" type="Integer" value="[0:110]=*"/> <!-- any integer in range, random initial value -->
				//<parameter name="MaxCount" type="Integer" value="[0:110:2]"/> <!-- any integer in range, in steps of 2 -->
				//<parameter name="Interval" type="Float" value="[-2.5:2.5]"/> <!-- any float in range -->
				//<parameter name="Interval2" type="Float" value="[-2.5;1.5;1.8;2.5]"/> <!-- any float in list -->
				//<parameter name="Category" type="String" value="['str1';'str2';'str3']"/> <!-- any string in list -->
				ptype = PARAM_STRING;
				if (str = node.getAttribute("type")) {
					if (stricmp(str, "integer") == 0)
						ptype = PARAM_INTEGER;
					else if ((stricmp(str, "float") == 0) || (stricmp(str, "double") == 0))
						ptype = PARAM_FLOAT;
				}
				sinterval = NULL;
				if (str = node.getAttribute("interval")) {
					sinterval = str;
				}
				if (str = node.getAttribute("value")) {
					if (!utils::TextStartsWith(str, "[")) {
						svalue = str;
						if (ptype == PARAM_STRING)
							manager->componentMemory->createParameter(compID, pname, svalue);
						else if (ptype == PARAM_INTEGER)
							manager->componentMemory->createParameter(compID, pname, utils::Ascii2Int64(svalue), 0, 0, utils::Ascii2Int64(sinterval));
						else if (ptype == PARAM_FLOAT)
							manager->componentMemory->createParameter(compID, pname, utils::Ascii2Float64(svalue), 0, 0, utils::Ascii2Float64(sinterval));
					}
					else {
						const char* end;
						if (end = strchr(str, '='))
							svalue = end+1;
						else
							svalue = NULL;
						end = strchr(str, ']');
						strString = std::string(str+1, str-end-2);
						if (strchr(str, ':')) {
							// range
							values = utils::TextListSplit(strString.c_str(), ":", true, true);
							if ((ptype == PARAM_INTEGER) && (values.size() > 1)) {
								if (!svalue)
									ivalue = utils::Ascii2Int64(values[0].c_str());
								else if (svalue[0] == '*') {
									if (values.size() > 2)
										ivalue = (int64) utils::RandomValue(utils::Ascii2Float64(values[0].c_str()), utils::Ascii2Float64(values[1].c_str()), utils::Ascii2Float64(values[2].c_str()));
									else
										ivalue = (int64) utils::RandomValue(utils::Ascii2Float64(values[0].c_str()), utils::Ascii2Float64(values[1].c_str()));
								}
								else
									ivalue = utils::Ascii2Int64(svalue);
								if (values.size() > 2)
									manager->componentMemory->createParameter(compID, pname, ivalue, utils::Ascii2Int64(values[0].c_str()), utils::Ascii2Int64(values[1].c_str()), utils::Ascii2Int64(values[2].c_str()));
								else
									manager->componentMemory->createParameter(compID, pname, ivalue, utils::Ascii2Int64(values[0].c_str()), utils::Ascii2Int64(values[1].c_str()), utils::Ascii2Int64(sinterval));
							}
							else if ((ptype == PARAM_FLOAT) && (values.size() > 1)) {
								if (!svalue)
									fvalue = utils::Ascii2Float64(values[0].c_str());
								else if (svalue[0] == '*') {
									if (values.size() > 2)
										fvalue = utils::RandomValue(utils::Ascii2Float64(values[0].c_str()), utils::Ascii2Float64(values[1].c_str()), utils::Ascii2Float64(values[2].c_str()));
									else
										fvalue = utils::RandomValue(utils::Ascii2Float64(values[0].c_str()), utils::Ascii2Float64(values[1].c_str()));
								}
								else
									fvalue = utils::Ascii2Float64(svalue);
								if (values.size() > 2)
									manager->componentMemory->createParameter(compID, pname, fvalue, utils::Ascii2Float64(values[0].c_str()), utils::Ascii2Float64(values[1].c_str()), utils::Ascii2Float64(values[2].c_str()));
								else
									manager->componentMemory->createParameter(compID, pname, fvalue, utils::Ascii2Float64(values[0].c_str()), utils::Ascii2Float64(values[1].c_str()), utils::Ascii2Float64(sinterval));
							}
						}
						else if (strchr(str, ';')) {
							// collection
							values = utils::TextListSplit(strString.c_str(), ";", true, true);
							if (values.size() > 0) {
								if (!svalue)
									svalue = values[0].c_str();
								else if (svalue[0] == '*')
									svalue = values[(uint32)utils::RandomValue(0, (double)values.size()-1, 1)].c_str();
								if (ptype == PARAM_STRING)
									manager->componentMemory->createParameter(compID, pname, values, svalue);
								else if (ptype == PARAM_INTEGER)
									manager->componentMemory->createParameter(compID, pname, values, utils::Ascii2Int64(svalue));
								else if (ptype == PARAM_FLOAT)
									manager->componentMemory->createParameter(compID, pname, values, utils::Ascii2Float64(svalue));
							}
						}
					}
				}
			}
		}
		else if (stricmp("context", node.getName()) == 0) {
			if ( (context = getPsyContext(node.getAttribute("name"))) != NOCONTEXT) {
				// for each subnode:
				uint32 m, subCount = node.nChildNode();
				for(m=0; m<subCount; m++){
					subNode=node.getChildNode(m);
					getTriggersFromXML(subNode, spec, context, triggers, retrieves, queries, posts, signals, cranks);
				}
				if (triggers.size() > 0)
					addTriggersToComponent(spec, triggers, retrieves, queries, posts, signals, cranks);
			}
		}
		else { // default contexts, adding triggers, cranks, etc.
			if (!getTriggersFromXML(node, spec, defaultContext, triggers, retrieves, queries, posts, signals, cranks)) {
				delete(spec);
				return NULL;
			}
		}
	}

			//if ( (test = getComponentID(setup->name)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "6 Component '%s' not created correctly (%u != %u)\n%s...", setup->name, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return NULL;
			//}

	if (!cranks.size() && !strlen(setup->function)) {
		uint16 crankID;
		uint32 id;
		// No crank specified, using built-in Internal::Simple
		// using crank name modulename_default_crank
		std::string crankName = utils::StringFormat("%s_default_crank", setup->name);
		if (crankID = getCrankID(crankName.c_str())) {
			if (id = manager->dataMapsMemory->getCrankCompID(crankID)) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Default crank name '%s' is already in use by component '%s' (ID %u)...", str, manager->componentMemory->getComponentNameString(id).c_str(), id);
			}
			else
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Default crank name '%s' is already in use by crank ID %u...", str, crankID);
			delete(spec);
			return NULL;
		}
		if (!(crankID = syncCrankID(spec->setup->componentID, crankName.c_str(), "Simple", "", "", ""))) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve default crank id for crank '%s' for component '%s' ('Internal::Simple')...", crankName.c_str(), setup->name);
			delete(spec);
			return NULL;
		}
		cranks.push_back(crankID);
	}

	if (triggers.size() > 0) {
		// Add remaining triggers, cranks, etc. to default context
		addTriggersToComponent(spec, triggers, retrieves, queries, posts, signals, cranks);
	}


			//if ( (test = getComponentID(setup->name)) != compID) {
			//	LogPrint(0, LOG_NODE, 0, "7 Component '%s' not created correctly (%u != %u)\n%s...", setup->name, test, compID, manager->componentMemory->printFriendly().c_str());
			//	return NULL;
			//}

	// T1.4: launch/manage the external executable for this module's space
	if (hasExecutable) {
		if (!execSpec.cmdline.size())
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Module '%s' has an <executable> element with no command line, ignoring...", setup->name);
		else {
			const char* execSpaceName = (sspace && strlen(sspace)) ? sspace : setup->name;
			uint16 execSpaceID = 0;
			if (!createExternalSpace(execSpaceName, execSpaceID))
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not create external space '%s' for executable of module '%s'...", execSpaceName, setup->name);
			else if (manager->processMemory->getProcessOSID(execSpaceID))
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Space '%s' already has a running OS process; ignoring <executable> of module '%s'...", execSpaceName, setup->name);
			else {
				execSpec.spaceName = execSpaceName;
				execSpec.spaceID = execSpaceID;
				execSpec.moduleID = compID;
				execSpec.moduleName = setup->name;
				if (!externalExec)
					externalExec = new ExternalExecManager(this);
				externalExec->registerExecutable(execSpec);
			}
		}
	}

	// All done, add to component list
	// ###################################
	return spec;
}

PsyType Node::getPsyType(const char* typestring) {
	PsyType type = NOTYPE;

	std::vector<std::string> entries = utils::TextListSplit(typestring, ".", true, true);
	if (entries.size() < 1)
		return NOTYPE;

	uint16 subID;
	uint16 level = 0;
	std::vector<std::string>::iterator it, itEnd;
	for (it = entries.begin(), itEnd = entries.end(); it != itEnd; it++) {
		if (it->length() > 0) {
			if (strcmp(it->c_str(), "*") == 0) {
				subID = 0xFFFF;
			}
			// Check to see if subtype already exists
			else if (!manager->dataMapsMemory->getTypeLevelID(it->c_str(), subID)) {
				LogPrint(networking->localNodeID, LOG_NODE, 5,"Subtype '%s' unknown (in '%s'), registering as new type with node network...", it->c_str(), typestring);
				// Create a new id
				if (!(subID = syncTypeLevelID(it->c_str()))) {
					LogPrint(networking->localNodeID, LOG_NODE, 0,"Could not register subtype '%s' for type '%s' ID %u with node network...", it->c_str(), typestring, subID);
					return NOTYPE;
				}
				//if (!(subID = idManager->getNextTypeID())) {
				//	LogPrint(networking->localNodeID, LOG_NODE, 0,"Could not register subtype '%s' for type '%s' - no more IDs available...", it->c_str(), typestring);
				//	return NOTYPE;
				//}
				//if (!manager->dataMapsMemory->createNewTypeLevel(subID, it->c_str())) {
				//	LogPrint(networking->localNodeID, LOG_NODE, 0,"Could not register subtype '%s' for type '%s' ID %u...", it->c_str(), typestring, subID);
				//	return NOTYPE;
				//}
				// Sync new subtype to other nodes
			}
			type.setLevel(level, subID);
		}
		level++;
	}
	return type;
}

PsyContext Node::getPsyContext(const char* contextstring) {
	PsyContext context = NOCONTEXT;

	std::vector<std::string> entries = utils::TextListSplit(contextstring, ".", true, true);
	if (entries.size() < 1)
		return NOCONTEXT;

	uint16 subID;
	uint16 level = 0;
	std::vector<std::string>::iterator it, itEnd;
	for (it = entries.begin(), itEnd = entries.end(); it != itEnd; it++) {
		if (it->length() > 0) {
			if (strcmp(it->c_str(), "*") == 0) {
				subID = 0xFFFF;
			}
			// Check to see if subcontext already exists
			else if (!manager->dataMapsMemory->getContextLevelID(it->c_str(), subID)) {
				// Create a new id
				if (!(subID = syncContextLevelID(it->c_str()))) {
					LogPrint(networking->localNodeID, LOG_NODE, 0,"Could not register subcontext '%s' for context '%s' with node network...", it->c_str(), contextstring);
					return NOCONTEXT;
				}
				//if (!(subID = idManager->getNextContextID())) {
				//	LogPrint(networking->localNodeID, LOG_NODE, 0,"Could not register subcontext '%s' for context '%s' - no more IDs available...", it->c_str(), contextstring);
				//	return NOCONTEXT;
				//}
				//if (!manager->dataMapsMemory->createNewContextLevel(subID, it->c_str())) {
				//	LogPrint(networking->localNodeID, LOG_NODE, 0,"Could not register subcontext '%s' for context '%s'...", it->c_str(), contextstring);
				//	return NOCONTEXT;
				//}
				//LogPrint(networking->localNodeID, LOG_NODE, 2,"Created new subcontext id %u (%s) for context '%s'...", subID, it->c_str(), contextstring);
				// Sync new subcontext to other nodes
			}
			context.setLevel(level, subID);
		}
		level++;
	}
	return context;
}

uint16 Node::getCrankID(const char* name) {
	uint16 id;
	if (!manager->dataMapsMemory->getCrankID(name, id))
		return 0;
	return id;
}

uint32 Node::getTagID(const char* name) {
	uint32 id;
	if (manager->dataMapsMemory->getTagID(name, id))
		return id;

	if (!(id = syncTagID(name))) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not register tag for '%s' with node network...", name);
		return false;
	}
	//if (!(id = idManager->getNextTagID())) {
	//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not register tag id for '%s' - no more IDs available...", name);
	//	return false;
	//}
	//if (!manager->dataMapsMemory->createNewTag(id, name)) {
	//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not register tag id for '%s'...", name);
	//	return false;
	//}
	return id;
}

uint32 Node::getComponentID(const char* name, bool autoCreate) {
	uint32 id;
	if (manager->componentMemory->getComponentID(name, id))
		return id;

	if (!autoCreate)
		return 0;

	if (!(id = syncComponentID(0, 0, SELFTRIGGER_WARN, name, 0, 0))) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not register component for '%s' with node network...", name);
		return 0;
	}

	//if (!(id = idManager->getNextComponentID())) {
	//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve component id for '%s' - no more IDs available...", name);
	//	return 0;
	//}

	//if (!manager->componentMemory->createComponent(id, 0, 0, name, 1024, 0, 0)) {
	//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve component id for '%s'...", name);
	//	return 0;
	//}
	return id;
}

bool Node::getTriggersFromXML(XMLNode &node, ComponentSpec* spec, PsyContext context,
		std::list<TriggerSpec*>& triggers,
		std::list<RetrieveSpec*>& retrieves,
		std::list<QuerySpec*>& queries,
		std::list<PostSpec*>& posts,
		std::list<SignalSpec*>& signals,
		std::list<uint16>& cranks) {

	if (node.isEmpty() || !spec)
		return false;

	std::list<std::string> triggerNames, postNames, retrieveNames, queryNames;
	std::list<TriggerSpec*>::iterator triggerIt = triggers.begin(), triggerEnd = triggers.end();
	while (triggerIt != triggerEnd) { triggerNames.push_back((*triggerIt)->name); triggerIt++; }
	std::list<PostSpec*>::iterator postIt = posts.begin(), postEnd = posts.end();
	while (postIt != postEnd) { postNames.push_back((*postIt)->name); postIt++; }
	std::list<RetrieveSpec*>::iterator retrieveIt = retrieves.begin(), retrieveEnd = retrieves.end();
	while (retrieveIt != retrieveEnd) { retrieveNames.push_back((*retrieveIt)->name); retrieveIt++; }
	std::list<QuerySpec*>::iterator queryIt = queries.begin(), queryEnd = queries.end();
	while (queryIt != queryEnd) { queryNames.push_back((*queryIt)->name); queryIt++; }

	FilterSpec* filterSpec;
	std::list<FilterSpec*> filters;
	const char* str, *func, *lib, *language, *script;
	char* libraryFilename = new char[MAXKEYNAMELEN+1];
	uint16 crankID;
	uint32 id;
	std::vector<std::string> libfunc;
	XMLNode subNode;
	uint32 size;
	int n;

	if (stricmp("trigger", node.getName()) == 0) {
		for (n=0; n<node.nChildNode(); n++) {
			if ((!(subNode = node.getChildNode(n)).isEmpty()) && (stricmp(subNode.getName(), "filter") == 0)) {
				if (stricmp(str = subNode.getAttribute("type"), "equals") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_EQUALS;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					utils::strcpyavail(filterSpec->value, subNode.getAttribute("value"), MAXVALUENAMELEN, true);
					filters.push_back(filterSpec);
				}
				else if (stricmp(str, "notequals") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_NOTEQUALS;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					utils::strcpyavail(filterSpec->value, subNode.getAttribute("value"), MAXVALUENAMELEN, true);
					filters.push_back(filterSpec);
				}
				else if (stricmp(str, "equalsnumeric") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_EQUALSNUMERIC;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					utils::strcpyavail(filterSpec->value, subNode.getAttribute("value"), MAXVALUENAMELEN, true);
					filters.push_back(filterSpec);
				}
				else if (stricmp(str, "notequalsnumeric") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_NOTEQUALSNUMERIC;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					utils::strcpyavail(filterSpec->value, subNode.getAttribute("value"), MAXVALUENAMELEN, true);
					filters.push_back(filterSpec);
				}
				else if (stricmp(str, "greaterthan") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_GREATERTHAN;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					utils::strcpyavail(filterSpec->value, subNode.getAttribute("value"), MAXVALUENAMELEN, true);
					filters.push_back(filterSpec);
				}
				else if (stricmp(str, "lessthan") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_LESSTHAN;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					utils::strcpyavail(filterSpec->value, subNode.getAttribute("value"), MAXVALUENAMELEN, true);
					filters.push_back(filterSpec);
				}
				else if (stricmp(str, "haskey") == 0) {
					filterSpec = new FilterSpec;
					memset(filterSpec, 0, sizeof(FilterSpec));
					filterSpec->filter = TRIGGERFILTER_HASKEY;
					utils::strcpyavail(filterSpec->key, subNode.getAttribute("key"), MAXKEYNAMELEN, true);
					filters.push_back(filterSpec);
				}
			}
		}
		if (str = node.getAttribute("maxage")) {
			filterSpec = new FilterSpec;
			memset(filterSpec, 0, sizeof(FilterSpec));
			filterSpec->filter = TRIGGERFILTER_MAXAGE;
			filterSpec->time = utils::Ascii2Uint64(str);
			filters.push_back(filterSpec);
		}
		
		size = sizeof(TriggerSpec) + ((uint32)filters.size() * sizeof(FilterSpec));

		TriggerSpec* triggerSpec = (TriggerSpec*) malloc(size);
		triggerSpec->reset(size);
		triggerSpec->componentID = spec->setup->componentID;
		triggerSpec->context = context;
		if (str = node.getAttribute("name"))
			utils::strcpyavail(triggerSpec->name, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("type"))
			triggerSpec->type = getPsyType(str);
		if (str = node.getAttribute("from"))
			triggerSpec->from = getComponentID(str);
		if (str = node.getAttribute("to"))
			triggerSpec->to = getComponentID(str);
		if (str = node.getAttribute("interval"))
			triggerSpec->interval = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("delay"))
			triggerSpec->delay = utils::Ascii2Uint32(str);
		if (!str && (str = node.getAttribute("after")))
			triggerSpec->delay = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("context")) {
			triggerSpec->triggerContext = getPsyContext(str);
			triggerSpec->type = PsyAPI::CTRL_CONTEXT_CHANGE;
		}

		if (str = node.getAttribute("tag")) {
			if (strcmp(str, "*") == 0)
				triggerSpec->tag = getTagID(spec->setup->name);
			else
				triggerSpec->tag = getTagID(str);
		}

		triggerSpec->filterCount = (uint16)filters.size();
		if (triggerSpec->filterCount) {
			char* data = ((char*)triggerSpec) + sizeof(TriggerSpec);
			std::list<FilterSpec*>::iterator it, itEnd;
			for (it = filters.begin(), itEnd = filters.end(); it != itEnd; it++) {
				memcpy(data, *it, sizeof(FilterSpec));
				data += sizeof(FilterSpec);
			}

			itEnd = filters.end();
			while ((it = filters.begin()) != itEnd) {
				delete(*it);
				filters.pop_front();
			}
		}

		if (std::find(triggerNames.begin(), triggerNames.end(), triggerSpec->name) != triggerNames.end()) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Trigger name '%s' is already in use...", triggerSpec->name);
			delete(triggerSpec);
			return false;
		}

		if (triggerSpec->name[0] && ((triggerSpec->type.isValid()) || triggerSpec->from || triggerSpec->to))
			triggers.push_back(triggerSpec);
		else
			delete(triggerSpec);
	}
	else if (stricmp("triggers", node.getName()) == 0) {
		// register types
	}
	else if (stricmp("triggergroup", node.getName()) == 0) {
		// T1.6: declarative join. Parse the group + its member <trigger>s into a TriggerGroupSpec.
		// Member triggers are registered through the normal <trigger> path (so the subscription
		// engine still routes them), then tagged with this group's id and referenced by name.
		// See Docs/T1.6-triggergroup-plan.md for the locked design.
		const char* gname = node.getAttribute("name");
		if (!gname || !strlen(gname)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "<triggergroup> is missing a 'name' attribute; discarding group...");
			return true;
		}
		// maxage is REQUIRED (no default).
		const char* gmaxage = node.getAttribute("maxage");
		if (!gmaxage || !strlen(gmaxage)) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "<triggergroup> '%s' is missing the required 'maxage' attribute; discarding group...", gname);
			return true;
		}
		// Reject a duplicate group name within this module.
		bool dupGroup = false;
		std::list<TriggerGroupSpec*>::iterator gIt, gItEnd;
		for (gIt = spec->triggerGroups.begin(), gItEnd = spec->triggerGroups.end(); gIt != gItEnd; gIt++) {
			if (stricmp((*gIt)->name, gname) == 0) { dupGroup = true; break; }
		}
		if (dupGroup) {
			LogPrint(networking->localNodeID, LOG_NODE, 1, "<triggergroup> name '%s' is already in use in this module; discarding the duplicate...", gname);
			return true;
		}

		// Allocate a group id (also used as the grouping tag key) up front so members can carry it.
		uint32 groupID = getTagID(gname);

		// Collect member metadata while recursing to register each member trigger normally.
		std::vector<GroupMemberSpec> members;
		uint32 c, childCount = node.nChildNode();
		for (c = 0; c < childCount; c++) {
			XMLNode childNode = node.getChildNode(c);
			if (childNode.isEmpty() || stricmp(childNode.getName(), "trigger") != 0)
				continue;
			const char* mname = childNode.getAttribute("name");
			if (!mname || !strlen(mname)) {
				LogPrint(networking->localNodeID, LOG_NODE, 1, "<triggergroup> '%s' has a member <trigger> with no name; skipping member...", gname);
				continue;
			}
			// Record member metadata (per-member optional/maxage/primary).
			GroupMemberSpec m;
			memset(&m, 0, sizeof(GroupMemberSpec));
			utils::strcpyavail(m.name, mname, MAXKEYNAMELEN, true);
			if (str = childNode.getAttribute("maxage")) m.maxage = utils::Ascii2Uint32(str);
			if ((str = childNode.getAttribute("optional")) && (stricmp(str, "true") == 0 || stricmp(str, "yes") == 0 || stricmp(str, "1") == 0)) m.optional = 1;
			if ((str = childNode.getAttribute("primary")) && (stricmp(str, "true") == 0 || stricmp(str, "yes") == 0 || stricmp(str, "1") == 0)) m.primary = 1;
			members.push_back(m);
			// Register the member trigger through the normal <trigger> path so it routes.
			getTriggersFromXML(childNode, spec, context, triggers, retrieves, queries, posts, signals, cranks);
			// Tag the just-added trigger with this group's id so the JoinEngine can associate it.
			if (!triggers.empty() && stricmp(triggers.back()->name, mname) == 0)
				triggers.back()->group = groupID;
		}
		if (members.empty()) {
			LogPrint(networking->localNodeID, LOG_NODE, 1, "<triggergroup> '%s' has no valid member <trigger>s; discarding group...", gname);
			return true;
		}
		// If no member is flagged primary, the first member is primary (for mode="snapshot").
		bool anyPrimary = false;
		for (size_t mi = 0; mi < members.size(); mi++) if (members[mi].primary) { anyPrimary = true; break; }
		if (!anyPrimary) members[0].primary = 1;

		// Build the packed TriggerGroupSpec (struct + member tail).
		uint32 gsize = sizeof(TriggerGroupSpec) + ((uint32)members.size() * sizeof(GroupMemberSpec));
		TriggerGroupSpec* groupSpec = (TriggerGroupSpec*) malloc(gsize);
		groupSpec->reset(gsize);
		groupSpec->id = groupID;
		groupSpec->componentID = spec->setup->componentID;
		utils::strcpyavail(groupSpec->name, gname, MAXKEYNAMELEN, true);
		groupSpec->maxage = utils::Ascii2Uint32(gmaxage);
		// count: default all (0); accept an integer or the literal "all".
		if ((str = node.getAttribute("count")) && stricmp(str, "all") != 0)
			groupSpec->count = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("min")) groupSpec->minCount = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("within")) groupSpec->within = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("time")) { groupSpec->timeWindow = utils::Ascii2Uint32(str); groupSpec->hasTimeWindow = 1; }
		if (str = node.getAttribute("debounce")) groupSpec->debounce = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("timeout")) groupSpec->timeout = utils::Ascii2Uint32(str);
		if ((str = node.getAttribute("mode")) && stricmp(str, "snapshot") == 0) groupSpec->mode = TRIGGERGROUP_MODE_SNAPSHOT;
		if ((str = node.getAttribute("slide")) && (stricmp(str, "true") == 0 || stricmp(str, "yes") == 0 || stricmp(str, "1") == 0)) groupSpec->slide = 1;
		if ((str = node.getAttribute("dup")) && stricmp(str, "keep") == 0) groupSpec->dup = TRIGGERGROUP_DUP_KEEP;
		if ((str = node.getAttribute("order")) && stricmp(str, "strict") == 0) groupSpec->order = TRIGGERGROUP_ORDER_STRICT;
		if ((str = node.getAttribute("ontimeout")) && stricmp(str, "discard") == 0) groupSpec->ontimeout = TRIGGERGROUP_ONTIMEOUT_DISCARD;
		groupSpec->memberCount = (uint16)members.size();
		char* mdata = ((char*)groupSpec) + sizeof(TriggerGroupSpec);
		for (size_t mi = 0; mi < members.size(); mi++) {
			memcpy(mdata, &members[mi], sizeof(GroupMemberSpec));
			mdata += sizeof(GroupMemberSpec);
		}
		spec->triggerGroups.push_back(groupSpec);
		// TODO(T1.6 Step 3+): JoinEngine consumes spec->triggerGroups to accumulate/complete/evict;
		// crankID is stamped later (addTriggersToComponent stamps member triggers) — see handoff note.
	}
	else if (stricmp("retrieve", node.getName()) == 0) {
		RetrieveSpec* retrieveSpec = new RetrieveSpec;
		memset(retrieveSpec, 0, sizeof(RetrieveSpec));
		retrieveSpec->context = context;
		retrieveSpec->startInt = retrieveSpec->endInt = INT64_NOVALUE;
		retrieveSpec->startFloat = retrieveSpec->endFloat = FLOAT64_NOVALUE;
		if (str = node.getAttribute("name"))
			utils::strcpyavail(retrieveSpec->name, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("type"))
			retrieveSpec->type = getPsyType(str);
		if (str = node.getAttribute("from"))
			retrieveSpec->from = getComponentID(str);
		if (str = node.getAttribute("to"))
			retrieveSpec->to = getComponentID(str);
		if (str = node.getAttribute("source"))
			retrieveSpec->source = getComponentID(str);

		if (str = node.getAttribute("maxcount"))
			retrieveSpec->maxcount = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("maxage"))
			retrieveSpec->maxage = utils::Ascii2Uint64(str)*1000;

		if (str = node.getAttribute("tag")) {
			if (strcmp(str, "*") == 0)
				retrieveSpec->tag = getTagID(spec->setup->name);
			else
				retrieveSpec->tag = getTagID(str);
		}

		if (str = node.getAttribute("key"))
			utils::strcpyavail(retrieveSpec->key, str, MAXKEYNAMELEN, true);
		else
			utils::strcpyavail(retrieveSpec->key, "time", MAXKEYNAMELEN, true);
		retrieveSpec->keytype = INDEX_TIME;
		if (str = node.getAttribute("keytype")) {
			if (stricmp(str, "string") == 0)
				retrieveSpec->keytype = INDEX_STRING;
			else if (stricmp(str, "integer") == 0)
				retrieveSpec->keytype = INDEX_INTEGER;
			else if (stricmp(str, "float") == 0)
				retrieveSpec->keytype = INDEX_FLOAT;
			else if (stricmp(str, "double") == 0)
				retrieveSpec->keytype = INDEX_FLOAT;
			else if (stricmp(str, "time") == 0)
				retrieveSpec->keytype = INDEX_TIME;
			else
				retrieveSpec->keytype = 0;
		}
		if (str = node.getAttribute("start")) {
			switch(retrieveSpec->keytype) {
			case INDEX_STRING:
				utils::strcpyavail(retrieveSpec->startString, str, MAXVALUENAMELEN, true);
				break;
			case INDEX_INTEGER:
				retrieveSpec->startInt = utils::Ascii2Int64(str);
				break;
			case INDEX_FLOAT:
				retrieveSpec->startFloat = utils::Ascii2Float64(str);
				break;
			default:
				retrieveSpec->startTime = utils::Ascii2Uint64(str);
				break;
			}
		}
		if (str = node.getAttribute("end")) {
			switch(retrieveSpec->keytype) {
			case INDEX_STRING:
				utils::strcpyavail(retrieveSpec->endString, str, MAXVALUENAMELEN, true);
				break;
			case INDEX_INTEGER:
				retrieveSpec->endInt = utils::Ascii2Int64(str);
				break;
			case INDEX_FLOAT:
				retrieveSpec->endFloat = utils::Ascii2Float64(str);
				break;
			default:
				retrieveSpec->endTime = utils::Ascii2Uint64(str);
				break;
			}
		}

		if (std::find(retrieveNames.begin(), retrieveNames.end(), retrieveSpec->name) != retrieveNames.end()) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Retrieve name '%s' is already in use...", retrieveSpec->name);
			delete(retrieveSpec);
			return false;
		}

		if (retrieveSpec->name[0] && retrieveSpec->source && retrieveSpec->keytype)
			retrieves.push_back(retrieveSpec);
		else
			delete(retrieveSpec);
	}
	else if (stricmp("query", node.getName()) == 0) {
		QuerySpec* querySpec = new QuerySpec;
		memset(querySpec, 0, sizeof(QuerySpec));
		querySpec->context = context;
		if (str = node.getAttribute("name"))
			utils::strcpyavail(querySpec->name, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("type"))
			utils::strcpyavail(querySpec->type, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("source")) {
			querySpec->source = getComponentID(str);
			utils::strcpyavail(querySpec->sourceName, str, MAXKEYNAMELEN, true);
		}
		if (str = node.getAttribute("subdir"))
			utils::strcpyavail(querySpec->subdir, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("ext"))
			utils::strcpyavail(querySpec->ext, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("binary"))
			querySpec->binary = (stricmp(str, "yes") == 0);

		if (str = node.getAttribute("maxcount"))
			querySpec->maxcount = utils::Ascii2Uint32(str);
		if (str = node.getAttribute("maxage"))
			querySpec->maxage = utils::Ascii2Uint64(str)*1000;

		if (str = node.getAttribute("key"))
			utils::strcpyavail(querySpec->key, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("value"))
			utils::strcpyavail(querySpec->value, str, MAXVALUENAMELEN, true);

		if (str = node.getAttribute("operation"))
			utils::strcpyavail(querySpec->operation, str, MAXKEYNAMELEN, true);

		if (str = node.getAttribute("port"))
			querySpec->ipPort = utils::Ascii2Uint32(str);

		if (str = node.getAttribute("host")) {
			if (!utils::LookupIPAddress(str, querySpec->ipHost))
				querySpec->ipHost = 0;
			utils::strcpyavail(querySpec->hostName, str, MAXKEYNAMELEN, true);
		}

		if (std::find(queryNames.begin(), queryNames.end(), querySpec->name) != queryNames.end()) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Query name '%s' is already in use...", querySpec->name);
			delete(querySpec);
			return false;
		}

		if (querySpec->name[0] && querySpec->source)
			queries.push_back(querySpec);
		else
			delete(querySpec);
	}
	else if (stricmp("retrieves", node.getName()) == 0) {
		// register types
	}
	else if (stricmp("crank", node.getName()) == 0) {
		// <crank name="c1" function="lib::function" />
		// <crank name="c1" script="angel1.script" />
		// <crank name="c1">script</crank>
		// ########## More error handling and printing...
		str = node.getAttribute("name");
		crankID = 0;
		if (lib = node.getAttribute("function")) {
			if (func = strstr(lib, "::")) {
				memcpy(libraryFilename, lib, func-lib);
				libraryFilename[func-lib] = 0;
				func += 2;
				if (crankID = getCrankID(str)) {
					if (id = manager->dataMapsMemory->getCrankCompID(crankID)) {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Crank name '%s' is already in use by component '%s' (ID %u)...", str, manager->componentMemory->getComponentNameString(id).c_str(), id);
					}
					else
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Crank name '%s' is already in use by crank ID %u...", str, crankID);
					return false;
				}
				if (!(crankID = syncCrankID(spec->setup->componentID, str, func, libraryFilename, "", ""))) {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for '%s' ('%s::%s')...", str, libraryFilename, func);
					return false;
				}

				//if (!(crankID = idManager->getNextCrankID())) {
				//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for '%s::%s'...", libraryFilename, func);
				//	return false;
				//}
				//if (crankID && !manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID, str, func, libraryFilename, ""))
				//	crankID = 0;
			}
			else {
				// Internal crank without "<library>::"
				if (crankID = getCrankID(str)) {
					if (id = manager->dataMapsMemory->getCrankCompID(crankID)) {
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Crank name '%s' is already in use by component '%s' (ID %u)...", str, manager->componentMemory->getComponentNameString(id).c_str(), id);
					}
					else
						LogPrint(networking->localNodeID, LOG_NODE, 0, "Crank name '%s' is already in use by crank ID %u...", str, crankID);
					return false;
				}
				if (!(crankID = syncCrankID(spec->setup->componentID, str, lib, "", "", ""))) {
					LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for '%s' ('Internal::%s')...", str, lib);
					return false;
				}

				//if (!(crankID = idManager->getNextCrankID())) {
				//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for 'Internal::%s'...", lib);
				//	return false;
				//}
				//if (crankID && !manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID, str, lib, "", ""))
				//	crankID = 0;
				//if (crankID && !registerCrankID(crankID, spec->setup->componentID, str, lib, "", ""))
				//	crankID = 0;
			}
		}
		else if ( (language = node.getAttribute("language")) && (script = node.getAttribute("script")) ) {
			if (!(crankID = syncCrankID(spec->setup->componentID, str, "", script, language, ""))) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve inline script crank id for '%s' ('%s')...", str, script);
				return false;
			}

			//if (!(crankID = idManager->getNextCrankID())) {
			//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve script crank id for '%s'...", script);
			//	return false;
			//}
			//if (crankID && !manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID, str, "", script, ""))
			//	crankID = 0;
			//if (crankID && !registerCrankID(crankID, spec->setup->componentID, str, "", script, ""))
			//	crankID = 0;
		}
		else if ((language = node.getAttribute("language")) && node.nClear() && (script = node.getClear().lpszValue) && strlen(script)) {
			if (!(crankID = syncCrankID(spec->setup->componentID, str, "", "", language, utils::TextTrim(utils::TextUnindent(script).c_str()).c_str()))) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve file script crank id for '%s' ('%s')...", str, script);
				return false;
			}

			//if (!(crankID = idManager->getNextCrankID())) {
			//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve script crank id for '%s'...", script);
			//	return false;
			//}
			//if (crankID && !manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID, str, "", "", script))
			//	crankID = 0;
			//if (crankID && !registerCrankID(crankID, spec->setup->componentID, str, "", "", script))
			//	crankID = 0;
		}
		else {
			// This is an external module to be filled in later
			if (!(crankID = syncCrankID(spec->setup->componentID, str, "", "", "", ""))) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve external crank id for '%s'...", str);
				return false;
			}
			LogPrint(networking->localNodeID, LOG_NODE, 1, "Configured external crank '%s' (ID %u) for component '%s', waiting for connection...", str, crankID, spec->setup->name);

			//if (!(crankID = idManager->getNextCrankID())) {
			//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve external crank id for '%s'...", str);
			//	return false;
			//}
			//if (crankID && !manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID, str, "", "", ""))
			//	crankID = 0;
			//if (crankID && !registerCrankID(crankID, spec->setup->componentID, str, "", "", ""))
			//	crankID = 0;
		}
		if (crankID)
			cranks.push_back(crankID);
	}
	else if (stricmp("post", node.getName()) == 0) {
		PostSpec* postSpec = new PostSpec;
		memset(postSpec, 0, sizeof(PostSpec));
		postSpec->context = context;
		if (str = node.getAttribute("name"))
			utils::strcpyavail(postSpec->name, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("type"))
			postSpec->type = getPsyType(str);
		if (str = node.getAttribute("context"))
			postSpec->contextchange = getPsyContext(str);
		if (str = node.getAttribute("to"))
			postSpec->to = getComponentID(str);
		if (str = node.getAttribute("ttl"))
			postSpec->ttl = utils::Ascii2Uint64(str) * 1000;
		else
			postSpec->ttl = 0;
		if (str = node.getAttribute("guaranteed"))
			if (stricmp(str, "no") == 0)
				postSpec->policy |= MESSAGE_NON_GUARANTEED;

		if (str = node.getAttribute("tag")) {
			if (strcmp(str, "*") == 0)
				postSpec->tag = getTagID(spec->setup->name);
			else
				postSpec->tag = getTagID(str);
		}

		//if (str = node.getAttribute("contentkey")) {
		//	utils::strcpyavail(postSpec->contentKey, str, MAXKEYNAMELEN, true);
		//	if (str = node.getAttribute("content"))
		//		utils::strcpyavail(postSpec->content, html::DecodeHTML(str).c_str(), MAXVALUENAMELEN, true);
		//}
		postSpec->contentFromXML(node);

		if (std::find(postNames.begin(), postNames.end(), postSpec->name) != postNames.end()) {
			LogPrint(networking->localNodeID, LOG_NODE, 0, "Post name '%s' is already in use...", postSpec->name);
			delete(postSpec);
			return false;
		}

		if (postSpec->name[0] && ( postSpec->type.isValid() || postSpec->contextchange.isValid() ))
			posts.push_back(postSpec);
		else
			delete(postSpec);
	}
	else if (stricmp("posts", node.getName()) == 0) {
		// register types
	}
	else if (stricmp("signal", node.getName()) == 0) {
		SignalSpec* signalSpec = new SignalSpec;
		memset(signalSpec, 0, sizeof(SignalSpec));
		if (str = node.getAttribute("name"))
			utils::strcpyavail(signalSpec->name, str, MAXKEYNAMELEN, true);
		if (str = node.getAttribute("type"))
			signalSpec->type = getPsyType(str);

		if (signalSpec->name[0] && (signalSpec->type.isValid()))
			signals.push_back(signalSpec);
		else
			delete(signalSpec);
	}
	else if (stricmp("signals", node.getName()) == 0) {
		// register types
	}
	delete [] libraryFilename;
	return true;
}

bool Node::addTriggersToComponent(ComponentSpec* spec,
		std::list<TriggerSpec*>& triggers,
		std::list<RetrieveSpec*>& retrieves,
		std::list<QuerySpec*>& queries,
		std::list<PostSpec*>& posts,
		std::list<SignalSpec*>& signals,
		std::list<uint16>& cranks) {

	uint32 rCount = (uint32)retrieves.size();
	uint32 qCount = (uint32)queries.size();
	uint32 pCount = (uint32)posts.size();
	uint32 sCount = (uint32)signals.size();
	uint16 crankID = 0;
	if (cranks.size())
		crankID = cranks.front();

	char* func = NULL, *lib;
	if (crankID == 0) {
		// See if the component has a function attached
		lib = new char[MAXFILENAMELEN+1];
		if (func = strstr(spec->setup->function, "::")) {
			memcpy(lib, spec->setup->function, func-spec->setup->function);
			lib[func-spec->setup->function] = 0;
			func += 2;
		}

		if (func) {
			if (!(crankID = syncCrankID(spec->setup->componentID, utils::StringFormat("%sComponentCrank", spec->setup->name).c_str(), func, lib, "", ""))) {
				LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for '%s::%s'...", lib, func);
				return false;
			}
		}
		else {
			// No crank provided, leave id = 0 to use the built-in simple crank
			//if (!(crankID = syncCrankID(spec->setup->componentID, utils::StringFormat("%sComponentCrank", spec->setup->name).c_str(), "", "", ""))) {
			//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for '%s'...", spec->setup->function);
			//	return false;
			//}
		}
		delete [] lib;

		//if (crankID = idManager->getNextCrankID()) {

		//	if (func) {
		//		if (!manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID, 
		//			utils::StringFormat("%sComponentCrank", spec->setup->name).c_str(), func, lib, ""))
		//			crankID = 0;
		//		if (!registerCrankID(crankID, spec->setup->componentID, 
		//			utils::StringFormat("%sComponentCrank", spec->setup->name).c_str(), func, lib, ""))
		//			crankID = 0;
		//	}
		//	else {
		//		// Enter a fresh empty crank with a real crankID to be filled in later...
		//		if (!manager->dataMapsMemory->createNewCrank(crankID, spec->setup->componentID,
		//			utils::StringFormat("%sComponentCrank", spec->setup->name).c_str(), "", "", ""))
		//			crankID = 0;
		//		if (!registerCrankID(crankID, spec->setup->componentID,
		//			utils::StringFormat("%sComponentCrank", spec->setup->name).c_str(), "", "", ""))
		//			crankID = 0;
		//	}
		//}
		//else {
		//	LogPrint(networking->localNodeID, LOG_NODE, 0, "Could not reserve crank id for '%s'...", spec->setup->function);
		//}
	}

	TriggerSpec* triggerSpec;
	uint32 addSize = (rCount*sizeof(RetrieveSpec)) + (qCount*sizeof(QuerySpec)) + (pCount*sizeof(PostSpec)) + (sCount*sizeof(SignalSpec));

	char* data;
	uint32 origSize;
	std::list<TriggerSpec*>::iterator tIt, tItEnd;
	std::list<RetrieveSpec*>::iterator rIt, rItEnd;
	std::list<QuerySpec*>::iterator qIt, qItEnd;
	std::list<PostSpec*>::iterator pIt, pItEnd;
	std::list<SignalSpec*>::iterator sIt, sItEnd;

	// Add retrieves and posts to each trigger
	for (tIt = triggers.begin(), tItEnd = triggers.end(); tIt != tItEnd; tIt++) {
		origSize = (*tIt)->size;
		triggerSpec = (TriggerSpec*) malloc(origSize + addSize);
		data = (char*) triggerSpec;
		memcpy(data, *tIt, origSize);
		triggerSpec->cid = TRIGGERDATA;
		triggerSpec->size = origSize + addSize;
		triggerSpec->crankID = crankID;
		triggerSpec->retrieveCount = rCount;
		triggerSpec->queryCount = qCount;
		triggerSpec->postCount = pCount;
		triggerSpec->signalCount = sCount;
		// Filters are already in there
		data += origSize;
		for (rIt = retrieves.begin(), rItEnd = retrieves.end(); rIt != rItEnd; rIt++) {
			memcpy(data, *rIt, sizeof(RetrieveSpec));
			data += sizeof(RetrieveSpec);
		}
		for (qIt = queries.begin(), qItEnd = queries.end(); qIt != qItEnd; qIt++) {
			memcpy(data, *qIt, sizeof(QuerySpec));
			data += sizeof(QuerySpec);
		}
		for (pIt = posts.begin(), pItEnd = posts.end(); pIt != pItEnd; pIt++) {
			memcpy(data, *pIt, sizeof(PostSpec));
			data += sizeof(PostSpec);
		}
		for (sIt = signals.begin(), sItEnd = signals.end(); sIt != sItEnd; sIt++) {
			memcpy(data, *sIt, sizeof(SignalSpec));
			data += sizeof(SignalSpec);
		}
		// Add triggers to component
		spec->triggers.push_back(triggerSpec);
	}
	// Delete content of lists
	tItEnd = triggers.end();
	while ((tIt = triggers.begin()) != tItEnd) {
		delete(*tIt);
		triggers.pop_front();
	}
	rItEnd = retrieves.end();
	while ((rIt = retrieves.begin()) != rItEnd) {
		delete(*rIt);
		retrieves.pop_front();
	}
	qItEnd = queries.end();
	while ((qIt = queries.begin()) != qItEnd) {
		delete(*qIt);
		queries.pop_front();
	}
	pItEnd = posts.end();
	while ((pIt = posts.begin()) != pItEnd) {
		delete(*pIt);
		posts.pop_front();
	}
	sItEnd = signals.end();
	while ((sIt = signals.begin()) != sItEnd) {
		delete(*sIt);
		signals.pop_front();
	}
	cranks.clear();
	return true;
}

bool Node::registerNewActiveContext(const PsyContext& activeContext, const PsyContext& inactiveContext) {
	if (inactiveContext != NOCONTEXT)
		LogPrint(networking->localNodeID, LOG_NODE, 2, "*** New context: %s     Inactive: %s", manager->contextToText(activeContext).c_str(), manager->contextToText(inactiveContext).c_str());
	else
		LogPrint(networking->localNodeID, LOG_NODE, 2, "*** New context: %s", manager->contextToText(activeContext).c_str());
	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_CONTEXT_ACTIVE, nodeID);
	msg->setData("ActiveContext", (char*)&activeContext, sizeof(PsyContext));
//	if (inactiveContext != NOCONTEXT)
	msg->setData("InactiveContext", (char*)&inactiveContext, sizeof(PsyContext));
	networking->sendMessageToAllOtherNodes(msg, true);
	// Send out context signals
	msg = new DataMessage(*(PsyType*)&activeContext, nodeID);
	manager->processMemory->addToSigQ(0, msg); // just add to own queue and let thread handle it
	delete(msg);
	return true;
}



bool Node::gatherLocalPerformanceStats(std::list<PerfStats> &perfStats) {
	addLocalNodePerformanceStats(perfStats);
	manager->processMemory->addLocalPerformanceStats(perfStats);
	manager->componentMemory->addLocalPerformanceStats(perfStats);
	return true;
}

bool Node::addLocalNodePerformanceStats(std::list<PerfStats> &perfStats) {
	uint64 sysAlloc, sysUsage, dataAlloc, dataUsage;
	manager->getMemoryUsage(sysAlloc, sysUsage, dataAlloc, dataUsage);

	PerfStats stats;
	memset(&stats, 0, sizeof(PerfStats));
	stats.nodeID = nodeID;
	stats.osID = osProcID;
	// stats.type = stats.node;
	stats.currentCPUTicks = currentCPUTicks;
	//stats.currentMemoryBytes = sysAlloc + dataAlloc;
	stats.currentMemoryBytes = utils::GetProcessMemoryUsage();
	stats.totalInputBytes = procBytes;
	stats.totalInputCount = procCount;
	stats.totalOutputBytes = toBytes + deliverBytes;
	stats.totalOutputCount = toCount + deliverCount;
	//stats.runCount = 0;
	manager->processMemory->getQueueSizes(0, stats.totalQueueBytes, stats.totalQueueCount);
	//stats.currentRunStartTime = 0;
	stats.firstRunStartTime = createTime;
	stats.totalCycleCount = procCount;
	//stats.totalRunCount = 0;
	//stats.migrationCount = 0;
	perfStats.push_back(stats);
	return true;
}

std::map<std::string, std::string> Node::getComponentPrivateDataKeysAndTypes(uint32 compID) {
	std::map<std::string, std::string> map;
	uint16 compNodeID;
	if (!(compNodeID = manager->componentMemory->getComponentNodeID(compID)))
		return map;

	if (compNodeID == this->nodeID)
		return manager->componentMemory->getPrivateDataKeysAndTypes(compID);

	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
	msg->setReference(QUERYNODE_GETCOMPPRIVATEKEYSTYPES);
	msg->setInt("CompID", compID);
	DataMessage* replyMsg = NULL;
	if (networking->queryNode(compNodeID, msg, NODEQUERY_TIMEOUT_MS, true, &replyMsg) != SYNC_SUCCESS) {
		LogPrint(networking->localNodeID, LOG_NODE, 0, "Didn't get Component Private Data keys back from Node %u...", compNodeID);
		delete(replyMsg);
		return map;
	}

	map = msg->getStringMap("ComponentPrivateDataKeysAndTypes");
	delete(replyMsg);
	return map;
}


char* Node::getComponentPrivateData(uint32 compID, const char* dataName, uint64 &size, std::string &mimetype) {
	uint16 compNodeID;
	if (! (compNodeID = manager->componentMemory->getComponentNodeID(compID)))
		return NULL;

	if (compNodeID == this->nodeID) {
		mimetype = manager->componentMemory->getPrivateDataMimetype(compID, dataName);
		return manager->componentMemory->getPrivateDataCopy(compID, dataName, size);
	}

	DataMessage* msg = new DataMessage(Networking::CTRL_NODE_QUERY, nodeID);
	msg->setReference(QUERYNODE_GETCOMPPRIVATEDATA);
	msg->setInt("CompID", compID);
	msg->setString("DataName", dataName);
	DataMessage* replyMsg = NULL;
	if (networking->queryNode(compNodeID, msg, NODEQUERY_TIMEOUT_MS, true, &replyMsg) != SYNC_SUCCESS) {
		LogPrint(networking->localNodeID, LOG_NODE, 0,"Didn't get Component Private Data back from Node %u...", compNodeID);
		delete(replyMsg);
		return NULL;
	}

	char* data;
	uint32 size32;
	if (!(data = replyMsg->getDataCopy("ComponentPrivateData", size32))) {
		LogPrint(networking->localNodeID, LOG_NODE, 0,"Component Private Data Pack reply back from Node %u didn't contain any data...", compNodeID);
		delete(replyMsg);
		return NULL;
	}
	mimetype = replyMsg->getString("Mimetype");
	size = size32;
	delete(replyMsg);
	return data;
}

bool Node::initialiseControlTypes() {
	if (!manager)
		return false;

	uint64 alloc=0, usage=0;
	manager->processMemory->getMemoryUsage(alloc, usage);

	uint16 existingID = 0;
	manager->dataMapsMemory->createNewTypeLevel(10100, "CTRL_NODE_WELCOME", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10100);
	manager->dataMapsMemory->createNewTypeLevel(10101, "CTRL_NODE_TIMESYNC", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10101);
	manager->dataMapsMemory->createNewTypeLevel(10102, "CTRL_NODE_NODEMAP", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10102);
	manager->dataMapsMemory->createNewTypeLevel(10104, "CTRL_NODE_SYNC", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10104);
	manager->dataMapsMemory->createNewTypeLevel(10105, "CTRL_NODE_CONFIG", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10105);
	manager->dataMapsMemory->createNewTypeLevel(10106, "CTRL_NODE_SYNC_ID", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10106);
	manager->dataMapsMemory->createNewTypeLevel(10107, "CTRL_NODE_CONTEXT_ACTIVE", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10107);
	manager->dataMapsMemory->createNewTypeLevel(10108, "CTRL_NODE_SIGNAL", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10108);
	manager->dataMapsMemory->createNewTypeLevel(10109, "CTRL_NODE_QUERY", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10109);
	manager->dataMapsMemory->createNewTypeLevel(10110, "CTRL_NODE_SYNC_SUBSCRIPTION", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10110);

	manager->dataMapsMemory->createNewTypeLevel(10001, "CTRL_TEST", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10001);
	manager->dataMapsMemory->createNewTypeLevel(10002, "CTRL_LOGPRINT", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10002);
	manager->dataMapsMemory->createNewTypeLevel(10003, "CTRL_PING", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10003);
	manager->dataMapsMemory->createNewTypeLevel(10004, "CTRL_PING_REPLY", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10004);

	manager->dataMapsMemory->createNewTypeLevel(10200, "CTRL_SYSTEM_READY", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10200);
	manager->dataMapsMemory->createNewTypeLevel(10201, "CTRL_PROCESS_INITIALISE", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10201);
	manager->dataMapsMemory->createNewTypeLevel(10202, "CTRL_PROCESS_GREETING", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10202);
	manager->dataMapsMemory->createNewTypeLevel(10203, "CTRL_PROCESS_SHUTDOWN", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10203);
	manager->dataMapsMemory->createNewTypeLevel(10297, "ContextChange", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10297);
	manager->dataMapsMemory->createNewTypeLevel(10298, "CTRL_SYSTEM_SHUTDOWN", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10298);
	manager->dataMapsMemory->createNewTypeLevel(10299, "CTRL_SYSTEM_SHUTTINGDOWN", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10299);
	manager->dataMapsMemory->createNewTypeLevel(10300, "CTRL_TRIGGER", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10300);
	manager->dataMapsMemory->createNewTypeLevel(10307, "CTRL_TRIGGER_GROUP", GetTimeNow(), existingID);	// T1.6
	manager->dataMapsMemory->confirmTypeLevelID(10307);
	manager->dataMapsMemory->createNewTypeLevel(10301, "CTRL_QUERY", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10301);
	manager->dataMapsMemory->createNewTypeLevel(10302, "CTRL_QUERY_REPLY", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10302);
	manager->dataMapsMemory->createNewTypeLevel(10303, "CTRL_PULLCOMPONENTDATA", GetTimeNow(), existingID);
	manager->dataMapsMemory->confirmTypeLevelID(10303);

	manager->processMemory->getMemoryUsage(alloc, usage);
	return true;
}


bool Node::testTypesContexts() {
	PsyType type21 = getPsyType("my.one");
	PsyType type22 = getPsyType("my.two");
	PsyType type31 = getPsyType("my.test.one");
	PsyType type32 = getPsyType("my.test.two");
	PsyType type3s = getPsyType("my.test.*");

	if (!type21.matches(type21))
		return false;
	if (type21.matches(type22))
		return false;
	if (type21.matches(type31))
		return false;
	if (type31.matches(type21))
		return false;
	if (!type3s.matches(type31))
		return false;
	if (!type3s.matches(type32))
		return false;

	PsyContext con1 = getPsyContext("my.one");
	PsyContext con2 = getPsyContext("my.test");
	PsyContext con31 = getPsyContext("my.test.one");
	PsyContext con32 = getPsyContext("my.test.two");
	PsyContext con43 = getPsyContext("my.test.two.three");

	if (con1.compare(con1) != CONTEXT_EQUAL)
		return false;
	if (con2.compare(con1) != CONTEXT_SUPERSEDES)
		return false;
	if (con1.compare(con2) != CONTEXT_SUPERSEDES)
		return false;
	if (con2.compare(con31) != CONTEXT_INCLUDED)
		return false;
	if (con31.compare(con2) != CONTEXT_SUPERSEDES)
		return false;
	if (con32.compare(con31) != CONTEXT_SUPERSEDES)
		return false;
	if (con43.compare(con32) != CONTEXT_SUPERSEDES)
		return false;
	if (con2.compare(con43) != CONTEXT_INCLUDED)
		return false;
	if (con31.compare(con43) != CONTEXT_SUPERSEDES)
		return false;
	if (con32.compare(con43) != CONTEXT_INCLUDED)
		return false;
	if (con1.compare(con43) != CONTEXT_SUPERSEDES)
		return false;

	return true;
}

THREAD_RET THREAD_FUNCTION_CALL NodeMessageThread(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((Node*)arg)->runMessageQ() ? 0 : 1));
}
THREAD_RET THREAD_FUNCTION_CALL NodeSignalThread(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((Node*)arg)->runSignalQ() ? 0 : 1));
}
THREAD_RET THREAD_FUNCTION_CALL NodeRequestThread(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((Node*)arg)->runRequestQ() ? 0 : 1));
}

}
