<commit_before>// Copyright 2010-2014 RethinkDB, all rights reserved.
#include "clustering/administration/main/file_based_svs_by_namespace.hpp"

#include "errors.hpp"
#include <boost/bind.hpp>

#include "clustering/immediate_consistency/branch/multistore.hpp"
#include "clustering/reactor/reactor.hpp"
#include "logger.hpp"
#include "rdb_protocol/store.hpp"
#include "serializer/config.hpp"
#include "serializer/translator.hpp"
#include "serializer/merger.hpp"
#include "utils.hpp"

/* This object serves mostly as a container for arguments to the
 * do_construct_existing_store function because we hit the boost::bind argument
 * limit. */
struct store_args_t {
    store_args_t(io_backender_t *_io_backender,
                 const base_path_t &_base_path,
                 namespace_id_t _namespace_id,
                 cache_balancer_t *_balancer,
                 perfmon_collection_t *_serializers_perfmon_collection,
                 rdb_context_t *_ctx,
                 outdated_index_issue_tracker_t *_outdated_index_tracker,
                 namespace_id_t _ns_id)
        : io_backender(_io_backender), base_path(_base_path),
          namespace_id(_namespace_id), balancer(_balancer),
          serializers_perfmon_collection(_serializers_perfmon_collection),
          ctx(_ctx), outdated_index_tracker(_outdated_index_tracker), ns_id(_ns_id)
    { }

    io_backender_t *io_backender;
    base_path_t base_path;
    namespace_id_t namespace_id;
    cache_balancer_t *balancer;
    perfmon_collection_t *serializers_perfmon_collection;
    rdb_context_t *ctx;
    outdated_index_issue_tracker_t *outdated_index_tracker;
    namespace_id_t ns_id;
};

std::string hash_shard_perfmon_name(int hash_shard_number) {
    return strprintf("shard_%d", hash_shard_number);
}

void do_store(
    const std::vector<threadnum_t> &threads,
    int thread_offset,
    store_args_t store_args,
    serializer_multiplexer_t *multiplexer,
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores,
    store_view_t **store_views,
    bool create) {

    // TODO: Exceptions?  Can exceptions happen, and then this doesn't
    // catch it, and the caller doesn't handle it.
    on_thread_t th(threads[thread_offset]);

    // Only pass this down to the first store
    scoped_ptr_t<outdated_index_report_t> index_report;
    if (thread_offset == 0) {
        index_report = store_args.outdated_index_tracker->create_report(store_args.ns_id);
    }

    // TODO: Can we pass serializers_perfmon_collection across threads like this?
    store_t *store = new store_t(
        multiplexer->proxies[thread_offset], store_args.balancer,
        hash_shard_perfmon_name(thread_offset),
        create, store_args.serializers_perfmon_collection,
        store_args.ctx, store_args.io_backender, store_args.base_path,
        std::move(index_report), store_args.ns_id);
    (*stores_out_stores)[thread_offset].init(store);
    store_views[thread_offset] = store;
}

void do_construct_existing_store(
    const std::vector<threadnum_t> &threads,
    int thread_offset,
    store_args_t store_args,
    serializer_multiplexer_t *multiplexer,
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores,
    store_view_t **store_views) {

    do_store(threads, thread_offset, store_args, multiplexer, stores_out_stores,
    store_views, false);
}

void do_create_new_store(
    const std::vector<threadnum_t> &threads,
    int thread_offset,
    store_args_t store_args,
    serializer_multiplexer_t *multiplexer,
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores,
    store_view_t **store_views) {

    do_store(threads, thread_offset, store_args, multiplexer, stores_out_stores,
    store_views, true);
}

void
file_based_svs_by_namespace_t::get_svs(
            perfmon_collection_t *serializers_perfmon_collection,
            namespace_id_t namespace_id,
            stores_lifetimer_t *stores_out,
            scoped_ptr_t<multistore_ptr_t> *svs_out,
            rdb_context_t *ctx) {
    const int num_db_threads = get_num_db_threads();

    // TODO: If the server gets killed when starting up, we can
    // get a database in an invalid startup state.

    // TODO: This is quite suspicious in that we check if the file
    // exists and then assume it exists or does not exist when
    // loading or creating it.

    // TODO: We should use N slices on M serializers, not N slices
    // on N serializers.

    const int num_stores = CPU_SHARDING_FACTOR;
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores
        = stores_out->stores();
    stores_out_stores->init(num_stores);

    const threadnum_t serializer_thread = next_thread(num_db_threads);
    std::vector<threadnum_t> store_threads;
    for (int i = 0; i < num_stores; ++i) {
        store_threads.push_back(next_thread(num_db_threads));
    }

    scoped_ptr_t<serializer_t> serializer;
    scoped_ptr_t<serializer_multiplexer_t> multiplexer;
    scoped_ptr_t<multistore_ptr_t> mptr;
    {
        on_thread_t th(serializer_thread);
        scoped_array_t<store_view_t *> store_views(num_stores);

        const serializer_filepath_t serializer_filepath = file_name_for(namespace_id);
        int res = access(serializer_filepath.permanent_path().c_str(), R_OK | W_OK);
        store_args_t store_args(io_backender_, base_path_,
                                namespace_id, balancer_,
                                serializers_perfmon_collection, ctx,
                                &outdated_index_tracker, namespace_id);
        filepath_file_opener_t file_opener(serializer_filepath, io_backender_);
        if (res == 0) {
            // TODO: Could we handle failure when loading the serializer?  Right
            // now, we don't.

            {
                scoped_ptr_t<serializer_t> ser
                    = make_scoped<standard_serializer_t>(
                        standard_serializer_t::dynamic_config_t(),
                        &file_opener,
                        serializers_perfmon_collection);
                ser = make_scoped<merger_serializer_t>(std::move(ser),
                                                       MERGER_SERIALIZER_MAX_ACTIVE_WRITES);
                serializer = std::move(ser);
            }

            std::vector<serializer_t *> ptrs;
            ptrs.push_back(serializer.get());
            multiplexer.init(new serializer_multiplexer_t(ptrs));

            // TODO: Exceptions?  Can exceptions happen, and then
            // store_views' values would leak.  That is, are we handling
            // them in the pmap?  No.
            pmap(num_stores, boost::bind(do_construct_existing_store,
                                         store_threads, _1, store_args,
                                         multiplexer.get(),
                                         stores_out_stores, store_views.data()));
            mptr.init(new multistore_ptr_t(store_views.data(), num_stores));
        } else {
            standard_serializer_t::create(&file_opener,
                                          standard_serializer_t::static_config_t());
            {
                scoped_ptr_t<serializer_t> ser
                    = make_scoped<standard_serializer_t>(
                        standard_serializer_t::dynamic_config_t(),
                        &file_opener,
                        serializers_perfmon_collection);
                ser = make_scoped<merger_serializer_t>(std::move(ser),
                                                       MERGER_SERIALIZER_MAX_ACTIVE_WRITES);
                serializer = std::move(ser);
            }

            std::vector<serializer_t *> ptrs;
            ptrs.push_back(serializer.get());
            serializer_multiplexer_t::create(ptrs, num_stores);
            multiplexer.init(new serializer_multiplexer_t(ptrs));

            // TODO: How do we specify what the stores' regions are?
            // TODO: Exceptions?  Can exceptions happen, and then store_views'
            // values would leak.
            // The files do not exist, create them.
            // TODO: This should use pmap.
            pmap(num_stores, boost::bind(do_create_new_store,
                                         store_threads, _1, store_args,
                                         multiplexer.get(),
                                         stores_out_stores, store_views.data()));
            mptr.init(new multistore_ptr_t(store_views.data(), num_stores));

            // Initialize the metadata in the underlying stores.
            object_buffer_t<fifo_enforcer_sink_t::exit_write_t> write_token;
            mptr->new_write_token(&write_token);
            cond_t dummy_interruptor;
            order_source_t order_source;  // TODO: order_token_t::ignore.  Use the svs.
            guarantee(mptr->get_region() == region_t::universe());
            mptr->set_metainfo(
                region_map_t<binary_blob_t>(
                    mptr->get_region(),
                    binary_blob_t(version_range_t(version_t::zero()))),
                order_source.check_in("file_based_svs_by_namespace_t"),
                &write_token,
                &dummy_interruptor);

            // Finally, the store is created.
            file_opener.move_serializer_file_to_permanent_location();
        }
    } // back on calling thread

    svs_out->init(mptr.release());
    stores_out->serializer()->init(serializer.release());
    stores_out->multiplexer()->init(multiplexer.release());
}

void file_based_svs_by_namespace_t::destroy_svs(namespace_id_t namespace_id) {
    // TODO: Handle errors?  It seems like we can't really handle the error so
    // let's just ignore it?
    const std::string filepath = file_name_for(namespace_id).permanent_path();
    logNTC("Removing file %s\n", filepath.c_str());
    const int res = ::unlink(filepath.c_str());
    guarantee_err(res == 0 || get_errno() == ENOENT,
                  "unlink failed for file %s", filepath.c_str());
}

serializer_filepath_t file_based_svs_by_namespace_t::file_name_for(namespace_id_t namespace_id) {
    return serializer_filepath_t(base_path_, uuid_to_str(namespace_id));
}

threadnum_t file_based_svs_by_namespace_t::next_thread(int num_db_threads) {
    thread_counter_ = (thread_counter_ + 1) % num_db_threads;
    return threadnum_t(thread_counter_);
}
<commit_msg>Renamed new do_store to do_construct_store<commit_after>// Copyright 2010-2014 RethinkDB, all rights reserved.
#include "clustering/administration/main/file_based_svs_by_namespace.hpp"

#include "errors.hpp"
#include <boost/bind.hpp>

#include "clustering/immediate_consistency/branch/multistore.hpp"
#include "clustering/reactor/reactor.hpp"
#include "logger.hpp"
#include "rdb_protocol/store.hpp"
#include "serializer/config.hpp"
#include "serializer/translator.hpp"
#include "serializer/merger.hpp"
#include "utils.hpp"

/* This object serves mostly as a container for arguments to the
 * do_construct_existing_store function because we hit the boost::bind argument
 * limit. */
struct store_args_t {
    store_args_t(io_backender_t *_io_backender,
                 const base_path_t &_base_path,
                 namespace_id_t _namespace_id,
                 cache_balancer_t *_balancer,
                 perfmon_collection_t *_serializers_perfmon_collection,
                 rdb_context_t *_ctx,
                 outdated_index_issue_tracker_t *_outdated_index_tracker,
                 namespace_id_t _ns_id)
        : io_backender(_io_backender), base_path(_base_path),
          namespace_id(_namespace_id), balancer(_balancer),
          serializers_perfmon_collection(_serializers_perfmon_collection),
          ctx(_ctx), outdated_index_tracker(_outdated_index_tracker), ns_id(_ns_id)
    { }

    io_backender_t *io_backender;
    base_path_t base_path;
    namespace_id_t namespace_id;
    cache_balancer_t *balancer;
    perfmon_collection_t *serializers_perfmon_collection;
    rdb_context_t *ctx;
    outdated_index_issue_tracker_t *outdated_index_tracker;
    namespace_id_t ns_id;
};

std::string hash_shard_perfmon_name(int hash_shard_number) {
    return strprintf("shard_%d", hash_shard_number);
}

void do_construct_store(
    const std::vector<threadnum_t> &threads,
    int thread_offset,
    store_args_t store_args,
    serializer_multiplexer_t *multiplexer,
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores,
    store_view_t **store_views,
    bool create) {

    // TODO: Exceptions?  Can exceptions happen, and then this doesn't
    // catch it, and the caller doesn't handle it.
    on_thread_t th(threads[thread_offset]);

    // Only pass this down to the first store
    scoped_ptr_t<outdated_index_report_t> index_report;
    if (thread_offset == 0) {
        index_report = store_args.outdated_index_tracker->create_report(store_args.ns_id);
    }

    // TODO: Can we pass serializers_perfmon_collection across threads like this?
    store_t *store = new store_t(
        multiplexer->proxies[thread_offset], store_args.balancer,
        hash_shard_perfmon_name(thread_offset),
        create, store_args.serializers_perfmon_collection,
        store_args.ctx, store_args.io_backender, store_args.base_path,
        std::move(index_report), store_args.ns_id);
    (*stores_out_stores)[thread_offset].init(store);
    store_views[thread_offset] = store;
}

void do_construct_existing_store(
    const std::vector<threadnum_t> &threads,
    int thread_offset,
    store_args_t store_args,
    serializer_multiplexer_t *multiplexer,
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores,
    store_view_t **store_views) {

    do_construct_store(threads, thread_offset, store_args, multiplexer, stores_out_stores,
    store_views, false);
}

void do_create_new_store(
    const std::vector<threadnum_t> &threads,
    int thread_offset,
    store_args_t store_args,
    serializer_multiplexer_t *multiplexer,
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores,
    store_view_t **store_views) {

    do_construct_store(threads, thread_offset, store_args, multiplexer, stores_out_stores,
    store_views, true);
}

void
file_based_svs_by_namespace_t::get_svs(
            perfmon_collection_t *serializers_perfmon_collection,
            namespace_id_t namespace_id,
            stores_lifetimer_t *stores_out,
            scoped_ptr_t<multistore_ptr_t> *svs_out,
            rdb_context_t *ctx) {
    const int num_db_threads = get_num_db_threads();

    // TODO: If the server gets killed when starting up, we can
    // get a database in an invalid startup state.

    // TODO: This is quite suspicious in that we check if the file
    // exists and then assume it exists or does not exist when
    // loading or creating it.

    // TODO: We should use N slices on M serializers, not N slices
    // on N serializers.

    const int num_stores = CPU_SHARDING_FACTOR;
    scoped_array_t<scoped_ptr_t<store_t> > *stores_out_stores
        = stores_out->stores();
    stores_out_stores->init(num_stores);

    const threadnum_t serializer_thread = next_thread(num_db_threads);
    std::vector<threadnum_t> store_threads;
    for (int i = 0; i < num_stores; ++i) {
        store_threads.push_back(next_thread(num_db_threads));
    }

    scoped_ptr_t<serializer_t> serializer;
    scoped_ptr_t<serializer_multiplexer_t> multiplexer;
    scoped_ptr_t<multistore_ptr_t> mptr;
    {
        on_thread_t th(serializer_thread);
        scoped_array_t<store_view_t *> store_views(num_stores);

        const serializer_filepath_t serializer_filepath = file_name_for(namespace_id);
        int res = access(serializer_filepath.permanent_path().c_str(), R_OK | W_OK);
        store_args_t store_args(io_backender_, base_path_,
                                namespace_id, balancer_,
                                serializers_perfmon_collection, ctx,
                                &outdated_index_tracker, namespace_id);
        filepath_file_opener_t file_opener(serializer_filepath, io_backender_);
        if (res == 0) {
            // TODO: Could we handle failure when loading the serializer?  Right
            // now, we don't.

            {
                scoped_ptr_t<serializer_t> ser
                    = make_scoped<standard_serializer_t>(
                        standard_serializer_t::dynamic_config_t(),
                        &file_opener,
                        serializers_perfmon_collection);
                ser = make_scoped<merger_serializer_t>(std::move(ser),
                                                       MERGER_SERIALIZER_MAX_ACTIVE_WRITES);
                serializer = std::move(ser);
            }

            std::vector<serializer_t *> ptrs;
            ptrs.push_back(serializer.get());
            multiplexer.init(new serializer_multiplexer_t(ptrs));

            // TODO: Exceptions?  Can exceptions happen, and then
            // store_views' values would leak.  That is, are we handling
            // them in the pmap?  No.
            pmap(num_stores, boost::bind(do_construct_existing_store,
                                         store_threads, _1, store_args,
                                         multiplexer.get(),
                                         stores_out_stores, store_views.data()));
            mptr.init(new multistore_ptr_t(store_views.data(), num_stores));
        } else {
            standard_serializer_t::create(&file_opener,
                                          standard_serializer_t::static_config_t());
            {
                scoped_ptr_t<serializer_t> ser
                    = make_scoped<standard_serializer_t>(
                        standard_serializer_t::dynamic_config_t(),
                        &file_opener,
                        serializers_perfmon_collection);
                ser = make_scoped<merger_serializer_t>(std::move(ser),
                                                       MERGER_SERIALIZER_MAX_ACTIVE_WRITES);
                serializer = std::move(ser);
            }

            std::vector<serializer_t *> ptrs;
            ptrs.push_back(serializer.get());
            serializer_multiplexer_t::create(ptrs, num_stores);
            multiplexer.init(new serializer_multiplexer_t(ptrs));

            // TODO: How do we specify what the stores' regions are?
            // TODO: Exceptions?  Can exceptions happen, and then store_views'
            // values would leak.
            // The files do not exist, create them.
            // TODO: This should use pmap.
            pmap(num_stores, boost::bind(do_create_new_store,
                                         store_threads, _1, store_args,
                                         multiplexer.get(),
                                         stores_out_stores, store_views.data()));
            mptr.init(new multistore_ptr_t(store_views.data(), num_stores));

            // Initialize the metadata in the underlying stores.
            object_buffer_t<fifo_enforcer_sink_t::exit_write_t> write_token;
            mptr->new_write_token(&write_token);
            cond_t dummy_interruptor;
            order_source_t order_source;  // TODO: order_token_t::ignore.  Use the svs.
            guarantee(mptr->get_region() == region_t::universe());
            mptr->set_metainfo(
                region_map_t<binary_blob_t>(
                    mptr->get_region(),
                    binary_blob_t(version_range_t(version_t::zero()))),
                order_source.check_in("file_based_svs_by_namespace_t"),
                &write_token,
                &dummy_interruptor);

            // Finally, the store is created.
            file_opener.move_serializer_file_to_permanent_location();
        }
    } // back on calling thread

    svs_out->init(mptr.release());
    stores_out->serializer()->init(serializer.release());
    stores_out->multiplexer()->init(multiplexer.release());
}

void file_based_svs_by_namespace_t::destroy_svs(namespace_id_t namespace_id) {
    // TODO: Handle errors?  It seems like we can't really handle the error so
    // let's just ignore it?
    const std::string filepath = file_name_for(namespace_id).permanent_path();
    logNTC("Removing file %s\n", filepath.c_str());
    const int res = ::unlink(filepath.c_str());
    guarantee_err(res == 0 || get_errno() == ENOENT,
                  "unlink failed for file %s", filepath.c_str());
}

serializer_filepath_t file_based_svs_by_namespace_t::file_name_for(namespace_id_t namespace_id) {
    return serializer_filepath_t(base_path_, uuid_to_str(namespace_id));
}

threadnum_t file_based_svs_by_namespace_t::next_thread(int num_db_threads) {
    thread_counter_ = (thread_counter_ + 1) % num_db_threads;
    return threadnum_t(thread_counter_);
}

{'fraction_non_alphanumeric': 0.08523577392731309, 'fraction_numerical': 0.0014479463294560548, 'mean_word_length': 2.6370019308407935, 'pattern_counts': {'":': 0, '<': 63, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1890060', 'n_tokens_mistral': 6473, 'n_tokens_neox': 6329, 'n_words': 1156}