MemoryRequestConnection.cpp

Go to the documentation of this file.

 
#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