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

#include <RequestExecutor.h>

Inheritance diagram for cmlabs::RequestExecutor:

[legend]

Collaboration diagram for cmlabs::RequestExecutor:

[legend]

Public Member Functions
	RequestExecutor (uint32 id=0, const char *name=NULL)
virtual	~RequestExecutor ()
bool	shutdownNetwork ()
	Disconnect from all gateways and stop the network stack.
bool	addLongRequestName (const char *name)
	Register a request name to be treated as a long-running request.
bool	addGateway (uint32 id, std::string addr, uint16 port, uint8 encryption=NOENC)
	Register a gateway to connect to (repeatable for redundancy).
bool	setLongRequestLimit (uint32 limit)
	Cap the number of long requests processed concurrently.
bool	receiveNetworkEvent (NetworkEvent evt, NetworkChannel channel, uint64 conid)
	NetworkReceiver hook: gateway connect/disconnect events (drives reconnection).
bool	receiveMessage (DataMessage msg, NetworkChannel channel, uint64 conid)
	NetworkReceiver hook: incoming request messages, routed to the short/long queue.
bool	receiveHTTPRequest (HTTPRequest req, NetworkChannel channel, uint64 conid)
	HTTP is not served by executors; always returns false.
DataMessage *	waitForLongRequest (uint32 timeoutMS)
	Pull the next long request (worker thread API).
DataMessage *	waitForShortRequest (uint32 timeoutMS)
	Pull the next short request (worker thread API).
bool	replyToQuery (DataMessage *msg)
	Send a reply for a previously pulled request back to its gateway.
Public Member Functions inherited from cmlabs::Runnable
	Runnable ()
	Initialise flags: not running, allowed to continue.
virtual	~Runnable ()
	Destructor requests a stop (with the default timeout) before destruction proceeds.
virtual bool	stop (uint32 timeout=200)
	Ask the worker loop to finish and wait for it to do so.
Public Member Functions inherited from cmlabs::NetworkReceiver
	NetworkReceiver ()
virtual bool	receiveTelnetLine (TelnetLine line, NetworkChannel channel, uint64 conid)
	A Telnet line arrived.
virtual bool	receiveHTTPReply (HTTPReply reply, HTTPRequest req, NetworkChannel *channel, uint64 conid)
	An HTTP reply to an earlier async request arrived.
virtual bool	receiveWebsocketData (WebsocketData wsData, NetworkChannel channel, uint64 conid)
	A WebSocket message arrived.

Protected Member Functions
bool	run ()
	Worker loop: gateway upkeep, heartbeats, reply dispatch.
uint32	sendStatusNow ()
	Send a status heartbeat (queue sizes) to all gateways.
bool	sendMessageToGateway (DataMessage *msg, RequestGatewayConnection &con)
	Send a message on one gateway link.
bool	replyToGateway (DataMessage *msg)
	Route a finished reply to the gateway that sent the request.

Protected Attributes
std::list< RequestGatewayConnection >	connections
	Configured gateways and their state.
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()).
Protected Attributes inherited from cmlabs::Runnable
uint32	threadID
	ThreadManager slot ID of the worker thread (0 until known).
bool	shouldContinue
	Loop-continuation flag; cleared by stop().
bool	isRunning
	Set by the worker while its loop is active.

Friends
THREAD_RET THREAD_FUNCTION_CALL	RequestExecutorRun (THREAD_ARG arg)
	Thread entry point for the RequestExecutor worker loop.

Detailed Description

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().

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:

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
}

Warning: Ownership footgun: the DataMessage returned by waitForShortRequest()/waitForLongRequest() is the 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.

Definition at line 73 of file RequestExecutor.h.

Constructor & Destructor Documentation

◆ RequestExecutor()

cmlabs::RequestExecutor::RequestExecutor	(	uint32	id = 0,
		const char *	name = NULL )

Parameters

id	Numeric executor id (reported to gateways).
name	Optional human-readable name (combined with `id` for logging).

Definition at line 14 of file RequestExecutor.cpp.

References channel, cmlabs::ThreadManager::CreateThread(), executorID, executorName, heartbeatIntervalMS, cmlabs::Runnable::isRunning, lastHeartbeat, lastRefID, longReceivedCount, longReqLimit, manager, repliedCount, RequestExecutorRun, sentCount, shortReceivedCount, cmlabs::Runnable::shouldContinue, cmlabs::utils::StringFormat(), and threadID.

Referenced by RequestExecutorRun, and cmlabs::TestRequestExecutor::TestRequestExecutor().

◆ ~RequestExecutor()

cmlabs::RequestExecutor::~RequestExecutor ( )

virtual

Definition at line 43 of file RequestExecutor.cpp.

References longExecQ, manager, mutex, requestMap, shortExecQ, shutdownNetwork(), and cmlabs::Runnable::stop().

Member Function Documentation

◆ addGateway()

bool cmlabs::RequestExecutor::addGateway	(	uint32	id,
		std::string	addr,
		uint16	port,
		uint8	encryption = NOENC )

Register a gateway to connect to (repeatable for redundancy).

Parameters

id	Gateway id.
addr	Host.
port	Port.
encryption	NOENC or SSLENC.

Returns: true on success.

Definition at line 73 of file RequestExecutor.cpp.

References cmlabs::RequestGatewayConnection::addr, channel, cmlabs::RequestGatewayConnection::clear(), cmlabs::RequestGatewayConnection::conID, connections, cmlabs::DataMessage::data, cmlabs::RequestGatewayConnection::encryption, cmlabs::DataMessage::getSize(), cmlabs::RequestGatewayConnection::id, cmlabs::RequestGatewayConnection::location, longReqNames, manager, mutex, cmlabs::RequestGatewayConnection::port, PROTOCOL_MESSAGE, cmlabs::DataMessage::setString(), and cmlabs::utils::TextVectorConcat().

Referenced by cmlabs::RequestGateway::UnitTest().

◆ addLongRequestName()

bool cmlabs::RequestExecutor::addLongRequestName ( const char * name )

Register a request name to be treated as a long-running request.

Parameters

name	Request name (matched against incoming requests).

Returns: true on success.

Definition at line 66 of file RequestExecutor.cpp.

References channel, and longReqNames.

Referenced by cmlabs::RequestGateway::UnitTest().

◆ receiveHTTPRequest()

bool cmlabs::RequestExecutor::receiveHTTPRequest	(	HTTPRequest *	req,
		NetworkChannel *	channel,
		uint64	conid )

inlinevirtual

HTTP is not served by executors; always returns false.

Reimplemented from cmlabs::NetworkReceiver.

Definition at line 100 of file RequestExecutor.h.

References channel.

◆ receiveMessage()

bool cmlabs::RequestExecutor::receiveMessage	(	DataMessage *	msg,
		NetworkChannel *	channel,
		uint64	conid )

virtual

NetworkReceiver hook: incoming request messages, routed to the short/long queue.

Reimplemented from cmlabs::NetworkReceiver.

Definition at line 198 of file RequestExecutor.cpp.

References channel, connections, cmlabs::RequestReply::execRef, cmlabs::RequestReply::gatewayRef, cmlabs::DataMessage::getReference(), cmlabs::DataMessage::getSize(), cmlabs::DataMessage::getString(), cmlabs::GetTimeNow(), cmlabs::RequestReply::giveRequestMessage(), lastRefID, LOG_SYSTEM, LogPrint, longExecQ, longReceivedCount, longReqLimit, longReqNames, mutex, cmlabs::RequestReply::origin, cmlabs::QUEUED, requestMap, sendStatusNow(), cmlabs::DataMessage::setReference(), cmlabs::RequestReply::setStatus(), shortExecQ, shortReceivedCount, and cmlabs::utils::stristr().

◆ receiveNetworkEvent()

bool cmlabs::RequestExecutor::receiveNetworkEvent	(	NetworkEvent *	evt,
		NetworkChannel *	channel,
		uint64	conid )

virtual

NetworkReceiver hook: gateway connect/disconnect events (drives reconnection).

Reimplemented from cmlabs::NetworkReceiver.

Definition at line 143 of file RequestExecutor.cpp.

References channel, connections, executorName, cmlabs::GetTimeNow(), LOG_SYSTEM, LogPrint, mutex, NETWORKEVENT_CONNECT, NETWORKEVENT_DISCONNECT, NETWORKEVENT_DISCONNECT_RETRYING, NETWORKEVENT_RECONNECT, and cmlabs::NetworkEvent::type.

◆ replyToGateway()

bool cmlabs::RequestExecutor::replyToGateway ( DataMessage * msg )

protected

Route a finished reply to the gateway that sent the request.

Returns: true if sent.

Definition at line 323 of file RequestExecutor.cpp.

References connections, cmlabs::RequestReply::gatewayRef, cmlabs::DataMessage::getReference(), LOG_SYSTEM, LogPrint, mutex, cmlabs::RequestReply::origin, requestMap, sendMessageToGateway(), sentCount, and cmlabs::DataMessage::setReference().

Referenced by run().

◆ replyToQuery()

bool cmlabs::RequestExecutor::replyToQuery ( DataMessage * msg )

Send a reply for a previously pulled request back to its gateway.

Parameters

msg	Reply message carrying the original request reference id (ownership transfers).

Returns: 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 msg).

Definition at line 291 of file RequestExecutor.cpp.

References connections, cmlabs::DataMessage::getReference(), mutex, repliedCount, replyQ, and requestMap.

Referenced by cmlabs::TestRequestExecutor::longRequestRun(), and cmlabs::TestRequestExecutor::shortRequestRun().

◆ run()

bool cmlabs::RequestExecutor::run ( )

protected

Worker loop: gateway upkeep, heartbeats, reply dispatch.

Returns: true to continue.

Definition at line 375 of file RequestExecutor.cpp.

References cmlabs::GetTimeAgeMS(), heartbeatIntervalMS, cmlabs::Runnable::isRunning, lastHeartbeat, mutex, replyQ, replyToGateway(), sendStatusNow(), and cmlabs::Runnable::shouldContinue.

Referenced by RequestExecutorRun.

◆ sendMessageToGateway()

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

protected

Send a message on one gateway link.

Returns: true if sent.

Definition at line 412 of file RequestExecutor.cpp.

References cmlabs::RequestGatewayConnection::addr, channel, cmlabs::RequestGatewayConnection::conID, executorName, cmlabs::GetTimeNow(), cmlabs::RequestGatewayConnection::id, cmlabs::RequestGatewayConnection::lastFailTime, LOG_SYSTEM, LogPrint, and cmlabs::RequestGatewayConnection::port.

Referenced by replyToGateway(), and sendStatusNow().

◆ sendStatusNow()

uint32 cmlabs::RequestExecutor::sendStatusNow ( )

protected

Send a status heartbeat (queue sizes) to all gateways.

Returns: Gateways notified.

Definition at line 468 of file RequestExecutor.cpp.

References conMutex, connections, executorID, cmlabs::GetTimeNow(), lastHeartbeat, LOG_SYSTEM, LogPrint, longExecQ, sendMessageToGateway(), cmlabs::DataMessage::setInt(), cmlabs::DataMessage::setString(), and shortExecQ.

Referenced by receiveMessage(), and run().

◆ setLongRequestLimit()

bool cmlabs::RequestExecutor::setLongRequestLimit ( uint32 limit )

Cap the number of long requests processed concurrently.

Parameters

limit Maximum concurrent long requests.

Returns: true on success.

Definition at line 193 of file RequestExecutor.cpp.

References longReqLimit.

◆ shutdownNetwork()

bool cmlabs::RequestExecutor::shutdownNetwork ( )

Disconnect from all gateways and stop the network stack.

Returns: true when done.

Definition at line 60 of file RequestExecutor.cpp.

References channel.

Referenced by ~RequestExecutor().

◆ waitForLongRequest()

DataMessage * cmlabs::RequestExecutor::waitForLongRequest ( uint32 timeoutMS )

Pull the next long request (worker thread API).

Parameters

timeoutMS Max wait in ms.

Returns: 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.

Definition at line 271 of file RequestExecutor.cpp.

References cmlabs::GetTimeNow(), longExecQ, and cmlabs::DataMessage::setSendTime().

Referenced by cmlabs::TestRequestExecutor::longRequestRun().

◆ waitForShortRequest()

DataMessage * cmlabs::RequestExecutor::waitForShortRequest ( uint32 timeoutMS )

Pull the next short request (worker thread API).

Parameters

timeoutMS Max wait in ms.

Returns: 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.

Definition at line 281 of file RequestExecutor.cpp.

References cmlabs::GetTimeNow(), cmlabs::DataMessage::setSendTime(), and shortExecQ.

Referenced by cmlabs::TestRequestExecutor::shortRequestRun().

◆ RequestExecutorRun

THREAD_RET THREAD_FUNCTION_CALL RequestExecutorRun ( THREAD_ARG arg )

friend

Thread entry point for the RequestExecutor worker loop.

Definition at line 511 of file RequestExecutor.cpp.

References RequestExecutor(), run(), THREAD_ARG, THREAD_FUNCTION_CALL, THREAD_RET, and thread_ret_val.

Referenced by RequestExecutor().

Member Data Documentation

◆ channel

NetworkChannel* cmlabs::RequestExecutor::channel

protected

Channel carrying the gateway links.

Definition at line 138 of file RequestExecutor.h.

Referenced by addGateway(), addLongRequestName(), receiveHTTPRequest(), receiveMessage(), receiveNetworkEvent(), RequestExecutor(), sendMessageToGateway(), and shutdownNetwork().

◆ conMutex

utils::Mutex cmlabs::RequestExecutor::conMutex

protected

Guards ::connections.

Definition at line 145 of file RequestExecutor.h.

Referenced by sendStatusNow().

◆ connections

std::list<RequestGatewayConnection> cmlabs::RequestExecutor::connections

protected

Configured gateways and their state.

Definition at line 129 of file RequestExecutor.h.

Referenced by addGateway(), receiveMessage(), receiveNetworkEvent(), replyToGateway(), replyToQuery(), and sendStatusNow().

◆ executorID

uint32 cmlabs::RequestExecutor::executorID

protected

Numeric id reported to gateways.

Definition at line 135 of file RequestExecutor.h.

Referenced by RequestExecutor(), and sendStatusNow().

◆ executorName

std::string cmlabs::RequestExecutor::executorName

protected

Display name for logging.

Definition at line 136 of file RequestExecutor.h.

Referenced by receiveNetworkEvent(), RequestExecutor(), and sendMessageToGateway().

◆ heartbeatIntervalMS

int32 cmlabs::RequestExecutor::heartbeatIntervalMS

protected

Interval between status heartbeats (ms).

Definition at line 154 of file RequestExecutor.h.

Referenced by RequestExecutor(), and run().

◆ lastHeartbeat

uint64 cmlabs::RequestExecutor::lastHeartbeat

protected

Time of last heartbeat (ms epoch).

Definition at line 155 of file RequestExecutor.h.

Referenced by RequestExecutor(), run(), and sendStatusNow().

◆ lastRefID

uint64 cmlabs::RequestExecutor::lastRefID

protected

Last issued executor reference id.

Definition at line 149 of file RequestExecutor.h.

Referenced by receiveMessage(), and RequestExecutor().

◆ longExecQ

utils::WaitQueuePointer<DataMessage*> cmlabs::RequestExecutor::longExecQ

protected

Pending long requests.

Definition at line 140 of file RequestExecutor.h.

Referenced by receiveMessage(), sendStatusNow(), waitForLongRequest(), and ~RequestExecutor().

◆ longExecQSemaphore

utils::Semaphore cmlabs::RequestExecutor::longExecQSemaphore

protected

Signals long-queue arrivals.

Definition at line 147 of file RequestExecutor.h.

◆ longReceivedCount

uint64 cmlabs::RequestExecutor::longReceivedCount

protected

Total long requests received.

Definition at line 153 of file RequestExecutor.h.

Referenced by receiveMessage(), and RequestExecutor().

◆ longReqLimit

uint32 cmlabs::RequestExecutor::longReqLimit

protected

Max concurrent long requests (see setLongRequestLimit()).

Definition at line 156 of file RequestExecutor.h.

Referenced by receiveMessage(), RequestExecutor(), and setLongRequestLimit().

◆ longReqNames

std::vector<std::string> cmlabs::RequestExecutor::longReqNames

protected

Request names classified as long.

Definition at line 143 of file RequestExecutor.h.

Referenced by addGateway(), addLongRequestName(), and receiveMessage().

◆ manager

NetworkManager* cmlabs::RequestExecutor::manager

protected

Owned network stack.

Definition at line 137 of file RequestExecutor.h.

Referenced by addGateway(), RequestExecutor(), and ~RequestExecutor().

◆ mutex

utils::Mutex cmlabs::RequestExecutor::mutex

protected

Guards request bookkeeping.

Definition at line 144 of file RequestExecutor.h.

Referenced by addGateway(), receiveMessage(), receiveNetworkEvent(), replyToGateway(), replyToQuery(), run(), and ~RequestExecutor().

◆ repliedCount

uint64 cmlabs::RequestExecutor::repliedCount

protected

Total replies sent.

Definition at line 150 of file RequestExecutor.h.

Referenced by replyToQuery(), and RequestExecutor().

◆ replyQ

utils::WaitQueuePointer<DataMessage*> cmlabs::RequestExecutor::replyQ

protected

Outgoing replies awaiting dispatch.

Definition at line 141 of file RequestExecutor.h.

Referenced by replyToQuery(), and run().

◆ requestMap

std::map<uint64,RequestReply*> cmlabs::RequestExecutor::requestMap

protected

In-flight request bookkeeping by ref id.

Definition at line 142 of file RequestExecutor.h.

Referenced by receiveMessage(), replyToGateway(), replyToQuery(), and ~RequestExecutor().

◆ sentCount

uint64 cmlabs::RequestExecutor::sentCount

protected

Total messages sent to gateways.

Definition at line 151 of file RequestExecutor.h.

Referenced by replyToGateway(), and RequestExecutor().

◆ shortExecQ

utils::WaitQueuePointer<DataMessage*> cmlabs::RequestExecutor::shortExecQ

protected

Pending short requests.

Definition at line 139 of file RequestExecutor.h.

Referenced by receiveMessage(), sendStatusNow(), waitForShortRequest(), and ~RequestExecutor().

◆ shortExecQSemaphore

utils::Semaphore cmlabs::RequestExecutor::shortExecQSemaphore

protected

Signals short-queue arrivals.

Definition at line 146 of file RequestExecutor.h.

◆ shortReceivedCount

uint64 cmlabs::RequestExecutor::shortReceivedCount

protected

Total short requests received.

Definition at line 152 of file RequestExecutor.h.

Referenced by receiveMessage(), and RequestExecutor().

◆ threadID

uint32 cmlabs::RequestExecutor::threadID

protected

Worker thread id.

Definition at line 148 of file RequestExecutor.h.

Referenced by RequestExecutor().

The documentation for this class was generated from the following files:

/home/ubuntu/c/partner/psyclone2/CMSDK/include/RequestExecutor.h
/home/ubuntu/c/partner/psyclone2/CMSDK/src/RequestExecutor.cpp

Public Member Functions

Protected Member Functions

Protected Attributes

Friends

Detailed Description

Constructor & Destructor Documentation

◆ RequestExecutor()

◆ ~RequestExecutor()

Member Function Documentation

◆ addGateway()

◆ addLongRequestName()

◆ receiveHTTPRequest()

◆ receiveMessage()

◆ receiveNetworkEvent()

◆ replyToGateway()

◆ replyToQuery()

◆ run()

◆ sendMessageToGateway()

◆ sendStatusNow()

◆ setLongRequestLimit()

◆ shutdownNetwork()

◆ waitForLongRequest()

◆ waitForShortRequest()

Friends And Related Symbol Documentation

◆ RequestExecutorRun

Member Data Documentation

◆ channel

◆ conMutex

◆ connections

◆ executorID

◆ executorName

◆ heartbeatIntervalMS

◆ lastHeartbeat

◆ lastRefID

◆ longExecQ

◆ longExecQSemaphore

◆ longReceivedCount

◆ longReqLimit

◆ longReqNames

◆ manager

◆ mutex

◆ repliedCount

◆ replyQ

◆ requestMap

◆ sentCount

◆ shortExecQ

◆ shortExecQSemaphore

◆ shortReceivedCount

◆ threadID