<commit_before>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2011 Couchbase, Inc
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#include "config.h"

#include <platform/checked_snprintf.h>
#include <string>
#include <utility>
#include <vector>

#include "checkpoint.h"
#include "ep_engine.h"
#define STATWRITER_NAMESPACE checkpoint
#include "statwriter.h"
#undef STATWRITER_NAMESPACE
#include "vbucket.h"

const std::string CheckpointManager::pCursorName("persistence");

/**
 * A listener class to update checkpoint related configs at runtime.
 */
class CheckpointConfigChangeListener : public ValueChangedListener {
public:
    CheckpointConfigChangeListener(CheckpointConfig &c) : config(c) { }
    virtual ~CheckpointConfigChangeListener() { }

    virtual void sizeValueChanged(const std::string &key, size_t value) {
        if (key.compare("chk_period") == 0) {
            config.setCheckpointPeriod(value);
        } else if (key.compare("chk_max_items") == 0) {
            config.setCheckpointMaxItems(value);
        } else if (key.compare("max_checkpoints") == 0) {
            config.setMaxCheckpoints(value);
        }
    }

    virtual void booleanValueChanged(const std::string &key, bool value) {
        if (key.compare("item_num_based_new_chk") == 0) {
            config.allowItemNumBasedNewCheckpoint(value);
        } else if (key.compare("keep_closed_chks") == 0) {
            config.allowKeepClosedCheckpoints(value);
        } else if (key.compare("enable_chk_merge") == 0) {
            config.allowCheckpointMerge(value);
        }
    }

private:
    CheckpointConfig &config;
};

void CheckpointCursor::decrOffset(size_t decr) {
    if (offset >= decr) {
        offset.fetch_sub(decr);
    } else {
        offset = 0;
        LOG(EXTENSION_LOG_INFO, "%s cursor offset is negative. Reset it to 0.",
            name.c_str());
    }
}

void CheckpointCursor::decrPos() {
    if (currentPos != (*currentCheckpoint)->begin()) {
        --currentPos;
    }
}

MustSendCheckpointEnd CheckpointCursor::shouldSendCheckpointEndMetaItem() const
{
    return sendCheckpointEndMetaItem;
}

Checkpoint::~Checkpoint() {
    LOG(EXTENSION_LOG_INFO,
        "Checkpoint %" PRIu64 " for vbucket %d is purged from memory",
        checkpointId, vbucketId);
    stats.memOverhead.fetch_sub(memorySize());
    if (stats.memOverhead.load() >= GIGANTOR) {
        LOG(EXTENSION_LOG_WARNING,
            "Checkpoint::~Checkpoint: stats.memOverhead (which is %" PRId64
            ") is greater than %" PRId64, uint64_t(stats.memOverhead.load()),
            uint64_t(GIGANTOR));
    }
}

void Checkpoint::setState(checkpoint_state state) {
    checkpointState = state;
}

void Checkpoint::popBackCheckpointEndItem() {
    if (!toWrite.empty() &&
        toWrite.back()->getOperation() == queue_op_checkpoint_end) {
        metaKeyIndex.erase(toWrite.back()->getKey());
        toWrite.pop_back();
    }
}

bool Checkpoint::keyExists(const std::string &key) {
    return keyIndex.find(key) != keyIndex.end();
}

queue_dirty_t Checkpoint::queueDirty(const queued_item &qi,
                                     CheckpointManager *checkpointManager) {
    if (checkpointState != CHECKPOINT_OPEN) {
        throw std::logic_error("Checkpoint::queueDirty: checkpointState "
                        "(which is" + std::to_string(checkpointState) +
                        ") is not OPEN");
    }
    queue_dirty_t rv;

    checkpoint_index::iterator it = keyIndex.find(qi->getKey());

    // Check if the item is a meta item
    if (qi->isCheckPointMetaItem()) {
        rv = NEW_ITEM;
        toWrite.push_back(qi);
    } else {
        // Check if this checkpoint already had an item for the same key
        if (it != keyIndex.end()) {
            rv = EXISTING_ITEM;
            std::list<queued_item>::iterator currPos = it->second.position;
            uint64_t currMutationId = it->second.mutation_id;

            cursor_index::iterator map_it =
                                        checkpointManager->connCursors.begin();
            for (; map_it != checkpointManager->connCursors.end(); ++map_it) {

                if (*(map_it->second.currentCheckpoint) == this) {
                    queued_item &tqi = *(map_it->second.currentPos);
                    const std::string &key = tqi->getKey();
                    checkpoint_index::iterator ita = keyIndex.find(key);
                    if (ita != keyIndex.end() && (!tqi->isCheckPointMetaItem()))
                    {
                        uint64_t mutationId = ita->second.mutation_id;
                        if (currMutationId <= mutationId) {
                            map_it->second.decrOffset(1);
                            if (map_it->second.name.compare(CheckpointManager::pCursorName)
                                == 0) {
                                rv = PERSIST_AGAIN;
                            }
                        }
                    }
                    /* If an TAP cursor points to the existing item for the same
                       key, shift it left by 1 */
                    if (map_it->second.currentPos == currPos) {
                        map_it->second.decrPos();
                    }
                }
            }

            toWrite.push_back(qi);
            // Remove the existing item for the same key from the list.
            toWrite.erase(currPos);
        } else {
            ++numItems;
            rv = NEW_ITEM;
            // Push the new item into the list
            toWrite.push_back(qi);
        }
    }

    if (qi->getNKey() > 0) {
        std::list<queued_item>::iterator last = toWrite.end();
        // --last is okay as the list is not empty now.
        index_entry entry = {--last, qi->getBySeqno()};
        // Set the index of the key to the new item that is pushed back into
        // the list.
        if (qi->isCheckPointMetaItem()) {
            // We add a meta item only once to a checkpoint
            metaKeyIndex[qi->getKey()] = entry;
        } else {
            keyIndex[qi->getKey()] = entry;
        }
        if (rv == NEW_ITEM) {
            size_t newEntrySize = qi->getNKey() + sizeof(index_entry) +
                                  sizeof(queued_item);
            memOverhead += newEntrySize;
            stats.memOverhead.fetch_add(newEntrySize);
            if (stats.memOverhead.load() >= GIGANTOR) {
                LOG(EXTENSION_LOG_WARNING,
                    "Checkpoint::queueDirty: stats.memOverhead (which is %" PRId64
                    ") is greater than %" PRId64, uint64_t(stats.memOverhead.load()),
                    uint64_t(GIGANTOR));
            }
        }
    }

    // Notify flusher if in case queued item is a checkpoint meta item
    if (qi->getOperation() == queue_op_checkpoint_start ||
        qi->getOperation() == queue_op_checkpoint_end) {
        checkpointManager->notifyFlusher();
    }

    return rv;
}

size_t Checkpoint::mergePrevCheckpoint(Checkpoint *pPrevCheckpoint) {
    size_t numNewItems = 0;
    size_t newEntryMemOverhead = 0;
    std::list<queued_item>::reverse_iterator rit = pPrevCheckpoint->rbegin();

    LOG(EXTENSION_LOG_INFO,
        "Collapse the checkpoint %" PRIu64 " into the checkpoint %" PRIu64
        " for vbucket %d",
        pPrevCheckpoint->getId(), checkpointId, vbucketId);

    std::list<queued_item>::iterator itr = toWrite.begin();
    uint64_t seqno = pPrevCheckpoint->getMutationIdForKey("dummy_key", true);
    metaKeyIndex["dummy_key"].mutation_id = seqno;
    (*itr)->setBySeqno(seqno);

    seqno = pPrevCheckpoint->getMutationIdForKey("checkpoint_start", true);
    metaKeyIndex["checkpoint_start"].mutation_id = seqno;
    ++itr;
    (*itr)->setBySeqno(seqno);

    for (; rit != pPrevCheckpoint->rend(); ++rit) {
        const std::string &key = (*rit)->getKey();
        if ((*rit)->getOperation() != queue_op_del &&
            (*rit)->getOperation() != queue_op_set) {
            continue;
        }
        checkpoint_index::iterator it = keyIndex.find(key);
        if (it == keyIndex.end()) {
            std::list<queued_item>::iterator pos = toWrite.begin();
            // Skip the first two meta items
            ++pos; ++pos;
            toWrite.insert(pos, *rit);
            index_entry entry = {--pos, static_cast<int64_t>(pPrevCheckpoint->
                                            getMutationIdForKey(key, false))};
            keyIndex[key] = entry;
            newEntryMemOverhead += key.size() + sizeof(index_entry);
            ++numItems;
            ++numNewItems;

            // Update new checkpoint's memory usage
            incrementMemConsumption((*rit)->size());
        }
    }

    /**
     * Update snapshot start of current checkpoint to the first
     * item's sequence number, after merge completed, as items
     * from the previous checkpoint will be inserted into this
     * checkpoint.
     */
    setSnapshotStartSeqno(getLowSeqno());

    memOverhead += newEntryMemOverhead;
    stats.memOverhead.fetch_add(newEntryMemOverhead);
    LOG(EXTENSION_LOG_WARNING,
        "Checkpoint::mergePrevCheckpoint: stats.memOverhead (which is %" PRId64
        ") is greater than %" PRId64, uint64_t(stats.memOverhead.load()),
        uint64_t(GIGANTOR));
    return numNewItems;
}

uint64_t Checkpoint::getMutationIdForKey(const std::string &key, bool isMeta)
{
    uint64_t mid = 0;
    checkpoint_index& chkIdx = isMeta ? metaKeyIndex : keyIndex;

    checkpoint_index::iterator it = chkIdx.find(key);
    if (it != chkIdx.end()) {
        mid = it->second.mutation_id;
    } else {
        LOG(EXTENSION_LOG_WARNING, "%s not found in chk index", key.c_str());
    }
    return mid;
}

bool Checkpoint::isEligibleToBeUnreferenced() {
    const std::set<std::string> &cursors = getCursorNameList();
    std::set<std::string>::const_iterator cit = cursors.begin();
    for (; cit != cursors.end(); ++cit) {
        if ((*cit).compare(CheckpointManager::pCursorName) == 0) {
            // Persistence cursor is on current checkpoint
            return false;
        }
    }
    return true;
}

std::ostream& operator <<(std::ostream& os, const Checkpoint& c) {
    os << "Checkpoint[" << &c << "] with "
       << "seqno:{" << c.getLowSeqno() << "," << c.getHighSeqno() << "} "
       << "state:" << c.getState();
    return os;
}

CheckpointManager::~CheckpointManager() {
    LockHolder lh(queueLock);
    std::list<Checkpoint*>::iterator it = checkpointList.begin();
    while(it != checkpointList.end()) {
        delete *it;
        ++it;
    }
}

uint64_t CheckpointManager::getOpenCheckpointId_UNLOCKED() {
    if (checkpointList.empty()) {
        return 0;
    }

    uint64_t id = checkpointList.back()->getId();
    return checkpointList.back()->getState() == CHECKPOINT_OPEN ? id : id + 1;
}

uint64_t CheckpointManager::getOpenCheckpointId() {
    LockHolder lh(queueLock);
    return getOpenCheckpointId_UNLOCKED();
}

uint64_t CheckpointManager::getLastClosedCheckpointId_UNLOCKED() {
    if (!isCollapsedCheckpoint) {<commit_msg>MB-20351: Fix lock-order inversion in ~CheckpointManager<commit_after>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2011 Couchbase, Inc
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#include "config.h"

#include <platform/checked_snprintf.h>
#include <string>
#include <utility>
#include <vector>

#include "checkpoint.h"
#include "ep_engine.h"
#define STATWRITER_NAMESPACE checkpoint
#include "statwriter.h"
#undef STATWRITER_NAMESPACE
#include "vbucket.h"

const std::string CheckpointManager::pCursorName("persistence");

/**
 * A listener class to update checkpoint related configs at runtime.
 */
class CheckpointConfigChangeListener : public ValueChangedListener {
public:
    CheckpointConfigChangeListener(CheckpointConfig &c) : config(c) { }
    virtual ~CheckpointConfigChangeListener() { }

    virtual void sizeValueChanged(const std::string &key, size_t value) {
        if (key.compare("chk_period") == 0) {
            config.setCheckpointPeriod(value);
        } else if (key.compare("chk_max_items") == 0) {
            config.setCheckpointMaxItems(value);
        } else if (key.compare("max_checkpoints") == 0) {
            config.setMaxCheckpoints(value);
        }
    }

    virtual void booleanValueChanged(const std::string &key, bool value) {
        if (key.compare("item_num_based_new_chk") == 0) {
            config.allowItemNumBasedNewCheckpoint(value);
        } else if (key.compare("keep_closed_chks") == 0) {
            config.allowKeepClosedCheckpoints(value);
        } else if (key.compare("enable_chk_merge") == 0) {
            config.allowCheckpointMerge(value);
        }
    }

private:
    CheckpointConfig &config;
};

void CheckpointCursor::decrOffset(size_t decr) {
    if (offset >= decr) {
        offset.fetch_sub(decr);
    } else {
        offset = 0;
        LOG(EXTENSION_LOG_INFO, "%s cursor offset is negative. Reset it to 0.",
            name.c_str());
    }
}

void CheckpointCursor::decrPos() {
    if (currentPos != (*currentCheckpoint)->begin()) {
        --currentPos;
    }
}

MustSendCheckpointEnd CheckpointCursor::shouldSendCheckpointEndMetaItem() const
{
    return sendCheckpointEndMetaItem;
}

Checkpoint::~Checkpoint() {
    LOG(EXTENSION_LOG_INFO,
        "Checkpoint %" PRIu64 " for vbucket %d is purged from memory",
        checkpointId, vbucketId);
    stats.memOverhead.fetch_sub(memorySize());
    if (stats.memOverhead.load() >= GIGANTOR) {
        LOG(EXTENSION_LOG_WARNING,
            "Checkpoint::~Checkpoint: stats.memOverhead (which is %" PRId64
            ") is greater than %" PRId64, uint64_t(stats.memOverhead.load()),
            uint64_t(GIGANTOR));
    }
}

void Checkpoint::setState(checkpoint_state state) {
    checkpointState = state;
}

void Checkpoint::popBackCheckpointEndItem() {
    if (!toWrite.empty() &&
        toWrite.back()->getOperation() == queue_op_checkpoint_end) {
        metaKeyIndex.erase(toWrite.back()->getKey());
        toWrite.pop_back();
    }
}

bool Checkpoint::keyExists(const std::string &key) {
    return keyIndex.find(key) != keyIndex.end();
}

queue_dirty_t Checkpoint::queueDirty(const queued_item &qi,
                                     CheckpointManager *checkpointManager) {
    if (checkpointState != CHECKPOINT_OPEN) {
        throw std::logic_error("Checkpoint::queueDirty: checkpointState "
                        "(which is" + std::to_string(checkpointState) +
                        ") is not OPEN");
    }
    queue_dirty_t rv;

    checkpoint_index::iterator it = keyIndex.find(qi->getKey());

    // Check if the item is a meta item
    if (qi->isCheckPointMetaItem()) {
        rv = NEW_ITEM;
        toWrite.push_back(qi);
    } else {
        // Check if this checkpoint already had an item for the same key
        if (it != keyIndex.end()) {
            rv = EXISTING_ITEM;
            std::list<queued_item>::iterator currPos = it->second.position;
            uint64_t currMutationId = it->second.mutation_id;

            cursor_index::iterator map_it =
                                        checkpointManager->connCursors.begin();
            for (; map_it != checkpointManager->connCursors.end(); ++map_it) {

                if (*(map_it->second.currentCheckpoint) == this) {
                    queued_item &tqi = *(map_it->second.currentPos);
                    const std::string &key = tqi->getKey();
                    checkpoint_index::iterator ita = keyIndex.find(key);
                    if (ita != keyIndex.end() && (!tqi->isCheckPointMetaItem()))
                    {
                        uint64_t mutationId = ita->second.mutation_id;
                        if (currMutationId <= mutationId) {
                            map_it->second.decrOffset(1);
                            if (map_it->second.name.compare(CheckpointManager::pCursorName)
                                == 0) {
                                rv = PERSIST_AGAIN;
                            }
                        }
                    }
                    /* If an TAP cursor points to the existing item for the same
                       key, shift it left by 1 */
                    if (map_it->second.currentPos == currPos) {
                        map_it->second.decrPos();
                    }
                }
            }

            toWrite.push_back(qi);
            // Remove the existing item for the same key from the list.
            toWrite.erase(currPos);
        } else {
            ++numItems;
            rv = NEW_ITEM;
            // Push the new item into the list
            toWrite.push_back(qi);
        }
    }

    if (qi->getNKey() > 0) {
        std::list<queued_item>::iterator last = toWrite.end();
        // --last is okay as the list is not empty now.
        index_entry entry = {--last, qi->getBySeqno()};
        // Set the index of the key to the new item that is pushed back into
        // the list.
        if (qi->isCheckPointMetaItem()) {
            // We add a meta item only once to a checkpoint
            metaKeyIndex[qi->getKey()] = entry;
        } else {
            keyIndex[qi->getKey()] = entry;
        }
        if (rv == NEW_ITEM) {
            size_t newEntrySize = qi->getNKey() + sizeof(index_entry) +
                                  sizeof(queued_item);
            memOverhead += newEntrySize;
            stats.memOverhead.fetch_add(newEntrySize);
            if (stats.memOverhead.load() >= GIGANTOR) {
                LOG(EXTENSION_LOG_WARNING,
                    "Checkpoint::queueDirty: stats.memOverhead (which is %" PRId64
                    ") is greater than %" PRId64, uint64_t(stats.memOverhead.load()),
                    uint64_t(GIGANTOR));
            }
        }
    }

    // Notify flusher if in case queued item is a checkpoint meta item
    if (qi->getOperation() == queue_op_checkpoint_start ||
        qi->getOperation() == queue_op_checkpoint_end) {
        checkpointManager->notifyFlusher();
    }

    return rv;
}

size_t Checkpoint::mergePrevCheckpoint(Checkpoint *pPrevCheckpoint) {
    size_t numNewItems = 0;
    size_t newEntryMemOverhead = 0;
    std::list<queued_item>::reverse_iterator rit = pPrevCheckpoint->rbegin();

    LOG(EXTENSION_LOG_INFO,
        "Collapse the checkpoint %" PRIu64 " into the checkpoint %" PRIu64
        " for vbucket %d",
        pPrevCheckpoint->getId(), checkpointId, vbucketId);

    std::list<queued_item>::iterator itr = toWrite.begin();
    uint64_t seqno = pPrevCheckpoint->getMutationIdForKey("dummy_key", true);
    metaKeyIndex["dummy_key"].mutation_id = seqno;
    (*itr)->setBySeqno(seqno);

    seqno = pPrevCheckpoint->getMutationIdForKey("checkpoint_start", true);
    metaKeyIndex["checkpoint_start"].mutation_id = seqno;
    ++itr;
    (*itr)->setBySeqno(seqno);

    for (; rit != pPrevCheckpoint->rend(); ++rit) {
        const std::string &key = (*rit)->getKey();
        if ((*rit)->getOperation() != queue_op_del &&
            (*rit)->getOperation() != queue_op_set) {
            continue;
        }
        checkpoint_index::iterator it = keyIndex.find(key);
        if (it == keyIndex.end()) {
            std::list<queued_item>::iterator pos = toWrite.begin();
            // Skip the first two meta items
            ++pos; ++pos;
            toWrite.insert(pos, *rit);
            index_entry entry = {--pos, static_cast<int64_t>(pPrevCheckpoint->
                                            getMutationIdForKey(key, false))};
            keyIndex[key] = entry;
            newEntryMemOverhead += key.size() + sizeof(index_entry);
            ++numItems;
            ++numNewItems;

            // Update new checkpoint's memory usage
            incrementMemConsumption((*rit)->size());
        }
    }

    /**
     * Update snapshot start of current checkpoint to the first
     * item's sequence number, after merge completed, as items
     * from the previous checkpoint will be inserted into this
     * checkpoint.
     */
    setSnapshotStartSeqno(getLowSeqno());

    memOverhead += newEntryMemOverhead;
    stats.memOverhead.fetch_add(newEntryMemOverhead);
    LOG(EXTENSION_LOG_WARNING,
        "Checkpoint::mergePrevCheckpoint: stats.memOverhead (which is %" PRId64
        ") is greater than %" PRId64, uint64_t(stats.memOverhead.load()),
        uint64_t(GIGANTOR));
    return numNewItems;
}

uint64_t Checkpoint::getMutationIdForKey(const std::string &key, bool isMeta)
{
    uint64_t mid = 0;
    checkpoint_index& chkIdx = isMeta ? metaKeyIndex : keyIndex;

    checkpoint_index::iterator it = chkIdx.find(key);
    if (it != chkIdx.end()) {
        mid = it->second.mutation_id;
    } else {
        LOG(EXTENSION_LOG_WARNING, "%s not found in chk index", key.c_str());
    }
    return mid;
}

bool Checkpoint::isEligibleToBeUnreferenced() {
    const std::set<std::string> &cursors = getCursorNameList();
    std::set<std::string>::const_iterator cit = cursors.begin();
    for (; cit != cursors.end(); ++cit) {
        if ((*cit).compare(CheckpointManager::pCursorName) == 0) {
            // Persistence cursor is on current checkpoint
            return false;
        }
    }
    return true;
}

std::ostream& operator <<(std::ostream& os, const Checkpoint& c) {
    os << "Checkpoint[" << &c << "] with "
       << "seqno:{" << c.getLowSeqno() << "," << c.getHighSeqno() << "} "
       << "state:" << c.getState();
    return os;
}

CheckpointManager::~CheckpointManager() {
    std::list<Checkpoint*>::iterator it = checkpointList.begin();
    while(it != checkpointList.end()) {
        delete *it;
        ++it;
    }
}

uint64_t CheckpointManager::getOpenCheckpointId_UNLOCKED() {
    if (checkpointList.empty()) {
        return 0;
    }

    uint64_t id = checkpointList.back()->getId();
    return checkpointList.back()->getState() == CHECKPOINT_OPEN ? id : id + 1;
}

uint64_t CheckpointManager::getOpenCheckpointId() {
    LockHolder lh(queueLock);
    return getOpenCheckpointId_UNLOCKED();
}

uint64_t CheckpointManager::getLastClosedCheckpointId_UNLOCKED() {
    if (!isCollapsedCheckpoint) {
{'fraction_non_alphanumeric': 0.12180451127819548, 'fraction_numerical': 0.0069172932330827065, 'mean_word_length': 2.376269219611256, 'pattern_counts': {'":': 0, '<': 75, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7012393', 'n_tokens_mistral': 7079, 'n_tokens_neox': 6856, 'n_words': 1523}