MessageIndex.cpp

Go to the documentation of this file.

#include "MessageIndex.h"
#include "UnitTestFramework.h"
 
namespace   cmlabs {
 
MessageIndex::MessageIndex() {
    lastMaintenance = GetTimeNow();
    buffertime = 10000;
    maintenanceInterval = 10000;
    defaultTTL = PSYMINUTE;
    typeMap = NULL;
    timeMap = NULL;
}
 
MessageIndex::~MessageIndex() {
    timeMaps.clear();
    stringMaps.clear();
    integerMaps.clear();
    floatMaps.clear();
 
    if (typeMap)
        delete(typeMap);
    typeMap = NULL;
 
    if (timeMap)
        delete(timeMap);
    timeMap = NULL;
 
    std::multimap<uint64, DataMessage*>::iterator i = eolMap.begin(), e = eolMap.end();
    while (i != e) {
        delete(i->second);
        i++;
    }
    eolMap.clear();
}
 
bool MessageIndex::setDefaultTTL(uint64 defaultTTL) {
    this->defaultTTL = defaultTTL;
    return true;
}
 
bool MessageIndex::addIndexKey(const char* key, uint8 type) {
    switch(type) {
    case INDEX_TIME:
        if ((stricmp(key, "time") == 0) && !timeMap) {
            timeMap = new std::multimap<uint64, DataMessage*>;
            return true;
        }
        else {
            timeMaps[key];
            return true;
        }
    case INDEX_STRING:
        stringMaps[key];
        return true;
    case INDEX_INTEGER:
        integerMaps[key];
        return true;
    case INDEX_FLOAT:
        floatMaps[key];
        return true;
    default:
        return false;
    }
}
 
bool MessageIndex::removeIndexKey(const char* key, uint8 type) {
    switch(type) {
    case INDEX_TIME:
        if ((stricmp(key, "time") == 0) && timeMap) {
            delete(timeMap);
            timeMap = NULL;
            return true;
        }
        else
            return deleteIndexKey<uint64>(key, timeMaps);
    case INDEX_STRING:
        return deleteIndexKey<std::string>(key, stringMaps);
    case INDEX_INTEGER:
        return deleteIndexKey<int64>(key, integerMaps);
    case INDEX_FLOAT:
        return deleteIndexKey<float64>(key, floatMaps);
    default:
        return false;
    }
}
 
 
bool MessageIndex::setFileStorage(const char* dir, uint32 buffertime, const char* indexfilename) {
    this->dir = dir;
    this->indexfilename = indexfilename;
    this->buffertime = buffertime;
    return true;
}
 
uint32 MessageIndex::getCount() {
    return (uint32)eolMap.size();
}
 
bool MessageIndex::addMessage(DataMessage* msg) {
    if (GetTimeAgeMS(lastMaintenance) > maintenanceInterval)
        doMaintenance();
 
    // Add to each key in each type map
    addMessageToTime(msg);
    addMessageToString(msg);
    addMessageToInteger(msg);
    addMessageToFloat(msg);
 
    if (timeMap)
        timeMap->insert(Time_Pair(msg->getCreatedTime(), msg));
    if (typeMap)
        typeMap->insert(Type_Pair(msg->getType(), msg));
 
    // Add to EOL map
    // if msg->ttl is not set we need to use the default timeout to determine when to delete it
    if (!msg->getTTL())
        eolMap.insert(Time_Pair(msg->getCreatedTime() + defaultTTL, msg));
    else
        eolMap.insert(Time_Pair(msg->getEOL(), msg));
    return true;
}
 
char* MessageIndex::queryMessages(RetrieveSpec* spec, uint32 &size, uint32 &count) {
    if (GetTimeAgeMS(lastMaintenance) > maintenanceInterval)
        doMaintenance();
 
    size = count = 0;
    if (!spec)
        return NULL;
 
    std::map<std::string, std::multimap<uint64, DataMessage*> >::iterator it;
    std::map<std::string, std::multimap<std::string, DataMessage*> >::iterator is;
    std::map<std::string, std::multimap<int64, DataMessage*> >::iterator ii;
    std::map<std::string, std::multimap<float64, DataMessage*> >::iterator ifl;
 
//  printf("Querying for key '%s'...\n", spec->key);
 
    switch(spec->keytype) {
    case INDEX_TIME:
        if ((stricmp(spec->key, "time") == 0) && timeMap) {
            if (spec->startTime && spec->endTime)
                return queryMessages<uint64>(spec->startTime, spec->endTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, *timeMap);
            else if (spec->startTime)
                return queryMessages<uint64>(spec->startTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, *timeMap);
            else
                return queryMessages<uint64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, *timeMap);
        }
        else {
            it = timeMaps.find(spec->key);
            if (it != timeMaps.end()) {
                if (spec->startTime && spec->endTime)
                    return queryMessages<uint64>(spec->startTime, spec->endTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, it->second);
                else if (spec->startTime)
                    return queryMessages<uint64>(spec->startTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, it->second);
                else
                    return queryMessages<uint64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, it->second);
            }
            else
                return NULL;
        }
    case INDEX_STRING:
        is = stringMaps.find(spec->key);
        if (is != stringMaps.end()) {
            std::string start = spec->startString;
            std::string end = spec->endString;
            if (start.size() && end.size())
                return queryMessages<std::string>(start, end, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, is->second);
            else if (spec->startTime)
                return queryMessages<std::string>(start, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, is->second);
            else
                return queryMessages<std::string>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, is->second);
        }
        else
            return NULL;
    case INDEX_INTEGER:
        ii = integerMaps.find(spec->key);
        if (ii != integerMaps.end()) {
        //  printf("Querying for key '%s' (of %u)...\n", spec->key, ii->second.size());
            if ((spec->startInt != INT64_NOVALUE) && (spec->endInt != INT64_NOVALUE))
                return queryMessages<int64>(spec->startInt, spec->endInt, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ii->second);
            else if (spec->startInt != INT64_NOVALUE)
                return queryMessages<int64>(spec->startInt, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ii->second);
            else
                return queryMessages<int64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ii->second);
        }
        else
            return NULL;
    case INDEX_FLOAT:
        ifl = floatMaps.find(spec->key);
        if (ifl != floatMaps.end()) {
            if ((spec->startFloat != FLOAT64_NOVALUE) && (spec->endFloat != FLOAT64_NOVALUE))
                return queryMessages<float64>(spec->startFloat, spec->endFloat, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ifl->second);
            else if (spec->startFloat != FLOAT64_NOVALUE)
                return queryMessages<float64>(spec->startFloat, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ifl->second);
            else
                return queryMessages<float64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ifl->second);
        }
        else
            return NULL;
    default:
        return NULL;
    }
}
 
 
 
 
 
 
 
 
 
bool MessageIndex::doMaintenance() {
    uint64 now = lastMaintenance = GetTimeNow();
 
    std::map<std::string, std::multimap<uint64, DataMessage*> >::iterator it, et = timeMaps.end();
    std::map<std::string, std::multimap<std::string, DataMessage*> >::iterator is, es = stringMaps.end();
    std::map<std::string, std::multimap<int64, DataMessage*> >::iterator ii, ei = integerMaps.end();
    std::map<std::string, std::multimap<float64, DataMessage*> >::iterator ifl, efl = floatMaps.end();
 
    std::multimap<uint64, DataMessage*>::iterator i = eolMap.begin(), e = eolMap.end();
    while (i != e) {
        if (i->first < now) {
            // remove message from all other maps
            if (timeMap) {
                uint64 ctval = i->second->getCreatedTime();
                if (ctval)
                    removeMessage<uint64>(i->second, *timeMap, ctval);
            }
 
            it = timeMaps.begin();
            while (it != et) {
                uint64 tval = i->second->getTime(it->first.c_str());
                if (tval)
                    removeMessage<uint64>(i->second, it->second, tval);
                it++;
            }
 
            is = stringMaps.begin();
            while (is != es) {
                std::string sval = i->second->getString(is->first.c_str());
                removeMessage<std::string>(i->second, is->second, sval);
                is++;
            }
 
            ii = integerMaps.begin();
            while (ii != ei) {
                int64 ival;
                if (i->second->getInt(ii->first.c_str(), ival))
                    removeMessage<int64>(i->second, ii->second, ival);
                ii++;
            }
 
            ifl = floatMaps.begin();
            while (ifl != efl) {
                float64 fval;
                if (i->second->getDouble(ifl->first.c_str(), fval))
                    removeMessage<float64>(i->second, ifl->second, fval);
                ifl++;
            }
            // delete message itself
            delete(i->second);
            // remove entry
            eolMap.erase(i++);
        }
        else
            break;
    }
 
    return true;
}
 
 
 
 
 
 
 
 
 
 
 
bool MessageIndex::addMessageToTime(DataMessage* msg) {
    uint64 val;
    std::map<std::string, std::multimap<uint64, DataMessage*> >::iterator i = timeMaps.begin(), e = timeMaps.end();
    while (i != e) {
        if (val = msg->getTime(i->first.c_str()))
            i->second.insert(Time_Pair(val, msg));
        i++;
    }
    return true;
}
 
bool MessageIndex::addMessageToString(DataMessage* msg) {
    const char* val;
    std::map<std::string, std::multimap<std::string, DataMessage*> >::iterator i = stringMaps.begin(), e = stringMaps.end();
    while (i != e) {
        if (val = msg->getString(i->first.c_str()))
            i->second.insert(String_Pair(val, msg));
        i++;
    }
    return true;
}
 
bool MessageIndex::addMessageToInteger(DataMessage* msg) {
    int64 val;
    std::map<std::string, std::multimap<int64, DataMessage*> >::iterator i = integerMaps.begin(), e = integerMaps.end();
    while (i != e) {
        if (msg->getInt(i->first.c_str(), val)) {
            i->second.insert(Int_Pair(val, msg));
        //  printf("------------------- WB Added %lld to key '%s'   %s\n", val, i->first.c_str(), PrintTimeString(msg->getCreatedTime()).c_str());
        }
        i++;
    }
    return true;
}
 
bool MessageIndex::addMessageToFloat(DataMessage* msg) {
    float64 val;
    std::map<std::string, std::multimap<float64, DataMessage*> >::iterator i = floatMaps.begin(), e = floatMaps.end();
    while (i != e) {
        if (msg->getDouble(i->first.c_str(), val))
            i->second.insert(Float_Pair(val, msg));
        i++;
    }
    return true;
}
 
 
 
 
template <typename T>
bool MessageIndex::deleteIndexKey(const char* key, std::map<std::string, std::multimap<T, DataMessage*> > &maps) {
 
    std::multimap<T, DataMessage*>* map = &(maps[key]);
 
    typename std::multimap<T, DataMessage*>::iterator i = map->begin(), e = map->end();
    while (i != e) {
        delete(i->second);
        i++;
    }
    maps.erase(key);
    return true;
}
 
template <typename T>
bool MessageIndex::removeMessage(DataMessage* msg, std::multimap<T, DataMessage*> &map, T &val) {
    typename std::multimap<T, DataMessage*>::iterator i, e = map.end();
    i = map.lower_bound(val);
    while ((i != e) && (i->first == val)) {
        if (((void*)msg) == ((void*)i->second))
            map.erase(i++);
        else
            i++;
    }
    return true;
}
 
template <typename T>
char* MessageIndex::queryMessages(T start, T end, uint32 from, uint32 to, PsyType &type, uint32 maxcount, uint64 maxage, uint32 &size, uint32 &count, std::multimap<T, DataMessage*> &map) {
    count = size = 0;
    typename std::multimap<T, DataMessage*>::iterator i1, i2, e = map.end();
 
    uint64 now = GetTimeNow();
    uint64 createdAfter = 0;
    if (maxage)
        createdAfter = now - maxage;
 
    if ( (i1 = map.lower_bound(start)) == e) {
    //  printf("queryMessages found (%u) results (of %u)...\n", count, map.size());
        return NULL;
    }
    i2 = map.upper_bound(end);
 
    uint32 allocSize = 4096;
    char* result = (char*) malloc(allocSize);
 
    DataMessage* msg;
    uint32 msgSize;
    while ((i1 != i2) && (i1 != e)) {
        if (msg = i1->second) {
            // Filters...
            if (msg->getEOL() < now) {
                i1++;
                continue;
            }
            if (from && (from != msg->getFrom())) {
                i1++;
                continue;
            }
            if (to && (to != msg->getTo())) {
                i1++;
                continue;
            }
            if ((type != NOTYPE) && (!msg->getType().matches(type))) {
                i1++;
                continue;
            }
            if (maxcount && (count >= maxcount))
                break;
            if (createdAfter && (msg->getCreatedTime() < createdAfter)) {
                i1++;
                continue;
            }
 
            msgSize = i1->second->getSize();
            while (size + msgSize > allocSize) {
                allocSize *= 2;
                void* newmem = realloc(result, allocSize);
                if (!newmem) {
                    free(result);
                    count = size = 0;
                    return NULL;
                }
                result = (char*) newmem;
            }
            memcpy(result+size, msg->data, msgSize);
            size += msgSize;
            count++;
        }
        i1++;
    }
    // printf("queryMessages found %u results (of %u)...\n", count, map.size());
    return result;
}
 
template <typename T>
char* MessageIndex::queryMessages(T start, uint32 from, uint32 to, PsyType &type, uint32 maxcount, uint64 maxage, uint32 &size, uint32 &count, std::multimap<T, DataMessage*> &map) {
    count = size = 0;
    typename std::multimap<T, DataMessage*>::iterator i1, i2, e = map.end();
 
    uint64 now = GetTimeNow();
    uint64 createdAfter = 0;
    if (maxage)
        createdAfter = now - maxage;
 
    if ( (i1 = map.lower_bound(start)) == e)
        return NULL;
 
    uint32 allocSize = 4096;
    char* result = (char*) malloc(allocSize);
 
    DataMessage* msg;
    uint32 msgSize;
    while (i1 != e) {
        if (msg = i1->second) {
            // Filters...
            if (msg->getEOL() < now) {
                i1++;
                continue;
            }
            if (from && (from != msg->getFrom())) {
                i1++;
                continue;
            }
            if (to && (to != msg->getTo())) {
                i1++;
                continue;
            }
            if ((type != NOTYPE) && (!msg->getType().matches(type))) {
                i1++;
                continue;
            }
            if (maxcount && (count >= maxcount))
                break;
            if (createdAfter && (msg->getCreatedTime() < createdAfter)) {
                i1++;
                continue;
            }
 
            msgSize = i1->second->getSize();
            while (size + msgSize > allocSize) {
                allocSize *= 2;
                void* newmem = realloc(result, allocSize);
                if (!newmem) {
                    free(result);
                    count = size = 0;
                    return NULL;
                }
                result = (char*) newmem;
            }
            memcpy(result+size, msg->data, msgSize);
            size += msgSize;
            count++;
        }
        i1++;
    }
//  printf("queryMessages found %u results...\n", count);
    return result;
}
 
template <typename T>
char* MessageIndex::queryMessages(uint32 from, uint32 to, PsyType &type, uint32 maxcount, uint64 maxage, uint32 &size, uint32 &count, std::multimap<T, DataMessage*> &map) {
    count = size = 0;
    typename std::multimap<T, DataMessage*>::reverse_iterator i1, e = map.rend();
 
    uint64 now = GetTimeNow();
    uint64 createdAfter = 0;
    if (maxage)
        createdAfter = now - maxage;
 
    if ( (i1 = map.rbegin()) == e) {
        // printf("queryMessages found (%u) results (of %u)...\n", count, map.size());
        return NULL;
    }
 
    uint32 allocSize = 4096;
    char* result = (char*) malloc(allocSize);
 
    DataMessage* msg;
    uint32 msgSize;
    while (i1 != e) {
        if (msg = i1->second) {
            // Filters...
            if (msg->getTTL() && msg->getEOL() < now) {
                i1++;
                continue;
            }
            if (from && (from != msg->getFrom())) {
                i1++;
                continue;
            }
            if (to && (to != msg->getTo())) {
                i1++;
                continue;
            }
            if ((type != NOTYPE) && (!msg->getType().matches(type))) {
                i1++;
                continue;
            }
            if (maxcount && (count >= maxcount))
                break;
            if (createdAfter && (msg->getCreatedTime() < createdAfter)) {
                i1++;
                continue;
            }
 
            msgSize = i1->second->getSize();
            while (size + msgSize > allocSize) {
                allocSize *= 2;
                void* newmem = realloc(result, allocSize);
                if (!newmem) {
                    free(result);
                    count = size = 0;
                    return NULL;
                }
                result = (char*) newmem;
            }
            memcpy(result+size, msg->data, msgSize);
            size += msgSize;
            count++;
        }
        i1++;
    }
    // printf("queryMessages found %u results (of %u)...\n", count, map.size());
    return result;
}
 
 
bool MessageIndex::UnitTest() {
 
    // 1. Setup: build an index and populate it with messages
    unittest::progress(5, "populate index");
    MessageIndex* index = new MessageIndex();
 
    index->addIndexKey("time", INDEX_TIME);
 
    uint32 n, size, count;
    int64 val;
    DataMessage* msg;
    char* result, *data;
    RetrieveSpec spec;
 
    uint64 now = GetTimeNow();
    uint64 t0 = GetTimeNow();
    for (n=0; n<100; n++) {
        msg = new DataMessage();
        msg->setCreatedTime(now - 99000 + (n*1000));
        msg->setTTL(1000000);
        msg->setInt("Count", n);
        index->addMessage(msg);
    }
    double insertUs = (double)(GetTimeNow() - t0);
    unittest::detail("inserted 100 messages in %.0f us", insertUs);
    if (insertUs > 0)
        unittest::metric("insert_throughput", 100.0 / insertUs * 1e6, "msg/s", true);
 
    // 2. Query by maxage
    unittest::progress(35, "query by maxage");
    memset(&spec, 0, sizeof(RetrieveSpec));
    utils::strcpyavail(spec.key, "time", MAXKEYNAMELEN, true);
    spec.keytype = INDEX_TIME;
    spec.maxage = 50000;
 
    result = index->queryMessages(&spec, size, count);
    unittest::detail("maxage query returned %u msgs", count);
    if ( (count < 40) || (count > 50) ) {
        unittest::fail("MessageIndex test: maxage query didn't return around 50 msgs (got %u)", count);
        free(result);
        delete(index);
        return false;
    }
 
    data = result;
    for (n=0; n<count; n++) {
        if (GetObjID(data) != DATAMESSAGEID) {
            unittest::fail("MessageIndex test: maxage query result %u not a DataMessage", n);
            free(result);
            delete(index);
            return false;
        }
        msg = new DataMessage(data, true);
        if (!msg->getInt("Count", val) || (val != 99 - n)) {
            unittest::fail("MessageIndex test: maxage query ordering mismatch at %u (expected %lld)", n, (int64)(99 - n));
            free(result);
            delete(index);
            delete(msg);
            return false;
        }
        data += msg->getSize();
        delete(msg);
    }
    free(result);
 
    // 3. Query by maxcount
    unittest::progress(60, "query by maxcount");
    memset(&spec, 0, sizeof(RetrieveSpec));
    utils::strcpyavail(spec.key, "time", MAXKEYNAMELEN, true);
    spec.keytype = INDEX_TIME;
    spec.maxcount = 50;
 
    result = index->queryMessages(&spec, size, count);
    unittest::detail("maxcount query returned %u msgs", count);
    if (count != 50) {
        unittest::fail("MessageIndex test: maxcount query didn't return 50 msgs (got %u)", count);
        free(result);
        delete(index);
        return false;
    }
 
    data = result;
    for (n=0; n<count; n++) {
        if (GetObjID(data) != DATAMESSAGEID) {
            unittest::fail("MessageIndex test: maxcount query result %u not a DataMessage", n);
            free(result);
            delete(index);
            return false;
        }
        msg = new DataMessage(data, true);
        if (!msg->getInt("Count", val) || (val != 99 - n)) {
            unittest::fail("MessageIndex test: maxcount query ordering mismatch at %u (expected %lld)", n, (int64)(99 - n));
            free(result);
            delete(index);
            delete(msg);
            return false;
        }
        data += msg->getSize();
        delete(msg);
    }
    free(result);
 
    // 4. Time advancement: older messages should age out of the maxage window
    unittest::progress(85, "age-out window");
    utils::Sleep(950);
 
    memset(&spec, 0, sizeof(RetrieveSpec));
    utils::strcpyavail(spec.key, "time", MAXKEYNAMELEN, true);
    spec.keytype = INDEX_TIME;
    spec.maxcount = 100;
 
    result = index->queryMessages(&spec, size, count);
    unittest::detail("post-sleep query returned %u msgs", count);
    if (count > 50) {
        unittest::fail("MessageIndex test: post-sleep query returned more than 50 msgs (got %u)", count);
        free(result);
        delete(index);
        return false;
    }
    free(result);
 
    delete(index);
    unittest::progress(100, "done");
    return true;
}
 
 
void Register_MessageIndex_Tests() {
    UnitTestRunner::instance().registerTest("messageindex", MessageIndex::UnitTest,
        "Message index time-based insert and query (maxage, maxcount, age-out)", "core");
}
 
 
} // namespace cmlabs