

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK 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/
//_/_/
//_/_/  and is released under the following
//_/_/  
//_/_/  EXTENDED BSD License
//_/_/
//_/_/  Redistribution and use in source and binary forms, with or without
//_/_/  modification, is permitted provided that the following conditions 
//_/_/  are met:
//_/_/
//_/_/  - Redistributions of source code must retain the above copyright 
//_/_/    and collaboration notice, this list of conditions and the following 
//_/_/    disclaimer.
//_/_/
//_/_/  - Redistributions in binary form must reproduce the above copyright 
//_/_/    notice, this list of conditions and the following disclaimer in the 
//_/_/    documentation and/or other materials provided with the distribution.
//_/_/
//_/_/  - Neither the name of its copyright holders nor the names of its 
//_/_/    contributors may be used to endorse or promote products derived from 
//_/_/    this software without specific prior written permission.
//_/_/
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
/*! @file RequestExecutor.cpp
	@brief Implementation of the request-system worker side: gateway link
	upkeep with heartbeats, short/long request queueing and the reply path,
	plus the delay-simulating TestRequestExecutor.
	@ingroup Requests */
#include "RequestExecutor.h"

namespace	cmlabs{

/////////////////////////////////////////////////////////////
// Request Executor
/////////////////////////////////////////////////////////////

RequestExecutor::RequestExecutor(uint32 id, const char* name) {
	executorID = id;
	if (name && id)
		executorName = utils::StringFormat("%s:%u", name, id);
	else if (name)
		executorName = name;
	else if (id)
		executorName = utils::StringFormat("Executor:%u", id);
	else
		executorName = "";

	lastRefID = 0;
	sentCount = 0;
	repliedCount = 0;
	shortReceivedCount = 0;
	longReceivedCount = 0;
	heartbeatIntervalMS = 3000;
	lastHeartbeat = 0;
	shouldContinue = true;
	isRunning = false;
	channel = NULL;
	manager = new NetworkManager();
	longReqLimit = 0;

	if (!ThreadManager::CreateThread(RequestExecutorRun, this, threadID, 0)) {
		shouldContinue = false;
	}
}

RequestExecutor::~RequestExecutor() {
	stop();
	shutdownNetwork();

	std::map<uint64, RequestReply*>::iterator i, e;
	mutex.enter(200, __FUNCTION__);
	for (i = requestMap.begin(), e = requestMap.end(); i != e; i++)
		delete(i->second);
	requestMap.clear();

	longExecQ.clear();
	shortExecQ.clear();

	delete(manager);
	mutex.leave();
}

bool RequestExecutor::shutdownNetwork() {
	if (channel)
		channel->shutdown();
	return true;
}

bool RequestExecutor::addLongRequestName(const char* name) {
	if (channel)
		return false;
	longReqNames.push_back(name);
	return true;
}

bool RequestExecutor::addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption) {
	RequestGatewayConnection con;
	con.clear();
	con.id = id;
	con.addr = addr;
	con.port = port;
	con.encryption = encryption;
	
	if (!mutex.enter(200, __FUNCTION__))
		return false;

	DataMessage* msgConnect = new DataMessage();
	msgConnect->setString("URI", "ExecutorConnect");
	msgConnect->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());

	if (!channel)
		channel = manager->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, 0, this, con.conID, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());
	else
		con.conID = channel->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());

	delete msgConnect;

	connections.push_back(con);
	mutex.leave();
	return true;

	//if (!con.conID) {
	//	mutex.leave();
	//	return false;
	//}

	

	//RequestGatewayConnection con;
	//con.clear();
	//con.id = id;
	//con.addr = addr;
	//con.port = port;
	//con.encryption = encryption;

	//if (!mutex.enter(200, __FUNCTION__))
	//	return false;

	//if (!channel)
	//	channel = manager->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.id, this, con.conID, con.location);
	//else
	//	con.conID = channel->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.location);

	//if (!con.conID) {
	//	mutex.leave();
	//	return false;
	//}

	//// Send connect message
	//DataMessage* msg = new DataMessage();
	//msg->setString("URI", "ExecutorConnect");
	//msg->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());
	//if (!channel->sendMessage(msg, con.conID)) {
	//	delete(msg);
	//	mutex.leave();
	//	return false;
	//}
	//
	//delete(msg);
	//gateways[con.conID] = con;
	//mutex.leave();
	//return true;
}


bool RequestExecutor::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
	uint64 now = GetTimeNow();
	std::list<RequestGatewayConnection>::iterator i, e = connections.end();
	if (!mutex.enter(200, __FUNCTION__))
		return false;
	i = connections.begin();
	while (i != e) {
		if ((*i).conID == conid) {
			switch (evt->type) {
			case NETWORKEVENT_DISCONNECT:
			case NETWORKEVENT_DISCONNECT_RETRYING:
				LogPrint(0, LOG_SYSTEM, 0, "[%s] Disconnected from gateway %u on %s:%u", executorName.c_str(), conid, (*i).addr.c_str(), (*i).port);
				(*i).lastFailTime = now;
				//(*i).conID = 0;
				break;
			case NETWORKEVENT_CONNECT:
			case NETWORKEVENT_RECONNECT:
				LogPrint(0, LOG_SYSTEM, 0, "[%s] Connected to gateway %u on %s:%u", executorName.c_str(), conid, (*i).addr.c_str(), (*i).port);
				(*i).lastFailTime = 0;
				(*i).lastConTime = now;
				break;
			default:
				break;
			}
		}
		i++;
	}
	delete evt;
	mutex.leave();
	return true;

	//std::map<uint64,RequestGatewayConnection>::iterator i;
	//if (!mutex.enter(200, __FUNCTION__))
	//	return false;
	//if ( (i = gateways.find(conid)) != gateways.end()) {
	//	switch(evt->type) {
	//		case NETWORKEVENT_DISCONNECT:
	//			i->second.lastFailTime = now;
	//			break;
	//		case NETWORKEVENT_RECONNECT:
	//			i->second.lastFailTime = 0;
	//			break;
	//		default:
	//			break;
	//	}
	//}
	//mutex.leave();
	//return true;
}

bool RequestExecutor::setLongRequestLimit(uint32 limit) {
	longReqLimit = limit;
	return true;
}

bool RequestExecutor::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
	// Network-thread ingress: wrap each request in a RequestReply that records
	// the gateway's reference id (gatewayRef) and connection (origin), restamp
	// the message with a local execRef used as the requestMap key, then
	// classify short vs long — by size when longReqLimit is set, otherwise by
	// substring match against the registered long-request names — and queue it
	// for the worker threads. Note the queued pointer is the same object owned
	// by the RequestReply; it is freed when the reply is sent (replyToGateway).
	if (!msg) return false;
	RequestReply* reply;
	RequestGatewayConnection con;
	bool isLongReq;
	//std::map<uint64,RequestGatewayConnection>::iterator i;
	uint64 now = GetTimeNow();
	const char* req = msg->getString("URI");
	//LogPrint(0, LOG_SYSTEM, 0, "      * Executor received message from Gateway %llu (%llu)", conid, msg->getReference());

	if (!mutex.enter(1000, __FUNCTION__)) {
		LogPrint(0, LOG_SYSTEM, 0, "Error locking Executor mutex");
		delete(msg);
		return true;
	}

	std::list<RequestGatewayConnection>::iterator i, e = connections.end();
	i = connections.begin();
	while (i != e) {
		if ((*i).conID == conid) {
			reply = new RequestReply();
			reply->origin = conid;
			reply->execRef = ++lastRefID;
			reply->gatewayRef = msg->getReference();
			msg->setReference(reply->execRef);
			reply->giveRequestMessage(msg);
			reply->setStatus(QUEUED);
			requestMap[reply->execRef] = reply;

			if (longReqLimit)
				isLongReq = (msg->getSize() >= longReqLimit);
			else
				isLongReq = false;
			if (!isLongReq) {
				std::vector<std::string>::iterator il = longReqNames.begin(), el = longReqNames.end();
				while (il != el) {
					if (utils::stristr(req, il->c_str())) {
						isLongReq = true;
						break;
					}
					il++;
				}
			}
			//isLongReq = ( std::find(longReqNames.begin(), longReqNames.end(), req) != longReqNames.end());
			if (isLongReq) {
				longExecQ.add(msg);
				//longExecQSemaphore.signal();
				longReceivedCount++;
			}
			else {
				shortExecQ.add(msg);
				//shortExecQSemaphore.signal();
				shortReceivedCount++;
			}
			sendStatusNow();
			mutex.leave();
			return true;
		}
		i++;
	}

	delete(msg);
	mutex.leave();
	return true;
}

DataMessage* RequestExecutor::waitForLongRequest(uint32 timeoutMS) {
	DataMessage* msg;
	if (msg = longExecQ.waitForAndTakeFirst(timeoutMS)) {
		msg->setSendTime(GetTimeNow());
		return msg;
	}
	else
		return NULL;
}

DataMessage* RequestExecutor::waitForShortRequest(uint32 timeoutMS) {
	DataMessage* msg;
	if (msg = shortExecQ.waitForAndTakeFirst(timeoutMS)) {
		msg->setSendTime(GetTimeNow());
		return msg;
	}
	else
		return NULL;
}

bool RequestExecutor::replyToQuery(DataMessage* msg) {
	if (!msg) return false;
	uint64 execRef = msg->getReference();

	if (!mutex.enter(200, __FUNCTION__))
		return false;

	std::map<uint64,RequestReply*>::iterator i = requestMap.find(execRef);
	//std::map<uint64,RequestGatewayConnection>::iterator con;

	if (i == requestMap.end()) {
		mutex.leave();
		return false;
	}

	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		if ((*iCon).conID == i->second->origin) {
			replyQ.add(msg);
			repliedCount++;
			mutex.leave();
			return true;
		}
		iCon++;
	}

	mutex.leave();
	return false;
}


bool RequestExecutor::replyToGateway(DataMessage* msg) {
	if (!msg)
		return false;
	uint64 execRef = msg->getReference();

	if (!mutex.enter(200, __FUNCTION__))
		return false;

	std::map<uint64,RequestReply*>::iterator i = requestMap.find(execRef);
	std::map<uint64,RequestGatewayConnection>::iterator con;

	if (i == requestMap.end()) {
		LogPrint(0, LOG_SYSTEM, 0, "Executor couldn't find request to reply to (%llu)", execRef);
		mutex.leave();
		//delete(msg);
		return false;
	}
	RequestReply* reply = i->second;

	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		if ((*iCon).conID == reply->origin) {
			msg->setReference(reply->gatewayRef);
			if (sendMessageToGateway(msg, (*iCon))) {
				// success
				//delete(msg);
				requestMap.erase(i);
				delete(reply);
				sentCount++;
				mutex.leave();
				return true;
			}
			else {
				LogPrint(0, LOG_SYSTEM, 0, "   $ Executor couldn't reply to Gateway %llu (%llu)", iCon->conID, reply->gatewayRef);
				requestMap.erase(i);
				delete(reply);
				mutex.leave();
				return false;
			}
		}
		iCon++;
	}

	LogPrint(0, LOG_SYSTEM, 0, "   $ Executor couldn't find any Gateways to reply to (%llu)", reply->gatewayRef);
	requestMap.erase(i);
	delete(reply);
	mutex.leave();
	return false;
}


bool RequestExecutor::run() {
	// Worker loop with two duties: (1) drain replyQ — replies queued by
	// replyToQuery() from application worker threads — and push each back to
	// its originating gateway via replyToGateway(), which also erases the
	// request bookkeeping (deleting the RequestReply and the original request
	// message); (2) emit an ExecutorStatus heartbeat every heartbeatIntervalMS
	// so gateways can track liveness and queue depth for load balancing.
	// Incoming requests do NOT pass through here — they arrive on the network
	// thread in receiveMessage() and go straight into shortExecQ/longExecQ.
	DataMessage* msg;
	isRunning = true;

	sendStatusNow();

	while (shouldContinue) {
		while (msg = replyQ.waitForAndTakeFirst(50)) {
			if (!mutex.enter(200, __FUNCTION__)) {
				delete(msg);
				break;
			}
			if (!replyToGateway(msg)) {
				// #######
			}
			delete(msg);
			mutex.leave();
		}

		// Send status heartbeat
		if (!lastHeartbeat || (GetTimeAgeMS(lastHeartbeat) > heartbeatIntervalMS)) {
			sendStatusNow();
		}
	}
	isRunning = false;
	return true;
}


bool RequestExecutor::sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con) {

	uint64 now = GetTimeNow();

	// if (!channel || !channel->isConnected(con.conID))
	if (!channel)
		return false;

	// LogPrint(0, LOG_SYSTEM, 0, "Executor sending message to gateway...");

	//if (!con.conID || !channel) {
	//	if (!channel)
	//		channel = manager->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.id, this, con.conID, con.location);
	//	else
	//		con.conID = channel->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.location);
	//	if (!channel || !con.conID) {
	//		if (!con.lastFailTime)
	//			LogPrint(0, LOG_SYSTEM, 0, "[%s] Unable to connect to gateway %u on %s:%u, will retry later", executorName.c_str(), con.id, con.addr.c_str(), con.port);
	//		con.lastFailTime = now;
	//		return false;
	//	}
	//	DataMessage* msgConnect = new DataMessage();
	//	msgConnect->setString("URI", "ExecutorConnect");
	//	msgConnect->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());
	//	if (!channel->sendMessage(msgConnect, con.conID)) {
	//		LogPrint(0, LOG_SYSTEM, 0, "[%s] Unable to communicate with gateway %u on %s:%u, will retry later", executorName.c_str(), con.id, con.addr.c_str(), con.port);
	//		con.lastFailTime = now;
	//		// clearly not working, disconnect and try again later
	//		channel->endConnection(con.conID);
	//		con.conID = 0;
	//		delete msgConnect;
	//		return false;
	//	}
	//	delete msgConnect;
	//	if (!con.lastConTime)
	//		LogPrint(0, LOG_SYSTEM, 0, "[%s] Connected to gateway %u on %s:%u", executorName.c_str(), con.id, con.addr.c_str(), con.port);
	//	else
	//		LogPrint(0, LOG_SYSTEM, 0, "[%s] Reconnected to gateway %u on %s:%u", executorName.c_str(), con.id, con.addr.c_str(), con.port);
	//	con.lastConTime = now;
	//}
	if (!channel->sendMessage(msg, con.conID)) {
		if (!con.lastFailTime)
			LogPrint(0, LOG_SYSTEM, 0, "[%s] Unable to send data to gateway %u on %s:%u, disconnecting...", executorName.c_str(), con.id, con.addr.c_str(), con.port);
		con.lastFailTime = GetTimeNow();
		// clearly not working, disconnect and try again later
		//channel->endConnection(con.conID);
		//con.conID = 0;
		return false;
	}
	//if (msg->getReference())
	//	LogPrint(0, LOG_SYSTEM, 0, "      + Executor replying to Gateway %u (%llu)", con.id, msg->getReference());
	//else
	//	LogPrint(0, LOG_SYSTEM, 0, "      + Executor sending Heartbeat to Gateway %u", con.id);
	return true;
}

uint32 RequestExecutor::sendStatusNow() {
	// Heartbeat: broadcasts an ExecutorStatus message carrying the current
	// short/long queue depths to every configured gateway. The gateway feeds
	// these into its RequestConnection records for load balancing, and treats
	// a missing heartbeat (executorHeartbeatTimeout) as executor death.

	if (!conMutex.enter(1000, __FUNCTION__)) {
		LogPrint(0, LOG_SYSTEM, 0, "Mutex error sending heartbeat from Executor %u...", executorID);
		return 0;
	}
	if (!connections.size()) {
		//LogPrint(0, LOG_SYSTEM, 0, "No connections sending heartbeat from Executor %u...", executorID);
		conMutex.leave();
		return 0;
	}

	DataMessage* msg = new DataMessage();
	msg->setString("URI", "ExecutorStatus");
	msg->setInt("ShortQSize", shortExecQ.getCount());
	msg->setInt("LongQSize", longExecQ.getCount());

	uint32 success = 0, failed = 0, notConnected = 0;

	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		if (sendMessageToGateway(msg, (*iCon)))
			success++;
		else
			failed++;
		iCon++;
	}
	conMutex.leave();

	delete(msg);
	lastHeartbeat = GetTimeNow();
	if (failed)
		LogPrint(0, LOG_SYSTEM, 2, "Failed sending heartbeat from Executor %u to %u gateways (%u succeeded)...", executorID, failed, success);
//	else
//		LogPrint(0, LOG_SYSTEM, 0, "Sent heartbeat from Executor %u to %u gateways...", executorID, success);
	return success;
}

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





TestRequestExecutor::TestRequestExecutor(uint32 id, uint32 processTimeMS) : RequestExecutor(id) {
	this->id = id;
	this->processTimeMS = processTimeMS;
	shouldContinue = true;
	uint32 myThreadID;
	ThreadManager::CreateThread(ShortExecutorRun, this, myThreadID, 0);
	ThreadManager::CreateThread(LongExecutorRun, this, myThreadID, 0);
}


bool TestRequestExecutor::shortRequestRun() {
	uint32 count = 0;
	DataMessage* replyMsg, *msg;
	while (shouldContinue) {
		if (msg = waitForShortRequest(20)) {
			replyMsg = new DataMessage();
			replyMsg->setReference(msg->getReference());
			replyToQuery(replyMsg);
		}
	}
	return true;
}

bool TestRequestExecutor::longRequestRun() {
	uint32 count = 0;
	DataMessage* replyMsg, *msg;
	std::list<DataMessage*> storedMessages;
	std::list<DataMessage*>::iterator i, e = storedMessages.end();

	while (shouldContinue) {
		if (msg = waitForLongRequest(20)) {
			if (!processTimeMS) {
				replyMsg = new DataMessage();
				replyMsg->setReference(msg->getReference());
				replyToQuery(replyMsg);
				//printf("--- Replying immediately ---\n");
			}
			else {
				msg->setRecvTime(GetTimeNow());
				storedMessages.push_back(msg);
			}
		}

		i = storedMessages.begin();
		while (i != e) {
			if ((msg = (*i)) && (GetTimeAgeMS(msg->getRecvTime()) >= (int32)processTimeMS)) {
				i = storedMessages.erase(i);
				replyMsg = new DataMessage();
				replyMsg->setReference(msg->getReference());
				replyToQuery(replyMsg);
				//printf("--- Replying delayed ---\n");
			}
			else
				i++;
		}
	}
	return true;
}

THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((TestRequestExecutor*)arg)->shortRequestRun() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((TestRequestExecutor*)arg)->longRequestRun() ? 0 : 1));
}

}
