

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 ProcessMemory.h
    @brief Shared-memory process ("space") table plus per-process message queues.
    @ingroup Memory

    Each Psyclone space runs as its own OS process; this segment is how they see each
    other. It holds a fixed array of ProcessInfoStruct records (status, heartbeat, OS pid,
    stats) and up to four queues per process — command, message, signal and request —
    implemented as circular buffers in the same segment and signalled with named
    cross-process semaphores. The node monitors process heartbeats here to detect crashed
    spaces (crash isolation: a dead process only leaves a stale table entry behind).

    @warning Shared-memory layouts; changing any struct here requires a full rebuild of
    all binaries (`make clean && make`) — see MemoryManager.h for the ABI warning. */

#if !defined(_PROCESSMEMORY_H_)
#define _PROCESSMEMORY_H_

#include "MemoryManager.h"
#include "PsyTime.h"
#include "HTML.h"

namespace cmlabs {

/*! @name Per-process queue slot indices
    @{ */
#define SIGNALQ_ID 0	/*!< Signal queue slot. */
#define MSGQ_ID 1		/*!< Data-message queue slot. */
#define REQQ_ID 2		/*!< Request queue slot. */
#define CMDQ_ID 3		/*!< Command queue slot. */
/*! @} */

#define MAXPROC 64		/*!< Maximum processes (spaces) per node. */
#define MAXPROC4 256	/*!< Maximum queues per node (MAXPROC x 4 queue types). */

/*! @name Process lifecycle status values (ProcessInfoStruct::status)
    @{ */
#define PSYPROC_ERROR			0	/*!< Process is in an error state. */
#define PSYPROC_CREATED			1	/*!< Table entry created; process not yet started. */
#define PSYPROC_INIT			2	/*!< Process is initialising. */
#define PSYPROC_IDLE			3	/*!< Process is idle. */
#define PSYPROC_READY			4	/*!< Process is ready to run components. */
#define PSYPROC_ACTIVE			5	/*!< Process is actively running. */
#define PSYPROC_SHUTTING_DOWN	6	/*!< Process is shutting down. */
#define PSYPROC_TERMINATED		7	/*!< Process has terminated. */
/*! @} */

/*! @name Queue type selectors (used by the private addToQ/waitForQ helpers)
    @{ */
#define CMDQ_TYPE		1	/*!< Command queue. */
#define MSGQ_TYPE		2	/*!< Data-message queue. */
#define SIGQ_TYPE		3	/*!< Signal queue. */
#define REQQ_TYPE		4	/*!< Request queue. */
/*! @} */

THREAD_RET THREAD_FUNCTION_CALL QueueTest(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL ProcessMemoryPerfTest(THREAD_ARG arg);

/*! @brief Per-process throughput counters and recent-message rings, kept in shared memory.
    @ingroup Memory
    The recentInMsg/recentOutMsg buffers are rings of 10 DRAFTMSGSIZE-byte serialized
    message drafts, indexed by recentInPos/recentOutPos, used for live inspection tools.
    @warning Shared-memory layout; see MemoryManager.h for the ABI/rebuild warning. */
struct ProcessStats {
	uint64 time;		/*!< Timestamp of the last update (µs). */
	uint64 msgInCount;	/*!< Total messages received. */
	uint64 msgInBytes;	/*!< Total bytes received. */
	char recentInMsg[10 * DRAFTMSGSIZE];	/*!< Ring of the 10 most recent inbound message drafts. */
	uint64 msgOutCount;	/*!< Total messages sent. */
	uint64 msgOutBytes;	/*!< Total bytes sent. */
	char recentOutMsg[10 * DRAFTMSGSIZE];	/*!< Ring of the 10 most recent outbound message drafts. */
	uint64 procMemUsage;		/*!< Current process memory footprint in bytes. */
	uint64 currentCPUTicks;		/*!< Cumulative CPU ticks consumed. */
	uint32 recentInPos;	 // circular write index into recentInMsg ring (newest written here, then advanced)
	uint32 recentOutPos; // circular write index into recentOutMsg ring

	/*! Serialize counters and recent messages as a \<processstats\> XML element. @return XML fragment. */
	std::string toXML() {
		std::string xmlIn, xmlOut;
		int n;
		const char* oldData = NULL;
		DataMessage* msg = new DataMessage();
		char* msgData = recentInMsg;
		for (n = 0; n < 10; n++) {
			if (((DataMessageHeader*)msgData)->time) {
				if (!oldData) oldData = msg->swapMessageData(msgData);
				else msg->swapMessageData(msgData);
				xmlIn += msg->toXML();
			}
			msgData += DRAFTMSGSIZE;
		}
		msgData = recentOutMsg;
		for (n = 0; n < 10; n++) {
			if (((DataMessageHeader*)msgData)->time) {
				if (!oldData) oldData = msg->swapMessageData(msgData);
				else msg->swapMessageData(msgData);
				xmlOut += msg->toXML();
			}
			msgData += DRAFTMSGSIZE;
		}
		if (oldData)
			msg->swapMessageData(oldData);
		delete(msg);
		return utils::StringFormat("<processstats msgincount=\"%llu\" msginbytes=\"%llu\" msgoutcount=\"%llu\" msgoutbytes=\"%llu\" time=\"%llu\" memusage=\"%llu\">\n<inmsg>\n%s</inmsg>\n<outmsg>\n%s</outmsg>\n</processstats>\n",
			msgInCount, msgInBytes, msgOutCount, msgOutBytes, time, procMemUsage, xmlIn.c_str(), xmlOut.c_str());
	}
};

/*! @brief One process (space) record in the shared process table.
    @ingroup Memory
    The four *QID fields index into ProcessMemoryStruct::qIndex to locate this process's
    command/message/signal/request queues within the segment.
    @warning Shared-memory layout; see MemoryManager.h for the ABI/rebuild warning. */
struct ProcessInfoStruct {
	uint32 id;								/*!< Process id; corresponds to the index in the table. */
	uint16 nodeID;							/*!< Id of the node this process belongs to. */
	uint64 createdTime;						/*!< Creation timestamp (µs). */
	uint64 lastseen;						/*!< Last heartbeat/status update (µs); staleness implies a crashed space. */
	uint8 status;							/*!< PSYPROC_* lifecycle status. */
	uint32 osID;							/*!< OS process id (pid). */
	uint8 type;								/*!< 0: normal space process, 1: runs inside the node process. */
//	stats(activity)
	char name[MAXKEYNAMELEN+1];				/*!< Process (space) name. */
	char commandline[MAXCOMMANDLINELEN+1];	/*!< Command line used to launch the process. */
	ProcessStats stats;						/*!< Throughput counters and recent-message rings. */
	AveragePerfStats perfStats;				/*!< Windowed performance averages. */
	uint16 cmdQID;		/*!< Queue index of this process's command queue. */
	uint16 msgQID;		/*!< Queue index of this process's data-message queue. */
	uint16 sigQID;		/*!< Queue index of this process's signal queue. */
	uint16 reqQID;		/*!< Queue index of this process's request queue. */

	/*! Serialize this record (including stats) as a \<space\> XML element. @return XML fragment. */
	std::string toXML() {
		return utils::StringFormat("<space id=\"%u\" name=\"%s\" node=\"%u\" status=\"%u\" osid=\"%u\" commandline=\"%s\" created=\"%llu\" lastseen=\"%llu\">\n%s%s</space>",
			id, name, nodeID, status, osID, html::EncodeHTML(commandline).c_str(), createdTime, lastseen, stats.toXML().c_str(), perfStats.toXML().c_str());
	}
};

/*! @brief Root header of the process-memory segment: fixed process table plus queue index.
    qIndex holds byte offsets (within the segment) of up to MAXPROC*4 queues; the queues
    themselves follow in the remainder of the segment and are resized on demand.
    @warning Shared-memory layout; see MemoryManager.h for the ABI/rebuild warning. */
struct ProcessMemoryStruct {
	uint64 size;			/*!< Total segment size in bytes. */
	uint32 cid;				/*!< Check/magic id validated on attach. */
	uint64 createdTime;		// time of creation
	uint32 count;			/*!< Number of process entries in use. */
	uint64 usage;			/*!< Bytes currently in use in the segment. */
	ProcessInfoStruct processes[MAXPROC];	/*!< Fixed-size process table. */
	uint64 qIndex[MAXPROC*4];	/*!< Byte offsets of each process's four queues (0 = none). */
};

/*! Initial byte size of a newly created per-process queue. */
#define INITIALQSIZE sizeof(MessageQueueHeader) + 2048
/*! @brief On-segment header of one per-process circular message queue (64-bit layout).
    @note Distinct from the 32-bit MessageQueueHeader in MemoryQueues.h; both exist and
    are not interchangeable (this one shadows/replaces it wherever ProcessMemory.h is included).
    @warning Shared-memory layout; see MemoryManager.h for the ABI/rebuild warning. */
struct MessageQueueHeader {
	uint64 size;		/*!< Total queue region size in bytes (header + circular buffer). */
	uint32 id;			/*!< Queue index within qIndex. */
	uint32 count;		/*!< Messages currently queued. */
	uint64 startPos;	/*!< Read offset into the circular buffer. */
	uint64 endPos;		/*!< Write offset into the circular buffer. */
	uint64 padding;		/*!< Alignment/reserved. */
	char name[MAXKEYNAMELEN+1];				/*!< Queue name, used to derive the names of its cross-process mutex/semaphore. */
};

/*! Re-open the process segment if the master's resize serial no longer matches ours
    (another process re-created the segment); bails out of the calling method on failure. */
#define CHECKPROCESSMEMORYSERIAL 	if (serial != master->getProcessShmemSerial()) {if (!open()) {mutex->leave();return 0;}}

class MasterMemory;
/*! @brief Accessor for the shared process table and per-process queues.
    @ingroup Memory
    One instance per process (created by MemoryManager). create() is called by the node
    master; other processes open(). All calls lock the segment's named cross-process
    mutex and re-attach automatically if the segment was resized (serial check). Wait
    calls block on named semaphores up to the given timeout (ms) and return caller-owned
    message copies. The @a master pointer is borrowed; mapped memory is owned by the OS.
    @code
    uint16 pid;
    processMemory->createNewProcess("VisionSpace", pid);
    processMemory->addToCmdQ(pid, cmdMsg);                    // any process
    DataMessage* m = processMemory->waitForCmdQ(pid, 100);    // inside that process
    if (m) { ...; delete m; }
    @endcode */
class ProcessMemory {
public:
	static bool UnitTest();		/*!< Self-test. @return true if all checks pass. */
	static bool PerfTest();		/*!< Queue throughput benchmark. @return true on success. */

	/*! @param master Segment bookkeeping interface (borrowed, not owned). */
	ProcessMemory(MasterMemory* master);
	~ProcessMemory();

	/*! Report allocation/usage of the process segment. @return true on success. */
	bool getMemoryUsage(uint64& alloc, uint64& usage);
	/*! Snapshot all process records. @return New vector owned by the caller (delete it). */
	std::vector<ProcessInfoStruct>* getAllProcesses();
	/*! Find processes whose heartbeat is older than @a timeoutMS.
	    @param timeoutMS Staleness threshold in ms. @param procIssues Receives copies of stale entries.
	    @return true if the check ran. */
	bool checkProcessHeartbeats(uint32 timeoutMS, std::list<ProcessInfoStruct>& procIssues);

	bool open();	/*!< Attach to the existing process segment. @return true on success. */
	/*! Create the process segment (master only). @param initialProcCount Initial table capacity. @return true on success. */
	bool create(uint32 initialProcCount);

	// Processes
	/*! Register a new process (space) in the table. @param name Process name. @param id Receives the new process id. @return true on success. */
	bool createNewProcess(const char* name, uint16 &id);
	/*! Remove a process entry and free its queues. @return true on success. */
	bool deleteProcess(uint16 id);

	// Process Info
	bool getProcessName(uint16 id, char* name, uint32 maxSize);	/*!< Copy the process name into @a name (max @a maxSize bytes). @return true if found. */
	bool getProcessID(const char* name, uint16 &id);			/*!< Look up a process id by name. @return true if found. */
	uint64 getProcessCreateTime(uint16 id);						/*!< @return Creation timestamp (µs), 0 if unknown. */

	uint8 getProcessStatus(uint16 id, uint64& lastseen);		/*!< @return PSYPROC_* status; @a lastseen gets the last heartbeat. */
	uint8 getProcessStatus(uint16 id, uint64& lastseen, uint64& createTime);	/*!< Overload also returning creation time. @return PSYPROC_* status. */
	bool getProcessCommandLine(uint16 id, char* cmdline, uint32 maxSize);		/*!< Copy the launch command line. @return true if found. */
	uint16 getProcessIDFromOSID(uint32 osid);	/*!< Map an OS pid to a Psyclone process id. @return Id or 0. */
	uint32 getProcessOSID(uint16 id);			/*!< @return OS pid of process @a id, 0 if unknown. */

	/*! Update status + heartbeat (and optionally CPU ticks) for process @a id. @return true on success. */
	bool setProcessStatus(uint16 id, uint8 status, uint64 currentCPUTicks = 0);
	bool setProcessCommandLine(uint16 id, const char* cmdline);	/*!< Record the launch command line. @return true on success. */
	bool setProcessOSID(uint16 id, uint32 osid);				/*!< Record the OS pid. @return true on success. */
	bool setProcessType(uint16 id, uint8 type);					/*!< Set type (0 = normal, 1 = inside node). @return true on success. */

	AveragePerfStats getProcessPerfStats(uint16 procID);		/*!< @return Copy of the process's windowed performance averages. */
	bool setProcessPerfStats(uint16 procID, AveragePerfStats &perfStruct);	/*!< Store new performance averages. @return true on success. */

	/*! Add message traffic to the process's stats/rings. Caller keeps ownership of both messages. @return true on success. */
	bool addToProcessStats(uint16 id, DataMessage* inputMsg, DataMessage* outputMsg);

	// Process Queues
	/*! @name Queue producers
	    Copy @a msg into the target process's queue and signal its semaphore; the caller
	    keeps ownership of @a msg. Queues grow on demand. @return true on success.
	    @{ */
	bool addToCmdQ(uint16 procID, DataMessage* msg);	/*!< Enqueue on the command queue. */
	bool addToMsgQ(uint16 procID, DataMessage* msg);	/*!< Enqueue on the data-message queue. */
	bool addToSigQ(uint16 procID, DataMessage* msg);	/*!< Enqueue on the signal queue. */
	bool addToReqQ(uint16 procID, DataMessage* msg);	/*!< Enqueue on the request queue. */
	bool addToAllSignalQs(DataMessage* msg);			/*!< Broadcast to every process's signal queue. */
	bool addToAllSignalQsExcept(DataMessage* msg, uint16 except);	/*!< Broadcast to all signal queues except process @a except. */
	/*! @} */

	/*! @name Queue consumers (blocking)
	    Block up to @a timeout ms for the next message on this process's queue.
	    @return Caller-owned DataMessage copy (delete after use), or NULL on timeout.
	    @{ */
	DataMessage* waitForCmdQ(uint16 procID, uint32 timeout);	/*!< Wait on the command queue. */
	DataMessage* waitForMsgQ(uint16 procID, uint32 timeout);	/*!< Wait on the data-message queue. */
	DataMessage* waitForSigQ(uint16 procID, uint32 timeout);	/*!< Wait on the signal queue. */
	DataMessage* waitForReqQ(uint16 procID, uint32 timeout);	/*!< Wait on the request queue. */
	/*! @} */

	uint32 getCmdQCount(uint16 procID);		/*!< @return Messages waiting in the command queue. */
	uint32 getMsgQCount(uint16 procID);		/*!< @return Messages waiting in the data-message queue. */
	uint32 getSigQCount(uint16 procID);		/*!< @return Messages waiting in the signal queue. */
	uint32 getReqQCount(uint16 procID);		/*!< @return Messages waiting in the request queue. */

	/*! Total queued bytes/messages across all four queues of @a procID. @return true on success. */
	bool getQueueSizes(uint16 procID, uint64 &bytes, uint32 &count);
	/*! Append PerfStats snapshots for all local processes to @a perfStats. @return true on success. */
	bool addLocalPerformanceStats(std::list<PerfStats> &perfStats);

private:
	bool resize(uint64 newMemorySize);
	bool setupNextAvailableQ(uint16& qID);
	bool deleteQ(uint16 id);
	bool resizeQ(uint16 id, uint64 newSize);

	bool printQ(MessageQueueHeader* qHeader);

	MessageQueueHeader* getQHeader(uint16 procID, uint8 qType);
	bool addToQ(uint16 procID, uint8 qType, DataMessage* msg);
	DataMessage* waitForQ(uint16 procID, uint8 qType, uint32 timeout);

	utils::Mutex* mutex;			/*!< Named cross-process mutex guarding the segment (owned). */
	MasterMemory* master;			/*!< Bookkeeping interface for size/serial (borrowed). */
	ProcessMemoryStruct* header;	/*!< Mapped segment header (owned by the OS). */
	uint64 memorySize;				/*!< Size of our current mapping in bytes. */
	uint16 port;					/*!< Node port used in segment naming. */
	uint32 serial;					/*!< Cached resize serial; mismatch triggers re-open. */

	utils::Semaphore* qSemaphores[MAXPROC4];	/*!< Named semaphores, one per queue, for cross-process wakeups. */
};

} // namespace cmlabs

#endif //_PROCESSMEMORY_H_

