lmdbal/src/cache.hpp

/*
 * LMDB Abstraction Layer.
 * Copyright (C) 2023  Yury Gubich <blue@macaw.me>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef LMDBAL_CACHE_HPP
#define LMDBAL_CACHE_HPP

#include "cache.h"
#include "exceptions.h"

template<class K, class V>
LMDBAL::Cache<K, V>::Cache(const std::string& p_name, Base* parent):
    Storage<K, V>(p_name, parent),
    mode(new Mode),
    cache(new std::map<K, V>()),
    abscent(new std::set<K>()),
    sizeDifference(new uint32_t),
    transactionCache(new TransactionCache)
{
    *mode = Mode::nothing;
    *sizeDifference = 0;
}

template<class K, class V>
LMDBAL::Cache<K, V>::~Cache() {
    delete transactionCache;
    delete sizeDifference;
    delete mode;
    delete cache;
    delete abscent;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::addRecord(const K& key, const V& value) {
    iStorage::ensureOpened(iStorage::addRecordMethodName);

    if (cache->count(key) > 0)
        iStorage::throwDuplicate(iStorage::toString(key));

    Storage<K, V>::addRecord(key, value);
    handleAddRecord(key, value);
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleAddRecord(const K& key, const V& value) {
    cache->insert(std::make_pair(key, value));

    if (*mode != Mode::full)
        abscent->erase(key);
}

template<class K, class V>
bool LMDBAL::Cache<K, V>::forceRecord(const K& key, const V& value) {
    iStorage::ensureOpened(iStorage::forceRecordMethodName);

    bool added = Storage<K, V>::forceRecord(key, value);
    handleForceRecord(key, value, added);

    return added;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleForceRecord(const K& key, const V& value, bool added) {
    if (*mode == Mode::full) {
        (*cache)[key] = value;
    } else {
        if (added)
            abscent->erase(key);

        std::pair<typename std::map<K, V>::iterator, bool> result = cache->insert(std::make_pair(key, value));
        if (!result.second)
            result.first->second = value;
        else if (!added)    //this way database had value but cache didn't, so, need to decrease sizeDifference
            handleMode();
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::changeRecord(const K& key, const V& value) {
    iStorage::ensureOpened(iStorage::changeRecordMethodName);

    if (*mode == Mode::full) {
        typename std::map<K, V>::iterator itr = cache->find(key);
        if (itr == cache->end())
            iStorage::throwNotFound(iStorage::toString(key));

        Storage<K, V>::changeRecord(key, value);
        itr->second = value;
    } else {
        if (abscent->count(key) > 0)
            iStorage::throwNotFound(iStorage::toString(key));

        try {
            Storage<K, V>::changeRecord(key, value);
            typename std::pair<typename std::map<K, V>::iterator, bool> res = cache->insert(std::make_pair(key, value));
            if (!res.second)
                res.first->second = value;
            else
                handleMode();
        } catch (const NotFound& error) {
            abscent->insert(key);
            throw error;
        }
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleChangeRecord(const K& key, const V& value) {
    if (*mode == Mode::full) {
        cache->at(key) = value;
    } else {
        typename std::pair<typename std::map<K, V>::iterator, bool> res = cache->insert(std::make_pair(key, value));
        if (!res.second)
            res.first->second = value;
        else
            handleMode();
    }
}

template<class K, class V>
V LMDBAL::Cache<K, V>::getRecord(const K& key) const {
    iStorage::ensureOpened(iStorage::getRecordMethodName);

    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end())
        return itr->second;

    if (*mode == Mode::full || abscent->count(key) != 0)
        iStorage::throwNotFound(iStorage::toString(key));

    try {
        V value = Storage<K, V>::getRecord(key);
        cache->insert(std::make_pair(key, value));
        handleMode();
        return value;
    } catch (const NotFound& error) {
        if (*mode != Mode::full)
            abscent->insert(key);

        throw error;
    }
}

template<class K, class V>
bool LMDBAL::Cache<K, V>::checkRecord(const K& key) const {
    iStorage::ensureOpened(iStorage::checkRecordMethodName);

    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end())
        return true;

    if (*mode == Mode::full || abscent->count(key) != 0)
        return false;

    try {
        V value = Storage<K, V>::getRecord(key);
        cache->insert(std::make_pair(key, value));
        handleMode();
        return true;
    } catch (const NotFound& error) {
        if (*mode != Mode::full)
            abscent->insert(key);

        return false;
    }
}

template<class K, class V>
std::map<K, V> LMDBAL::Cache<K, V>::readAll() const {
    iStorage::ensureOpened(iStorage::readAllMethodName);

    if (*mode != Mode::full) {              //there is a room for optimization
        *mode = Mode::full;                 //I can read and deserialize only those values
        *cache = Storage<K, V>::readAll();  //that are missing in the cache
        abscent->clear();
        *sizeDifference = 0;
    }

    return *cache;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::replaceAll(const std::map<K, V>& data) {
    Storage<K, V>::replaceAll(data);
    *cache = data;

    if (*mode != Mode::full) {
        *mode = Mode::full;
        abscent->clear();
        *sizeDifference = 0;
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleReplaceAll(std::map<K, V>* data) {
    delete cache;
    cache = data;

    if (*mode != Mode::full) {
        *mode = Mode::full;
        abscent->clear();
        *sizeDifference = 0;
    }
}

template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::addRecords(const std::map<K, V>& data, bool overwrite) {
    SizeType newSize = Storage<K, V>::addRecords(data, overwrite);
    handleAddRecords(data, overwrite, newSize);

    return newSize;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleAddRecords(const std::map<K, V>& data, bool overwrite, SizeType newSize) {
    Mode& m = *mode;
    if (m == Mode::nothing)
        m = Mode::size;

    std::map<K, V>& c = *cache;
    std::set<K>& a = *abscent;
    for (const std::pair<const K, V>& pair : data) {
        std::pair<typename std::map<K, V>::iterator, bool> res = c.insert(pair);
        if (!res.second) {
            if (overwrite)
                res.first->second = pair.second;
        } else if (m != Mode::full) {
            a.erase(pair.first);
        }
    }

    if (m != Mode::full) {
        *sizeDifference = newSize - c.size();
        if (*sizeDifference == 0) {
            *mode = Mode::full;
            abscent->clear();
        }
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::removeRecord(const K& key) {
    iStorage::ensureOpened(iStorage::removeRecordMethodName);

    bool noKey = false;
    if (*mode != Mode::full)
        noKey = cache->count(key) == 0;
    else
        noKey = abscent->count(key) > 0;

    if (noKey)
        iStorage::throwNotFound(iStorage::toString(key));

    Storage<K, V>::removeRecord(key);
    handleRemoveRecord(key);
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleRemoveRecord(const K& key)  {
    if (cache->erase(key) == 0)     //if it was not cached and we are now in size mode then the sizeDifference would decrease
        handleMode();

    if (*mode != Mode::full)
        abscent->insert(key);
}

template<class K, class V>
uint32_t LMDBAL::Cache<K, V>::count() const {
    switch (*mode) {
        case Mode::nothing:
        {
            uint32_t sz = Storage<K, V>::count();
            *sizeDifference = sz - cache->size();
            if (sz == 0) {
                *mode = Mode::full;
                abscent->clear();
            } else {
                *mode = Mode::size;
            }
            return sz;
        }
        case Mode::size:
            return cache->size() + *sizeDifference;
        case Mode::full:
            return cache->size();
        default:
            return 0;   //unreachable, no such state, just to suppress the waring
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleMode() const {
    if (*mode == Mode::size) {
        --(*sizeDifference);
        if (*sizeDifference == 0) {
            *mode = Mode::full;
            abscent->clear();
        }
    }
}

template<class K, class V>
int LMDBAL::Cache<K, V>::drop(TransactionID transaction) {
    int res = Storage<K, V>::drop(transaction);

    transactionCache->at(transaction).emplace_back(Operation::drop, nullptr);

    return res;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleDrop() {
    cache->clear();
    abscent->clear();
    *mode = Mode::full;
    *sizeDifference = 0;
}

template<class K, class V>
void LMDBAL::Cache<K, V>::transactionStarted(TransactionID txn, bool readOnly) const {
    if (!readOnly)
        transactionCache->emplace(txn, Queue());
}

template<class K, class V>
void LMDBAL::Cache<K, V>::transactionCommited(TransactionID txn) {
    typename TransactionCache::iterator itr = transactionCache->find(txn);
    if (itr != transactionCache->end()) {
        Queue& queue = itr->second;
        for (const Entry& entry : queue)
            handleTransactionEntry(entry);

        transactionCache->erase(itr);
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::transactionAborted(TransactionID txn) const {
    typename TransactionCache::iterator itr = transactionCache->find(txn);
    if (itr != transactionCache->end()) {
        Queue& queue = itr->second;
        for (const Entry& entry : queue)
            destroyTransactionEntry(entry);

        transactionCache->erase(itr);
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::handleTransactionEntry(const Entry& entry) {
    switch (entry.first) {
        case Operation::add: {
            std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
            handleAddRecord(pair->first, pair->second);
            delete pair;
        }
            break;
        case Operation::remove: {
            K* key = static_cast<K*>(entry.second);
            handleRemoveRecord(*key);
            delete key;
        }

            break;
        case Operation::change: {
            std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
            handleChangeRecord(pair->first, pair->second);
            delete pair;
        }
        case Operation::force: {
            std::tuple<bool, K, V>* tuple = static_cast<std::tuple<bool, K, V>*>(entry.second);
            const std::tuple<bool, K, V>& t = *tuple;
            handleForceRecord(std::get<1>(t), std::get<2>(t), std::get<0>(t));
            delete tuple;
        }
            break;
        case Operation::drop:
            handleDrop();
            break;
        case Operation::replace:
            handleReplaceAll(static_cast<std::map<K, V>*>(entry.second));   //I take ownership, no need to delete
            break;
        case Operation::addMany: {
            std::tuple<bool, SizeType, std::map<K, V>>* tuple = static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
            const std::tuple<bool, SizeType, std::map<K, V>>& t = * tuple;
            handleAddRecords(std::get<2>(t), std::get<0>(t), std::get<1>(t));
            delete tuple;
        }
            break;
    }
}

template<class K, class V>
void LMDBAL::Cache<K, V>::destroyTransactionEntry(const Entry& entry) const {
    switch (entry.first) {
        case Operation::add:
            delete static_cast<std::pair<K, V>*>(entry.second);
            break;
        case Operation::remove:
            delete static_cast<K*>(entry.second);
            break;
        case Operation::change:
            delete static_cast<std::pair<K, V>*>(entry.second);
            break;
        case Operation::force:
            delete static_cast<std::tuple<bool, K, V>*>(entry.second);
            break;
        case Operation::drop:
            break;
        case Operation::replace:
            delete static_cast<std::map<K, V>*>(entry.second);
            break;
        case Operation::addMany:
            delete static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
            break;
    }
}

#endif //LMDBAL_CACHE_HPP