<commit_before>// Copyright 2010-2013 RethinkDB, all rights reserved.
#include "rdb_protocol/protocol.hpp"

#include <algorithm>

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

#include "arch/io/disk.hpp"
#include "btree/erase_range.hpp"
#include "btree/parallel_traversal.hpp"
#include "btree/slice.hpp"
#include "btree/superblock.hpp"
#include "clustering/administration/metadata.hpp"
#include "concurrency/cross_thread_watchable.hpp"
#include "concurrency/pmap.hpp"
#include "concurrency/wait_any.hpp"
#include "containers/archive/archive.hpp"
#include "containers/archive/vector_stream.hpp"
#include "protob/protob.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/func.hpp"
#include "rdb_protocol/transform_visitors.hpp"
#include "rdb_protocol/minidriver.hpp"
#include "rdb_protocol/term_walker.hpp"
#include "rpc/semilattice/view/field.hpp"
#include "rpc/semilattice/watchable.hpp"
#include "serializer/config.hpp"

// Ignore clang errors for ql macros
#pragma GCC diagnostic ignored "-Wshadow"

typedef rdb_protocol_details::backfill_atom_t rdb_backfill_atom_t;
typedef rdb_protocol_details::range_key_tester_t range_key_tester_t;

typedef rdb_protocol_t::context_t context_t;

typedef rdb_protocol_t::store_t store_t;
typedef rdb_protocol_t::region_t region_t;

typedef rdb_protocol_t::read_t read_t;
typedef rdb_protocol_t::read_response_t read_response_t;

typedef rdb_protocol_t::point_read_t point_read_t;
typedef rdb_protocol_t::point_read_response_t point_read_response_t;

typedef rdb_protocol_t::rget_read_t rget_read_t;
typedef rdb_protocol_t::rget_read_response_t rget_read_response_t;

typedef rdb_protocol_t::distribution_read_t distribution_read_t;
typedef rdb_protocol_t::distribution_read_response_t distribution_read_response_t;

typedef rdb_protocol_t::sindex_list_t sindex_list_t;
typedef rdb_protocol_t::sindex_list_response_t sindex_list_response_t;

typedef rdb_protocol_t::write_t write_t;
typedef rdb_protocol_t::write_response_t write_response_t;

typedef rdb_protocol_t::batched_replace_t batched_replace_t;
typedef rdb_protocol_t::batched_insert_t batched_insert_t;

typedef rdb_protocol_t::point_write_t point_write_t;
typedef rdb_protocol_t::point_write_response_t point_write_response_t;

typedef rdb_protocol_t::point_delete_t point_delete_t;
typedef rdb_protocol_t::point_delete_response_t point_delete_response_t;

typedef rdb_protocol_t::sindex_create_t sindex_create_t;
typedef rdb_protocol_t::sindex_create_response_t sindex_create_response_t;

typedef rdb_protocol_t::sindex_drop_t sindex_drop_t;
typedef rdb_protocol_t::sindex_drop_response_t sindex_drop_response_t;

typedef rdb_protocol_t::sync_t sync_t;
typedef rdb_protocol_t::sync_response_t sync_response_t;

typedef rdb_protocol_t::backfill_chunk_t backfill_chunk_t;

typedef rdb_protocol_t::backfill_progress_t backfill_progress_t;

typedef rdb_protocol_t::rget_read_response_t::stream_t stream_t;

typedef btree_store_t<rdb_protocol_t>::sindex_access_vector_t sindex_access_vector_t;

const std::string rdb_protocol_t::protocol_name("rdb");

RDB_IMPL_PROTOB_SERIALIZABLE(Term);
RDB_IMPL_PROTOB_SERIALIZABLE(Datum);
RDB_IMPL_PROTOB_SERIALIZABLE(Backtrace);


RDB_IMPL_SERIALIZABLE_2(filter_transform_t, filter_func, default_filter_val);

namespace rdb_protocol_details {

RDB_IMPL_SERIALIZABLE_3(backfill_atom_t, key, value, recency);

void post_construct_and_drain_queue(
        auto_drainer_t::lock_t lock,
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue)
    THROWS_NOTHING;

/* Creates a queue of operations for the sindex, runs a post construction for
 * the data already in the btree and finally drains the queue. */
void bring_sindexes_up_to_date(
        const std::set<std::string> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        buf_lock_t *sindex_block,
        transaction_t *txn)
    THROWS_NOTHING
{
    with_priority_t p(-2);

    /* We register our modification queue here. An important point about
     * correctness here: we've held the superblock this whole time and will
     * continue to do so until the call to post_construct_secondary_indexes
     * begins a parallel traversal which releases the superblock. This
     * serves to make sure that every changes which we don't learn about in
     * the parallel traversal we do learn about from the mod queue. */
    uuid_u post_construct_id = generate_uuid();

    boost::shared_ptr<internal_disk_backed_queue_t> mod_queue(
            new internal_disk_backed_queue_t(
                store->io_backender_,
                serializer_filepath_t(
                    store->base_path_,
                    "post_construction_" + uuid_to_str(post_construct_id)),
                &store->perfmon_collection));

    {
        mutex_t::acq_t acq;
        store->lock_sindex_queue(sindex_block, &acq);
        store->register_sindex_queue(mod_queue.get(), &acq);
    }

    std::map<std::string, secondary_index_t> sindexes;
    store->get_sindexes(sindex_block, txn, &sindexes);
    std::set<uuid_u> sindexes_to_bring_up_to_date_uuid;

    for (auto it = sindexes_to_bring_up_to_date.begin();
         it != sindexes_to_bring_up_to_date.end(); ++it) {
        guarantee(std_contains(sindexes, *it));
        sindexes_to_bring_up_to_date_uuid.insert(sindexes[*it].id);
    }

    coro_t::spawn_sometime(boost::bind(
                &post_construct_and_drain_queue,
                auto_drainer_t::lock_t(&store->drainer),
                sindexes_to_bring_up_to_date_uuid,
                store,
                mod_queue));
}

class apply_sindex_change_visitor_t : public boost::static_visitor<> {
public:
    apply_sindex_change_visitor_t(const sindex_access_vector_t *sindexes,
            transaction_t *txn,
            signal_t *interruptor)
        : sindexes_(sindexes), txn_(txn), interruptor_(interruptor) { }
    void operator()(const rdb_modification_report_t &mod_report) const {
        rdb_update_sindexes(*sindexes_, &mod_report, txn_);
        coro_t::yield();
    }

    void operator()(const rdb_erase_range_report_t &erase_range_report) const {
        rdb_erase_range_sindexes(*sindexes_, &erase_range_report, txn_, interruptor_);
        coro_t::yield();
    }

private:
    const sindex_access_vector_t *sindexes_;
    transaction_t *txn_;
    signal_t *interruptor_;
};

/* This function is really part of the logic of bring_sindexes_up_to_date
 * however it needs to be in a seperate function so that it can be spawned in a
 * coro. 
 * NOTE: the auto_drainer lock must be in front of the
 * mod_queue shared pointer, because we must not release
 * the drainer lock before the mod_queue is released.
 */
void post_construct_and_drain_queue(
        auto_drainer_t::lock_t lock,
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue)<commit_msg>These yields were not necessary<commit_after>// Copyright 2010-2013 RethinkDB, all rights reserved.
#include "rdb_protocol/protocol.hpp"

#include <algorithm>

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

#include "arch/io/disk.hpp"
#include "btree/erase_range.hpp"
#include "btree/parallel_traversal.hpp"
#include "btree/slice.hpp"
#include "btree/superblock.hpp"
#include "clustering/administration/metadata.hpp"
#include "concurrency/cross_thread_watchable.hpp"
#include "concurrency/pmap.hpp"
#include "concurrency/wait_any.hpp"
#include "containers/archive/archive.hpp"
#include "containers/archive/vector_stream.hpp"
#include "protob/protob.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/func.hpp"
#include "rdb_protocol/transform_visitors.hpp"
#include "rdb_protocol/minidriver.hpp"
#include "rdb_protocol/term_walker.hpp"
#include "rpc/semilattice/view/field.hpp"
#include "rpc/semilattice/watchable.hpp"
#include "serializer/config.hpp"

// Ignore clang errors for ql macros
#pragma GCC diagnostic ignored "-Wshadow"

typedef rdb_protocol_details::backfill_atom_t rdb_backfill_atom_t;
typedef rdb_protocol_details::range_key_tester_t range_key_tester_t;

typedef rdb_protocol_t::context_t context_t;

typedef rdb_protocol_t::store_t store_t;
typedef rdb_protocol_t::region_t region_t;

typedef rdb_protocol_t::read_t read_t;
typedef rdb_protocol_t::read_response_t read_response_t;

typedef rdb_protocol_t::point_read_t point_read_t;
typedef rdb_protocol_t::point_read_response_t point_read_response_t;

typedef rdb_protocol_t::rget_read_t rget_read_t;
typedef rdb_protocol_t::rget_read_response_t rget_read_response_t;

typedef rdb_protocol_t::distribution_read_t distribution_read_t;
typedef rdb_protocol_t::distribution_read_response_t distribution_read_response_t;

typedef rdb_protocol_t::sindex_list_t sindex_list_t;
typedef rdb_protocol_t::sindex_list_response_t sindex_list_response_t;

typedef rdb_protocol_t::write_t write_t;
typedef rdb_protocol_t::write_response_t write_response_t;

typedef rdb_protocol_t::batched_replace_t batched_replace_t;
typedef rdb_protocol_t::batched_insert_t batched_insert_t;

typedef rdb_protocol_t::point_write_t point_write_t;
typedef rdb_protocol_t::point_write_response_t point_write_response_t;

typedef rdb_protocol_t::point_delete_t point_delete_t;
typedef rdb_protocol_t::point_delete_response_t point_delete_response_t;

typedef rdb_protocol_t::sindex_create_t sindex_create_t;
typedef rdb_protocol_t::sindex_create_response_t sindex_create_response_t;

typedef rdb_protocol_t::sindex_drop_t sindex_drop_t;
typedef rdb_protocol_t::sindex_drop_response_t sindex_drop_response_t;

typedef rdb_protocol_t::sync_t sync_t;
typedef rdb_protocol_t::sync_response_t sync_response_t;

typedef rdb_protocol_t::backfill_chunk_t backfill_chunk_t;

typedef rdb_protocol_t::backfill_progress_t backfill_progress_t;

typedef rdb_protocol_t::rget_read_response_t::stream_t stream_t;

typedef btree_store_t<rdb_protocol_t>::sindex_access_vector_t sindex_access_vector_t;

const std::string rdb_protocol_t::protocol_name("rdb");

RDB_IMPL_PROTOB_SERIALIZABLE(Term);
RDB_IMPL_PROTOB_SERIALIZABLE(Datum);
RDB_IMPL_PROTOB_SERIALIZABLE(Backtrace);


RDB_IMPL_SERIALIZABLE_2(filter_transform_t, filter_func, default_filter_val);

namespace rdb_protocol_details {

RDB_IMPL_SERIALIZABLE_3(backfill_atom_t, key, value, recency);

void post_construct_and_drain_queue(
        auto_drainer_t::lock_t lock,
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue)
    THROWS_NOTHING;

/* Creates a queue of operations for the sindex, runs a post construction for
 * the data already in the btree and finally drains the queue. */
void bring_sindexes_up_to_date(
        const std::set<std::string> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        buf_lock_t *sindex_block,
        transaction_t *txn)
    THROWS_NOTHING
{
    with_priority_t p(-2);

    /* We register our modification queue here. An important point about
     * correctness here: we've held the superblock this whole time and will
     * continue to do so until the call to post_construct_secondary_indexes
     * begins a parallel traversal which releases the superblock. This
     * serves to make sure that every changes which we don't learn about in
     * the parallel traversal we do learn about from the mod queue. */
    uuid_u post_construct_id = generate_uuid();

    boost::shared_ptr<internal_disk_backed_queue_t> mod_queue(
            new internal_disk_backed_queue_t(
                store->io_backender_,
                serializer_filepath_t(
                    store->base_path_,
                    "post_construction_" + uuid_to_str(post_construct_id)),
                &store->perfmon_collection));

    {
        mutex_t::acq_t acq;
        store->lock_sindex_queue(sindex_block, &acq);
        store->register_sindex_queue(mod_queue.get(), &acq);
    }

    std::map<std::string, secondary_index_t> sindexes;
    store->get_sindexes(sindex_block, txn, &sindexes);
    std::set<uuid_u> sindexes_to_bring_up_to_date_uuid;

    for (auto it = sindexes_to_bring_up_to_date.begin();
         it != sindexes_to_bring_up_to_date.end(); ++it) {
        guarantee(std_contains(sindexes, *it));
        sindexes_to_bring_up_to_date_uuid.insert(sindexes[*it].id);
    }

    coro_t::spawn_sometime(boost::bind(
                &post_construct_and_drain_queue,
                auto_drainer_t::lock_t(&store->drainer),
                sindexes_to_bring_up_to_date_uuid,
                store,
                mod_queue));
}

class apply_sindex_change_visitor_t : public boost::static_visitor<> {
public:
    apply_sindex_change_visitor_t(const sindex_access_vector_t *sindexes,
            transaction_t *txn,
            signal_t *interruptor)
        : sindexes_(sindexes), txn_(txn), interruptor_(interruptor) { }
    void operator()(const rdb_modification_report_t &mod_report) const {
        rdb_update_sindexes(*sindexes_, &mod_report, txn_);
    }

    void operator()(const rdb_erase_range_report_t &erase_range_report) const {
        rdb_erase_range_sindexes(*sindexes_, &erase_range_report, txn_, interruptor_);
    }

private:
    const sindex_access_vector_t *sindexes_;
    transaction_t *txn_;
    signal_t *interruptor_;
};

/* This function is really part of the logic of bring_sindexes_up_to_date
 * however it needs to be in a seperate function so that it can be spawned in a
 * coro. 
 * NOTE: the auto_drainer lock must be in front of the
 * mod_queue shared pointer, because we must not release
 * the drainer lock before the mod_queue is released.
 */
void post_construct_and_drain_queue(
        auto_drainer_t::lock_t lock,
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue)
{'fraction_non_alphanumeric': 0.08037992628296002, 'fraction_numerical': 0.0015593989225971081, 'mean_word_length': 5.335428828019758, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1854019', 'n_tokens_mistral': 5137, 'n_tokens_neox': 4921, 'n_words': 891}