src/main/Space.cpp

Go to the documentation of this file.

#include <cstring>
#include <list>
#include <utility>
#include <iomanip>
#include <cassert>
#include <cmath>
#include <cerrno>
#include <climits> /* for UINT_MAX */

#include "Space.h"
#include "Value.h"
#include "Cache.h"
#include "functions.h"
#include "exceptions.h"
#include "LightString.h"

#define SPACE_DEBUG 0

unsigned int Space::NUMERIC_OBJECT_SIZE = UINT_MAX / 2;
unsigned int Space::INCR_MAX_STORAGE_LEN = 24;
const int Space::MUTEXES_COUNT = 32;

Space::Space(const std::string & name, unsigned long long softQuota, unsigned long long hardQuota,
                Cache *memcache, User *user) :
        name(name), stats(hardQuota, softQuota), hashTab(MUTEXES_COUNT), LRUList(MUTEXES_COUNT),
                        memcache(memcache), user(user) {

        pthread_mutex_init(&globalMutex, NULL);

        for (int i = 0; i < MUTEXES_COUNT; ++i) {
                mapMutexes.push_back((pthread_rwlock_t *) malloc(sizeof(pthread_rwlock_t)));

                if (pthread_rwlock_init(mapMutexes[i], NULL)) {
                        perror("pthread_rwlock_init");
                        exit(1);
                }
        }

        fragmentedStats = new Stats[MUTEXES_COUNT];

        current_cas_id = 1;

}

Space::~Space() {
        cleanSpace();

        pthread_mutex_destroy(&globalMutex);

        for (int i = 0; i < MUTEXES_COUNT; ++i) {
                pthread_rwlock_destroy(mapMutexes[i]);

                free(mapMutexes[i]);
        }

        delete[] fragmentedStats;
}

bool Space::get(const LightString &key, Value& value, unsigned timeout) {

        int tableCookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[tableCookie]);

        SpaceHashTab::iterator it = hashTab[tableCookie].find(key);
        if ((it == hashTab[tableCookie].end()) || (Cache::getCurrentTime()
                        > it->second.iteratorLRUList->expiryTime)) {

                ++fragmentedStats[tableCookie].missCount;
                ++fragmentedStats[tableCookie].getsCount;
                pthread_rwlock_unlock(mapMutexes[tableCookie]);

                return false;
        }

        if (it->second.value.size != NUMERIC_OBJECT_SIZE) {
                fragmentedStats[tableCookie].bytesRead += it->second.value.size;
        } else {
                fragmentedStats[tableCookie].bytesRead += sizeof(unsigned long long);
        }
        ++fragmentedStats[tableCookie].hitCount;
        ++fragmentedStats[tableCookie].getsCount;

        if (timeout != 0) {
                it->second.iteratorLRUList->expiryTime = computeExpiryTime(timeout);
        }

        // moving the current LRUelement to the beginning
        LRUListElement LRUe = *it->second.iteratorLRUList;
        LRUList[tableCookie].erase(it->second.iteratorLRUList);

        LRUList[tableCookie].push_front(LRUe);

        it->second.iteratorLRUList = LRUList[tableCookie].begin();

        if (!value.memoryPointer) {
                if (!(value.memoryPointer = malloc(it->second.value.size))) {
                        pthread_rwlock_unlock(mapMutexes[tableCookie]);
                        return false;
                }
        } else if (value.size < it->second.value.size && it->second.value.size != NUMERIC_OBJECT_SIZE) {
                pthread_rwlock_unlock(mapMutexes[tableCookie]);
                return false;
        }

        value.cas_id = it->second.value.cas_id;
        value.flag = it->second.value.flag;
        value.size = it->second.value.size;

        if (value.size != NUMERIC_OBJECT_SIZE) {
                memcpy(value.memoryPointer, it->second.value.memoryPointer, value.size);
                pthread_rwlock_unlock(mapMutexes[tableCookie]);
                return true;
        }

        //now we have to convert the numeric value to its text representation which is expected by server


        unsigned long long numVal = *((unsigned long long*) it->second.value.memoryPointer);
        pthread_rwlock_unlock(mapMutexes[tableCookie]);
        value.size = NUM_LENGTH(numVal);
        char * end = (char*) value.memoryPointer + value.size - 1;

        while (numVal) {
                *end-- = numVal % 10 + '0';
                numVal /= 10;
        };

        return true;
}

Value Space::get(const LightString &key, unsigned timeout) {

        Value v;
        get(key, v, timeout);
        return v;

}

SpaceResult Space::doSet(const LightString &key, const Value &value, unsigned timeout,
                int tableCookie, SpaceHashTab::iterator found, uint64_t cas_id, uint64_t * new_cas_id) {

        unsigned expiryTime = computeExpiryTime(timeout);

        unsigned int valueSize = (value.size == NUMERIC_OBJECT_SIZE ? sizeof(uint64_t) : value.size);

        int64_t spaceTaken = 0;

        if (found != hashTab[tableCookie].end()) {

                // NUMERIC_OBJECT_SIZE can occur only when inserting element does not exist
                // it is called from crement() then
                assert(value.size != NUMERIC_OBJECT_SIZE);

                if (cas_id) {
                        if (found->second.value.cas_id != cas_id) {
                                return RESULT_CAS_FAIL;
                        }
                }

                if (found->second.value.size != value.size) {
                        found->second.value.memoryPointer = realloc(found->second.value.memoryPointer,
                                        value.size);
                        if (found->second.value.memoryPointer == NULL) {
                                return RESULT_ERROR;
                        }
                }

                memcpy(found->second.value.memoryPointer, value.memoryPointer, value.size);
                found->second.value.flag = value.flag;
                found->second.iteratorLRUList->expiryTime = expiryTime;

                // synchronization of the space counters and the list

                if (found->second.value.size != NUMERIC_OBJECT_SIZE) {
                        spaceTaken = (long long) value.size - (long long) found->second.value.size;
                } else {
                        spaceTaken = (long long) value.size - (long long) sizeof(unsigned long long);
                }

                found->second.value.size = value.size;

                // moving the current LRUelement to the beginning
                LRUListElement LRUe = *found->second.iteratorLRUList;
                LRUList[tableCookie].erase(found->second.iteratorLRUList);

                LRUList[tableCookie].push_front(LRUe);

                found->second.iteratorLRUList = LRUList[tableCookie].begin();

        } else {

                //it may exceed hardQuota for a 'very short' moment
                void *memoryPointer = malloc(valueSize);

                if (!memoryPointer) {
                        return RESULT_ERROR;
                }

                //copy given value into the newly allocated block
                memcpy(memoryPointer, value.memoryPointer, valueSize);

                //update statistics
                ++fragmentedStats[tableCookie].itemsCount;

                spaceTaken = valueSize + 2 * key.size;

                SpaceLRUList::iterator newHead = LRUList[tableCookie].insert(LRUList[tableCookie].begin(),
                                LRUListElement(tableCookie, key, expiryTime));

                found = hashTab[tableCookie].insert(std::make_pair(key, HashTabElement(Value(memoryPointer,
                                value.size, value.flag, cas_id), newHead))).first;

        }

        ++fragmentedStats[tableCookie].setsCount;
        fragmentedStats[tableCookie].bytesWritten += valueSize;

        pthread_mutex_lock(&globalMutex);
        //update statistics
        stats.usedQuota += spaceTaken;
        cas_id = current_cas_id++;

        found->second.value.cas_id = cas_id;

        if (stats.usedQuota > stats.hardQuota && spaceTaken > 0) {
                pthread_mutex_unlock(&globalMutex);
                unsigned removed = 0;
                while (removed < valueSize + 2 * key.size) {
                        //czy usuwanie od siebie jest zawsze dobre?
                        unsigned which = tableCookie;//rand() % MUTEXES_COUNT;

                        LRUListElement LRUe = LRUList[which].back();
                        SpaceHashTab::iterator it = hashTab[which].find(LRUe.key); //LRUe.iteratorHashTab;
                        free(it->second.value.memoryPointer);
                        removed += (it->second.value.size + 2 * it->first.size);
                        hashTab[which].erase(it);
                        LRUList[which].pop_back();

                }
                pthread_mutex_lock(&globalMutex);
                stats.usedQuota -= removed;
        }
        pthread_mutex_unlock(&globalMutex);

        if (new_cas_id) {
                *new_cas_id = cas_id;
        }

        return RESULT_SUCCESS;
}

SpaceResult Space::set(const LightString &key, const Value &value, unsigned timeout,
                uint64_t * new_cas_id) {

        //check if the given value can be stored
        if (value.size == 0 || value.size > Cache::MAX_OBJECT_SIZE || value.size > stats.hardQuota) {
                return RESULT_BAD_SIZE;
        }

        int cookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[cookie]);

        SpaceHashTab::iterator it = hashTab[cookie].find(key);

        SpaceResult output = doSet(key, value, timeout, cookie, it, 0, new_cas_id);

        pthread_rwlock_unlock(mapMutexes[cookie]);

        return output;
}

SpaceResult Space::cas(const LightString &key, const Value &value, unsigned timeout,
                uint64_t cas_id, uint64_t * new_cas_id) {

        // check if the given value can be stored
        if (value.size == 0 || value.size > Cache::MAX_OBJECT_SIZE || value.size > stats.hardQuota) {
                return RESULT_BAD_SIZE;
        }

        int cookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[cookie]);

        SpaceHashTab::iterator it = hashTab[cookie].find(key);

        SpaceResult output = doSet(key, value, timeout, cookie, it, cas_id, new_cas_id);

        pthread_rwlock_unlock(mapMutexes[cookie]);

        return output;

}

SpaceResult Space::add(const LightString &key, const Value &value, unsigned timeout,
                uint64_t cas_id, uint64_t * new_cas_id) {

        //check if the given value can be stored
        if (value.size == 0 || value.size > Cache::MAX_OBJECT_SIZE || value.size > stats.hardQuota) {
                return RESULT_BAD_SIZE;
        }

        int cookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[cookie]);

        SpaceHashTab::iterator it = hashTab[cookie].find(key);
        SpaceResult ret;
        if (it != hashTab[cookie].end()) {
                ret = RESULT_EXISTS;
        } else {
                ret = doSet(key, value, timeout, cookie, it, cas_id, new_cas_id);
        }

        pthread_rwlock_unlock(mapMutexes[cookie]);

        return ret;
}

SpaceResult Space::replace(const LightString &key, const Value &value, unsigned timeout,
                uint64_t cas_id, uint64_t * new_cas_id) {

        //check if the given value can be stored
        if (value.size == 0 || value.size > Cache::MAX_OBJECT_SIZE || value.size > stats.hardQuota) {
                return RESULT_BAD_SIZE;
        }

        int cookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[cookie]);

        SpaceHashTab::iterator it = hashTab[cookie].find(key);
        SpaceResult ret;
        if (it != hashTab[cookie].end()) {
                ret = doSet(key, value, timeout, cookie, it, cas_id, new_cas_id);
        } else {
                ret = RESULT_MISS;
        }
        pthread_rwlock_unlock(mapMutexes[cookie]);

        return ret;
}

SpaceResult Space::append(const LightString &key, const Value &value, uint64_t * new_cas_id) {
        return pend(key, value, new_cas_id, true);
}

SpaceResult Space::prepend(const LightString &key, const Value &value, uint64_t * new_cas_id) {
        return pend(key, value, new_cas_id, false);
}

SpaceResult Space::pend(const LightString &key, const Value &value, uint64_t * new_cas_id,
                bool append) {

        // the base of append and prepend --> *pend ;)

        //check if the given value can be stored
        if (value.size > stats.hardQuota || value.size == 0) {
                return RESULT_BAD_SIZE;
        }

        // table exclusive lock
        int cookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[cookie]);
        SpaceResult res;

        SpaceHashTab::iterator it = hashTab[cookie].find(key);
        if (it == hashTab[cookie].end()) {
                res = RESULT_MISS;
        } else {

                unsigned realSize;
                char tmpbuf[INCR_MAX_STORAGE_LEN];
                if (it->second.value.size == NUMERIC_OBJECT_SIZE) {
                        sprintf(tmpbuf, "%llu", (*((unsigned long long*) it->second.value.memoryPointer)));
                        realSize = strlen(tmpbuf);
                } else {
                        realSize = it->second.value.size;
                }

                if (value.size + realSize > stats.hardQuota) {
                        res = RESULT_ERROR;
                } else {
                        //update statistics
                        fragmentedStats[cookie].setsCount++;
                        fragmentedStats[cookie].bytesWritten += value.size;

                        //allocate the memory for current value plus the given one
                        void *memoryPointer = malloc(value.size + realSize);

                        if (!memoryPointer) {
                                pthread_rwlock_unlock(mapMutexes[cookie]);
                                return RESULT_ERROR;
                        }

                        // a bit of the following code is copied between two if cases but it is the fastest
                        // way to do this
                        if (append) {
                                //copy current value into the new block
                                if (it->second.value.size == NUMERIC_OBJECT_SIZE) {
                                        memcpy(memoryPointer, tmpbuf, realSize);
                                        it->second.value.size = sizeof(unsigned long long);
                                } else {
                                        memcpy(memoryPointer, it->second.value.memoryPointer, realSize);
                                }

                                // append given value into the new block
                                memcpy((char*) memoryPointer + realSize, value.memoryPointer, value.size);
                        } else {
                                //copy current value into the new block
                                if (it->second.value.size == NUMERIC_OBJECT_SIZE) {
                                        memcpy((char*) memoryPointer + value.size, tmpbuf, realSize);
                                        it->second.value.size = sizeof(unsigned long long);
                                } else {
                                        memcpy((char*) memoryPointer + value.size, it->second.value.memoryPointer,
                                                        realSize);
                                }

                                // prepend given value into the new block
                                memcpy(memoryPointer, value.memoryPointer, value.size);
                        }

                        //free the block storing old value
                        free(it->second.value.memoryPointer);

                        //replace info at the hashtable about the old value with the appended one
                        pthread_mutex_lock(&globalMutex);
                        stats.usedQuota += value.size;
                        uint64_t cas_id = current_cas_id++;
                        pthread_mutex_unlock(&globalMutex);

                        it->second.value.memoryPointer = memoryPointer;
                        it->second.value.size = value.size + realSize;
                        it->second.value.flag = value.flag;
                        it->second.value.cas_id = cas_id;

                        if (new_cas_id) {
                                *new_cas_id = cas_id;
                        }
                        res = RESULT_SUCCESS;
                }
        }
        pthread_rwlock_unlock(mapMutexes[cookie]);
        return res;
}

SpaceResult Space::increment(const LightString &key, uint64_t step, uint64_t * out,
                unsigned timeout, bool force, uint64_t initial) {

        return crement(key, step, out, timeout, force, initial);

}

SpaceResult Space::decrement(const LightString &key, uint64_t step, uint64_t * out,
                unsigned timeout, bool force, uint64_t initial) {

        return crement(key, ~step + 1, out, timeout, force, initial);

}

SpaceResult Space::crement(const LightString &key, uint64_t step, uint64_t * out, unsigned timeout,
                bool force, uint64_t initial) {

        // the base of increment and decremet --> *crement ;)

        int cookie = hash(key);
        pthread_rwlock_wrlock(mapMutexes[cookie]);

        SpaceHashTab::iterator it = hashTab[cookie].find(key);

        SpaceResult res;
        void * memoryPointer;

        if (it != hashTab[cookie].end()) {

                //unsigned flag = it->second.value.flag;

                if (it->second.value.size == NUMERIC_OBJECT_SIZE) {
                        *out = ((*((uint64_t *) it->second.value.memoryPointer)) += step);
                        res = RESULT_SUCCESS;
                } else {
                        if (!safe_strtoull((string((char*) (it->second.value.memoryPointer),
                                        it->second.value.size).substr(0, it->second.value.size)).c_str(),
                                        (uint64_t*) out)) {

                                res = RESULT_ERROR;

                        } else if (!(memoryPointer = malloc(sizeof(initial)))) {

                                res = RESULT_ERROR;

                        } else {

                                *out += step;
                                memcpy(memoryPointer, (void*) out, sizeof(initial));
                                free(it->second.value.memoryPointer);
                                it->second.value.memoryPointer = memoryPointer;

                                pthread_mutex_lock(&globalMutex);
                                stats.usedQuota += sizeof(initial) - it->second.value.size;
                                pthread_mutex_unlock(&globalMutex);

                                it->second.value.size = NUMERIC_OBJECT_SIZE;
                                res = RESULT_SUCCESS;
                        }
                }
        } else {

                if (force) {
                        res = doSet(key, Value(&initial, NUMERIC_OBJECT_SIZE), timeout, cookie, it);
                        *out = initial;
                } else {
                        res = RESULT_MISS;
                }
        }
        pthread_rwlock_unlock(mapMutexes[cookie]);
        return res;
}

void Space::setStatus(unsigned nUsedQuota, unsigned nGetsCount, unsigned nSetsCount,
                unsigned nBytesWritten, unsigned nBytesRead, unsigned nMissCount, unsigned nHitCount,
                unsigned long long nItemsCount) {
        //TODO synchrnonizacja?
        //na razie dwa sa zakomentowane, gdyz dodawanie wpisow setami je podwaja
        //usedQuota = nUsedQuota;
        stats.getsCount = nGetsCount;
        stats.setsCount = nSetsCount;
        //bytesWritten = nBytesWritten;
        stats.bytesRead = nBytesRead;
        stats.missCount = nMissCount;
        stats.hitCount = nHitCount;
        stats.itemsCount = nItemsCount;
}

bool Space::saveSpace(const char *fileName, unsigned minTimeout) {
        /*      ofstream output;
         output.open(fileName, ios::out | ios::binary);

         time_t threshold = Cache::getCurrentTime() + minTimeout;

         long spaceMark;
         unsigned writtenCount = 0;

         output.write((char*) &stats.softQuota, sizeof(stats.softQuota));
         output.write((char*) &stats.hardQuota, sizeof(stats.hardQuota));

         spaceMark = output.tellp();
         output.write((char*) &writtenCount, sizeof(writtenCount));

         for (int i = 0; i < MUTEXES_COUNT; i++) {

         for (SpaceHashTab::const_iterator it = hashTab[i].begin(); it != hashTab[i].end(); ++it) {
         if (it->second.iteratorLRUList->expiryTime >= threshold) {

         unsigned key_length = it->first.size + 1;
         output.write((char*) &key_length, sizeof(key_length));

         output.write(it->first.memoryPointer, key_length);

         unsigned size = it->second.value.size;
         output.write((char*) &size, sizeof(size));

         //                     char * value = strdup((char*)it->second.value.memoryPointer);
         output.write((char*) it->second.value.memoryPointer, size);

         time_t expiryTime = it->second.iteratorLRUList->expiryTime;
         output.write((char*) &expiryTime, sizeof(expiryTime));

         output.write((char *) &it->second.value.flag, sizeof(it->second.value.flag));

         writtenCount++;
         }
         }

         }

         output.seekp(spaceMark);
         output.write((char*) &writtenCount, sizeof(writtenCount));

         output.flush();
         output.close();
         */
        return true;
}

//
void Space::cleanSpace() {
        //TODO should clear stats too?
        //      for (SpaceLRUList::iterator it = LRUList.begin(); it != LRUList.end();) {
        //              SpaceLRUList::iterator next = it;
        //              ++next;
        //              removeValue(hashTab.find(it->key));
        //              it = next;
        //      }
}

void Space::destroySpace() {

        cleanSpace(); //TODO what's the difference between cleaning and destroying?
}

Stats Space::getStats() {
        pthread_mutex_lock(&globalMutex);
        stats.getsCount = 0;
        stats.missCount = 0;
        stats.hitCount = 0;
        stats.bytesRead = 0;
        stats.itemsCount = 0;
        stats.bytesWritten = 0;
        stats.setsCount = 0;
        for (int i = 0; i < MUTEXES_COUNT; i++) {
                stats.getsCount += fragmentedStats[i].getsCount;
                stats.missCount += fragmentedStats[i].missCount;
                stats.hitCount += fragmentedStats[i].hitCount;
                stats.bytesRead += fragmentedStats[i].bytesRead;
                stats.bytesWritten += fragmentedStats[i].bytesWritten;
                stats.itemsCount += fragmentedStats[i].itemsCount;
                stats.setsCount += fragmentedStats[i].setsCount;
        }
        Stats tmp = stats;
        pthread_mutex_unlock(&globalMutex);
        return tmp;
}

void Space::clearStats() {

        stats.bytesRead = 0;
        stats.bytesWritten = 0;
        stats.getsCount = 0;
        stats.setsCount = 0;
        stats.hitCount = 0;
        stats.missCount = 0;
}

size_t Space::hash(const LightString & key) {
        return HASH(key) % MUTEXES_COUNT;
}

unsigned Space::computeExpiryTime(unsigned timeout) {

        if (timeout == 0) {
                return INT_MAX;
        } else if (timeout > 30 * 24 * 60 * 60) { //number of seconds in 30 days
                return timeout;
        } else {
                return timeout + Cache::getCurrentTime();
        }
}

---