

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "MemoryRequestServer.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() {

	uint32 count = 100, queueCount = 100, n, m;

	MemoryRequestServer* server = new MemoryRequestServer();
	if (!server->init("TestServer", 20, 10*1024*1024, 256)) {
		LogPrint(0,LOG_MEMORY,0,"Couldn't initialise MemoryRequestServer...");
		delete(server);
		return false;
	}

	uint64 createdTime, lastUpdateTime;
	uint64 reqID1 = 0, reqID2 = 0;
	DataMessage* reqMsg;
	uint64* reqIDs1 = new uint64[queueCount];
	uint64* reqIDs2 = new uint64[queueCount];
	MemoryRequestConnection* con1 = new MemoryRequestConnection();
	MemoryRequestConnection* con2 = new MemoryRequestConnection();
	if (!con1->connect("TestServer")) {
		LogPrint(0,LOG_MEMORY,0,"Couldn't initialise MemoryRequestConnection 1...");
		goto testfail;
	}

	if (!con2->connect("TestServer")) {
		LogPrint(0,LOG_MEMORY,0,"Couldn't initialise MemoryRequestConnection 2...");
		goto testfail;
	}

	if (reqMsg = server->waitForRequest(100)) {
		LogPrint(0,LOG_MEMORY,0,"Server got request when no request expected...");
		goto testfail;
	}

	for (n=0; n<count; n++) {
		reqMsg = new DataMessage();
		reqMsg->setTo(1);

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

		for (m=0; m<queueCount; m++) {
			if (con1->getRequestStatus(reqIDs1[m], createdTime, lastUpdateTime) < MEMORYREQUEST_SUBMITTED) {
				LogPrint(0,LOG_MEMORY,0,"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) {
				LogPrint(0,LOG_MEMORY,0,"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))) {
				LogPrint(0,LOG_MEMORY,0,"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)) {
				LogPrint(0,LOG_MEMORY,0,"Server couldn't reply to request from con1...[run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			delete(reqMsg);

			if (!(reqMsg = server->waitForRequest(100))) {
				LogPrint(0,LOG_MEMORY,0,"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)) {
				LogPrint(0,LOG_MEMORY,0,"Server couldn't reply to request from con2...[run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			delete(reqMsg);
		}

		for (m=0; m<queueCount; m++) {
			if (con1->getRequestStatus(reqIDs1[m], createdTime, lastUpdateTime) < MEMORYREQUEST_REPLIED) {
				LogPrint(0,LOG_MEMORY,0,"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) {
				LogPrint(0,LOG_MEMORY,0,"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))) {
				LogPrint(0,LOG_MEMORY,0,"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))) {
				LogPrint(0,LOG_MEMORY,0,"Con2 didn't get reply...[run %u of %u, entry %u of %u]", n, count, m, queueCount);
				goto testfail;
			}
			delete(reqMsg);
		}
	}

	delete [] reqIDs1;
	delete [] reqIDs2;
	delete(con2);
	delete(con1);
	delete(server);
	LogPrint(0,LOG_MEMORY,0,"Test of MemoryRequestServer success [%u runs of %u entries]", count, queueCount);
	return true;

testfail:
	delete [] reqIDs1;
	delete [] reqIDs2;
	delete(con2);
	delete(con1);
	delete(server);
	LogPrint(0,LOG_MEMORY,0,"Test of MemoryRequestServer failed");
	return false;
}

} // namespace cmlabs
