<commit_before>#include "btree/node.hpp"
#include "replication/backfill_out.hpp"
#include "concurrency/drain_semaphore.hpp"
#include "concurrency/pmap.hpp"
#include "btree/slice.hpp"
#include "concurrency/coro_pool.hpp"

namespace replication {

struct backfill_and_streaming_manager_t :
    public backfill_callback_t,
    public home_thread_mixin_t
{
    /* Realtime operation handlers */

    /* realtime_mutation_dispatcher_t is used to listen for realtime mutations on the key-value
    store. */
    
    struct realtime_mutation_dispatcher_t : public mutation_dispatcher_t {
    
        struct change_visitor_t : public boost::static_visitor<mutation_t> {
        
            backfill_and_streaming_manager_t *manager;
            castime_t castime;
        
            mutation_t operator()(const get_cas_mutation_t& m) {
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_get_cas, manager,
                        m.key, castime));
                return m;
            }
            mutation_t operator()(const sarc_mutation_t& m) {
                unique_ptr_t<data_provider_t> dps[2];
                duplicate_data_provider(m.data, 2, dps);
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_sarc, manager,
                        m.key, dps[0], m.flags, m.exptime, castime, m.add_policy, m.replace_policy, m.old_cas));
                sarc_mutation_t m2(m);
                m2.data = dps[1];
                return m2;
            }
            mutation_t operator()(const incr_decr_mutation_t& m) {
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_incr_decr, manager,
                        m.kind, m.key, m.amount, castime));
                return m;
            }
            mutation_t operator()(const append_prepend_mutation_t &m) {
                unique_ptr_t<data_provider_t> dps[2];
                duplicate_data_provider(m.data, 2, dps);
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_append_prepend, manager,
                        m.kind, m.key, dps[0], castime));
                append_prepend_mutation_t m2(m);
                m2.data = dps[1];
                return m2;
            }
            mutation_t operator()(const delete_mutation_t& m) {
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_delete_key, manager,
                        m.key, castime.timestamp));
                return m;
            }
        };
    
        mutation_t dispatch_change(const mutation_t& m, castime_t castime) {
            change_visitor_t functor;
            functor.manager = manager_;
            functor.castime = castime;
            return boost::apply_visitor(functor, m.mutation);
        }
    
        backfill_and_streaming_manager_t *manager_;
    };

    void realtime_get_cas(const store_key_t& key, castime_t castime) {
        handler_->realtime_get_cas(key, castime);
    }
    void realtime_sarc(const store_key_t& key, unique_ptr_t<data_provider_t> data,
            mcflags_t flags, exptime_t exptime, castime_t castime, add_policy_t add_policy,
            replace_policy_t replace_policy, cas_t old_cas) {
        handler_->realtime_sarc(key, data, flags, exptime, castime, add_policy, replace_policy, old_cas);
    }
    void realtime_incr_decr(incr_decr_kind_t kind, const store_key_t &key, uint64_t amount,
            castime_t castime) {
        handler_->realtime_incr_decr(kind, key, amount, castime);
    }
    void realtime_append_prepend(append_prepend_kind_t kind, const store_key_t &key,
            unique_ptr_t<data_provider_t> data, castime_t castime) {
        handler_->realtime_append_prepend(kind, key, data, castime);
    }
    void realtime_delete_key(const store_key_t &key, repli_timestamp timestamp) {
        handler_->realtime_delete_key(key, timestamp);
    }

    /* backfill_callback_t implementation */

    bool should_send_deletion_keys(bool can_send_deletion_keys) {
        on_thread_t th(home_thread);

        all_delete_queues_so_far_can_send_keys_ &= can_send_deletion_keys;

        delete_queues_can_send_keys_cond_.pulse();

        // Wait until all slices have pulsed it.
        delete_queues_can_send_keys_cond_.wait();

        return all_delete_queues_so_far_can_send_keys_;
    }
<commit_msg>Now we actually send a delete everything command, and perhaps we actually delete everything.<commit_after>#include "btree/node.hpp"
#include "replication/backfill_out.hpp"
#include "concurrency/drain_semaphore.hpp"
#include "concurrency/pmap.hpp"
#include "btree/slice.hpp"
#include "concurrency/coro_pool.hpp"

namespace replication {

struct backfill_and_streaming_manager_t :
    public backfill_callback_t,
    public home_thread_mixin_t
{
    /* Realtime operation handlers */

    /* realtime_mutation_dispatcher_t is used to listen for realtime mutations on the key-value
    store. */
    
    struct realtime_mutation_dispatcher_t : public mutation_dispatcher_t {
    
        struct change_visitor_t : public boost::static_visitor<mutation_t> {
        
            backfill_and_streaming_manager_t *manager;
            castime_t castime;
        
            mutation_t operator()(const get_cas_mutation_t& m) {
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_get_cas, manager,
                        m.key, castime));
                return m;
            }
            mutation_t operator()(const sarc_mutation_t& m) {
                unique_ptr_t<data_provider_t> dps[2];
                duplicate_data_provider(m.data, 2, dps);
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_sarc, manager,
                        m.key, dps[0], m.flags, m.exptime, castime, m.add_policy, m.replace_policy, m.old_cas));
                sarc_mutation_t m2(m);
                m2.data = dps[1];
                return m2;
            }
            mutation_t operator()(const incr_decr_mutation_t& m) {
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_incr_decr, manager,
                        m.kind, m.key, m.amount, castime));
                return m;
            }
            mutation_t operator()(const append_prepend_mutation_t &m) {
                unique_ptr_t<data_provider_t> dps[2];
                duplicate_data_provider(m.data, 2, dps);
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_append_prepend, manager,
                        m.kind, m.key, dps[0], castime));
                append_prepend_mutation_t m2(m);
                m2.data = dps[1];
                return m2;
            }
            mutation_t operator()(const delete_mutation_t& m) {
                manager->coro_pool.queue_task_on_home_thread(boost::bind(&backfill_and_streaming_manager_t::realtime_delete_key, manager,
                        m.key, castime.timestamp));
                return m;
            }
        };
    
        mutation_t dispatch_change(const mutation_t& m, castime_t castime) {
            change_visitor_t functor;
            functor.manager = manager_;
            functor.castime = castime;
            return boost::apply_visitor(functor, m.mutation);
        }
    
        backfill_and_streaming_manager_t *manager_;
    };

    void realtime_get_cas(const store_key_t& key, castime_t castime) {
        handler_->realtime_get_cas(key, castime);
    }
    void realtime_sarc(const store_key_t& key, unique_ptr_t<data_provider_t> data,
            mcflags_t flags, exptime_t exptime, castime_t castime, add_policy_t add_policy,
            replace_policy_t replace_policy, cas_t old_cas) {
        handler_->realtime_sarc(key, data, flags, exptime, castime, add_policy, replace_policy, old_cas);
    }
    void realtime_incr_decr(incr_decr_kind_t kind, const store_key_t &key, uint64_t amount,
            castime_t castime) {
        handler_->realtime_incr_decr(kind, key, amount, castime);
    }
    void realtime_append_prepend(append_prepend_kind_t kind, const store_key_t &key,
            unique_ptr_t<data_provider_t> data, castime_t castime) {
        handler_->realtime_append_prepend(kind, key, data, castime);
    }
    void realtime_delete_key(const store_key_t &key, repli_timestamp timestamp) {
        handler_->realtime_delete_key(key, timestamp);
    }

    /* backfill_callback_t implementation */

    bool should_send_deletion_keys(bool can_send_deletion_keys) {
        on_thread_t th(home_thread);

        if (all_delete_queues_so_far_can_send_keys_ && !can_send_deletion_keys) {
            all_delete_queues_so_far_can_send_keys_ = false;

            // We only send this once, and it is important that we
            // send it before the delete_queues_can_send_keys_cond_
            // gets pulsed, because then a delete queue could finish
            // and start sending sets before we sent a
            // delete_everything message.
            handler_->backfill_delete_everything();
        }

        delete_queues_can_send_keys_cond_.pulse();

        // Wait until all slices have pulsed it.
        delete_queues_can_send_keys_cond_.wait();

        return all_delete_queues_so_far_can_send_keys_;
    }

    void wait_and_maybe_send_delete_all_keys_message() {
        delete_queues_can_send_keys_cond_.wait();
    }

{'fraction_non_alphanumeric': 0.08194860150686345, 'fraction_numerical': 0.003302714418412633, 'mean_word_length': 2.62378459237098, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 34, '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': '4988522', 'n_tokens_mistral': 3169, 'n_tokens_neox': 3055, 'n_words': 519}