

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 MemoryRequestServer.cpp
    @brief Implementation of the request/reply server: segment creation, request dispatch and connection maintenance.
    @ingroup Memory */

#include "MemoryRequestServer.h"
#include "PsyTime.h"
#include "UnitTestFramework.h"

namespace cmlabs {

MemoryRequestServer::MemoryRequestServer() {
	serverHeader = NULL;
	mutex = NULL;
	requestCon = NULL;
}

MemoryRequestServer::~MemoryRequestServer() {
	shutdown();
}


bool MemoryRequestServer::init(const char* memoryName, uint32 maxConnectionCount, uint32 maxQueueSize, uint32 maxRequestCount) {
	uint32 masterSize = sizeof(RequestServerHeader) + 
		(maxConnectionCount * (sizeof(RequestConnectionHeader) + sizeof(RequestQueueHeader) + maxQueueSize)) +
		sizeof(RequestStatusHeader) + (maxRequestCount * sizeof(RequestStatusEntry));

	mutex = new utils::Mutex(utils::StringFormat("%s_Mutex", memoryName).c_str());
	if (!mutex || !mutex->enter(1000))
		return false;

	// Allocate main shared memory segment
	serverHeader = (RequestServerHeader*) utils::CreateSharedMemorySegment(memoryName, masterSize);
	if (!serverHeader) {
		mutex->leave();
		return false;
	}

	// Init memory
	memset(serverHeader, 0, masterSize);
	serverHeader->cid = MEMORYREQUESTSERVERID;
	serverHeader->createdTime = GetTimeNow();
	serverHeader->lastUpdateTime = serverHeader->createdTime;
	serverHeader->maxConnectionCount = maxConnectionCount;
	serverHeader->eachConnectionSize = sizeof(RequestConnectionHeader) + sizeof(RequestQueueHeader) + maxQueueSize;
	serverHeader->maxQueueSize = maxQueueSize;
	serverHeader->maxRequestCount = maxRequestCount;
	serverHeader->size = masterSize;
	serverHeader->status = MEMORYREQUESTSERVER_INIT;
	utils::strcpyavail(serverHeader->name, memoryName, 255, true);

	char* requestHeader = ((char*)serverHeader)+sizeof(RequestServerHeader)
		+ (serverHeader->maxConnectionCount * serverHeader->eachConnectionSize);
	if (!MemoryRequestQueues::InitRequestMap(requestHeader, serverHeader->maxRequestCount)) {
		shutdown();
		return false;
	}

	// All queues, requests and connections are zero
	mutex->leave();

	requestCon = new MemoryRequestConnection();
	if (!requestCon->connect(memoryName)) {
		shutdown();
		return false;
	}

	return true;
}

bool MemoryRequestServer::shutdown() {
	if (requestCon) {
		MemoryRequestConnection* temp = requestCon;
		requestCon = NULL;
		delete(temp);
	}
	if (!mutex)
		return true;
	if (!mutex->enter(1000))
		return false;
	if (serverHeader) {
		utils::CloseSharedMemorySegment((char*) serverHeader, serverHeader->size);
		serverHeader = NULL;
	}
	mutex->leave();
	return true;
}


DataMessage* MemoryRequestServer::waitForRequest(uint32 ms) {
	if (!requestCon) return NULL;
	return requestCon->waitForRequest(ms);
}

bool MemoryRequestServer::setRequestStatus(uint64 id, uint16 status) {
	if (!requestCon) return false;
	return requestCon->setRequestStatus(id, status);
}

bool MemoryRequestServer::replyToRequest(uint64 id, DataMessage* msg) {
	if (!requestCon) return false;
	return requestCon->replyToRequest(id, msg);
}



bool MemoryRequestServer::maintenance() {
	return true;
}

bool MemoryRequestServer::UnitTest() {

	// Self-contained: server + two client connections share an in-process named
	// shared-memory segment ("TestServer"). All waits are bounded (<=500ms), so
	// the test always terminates. Loop counts kept modest for a 1-2s runtime
	// while still producing a stable throughput metric.
	uint32 count = 10, queueCount = 10, n, m;

	unittest::progress(5, "init server");
	MemoryRequestServer* server = new MemoryRequestServer();
	// Modest sizing: a few connections with a small per-connection queue is
	// plenty for this test and keeps the shared segment small.
	if (!server->init("TestServer", 4, 256*1024, 256)) {
		unittest::fail("Couldn't initialise MemoryRequestServer");
		delete(server);
		return false;
	}

	uint64 createdTime, lastUpdateTime;
	DataMessage* reqMsg;
	uint64* reqIDs1 = new uint64[queueCount];
	uint64* reqIDs2 = new uint64[queueCount];
	MemoryRequestConnection* con1 = new MemoryRequestConnection();
	MemoryRequestConnection* con2 = new MemoryRequestConnection();

	uint64 startTime = 0;
	uint32 totalRequests = 0;

	unittest::progress(15, "connect clients");
	if (!con1->connect("TestServer")) {
		unittest::fail("Couldn't initialise MemoryRequestConnection 1");
		goto testfail;
	}

	if (!con2->connect("TestServer")) {
		unittest::fail("Couldn't initialise MemoryRequestConnection 2");
		goto testfail;
	}

	if (reqMsg = server->waitForRequest(100)) {
		unittest::fail("Server got request when no request expected");
		delete(reqMsg);
		goto testfail;
	}

	unittest::progress(25, "request/reply exchange");
	startTime = GetTimeNow();
	for (n=0; n<count; n++) {
		reqMsg = new DataMessage();
		reqMsg->setTo(1);

		for (m=0; m<queueCount; m++) {
			if (!(reqIDs1[m] = con1->makeRequest(reqMsg))) {
				unittest::fail("Con1 couldn't make request [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				delete(reqMsg);
				goto testfail;
			}
			if (!(reqIDs2[m] = con2->makeRequest(reqMsg))) {
				unittest::fail("Con2 couldn't make request [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				delete(reqMsg);
				goto testfail;
			}
		}
		delete(reqMsg);

		for (m=0; m<queueCount; m++) {
			if (con1->getRequestStatus(reqIDs1[m], createdTime, lastUpdateTime) < MEMORYREQUEST_SUBMITTED) {
				unittest::fail("Con1 request not submitted [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			if (con2->getRequestStatus(reqIDs2[m], createdTime, lastUpdateTime) < MEMORYREQUEST_SUBMITTED) {
				unittest::fail("Con2 request not submitted [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
		}

		for (m=0; m<queueCount; m++) {
			if (!(reqMsg = server->waitForRequest(100))) {
				unittest::fail("Server didn't receive request from con1 [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			reqMsg->setTo(reqMsg->getFrom());
			if (!server->replyToRequest(reqMsg->getReference(), reqMsg)) {
				unittest::fail("Server couldn't reply to request from con1 [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				delete(reqMsg);
				goto testfail;
			}
			delete(reqMsg);

			if (!(reqMsg = server->waitForRequest(100))) {
				unittest::fail("Server didn't receive request from con2 [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			reqMsg->setTo(reqMsg->getFrom());
			if (!server->replyToRequest(reqMsg->getReference(), reqMsg)) {
				unittest::fail("Server couldn't reply to request from con2 [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				delete(reqMsg);
				goto testfail;
			}
			delete(reqMsg);
		}

		for (m=0; m<queueCount; m++) {
			if (con1->getRequestStatus(reqIDs1[m], createdTime, lastUpdateTime) < MEMORYREQUEST_REPLIED) {
				unittest::fail("Con1 request not replied [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			if (con2->getRequestStatus(reqIDs2[m], createdTime, lastUpdateTime) < MEMORYREQUEST_REPLIED) {
				unittest::fail("Con2 request not replied [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
		}

		for (m=0; m<queueCount; m++) {
			if (!(reqMsg = con1->waitForRequestReply(reqIDs1[m], 500))) {
				unittest::fail("Con1 didn't get reply [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			delete(reqMsg);
			if (!(reqMsg = con2->waitForRequestReply(reqIDs2[m], 500))) {
				unittest::fail("Con2 didn't get reply [run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			delete(reqMsg);
		}

		totalRequests += 2 * queueCount;
		unittest::detail("run %u of %u complete (%u requests so far)", n + 1, count, totalRequests);
		unittest::progress(25 + (uint32)((uint64)(n + 1) * 65 / count), "request/reply exchange");
	}

	{
		double us = (double)(GetTimeNow() - startTime);
		if (us > 0)
			unittest::metric("request_reply_throughput", (double)totalRequests / us * 1e6, "req/s", true);
	}

	unittest::progress(95, "shutdown");
	delete [] reqIDs1;
	delete [] reqIDs2;
	delete(con2);
	delete(con1);
	delete(server);
	unittest::progress(100, "done");
	return true;

testfail:
	delete [] reqIDs1;
	delete [] reqIDs2;
	delete(con2);
	delete(con1);
	delete(server);
	return false;
}

void Register_MemoryRequestServer_Tests() {
	UnitTestRunner::instance().registerTest("memoryrequestserver", MemoryRequestServer::UnitTest,
		"Shared-memory request server/connection request-reply round-trip", "memory");
}

} // namespace cmlabs
