<commit_before>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "documentbucketmover.h"
#include "i_move_operation_limiter.h"
#include "idocumentmovehandler.h"
#include "maintenancedocumentsubdb.h"
#include <vespa/searchcore/proton/documentmetastore/i_document_meta_store.h>
#include <vespa/searchcore/proton/feedoperation/moveoperation.h>
#include <vespa/searchcore/proton/persistenceengine/i_document_retriever.h>
#include <vespa/searchcore/proton/bucketdb/bucket_db_owner.h>
#include <vespa/document/fieldvalue/document.h>
#include <vespa/vespalib/util/destructor_callbacks.h>

using document::BucketId;
using document::Document;
using document::GlobalId;
using storage::spi::Timestamp;

namespace proton::bucketdb {

typedef IDocumentMetaStore::Iterator Iterator;

MoveOperation::UP
BucketMover::createMoveOperation(const MoveKey &key) {
    if (_source->lidNeedsCommit(key._lid)) return {};

    const RawDocumentMetaData &metaNow = _source->meta_store()->getRawMetaData(key._lid);
    if (metaNow.getGid() != key._gid) return {};
    if (metaNow.getTimestamp() != key._timestamp) return {};

    Document::SP doc(_source->retriever()->getFullDocument(key._lid));
    if (!doc || doc->getId().getGlobalId() != key._gid) {
        // Failed to retrieve document, removed or changed identity
        return {};
    }
    BucketId bucketId = _bucket.stripUnused();
    return std::make_unique<MoveOperation>(bucketId, key._timestamp, std::move(doc),
                                           DbDocumentId(_source->sub_db_id(), key._lid), _targetSubDbId);
}

void
BucketMover::moveDocument(MoveOperationUP moveOp, IDestructorCallbackSP onDone) {
    _handler->handleMove(*moveOp, std::move(onDone));
}

BucketMover::MoveKey::MoveKey(uint32_t lid, const document::GlobalId &gid, Timestamp timestamp, MoveGuard guard) noexcept
    : _lid(lid),
      _gid(gid),
      _timestamp(timestamp),
      _guard(std::move(guard))
{ }

BucketMover::MoveKey::~MoveKey() = default;

BucketMover::MoveKeys::~MoveKeys() = default;

std::shared_ptr<BucketMover>
BucketMover::MoveKeys::stealMover() {
    return std::move(_mover);
}

BucketMover::GuardedMoveOps
BucketMover::MoveKeys::createMoveOperations() {
    auto & mover = *_mover;
    return mover.createMoveOperations(std::move(*this));
}

BucketMover::GuardedMoveOps::GuardedMoveOps(std::shared_ptr<BucketMover> mover) noexcept
    : _mover(std::move(mover)),
      _success(),
      _failed()
{}
BucketMover::GuardedMoveOps::~GuardedMoveOps() = default;

BucketMover::BucketMover(const BucketId &bucket, const MaintenanceDocumentSubDB *source,
                         uint32_t targetSubDbId, IDocumentMoveHandler &handler) noexcept
    : _source(source),
      _handler(&handler),
      _bucket(bucket),
      _targetSubDbId(targetSubDbId),
      _started(0),
      _completed(0),
      _needReschedule(false),
      _allScheduled(false),
      _lastGidValid(false),
      _lastGid()
{ }

BucketMover::~BucketMover() {
    assert(inSync());
}

std::pair<BucketMover::MoveKeys, bool>
BucketMover::getKeysToMove(size_t maxDocsToMove) {
    std::pair<MoveKeys, bool> result(MoveKeys(shared_from_this()), false);
    Iterator itr = (_lastGidValid ? _source->meta_store()->upperBound(_lastGid)
                                  : _source->meta_store()->lowerBound(_bucket));
    const Iterator end = _source->meta_store()->upperBound(_bucket);
    std::vector<MoveKey> toMove;
    for (size_t docsMoved(0); itr != end && docsMoved < maxDocsToMove; ++itr) {
        uint32_t lid = itr.getKey().get_lid();
        const RawDocumentMetaData &metaData = _source->meta_store()->getRawMetaData(lid);
        if (metaData.getBucketUsedBits() == _bucket.getUsedBits()) {
            result.first.keys().emplace_back(lid, metaData.getGid(), metaData.getTimestamp(), MoveGuard(*this));
            ++docsMoved;
        }
    }
    result.second = (itr == end);
    return result;
}

BucketMover::GuardedMoveOps
BucketMover::createMoveOperations(MoveKeys toMove) {
    GuardedMoveOps moveOps(toMove.stealMover());
    moveOps.success().reserve(toMove.size());
    for (MoveKey &key : toMove.keys()) {
        if (moveOps.failed().empty()) {
            auto moveOp = createMoveOperation(key);
            if (moveOp) {
                moveOps.success().emplace_back(std::move(moveOp), std::move(key._guard));
            } else {
                moveOps.failed().push_back(std::move(key._guard));
            }
        } else {
            moveOps.failed().push_back(std::move(key._guard));
        }
    }
    if ( ! moveOps.failed().empty()) {
        _needReschedule.store(true, std::memory_order_relaxed);
    }
    return moveOps;
}

void
BucketMover::moveDocuments(std::vector<GuardedMoveOp> moveOps, IDestructorCallbackSP onDone) {
    for (auto & moveOp : moveOps) {
        moveDocument(std::move(moveOp.first), std::move(onDone));
    }
}

void
BucketMover::cancel() {
    setAllScheduled();
    _needReschedule.store(true, std::memory_order_relaxed);
}

}

namespace proton {

using bucketdb::BucketMover;
using bucketdb::BucketDBOwner;

DocumentBucketMover::DocumentBucketMover(IMoveOperationLimiter &limiter, BucketDBOwner &bucketDb) noexcept
    : _limiter(limiter),<commit_msg>Copy onDone token.<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "documentbucketmover.h"
#include "i_move_operation_limiter.h"
#include "idocumentmovehandler.h"
#include "maintenancedocumentsubdb.h"
#include <vespa/searchcore/proton/documentmetastore/i_document_meta_store.h>
#include <vespa/searchcore/proton/feedoperation/moveoperation.h>
#include <vespa/searchcore/proton/persistenceengine/i_document_retriever.h>
#include <vespa/searchcore/proton/bucketdb/bucket_db_owner.h>
#include <vespa/document/fieldvalue/document.h>
#include <vespa/vespalib/util/destructor_callbacks.h>

using document::BucketId;
using document::Document;
using document::GlobalId;
using storage::spi::Timestamp;

namespace proton::bucketdb {

typedef IDocumentMetaStore::Iterator Iterator;

MoveOperation::UP
BucketMover::createMoveOperation(const MoveKey &key) {
    if (_source->lidNeedsCommit(key._lid)) return {};

    const RawDocumentMetaData &metaNow = _source->meta_store()->getRawMetaData(key._lid);
    if (metaNow.getGid() != key._gid) return {};
    if (metaNow.getTimestamp() != key._timestamp) return {};

    Document::SP doc(_source->retriever()->getFullDocument(key._lid));
    if (!doc || doc->getId().getGlobalId() != key._gid) {
        // Failed to retrieve document, removed or changed identity
        return {};
    }
    BucketId bucketId = _bucket.stripUnused();
    return std::make_unique<MoveOperation>(bucketId, key._timestamp, std::move(doc),
                                           DbDocumentId(_source->sub_db_id(), key._lid), _targetSubDbId);
}

void
BucketMover::moveDocument(MoveOperationUP moveOp, IDestructorCallbackSP onDone) {
    _handler->handleMove(*moveOp, std::move(onDone));
}

BucketMover::MoveKey::MoveKey(uint32_t lid, const document::GlobalId &gid, Timestamp timestamp, MoveGuard guard) noexcept
    : _lid(lid),
      _gid(gid),
      _timestamp(timestamp),
      _guard(std::move(guard))
{ }

BucketMover::MoveKey::~MoveKey() = default;

BucketMover::MoveKeys::~MoveKeys() = default;

std::shared_ptr<BucketMover>
BucketMover::MoveKeys::stealMover() {
    return std::move(_mover);
}

BucketMover::GuardedMoveOps
BucketMover::MoveKeys::createMoveOperations() {
    auto & mover = *_mover;
    return mover.createMoveOperations(std::move(*this));
}

BucketMover::GuardedMoveOps::GuardedMoveOps(std::shared_ptr<BucketMover> mover) noexcept
    : _mover(std::move(mover)),
      _success(),
      _failed()
{}
BucketMover::GuardedMoveOps::~GuardedMoveOps() = default;

BucketMover::BucketMover(const BucketId &bucket, const MaintenanceDocumentSubDB *source,
                         uint32_t targetSubDbId, IDocumentMoveHandler &handler) noexcept
    : _source(source),
      _handler(&handler),
      _bucket(bucket),
      _targetSubDbId(targetSubDbId),
      _started(0),
      _completed(0),
      _needReschedule(false),
      _allScheduled(false),
      _lastGidValid(false),
      _lastGid()
{ }

BucketMover::~BucketMover() {
    assert(inSync());
}

std::pair<BucketMover::MoveKeys, bool>
BucketMover::getKeysToMove(size_t maxDocsToMove) {
    std::pair<MoveKeys, bool> result(MoveKeys(shared_from_this()), false);
    Iterator itr = (_lastGidValid ? _source->meta_store()->upperBound(_lastGid)
                                  : _source->meta_store()->lowerBound(_bucket));
    const Iterator end = _source->meta_store()->upperBound(_bucket);
    std::vector<MoveKey> toMove;
    for (size_t docsMoved(0); itr != end && docsMoved < maxDocsToMove; ++itr) {
        uint32_t lid = itr.getKey().get_lid();
        const RawDocumentMetaData &metaData = _source->meta_store()->getRawMetaData(lid);
        if (metaData.getBucketUsedBits() == _bucket.getUsedBits()) {
            result.first.keys().emplace_back(lid, metaData.getGid(), metaData.getTimestamp(), MoveGuard(*this));
            ++docsMoved;
        }
    }
    result.second = (itr == end);
    return result;
}

BucketMover::GuardedMoveOps
BucketMover::createMoveOperations(MoveKeys toMove) {
    GuardedMoveOps moveOps(toMove.stealMover());
    moveOps.success().reserve(toMove.size());
    for (MoveKey &key : toMove.keys()) {
        if (moveOps.failed().empty()) {
            auto moveOp = createMoveOperation(key);
            if (moveOp) {
                moveOps.success().emplace_back(std::move(moveOp), std::move(key._guard));
            } else {
                moveOps.failed().push_back(std::move(key._guard));
            }
        } else {
            moveOps.failed().push_back(std::move(key._guard));
        }
    }
    if ( ! moveOps.failed().empty()) {
        _needReschedule.store(true, std::memory_order_relaxed);
    }
    return moveOps;
}

void
BucketMover::moveDocuments(std::vector<GuardedMoveOp> moveOps, IDestructorCallbackSP onDone) {
    for (auto & moveOp : moveOps) {
        moveDocument(std::move(moveOp.first), onDone);
    }
}

void
BucketMover::cancel() {
    setAllScheduled();
    _needReschedule.store(true, std::memory_order_relaxed);
}

}

namespace proton {

using bucketdb::BucketMover;
using bucketdb::BucketDBOwner;

DocumentBucketMover::DocumentBucketMover(IMoveOperationLimiter &limiter, BucketDBOwner &bucketDb) noexcept
    : _limiter(limiter),
{'fraction_non_alphanumeric': 0.136560966402416, 'fraction_numerical': 0.0028312570781426952, 'mean_word_length': 4.2201970443349754, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '830224', 'n_tokens_mistral': 3424, 'n_tokens_neox': 3379, 'n_words': 537}