

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 "MemoryRequestConnection.h"

namespace cmlabs {

MemoryRequestConnection::MemoryRequestConnection() {
	conID = 0;
	serverHeader = NULL;
	conHeader = NULL;
	queueHeader = NULL;
	requestHeader = NULL;
	mutex = NULL;
	oneWaitingForQueue = false;
}

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

bool MemoryRequestConnection::connect(const char* memoryName) {
	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::OpenSharedMemorySegment(memoryName, 0);
	if (!serverHeader) {
		mutex->leave();
		return false;
	}

	serverHeader->lastUpdateTime = serverHeader->createdTime;

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

	// Now find the first available connection id
	conID = 1;
	conHeader = (RequestConnectionHeader*) (((char*)serverHeader)+sizeof(RequestServerHeader));
	while (conID <= serverHeader->maxConnectionCount) {
		if (!conHeader->size || !conHeader->conid || !conHeader->createdTime) {
			conHeader->size = serverHeader->eachConnectionSize;
			conHeader->conid = conID;
			conHeader->createdTime = conHeader->lastUpdateTime = GetTimeNow();
			conHeader->status = 1;
			break;
		}
		else {
			conID++;
			conHeader = (RequestConnectionHeader*) (((char*)conHeader)+conHeader->size);
		}
	}
	if (conID > serverHeader->maxConnectionCount) {
		conHeader = NULL;
		mutex->leave();
		shutdown();
		return false;
	}
	queueHeader = ((char*)conHeader)+sizeof(RequestConnectionHeader);
	if (!MemoryRequestQueues::InitQueue(queueHeader, serverHeader->maxQueueSize)) {
		mutex->leave();
		shutdown();
		return false;
	}

	mutex->leave();
	return true;
}

bool MemoryRequestConnection::shutdown() {
	conID = 0;
	if (!mutex)
		return true;
	if (!mutex->enter(1000))
		return false;

	utils::Mutex* tempMutex = mutex;
	mutex = NULL;
	requestHeader = NULL;
	queueHeader = NULL;
	conHeader = NULL;
	if (serverHeader) {
		utils::CloseSharedMemorySegment((char*) serverHeader, serverHeader->size);
		serverHeader = NULL;
	}

	std::map<uint64, DataMessage*>::iterator i = replyCache.begin(), e = replyCache.end();
	while (i != e) {
		delete(i->second);
		i++;
	}
	replyCache.clear();

	tempMutex->leave();
	delete(tempMutex);
	return true;
}


uint64 MemoryRequestConnection::makeRequest(DataMessage* msg) {
	if (!mutex || !mutex->enter(1000))
		return false;
	uint32 to = msg->getTo();
	uint64 reqID = MemoryRequestQueues::AddNewRequest(requestHeader, MEMORYREQUEST_INIT, conID, to);
	if (!reqID) {
		mutex->leave();
		return 0;
	}

	char* destQueueHeader = getQueueHeader(to);
	if (!destQueueHeader || !MemoryRequestQueues::CheckQueue(destQueueHeader, serverHeader->maxQueueSize)) {
		MemoryRequestQueues::CloseRequest(requestHeader, reqID);
		mutex->leave();
		return 0;
	}

	msg->setFrom(conID);
	msg->setReference(reqID);
	if (!MemoryRequestQueues::AddToQueue(destQueueHeader, msg)) {
		MemoryRequestQueues::CloseRequest(requestHeader, reqID);
		mutex->leave();
		return 0;
	}
	MemoryRequestQueues::SetRequestStatus(requestHeader, reqID, MEMORYREQUEST_SUBMITTED);
	utils::SignalSemaphore(utils::StringFormat("%s_%u", serverHeader->name, to).c_str());
	mutex->leave();
	return reqID;
}

uint16 MemoryRequestConnection::getRequestStatus(uint64 id, uint64& createdTime, uint64& lastUpdateTime) {
	if (!mutex || !mutex->enter(1000))
		return MEMORYREQUEST_ERROR;
	uint16 status = MemoryRequestQueues::GetRequestStatus(requestHeader, id, createdTime, lastUpdateTime);
	mutex->leave();
	return status;
}


DataMessage* MemoryRequestConnection::waitForQueue(uint32 ms) {
	// We assume the mutex is entered
	// If return NULL the mutex will be free, otherwise locked
	DataMessage* msg = MemoryRequestQueues::GetNextQueueMessage(queueHeader);
	if (!msg) {
		mutex->leave();
		if (!utils::WaitForSemaphore(utils::StringFormat("%s_%u", serverHeader->name, conID).c_str(), ms, true))
			return NULL;
		if (!mutex || !mutex->enter(1000))
			return NULL;
		if (!(msg = MemoryRequestQueues::GetNextQueueMessage(queueHeader))) {
			mutex->leave();
			return NULL;
		}
	}
	return msg;
}

DataMessage* MemoryRequestConnection::waitForRequestReply(uint32 ms) {
	if (!mutex || !mutex->enter(1000))
		return NULL;

	DataMessage* msg = waitForQueue(ms);
	if (!msg) {
		// mutex already unlocked
		return NULL;
	}
	// mutex still locked
	
	MemoryRequestQueues::SetRequestStatus(requestHeader, msg->getReference(), MEMORYREQUEST_REPLYRECEIVED);
	MemoryRequestQueues::CloseRequest(requestHeader, msg->getReference());

	mutex->leave();
	return msg;
}

DataMessage* MemoryRequestConnection::waitForRequest(uint32 ms) {
	if (!mutex || !mutex->enter(1000))
		return NULL;

	DataMessage* msg = waitForQueue(ms);
	if (!msg) {
		// mutex already unlocked
		return NULL;
	}
	// mutex still locked
	
	MemoryRequestQueues::SetRequestStatus(requestHeader, msg->getReference(), MEMORYREQUEST_RECEIVED);
	mutex->leave();
	return msg;
}

bool MemoryRequestConnection::setRequestStatus(uint64 id, uint16 status) {
	if (!mutex || !mutex->enter(1000))
		return false;
	if (!MemoryRequestQueues::SetRequestStatus(requestHeader, id, status)) {
		mutex->leave();
		return false;
	}
	mutex->leave();
	return true;
}

bool MemoryRequestConnection::replyToRequest(uint64 id, DataMessage* msg) {
	if (!msg->getTo() || (msg->getTo() == conID))
		return false;
	if (!mutex || !mutex->enter(1000))
		return false;
	uint64 createdTime, lastUpdateTime;
	uint16 status = MemoryRequestQueues::GetRequestStatus(requestHeader, id, createdTime, lastUpdateTime);
	if (status < MEMORYREQUEST_SUBMITTED) {
		mutex->leave();
		return false;
	}

	uint32 to = msg->getTo();
	char* destQueueHeader = getQueueHeader(to);
	if (!destQueueHeader || !MemoryRequestQueues::CheckQueue(destQueueHeader, serverHeader->maxQueueSize)) {
		MemoryRequestQueues::SetRequestStatus(requestHeader, id, MEMORYREQUEST_ERROR);
		mutex->leave();
		return 0;
	}

	msg->setFrom(conID);
	msg->setReference(id);
	if (!MemoryRequestQueues::AddToQueue(destQueueHeader, msg)) {
		MemoryRequestQueues::SetRequestStatus(requestHeader, id, MEMORYREQUEST_ERROR);
		mutex->leave();
		return 0;
	}
	MemoryRequestQueues::SetRequestStatus(requestHeader, id, MEMORYREQUEST_REPLIED);
	utils::SignalSemaphore(utils::StringFormat("%s_%u", serverHeader->name, to).c_str());
	mutex->leave();

	return true;
}


char* MemoryRequestConnection::getQueueHeader(uint32 otherConID) {
	if (!otherConID || (otherConID > serverHeader->maxConnectionCount))
		return NULL;
	char* otherQueueHeader = ((char*)serverHeader)
		+ sizeof(RequestServerHeader)
		+ (otherConID-1) * (serverHeader->eachConnectionSize)
		+ sizeof(RequestConnectionHeader);
	if (!MemoryRequestQueues::CheckQueue(otherQueueHeader, serverHeader->maxQueueSize))
		return NULL;
	return otherQueueHeader;
}



DataMessage* MemoryRequestConnection::waitForRequestReply(uint64 id, uint32 ms) {
	if (!mutex || !mutex->enter(1000))
		return NULL;
	
	uint64 ref;
	DataMessage* msg = NULL;
	uint64 startTime = GetTimeNow();
	int32 timeLeft = ms;

	while (true) {
		if (msg = replyCache[id]) {
			replyCache.erase(id);
			MemoryRequestQueues::SetRequestStatus(requestHeader, msg->getReference(), MEMORYREQUEST_REPLYRECEIVED);
			MemoryRequestQueues::CloseRequest(requestHeader, msg->getReference());
			mutex->leave();
			return msg;
		}

		if ((timeLeft = ms - GetTimeAgeMS(startTime)) < 0) {
			mutex->leave();
			return NULL;
		}

		// is anyone already waiting for the queue?
		if (!oneWaitingForQueue) {
			// then wait for the queue
			oneWaitingForQueue = true;
			msg = waitForQueue((uint32)timeLeft);
			oneWaitingForQueue = false;
			if (!msg) {
				// mutex already unlocked
				return NULL;
			}
			// mutex still locked
			if ((ref = msg->getReference()) == id) {
				// this is for me, thanks
				MemoryRequestQueues::SetRequestStatus(requestHeader, msg->getReference(), MEMORYREQUEST_REPLYRECEIVED);
				MemoryRequestQueues::CloseRequest(requestHeader, msg->getReference());
				mutex->leave();
				return msg;
			}
			else {
				// this is for someone else
				replyCache[ref] = msg;
				cacheEvent.signal();
				continue;
			}
		}
		else {
			mutex->leave();
			// wait for the event that someone else got a message not for them
			if (!cacheEvent.waitNext(timeLeft))
				return NULL;
			else {
				if (!mutex || !mutex->enter(1000))
					return NULL;
				continue;
			}
		}
	};
}








} // namespace cmlabs
