src/main/GarbageCollector.cpp

Go to the documentation of this file.

#include <deque>
#include <iomanip>
#include <sstream>
#include <ctime>
#include <sys/time.h>
#include <cstdlib>
#include <csignal>
#include <climits>
#include <cstring>
#include <pthread.h>

#include <cassert>

#include "GarbageCollector.h"
#include "functions.h"

#define GC_DEBUG 20

#define GC_SYNCHRO_DEBUG 0

#define SPACE_MANAGER 0

GarbageCollector::GarbageCollector(Cache *cache, unsigned sleepInterval) :
        cache(cache), sleepInterval(sleepInterval) {

        quotaToDistribute = 0;
        nextQuotaToDistribute = 0;
        maxHardQuota = 0;
        // TODO - add to configuration file
        reqHistoryLimiter = 10;

        pthread_mutex_init(&addMutex, NULL);
        pthread_mutex_init(&removeMutex, NULL);

        int ret;
        if ((ret = pthread_create(&threadID, NULL, GCThread, this))) {
                fprintf(stderr, "GC:\tpthread_create: %s\n", strerror(ret));
                exit(1);
        }

}

GarbageCollector::~GarbageCollector() {

        pthread_kill(threadID, SIGUSR1);

        for (std::map<Space*, std::deque<unsigned> *>::iterator it = reqHistoryMap.begin(); it
                        != reqHistoryMap.end(); ++it) {
                delete it->second;
        }

}

void GarbageCollector::addObservedSpace(Space *space) {
        MUTEX_LOCK(addMutex);
        spacesToAdd.push_back(space);
        cout << "GC:\tadded space " << space->name << endl;
        MUTEX_UNLOCK(addMutex);
}

void GarbageCollector::removeObservedSpace(Space * space) {
        MUTEX_LOCK(removeMutex);
        spacesToRemove.insert(space);
        MUTEX_LOCK(addMutex);
        for (SpaceList::iterator it = spacesToAdd.begin(); it != spacesToAdd.end();) {
                if (*it == space) {
                        it = spacesToAdd.erase(it);
                } else {
                        it++;
                }
        }
        MUTEX_UNLOCK(addMutex);
        cout << "GC:\tremoved space " << space->name << endl;
        MUTEX_UNLOCK(removeMutex);
}

void GarbageCollector::setInterval(unsigned interval) {

        MUTEX_LOCK(removeMutex);
        sleepInterval = interval;
        MUTEX_UNLOCK(removeMutex);

}

void GarbageCollector::operator()() {

        cout << "GC:\tstarted" << endl;

        for (;;) {

                MUTEX_LOCK(addMutex);
                spaceList.insert(spaceList.end(), spacesToAdd.begin(), spacesToAdd.end());
                spacesToAdd.clear();
#if SPACE_MANAGER
                for (SpaceList::iterator it = spacesToAdd.begin(); it != spacesToAdd.end(); ++it) {
                        Space *space = *it;

                        unsigned long long reqMin;

                        Stats stats = space->getStats();

                        if (stats.softQuota > stats.usedQuota)
                        reqMin = stats.usedQuota;
                        else
                        reqMin = stats.softQuota;

                        nextQuotaToDistribute += stats.hardQuota - reqMin;
                        maxHardQuota += stats.hardQuota;
                        requiredMinimum.insert(std::pair<Space *, unsigned long long>(space, reqMin));

                        reqHistoryMap.insert(std::pair<Space *, std::deque<unsigned> *>(space, new std::deque<unsigned>()));
                }
#endif
                MUTEX_UNLOCK(addMutex);

#if SPACE_MANAGER
                // adding new elements to queues
                if (reqHistoryTotal.size() == reqHistoryLimiter)
                reqHistoryTotal.pop_back(); //AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
                reqHistoryTotal.push_front(0);

                quotaToDistribute = maxHardQuota;
                for (std::map<Space *, unsigned long long>::iterator minIt = requiredMinimum.begin(); minIt
                                != requiredMinimum.end(); ++minIt) {
                        quotaToDistribute -= minIt->second;
                }
#endif

                for (SpaceList::iterator it = spaceList.begin(); it != spaceList.end(); ++it) {

                        MUTEX_LOCK(removeMutex);

                        Space * space = *it;

                        if (spacesToRemove.find(space) != spacesToRemove.end() /*|| space->LRUList.empty()*/) {
                                MUTEX_UNLOCK(removeMutex);
                                continue;
                        }

#if SPACE_MANAGER
                        // we set new quota according to the statistics
                        determineNewQuota(space);
#endif

                        collect(space, true);

                        MUTEX_UNLOCK(removeMutex);

                }

                MUTEX_LOCK(removeMutex);
                for (SpaceList::iterator it = spaceList.begin(); it != spaceList.end();) {
                        if (spacesToRemove.find(*it) != spacesToRemove.end()) {
                                it = spaceList.erase(it);
                        } else {
                                ++it;
                        }
                }
                spacesToRemove.clear();

                unsigned sleepTime = sleepInterval;

                MUTEX_UNLOCK(removeMutex);

                usleep(1000 * sleepTime);

        }

}

void GarbageCollector::collect(Space * space, bool goBelowQuota) {

        time_t deadline = Cache::getCurrentTime();

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

                pthread_rwlock_wrlock(space->mapMutexes[i]);

                for (Space::SpaceLRUList::iterator it = space->LRUList[i].begin(); it != space->LRUList[i].end(); ){
                        if (it->expiryTime <= deadline){ //expired, removing right now!
                                Space::SpaceHashTab::iterator hIt = space->hashTab[i].find(it->key);//it->iteratorHashTab;
                                pthread_mutex_lock(&space->globalMutex);
                                space->stats.usedQuota -= (hIt->second.value.size + 2* hIt->first.size);
                                pthread_mutex_unlock(&space->globalMutex);
                                ++space->fragmentedStats[i].itemsCount;

                                free(hIt->second.value.memoryPointer);
                                space->hashTab[i].erase(hIt);
                                Space::SpaceLRUList::iterator nit = it++;
                                space->LRUList[i].erase(nit);
                        } else {
                                it++;
                        }
                }

                pthread_rwlock_unlock(space->mapMutexes[i]);

        }

}

#if SPACE_MANAGER
unsigned long long GarbageCollector::determineNewQuota(Space * space) {

        unsigned long long addedQuota = 0;

        //      std::cout<<"DISTR: "<<quotaToDistribute<<std::endl;

        std::map<Space*, std::deque<unsigned> *>::iterator it = reqHistoryMap.find(space);
        std::deque<unsigned> * reqHistory = it->second;

        if (reqHistory->size() == reqHistoryLimiter) {
                reqHistory->pop_back(); //AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
        }

        Stats stats = space->getStats();

        if (reqHistory->size()) {
                reqHistoryTotal.front() += stats.setsCount - reqHistory->front();
                reqHistory->push_front(stats.setsCount - reqHistory->front());
        } else {
                reqHistoryTotal.front() += stats.setsCount;
                reqHistory->push_front(stats.setsCount);
        }

        //liczymy proporcje
        unsigned spaceScore = 0;
        unsigned totalScore = 0;

        for (unsigned pos = 1; pos < reqHistory->size(); pos++) {
                totalScore += (reqHistory->size() - pos) * reqHistoryTotal[pos];
                spaceScore += (reqHistory->size() - pos) * ((*reqHistory)[pos]);
        }

        unsigned long long reqMin;

        if (stats.softQuota > stats.usedQuota)
        reqMin = stats.usedQuota;
        else
        reqMin = stats.softQuota;

        std::map<Space *, unsigned long long>::iterator minIt = requiredMinimum.find(space);

        if (totalScore) {
                addedQuota = (spaceScore * quotaToDistribute) / totalScore;

                //              std::cout<<"ADDED: "<< addedQuota <<std::endl;

                unsigned long long newHardQuota = minIt->second + addedQuota;

                space->lock();
                space->stats.hardQuota = newHardQuota;
                space->unlock();

        } else
        addedQuota = UINT_MAX;

        //      nextQuotaToDistribute += hardQuota - reqMin;
        minIt->second = reqMin;

        return addedQuota;
}
#endif

void * GCThread(void * arg) {

        GarbageCollector * gc = (GarbageCollector *) arg;

        gc->operator ()();

        return NULL;

}

---