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

#include "storagebucketdbinitializer.h"
#include "lockablemap.hpp"
#include "config-stor-bucket-init.h"
#include "storbucketdb.h"
#include <vespa/storage/common/nodestateupdater.h>
#include <vespa/storage/common/content_bucket_space_repo.h>
#include <vespa/storage/storageserver/storagemetricsset.h>
#include <vespa/vdslib/distribution/distribution.h>
#include <vespa/vespalib/io/fileutil.h>
#include <vespa/config/config.h>
#include <vespa/vespalib/stllike/hash_map.hpp>
#include <vespa/config/helper/configgetter.hpp>
#include <iomanip>

#include <vespa/log/bufferedlogger.h>
LOG_SETUP(".storage.bucketdb.initializer");

using document::BucketSpace;

namespace storage {

using BucketSet = vespalib::hash_set<document::BucketId, document::BucketId::hash>;

struct BucketReadState {
    using UP = std::unique_ptr<BucketReadState>;

    BucketSet _pending;
    document::BucketId _databaseIterator;
    bool _done;

    BucketReadState() : _done(false) {}
};

using vespa::config::content::core::StorBucketInitConfig;

StorageBucketDBInitializer::Config::Config(const config::ConfigUri & configUri)
    : _listPriority(0),
      _infoReadPriority(255),
      _minPendingInfoReadsPerDisk(16),
      _maxPendingInfoReadsPerDisk(32)
{
    auto config = config::ConfigGetter<StorBucketInitConfig>::getConfig(configUri.getConfigId(),
                                                                        configUri.getContext());
    _maxPendingInfoReadsPerDisk = config->maxPendingInfoReadsPerDisk;
    _minPendingInfoReadsPerDisk = config->minPendingInfoReadsPerDisk;
    _infoReadPriority = config->infoReadPriority;
    _listPriority = config->listPriority;
    if (config->completeListBeforeStartingRead) {
        LOG(warning, "This config option is currently not honored. Info "
                     "reading will always start on a directory as soon as "
                     "it is done listing.");
    }
    LOG(debug, "Initializing bucket database: List priority %u, info priority "
               "%u, min/max pending info per disk %u/%u.",
        _listPriority, _infoReadPriority,
        _minPendingInfoReadsPerDisk, _maxPendingInfoReadsPerDisk);
}

StorageBucketDBInitializer::System::System(
        const spi::PartitionStateList& partitions,
        DoneInitializeHandler& doneInitializeHandler,
        ServiceLayerComponentRegister& compReg,
        const Config&)
    : _doneInitializeHandler(doneInitializeHandler),
      _component(compReg, "storagebucketdbinitializer"),
      _partitions(partitions),
      _bucketSpaceRepo(_component.getBucketSpaceRepo()),
      _nodeIndex(_component.getIndex()),
      _distribution(*_component.getDistribution()),
      _nodeState()
{
    // Is this correct? We should get the node state from the node state updater
    // so it could work with disk capacities. Object is used to check for
    // correct disk further down (in the case of internal join, deciding which
    // should have it). Not that bad if wrong disk is picked though.
    _nodeState.setDiskCount(_partitions.size());
    for (uint32_t i=0; i<_partitions.size(); ++i) {
        if (!_partitions[i].isUp()) {
            _nodeState.setDiskState(i, lib::State::DOWN);
        }
    }
}

StorBucketDatabase &
StorageBucketDBInitializer::System::getBucketDatabase(document::BucketSpace bucketSpace) const
{
    return _component.getBucketDatabase(bucketSpace);
}

StorageBucketDBInitializer::Metrics::Metrics(framework::Component& component)
    : metrics::MetricSet("dbinit", "",
                         "Metrics for the storage bucket database initializer"),
      _wrongDisk("wrongdisk", "",
              "Number of buckets found on non-ideal disk.", this),
      _insertedCount("insertedcount", "",
              "Number of buckets inserted into database in list step.", this),
      _joinedCount("joinedcount", "",
              "Number of buckets found in list step already found "
              "(added from other disks).", this),
      _infoReadCount("infocount", "",
              "Number of buckets we have read bucket information from.", this),
      _infoSetByLoad("infosetbyload", "",
              "Number of buckets we did not need to request bucket info for "
              "due to load already having updated them.", this),
      _dirsListed("dirslisted", "",
              "Directories listed in list step of initialization.", this),
      _startTime(component.getClock()),
      _listLatency("listlatency", "",
              "Time used until list phase is done. (in ms)", this),
      _initLatency("initlatency", "",
              "Time used until initialization is complete. (in ms)", this)
{
    component.registerMetric(*this);
}

StorageBucketDBInitializer::Metrics::~Metrics() {}

StorageBucketDBInitializer::GlobalState::GlobalState()
        : _insertedCount(0), _infoReadCount(0),
          _infoSetByLoad(0), _dirsListed(0), _dirsToList(0),
          _gottenInitProgress(false), _doneListing(false),
          _doneInitializing(false)
{ }
StorageBucketDBInitializer::GlobalState::~GlobalState() { }

StorageBucketDBInitializer::StorageBucketDBInitializer(
        const config::ConfigUri & configUri,
        const spi::PartitionStateList& partitions,
        DoneInitializeHandler& doneInitializeHandler,
        ServiceLayerComponentRegister& compReg)
    : StorageLink("StorageBucketDBInitializer"),
      framework::HtmlStatusReporter("dbinit", "Bucket database initializer"),
      _config(configUri),
      _system(partitions, doneInitializeHandler, compReg, _config),
      _metrics(_system._component),
      _state(),
      _readState(_system._partitions.size())
{
        // Initialize read state for disks being available
    for (uint32_t i=0; i<_system._partitions.size(); ++i) {
        if (!_system._partitions[i].isUp()) continue;
        _readState[i] = std::make_unique<BucketSpaceReadState>();
        for (const auto &elem : _system._bucketSpaceRepo) {
            _readState[i]->insert(std::make_pair(elem.first, std::make_unique<BucketReadState>()));
        }
        _state._dirsToList += 1;
    }
    _system._component.registerStatusPage(*this);
}

StorageBucketDBInitializer::~StorageBucketDBInitializer()
{<commit_msg>Use emplace instead of insert.<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "storagebucketdbinitializer.h"
#include "lockablemap.hpp"
#include "config-stor-bucket-init.h"
#include "storbucketdb.h"
#include <vespa/storage/common/nodestateupdater.h>
#include <vespa/storage/common/content_bucket_space_repo.h>
#include <vespa/storage/storageserver/storagemetricsset.h>
#include <vespa/vdslib/distribution/distribution.h>
#include <vespa/vespalib/io/fileutil.h>
#include <vespa/config/config.h>
#include <vespa/vespalib/stllike/hash_map.hpp>
#include <vespa/config/helper/configgetter.hpp>
#include <iomanip>

#include <vespa/log/bufferedlogger.h>
LOG_SETUP(".storage.bucketdb.initializer");

using document::BucketSpace;

namespace storage {

using BucketSet = vespalib::hash_set<document::BucketId, document::BucketId::hash>;

struct BucketReadState {
    using UP = std::unique_ptr<BucketReadState>;

    BucketSet _pending;
    document::BucketId _databaseIterator;
    bool _done;

    BucketReadState() : _done(false) {}
};

using vespa::config::content::core::StorBucketInitConfig;

StorageBucketDBInitializer::Config::Config(const config::ConfigUri & configUri)
    : _listPriority(0),
      _infoReadPriority(255),
      _minPendingInfoReadsPerDisk(16),
      _maxPendingInfoReadsPerDisk(32)
{
    auto config = config::ConfigGetter<StorBucketInitConfig>::getConfig(configUri.getConfigId(),
                                                                        configUri.getContext());
    _maxPendingInfoReadsPerDisk = config->maxPendingInfoReadsPerDisk;
    _minPendingInfoReadsPerDisk = config->minPendingInfoReadsPerDisk;
    _infoReadPriority = config->infoReadPriority;
    _listPriority = config->listPriority;
    if (config->completeListBeforeStartingRead) {
        LOG(warning, "This config option is currently not honored. Info "
                     "reading will always start on a directory as soon as "
                     "it is done listing.");
    }
    LOG(debug, "Initializing bucket database: List priority %u, info priority "
               "%u, min/max pending info per disk %u/%u.",
        _listPriority, _infoReadPriority,
        _minPendingInfoReadsPerDisk, _maxPendingInfoReadsPerDisk);
}

StorageBucketDBInitializer::System::System(
        const spi::PartitionStateList& partitions,
        DoneInitializeHandler& doneInitializeHandler,
        ServiceLayerComponentRegister& compReg,
        const Config&)
    : _doneInitializeHandler(doneInitializeHandler),
      _component(compReg, "storagebucketdbinitializer"),
      _partitions(partitions),
      _bucketSpaceRepo(_component.getBucketSpaceRepo()),
      _nodeIndex(_component.getIndex()),
      _distribution(*_component.getDistribution()),
      _nodeState()
{
    // Is this correct? We should get the node state from the node state updater
    // so it could work with disk capacities. Object is used to check for
    // correct disk further down (in the case of internal join, deciding which
    // should have it). Not that bad if wrong disk is picked though.
    _nodeState.setDiskCount(_partitions.size());
    for (uint32_t i=0; i<_partitions.size(); ++i) {
        if (!_partitions[i].isUp()) {
            _nodeState.setDiskState(i, lib::State::DOWN);
        }
    }
}

StorBucketDatabase &
StorageBucketDBInitializer::System::getBucketDatabase(document::BucketSpace bucketSpace) const
{
    return _component.getBucketDatabase(bucketSpace);
}

StorageBucketDBInitializer::Metrics::Metrics(framework::Component& component)
    : metrics::MetricSet("dbinit", "",
                         "Metrics for the storage bucket database initializer"),
      _wrongDisk("wrongdisk", "",
              "Number of buckets found on non-ideal disk.", this),
      _insertedCount("insertedcount", "",
              "Number of buckets inserted into database in list step.", this),
      _joinedCount("joinedcount", "",
              "Number of buckets found in list step already found "
              "(added from other disks).", this),
      _infoReadCount("infocount", "",
              "Number of buckets we have read bucket information from.", this),
      _infoSetByLoad("infosetbyload", "",
              "Number of buckets we did not need to request bucket info for "
              "due to load already having updated them.", this),
      _dirsListed("dirslisted", "",
              "Directories listed in list step of initialization.", this),
      _startTime(component.getClock()),
      _listLatency("listlatency", "",
              "Time used until list phase is done. (in ms)", this),
      _initLatency("initlatency", "",
              "Time used until initialization is complete. (in ms)", this)
{
    component.registerMetric(*this);
}

StorageBucketDBInitializer::Metrics::~Metrics() {}

StorageBucketDBInitializer::GlobalState::GlobalState()
        : _insertedCount(0), _infoReadCount(0),
          _infoSetByLoad(0), _dirsListed(0), _dirsToList(0),
          _gottenInitProgress(false), _doneListing(false),
          _doneInitializing(false)
{ }
StorageBucketDBInitializer::GlobalState::~GlobalState() { }

StorageBucketDBInitializer::StorageBucketDBInitializer(
        const config::ConfigUri & configUri,
        const spi::PartitionStateList& partitions,
        DoneInitializeHandler& doneInitializeHandler,
        ServiceLayerComponentRegister& compReg)
    : StorageLink("StorageBucketDBInitializer"),
      framework::HtmlStatusReporter("dbinit", "Bucket database initializer"),
      _config(configUri),
      _system(partitions, doneInitializeHandler, compReg, _config),
      _metrics(_system._component),
      _state(),
      _readState(_system._partitions.size())
{
        // Initialize read state for disks being available
    for (uint32_t i=0; i<_system._partitions.size(); ++i) {
        if (!_system._partitions[i].isUp()) continue;
        _readState[i] = std::make_unique<BucketSpaceReadState>();
        for (const auto &elem : _system._bucketSpaceRepo) {
            _readState[i]->emplace(elem.first, std::make_unique<BucketReadState>());
        }
        _state._dirsToList += 1;
    }
    _system._component.registerStatusPage(*this);
}

StorageBucketDBInitializer::~StorageBucketDBInitializer()
{
{'fraction_non_alphanumeric': 0.109629044988161, 'fraction_numerical': 0.004104183109707972, 'mean_word_length': 3.6721976401179943, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 45, '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': '2496776', 'n_tokens_mistral': 3567, 'n_tokens_neox': 3522, 'n_words': 790}