

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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.h
	@brief Request-system worker side: RequestExecutor pulls requests from a RequestGateway, processes them and replies.

	The executor is the "server" role of the request fabric (see
	RequestClient.h for the conceptual overview and a round-trip example).
	It connects out to one or more gateways (addGateway()), announces itself
	with periodic status heartbeats — the gateway uses these both for
	liveness (an executor missing heartbeats is declared dead and its queued
	requests are redistributed) and for load balancing — and receives
	requests over the binary message protocol.

	Requests are classified @b short (interactive, cheap) or @b long
	(expensive, e.g. heavy computation) by matching the request name against
	names registered with addLongRequestName(); the two classes have separate
	queues so long jobs cannot starve short ones, and the number of
	concurrent long requests is capped by setLongRequestLimit().

	Application worker threads call waitForShortRequest() /
	waitForLongRequest() to pull work and replyToQuery() to answer.

	@ingroup Requests
*/
#if !defined(_REQUESTEXECUTOR_H_)
#define _REQUESTEXECUTOR_H_

#pragma once

#include "NetworkManager.h"
#include "RequestClient.h"


namespace	cmlabs{

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

/*! Worker node of the request system: receives requests from gateways,
	queues them by class (short/long) and sends back replies.

	Owns its NetworkManager/NetworkChannel and a worker thread that keeps
	gateway links alive (reconnecting on failure) and emits heartbeats every
	::heartbeatIntervalMS. Application threads consume work via
	waitForShortRequest()/waitForLongRequest() and answer via replyToQuery().
	@par Thread-safety
	Public methods may be called from any thread; the wait queues support
	multiple concurrent worker threads.

	A minimal executor with one worker loop looks like this:
	@code
	RequestExecutor exec(7, "ImageWorker");
	exec.addLongRequestName("render");          // "render" requests go to the long queue
	exec.addGateway(1, "gateway.host", 9000);   // dial out to the gateway

	while (running) {
	    DataMessage* req = exec.waitForShortRequest(500); // owned by the executor!
	    if (!req) continue;                                // timeout, poll again
	    DataMessage* reply = new DataMessage(*req);        // copy keeps routing refs
	    reply->setString("Result", "ok");
	    exec.replyToQuery(reply);                          // ownership transfers
	    // do NOT delete req — the executor deletes it when the reply is sent
	}
	@endcode
	@warning Ownership footgun: the DataMessage returned by
	waitForShortRequest()/waitForLongRequest() is the @e same object the
	executor keeps in its internal request map (keyed by the message
	reference id); it is deleted by the executor itself when the matching
	reply is dispatched to the gateway. Deleting it in the worker causes a
	double-free. The reply passed to replyToQuery() must carry the request's
	reference id (which a field-for-field copy of the request preserves) so
	it can be routed back; replyToQuery() takes ownership of the reply.
	@ingroup Requests */
class RequestExecutor : public Runnable, public NetworkReceiver {
public:
	friend THREAD_RET THREAD_FUNCTION_CALL RequestExecutorRun(THREAD_ARG arg);

	/*! @param id Numeric executor id (reported to gateways).
		@param name Optional human-readable name (combined with @p id for logging). */
	RequestExecutor(uint32 id = 0, const char* name = NULL);
	virtual ~RequestExecutor();
	/*! Disconnect from all gateways and stop the network stack. @return true when done. */
	bool shutdownNetwork();

	/*! Register a request name to be treated as a long-running request.
		@param name Request name (matched against incoming requests). @return true on success. */
	bool addLongRequestName(const char* name);
	/*! Register a gateway to connect to (repeatable for redundancy).
		@param id Gateway id. @param addr Host. @param port Port.
		@param encryption ::NOENC or ::SSLENC. @return true on success. */
	bool addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption = NOENC);
	/*! Cap the number of long requests processed concurrently.
		@param limit Maximum concurrent long requests. @return true on success. */
	bool setLongRequestLimit(uint32 limit);

	/*! NetworkReceiver hook: gateway connect/disconnect events (drives reconnection). */
	bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid);
	/*! NetworkReceiver hook: incoming request messages, routed to the short/long queue. */
	bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid);
	/*! HTTP is not served by executors; always returns false. */
	bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid) {return false;}

	/*! Pull the next long request (worker thread API).
		@param timeoutMS Max wait in ms.
		@return The request message, or NULL on timeout. Blocking.
		@warning The message remains owned by the executor (see the class
		documentation) — reply via replyToQuery() and do not delete it. */
	DataMessage* waitForLongRequest(uint32 timeoutMS);
	/*! Pull the next short request (worker thread API).
		@param timeoutMS Max wait in ms.
		@return The request message, or NULL on timeout. Blocking.
		@warning The message remains owned by the executor (see the class
		documentation) — reply via replyToQuery() and do not delete it. */
	DataMessage* waitForShortRequest(uint32 timeoutMS);
	/*! Send a reply for a previously pulled request back to its gateway.
		@param msg Reply message carrying the original request reference id
		(ownership transfers). @return true if queued for sending; false when
		the reference does not match a pending request or its gateway is gone
		(in which case the caller still owns @p msg). */
	bool replyToQuery(DataMessage* msg);

protected:
	/*! Worker loop: gateway upkeep, heartbeats, reply dispatch. @return true to continue. */
	bool run();
	/*! Send a status heartbeat (queue sizes) to all gateways. @return Gateways notified. */
	uint32 sendStatusNow();
	/*! Send a message on one gateway link. @return true if sent. */
	bool sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con);

	std::list<RequestGatewayConnection> connections;	/*!< Configured gateways and their state. */
	//std::map<uint64, RequestGatewayConnection> gateways;

	/*! Route a finished reply to the gateway that sent the request. @return true if sent. */
	bool replyToGateway(DataMessage* msg);

	uint32 executorID;			/*!< Numeric id reported to gateways. */
	std::string executorName;	/*!< Display name for logging. */
	NetworkManager* manager;	/*!< Owned network stack. */
	NetworkChannel* channel;	/*!< Channel carrying the gateway links. */
	utils::WaitQueuePointer<DataMessage*> shortExecQ;	/*!< Pending short requests. */
	utils::WaitQueuePointer<DataMessage*> longExecQ;	/*!< Pending long requests. */
	utils::WaitQueuePointer<DataMessage*> replyQ;		/*!< Outgoing replies awaiting dispatch. */
	std::map<uint64,RequestReply*> requestMap;			/*!< In-flight request bookkeeping by ref id. */
	std::vector<std::string> longReqNames;				/*!< Request names classified as long. */
	utils::Mutex mutex;			/*!< Guards request bookkeeping. */
	utils::Mutex conMutex;		/*!< Guards ::connections. */
	utils::Semaphore shortExecQSemaphore;	/*!< Signals short-queue arrivals. */
	utils::Semaphore longExecQSemaphore;	/*!< Signals long-queue arrivals. */
	uint32 threadID;			/*!< Worker thread id. */
	uint64 lastRefID;			/*!< Last issued executor reference id. */
	uint64 repliedCount;		/*!< Total replies sent. */
	uint64 sentCount;			/*!< Total messages sent to gateways. */
	uint64 shortReceivedCount;	/*!< Total short requests received. */
	uint64 longReceivedCount;	/*!< Total long requests received. */
	int32 heartbeatIntervalMS;	/*!< Interval between status heartbeats (ms). */
	uint64 lastHeartbeat;		/*!< Time of last heartbeat (ms epoch). */
	uint32 longReqLimit;		/*!< Max concurrent long requests (see setLongRequestLimit()). */
	//MovingAverage shortAvgStats;
	//MovingAverage longAvgStats;
};

/*! Thread entry point for the RequestExecutor worker loop. @ingroup Requests */
THREAD_RET THREAD_FUNCTION_CALL RequestExecutorRun(THREAD_ARG arg);

/*! Test executor that services requests with a configurable artificial
	processing delay; used by RequestGateway::UnitTest().
	@ingroup Requests */
class TestRequestExecutor : public RequestExecutor {
public:
	friend THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg);
	friend THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg);
	TestRequestExecutor(uint32 id, uint32 processTimeMS);
protected:
	bool shortRequestRun();
	bool longRequestRun();
	uint32 id;
	uint32 processTimeMS;
};

THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg);

}

#endif // _REQUESTEXECUTOR_H_
