MemoryRequestServer.cpp

Go to the documentation of this file.

 
#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