<commit_before>#include "unittest/gtest.hpp"

#include "clustering/immediate_consistency/branch/broadcaster.hpp"
#include "clustering/immediate_consistency/branch/listener.hpp"
#include "clustering/immediate_consistency/branch/replier.hpp"
#include "memcached/protocol.hpp"
#include "unittest/dummy_metadata_controller.hpp"
#include "unittest/clustering_utils.hpp"
#include "unittest/unittest_utils.hpp"

namespace unittest {

void run_with_broadcaster(
        boost::function< void(
            simple_mailbox_cluster_t *,
            boost::shared_ptr<semilattice_readwrite_view_t<branch_history_t<memcached_protocol_t> > >,
            clone_ptr_t<watchable_t<boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > > > >,
            boost::scoped_ptr<broadcaster_t<memcached_protocol_t> > *,
            test_store_t<memcached_protocol_t> *,
            boost::scoped_ptr<listener_t<memcached_protocol_t> > *
            )> fun)
{
    /* Set up a cluster so mailboxes can be created */
    simple_mailbox_cluster_t cluster;

    /* Set up metadata meeting-places */
    branch_history_t<memcached_protocol_t> initial_branch_history;
    dummy_semilattice_controller_t<branch_history_t<memcached_protocol_t> >
        branch_history_controller(initial_branch_history);

    /* Set up a broadcaster and initial listener */
    test_store_t<memcached_protocol_t> initial_store;
    cond_t interruptor;

    boost::scoped_ptr<broadcaster_t<memcached_protocol_t> > broadcaster(
        new broadcaster_t<memcached_protocol_t>(
            cluster.get_mailbox_manager(),
            branch_history_controller.get_view(),
            &initial_store.store,
            &interruptor
        ));

    // TODO: visit a psychiatrist
    watchable_variable_t<boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > > > broadcaster_business_card_watchable_variable(boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > >(boost::optional<broadcaster_business_card_t<memcached_protocol_t> >(broadcaster->get_business_card())));

    boost::scoped_ptr<listener_t<memcached_protocol_t> > initial_listener(
        new listener_t<memcached_protocol_t>(
            cluster.get_mailbox_manager(),
            broadcaster_business_card_watchable_variable.get_watchable(),
            branch_history_controller.get_view(),
            broadcaster.get(),
            &interruptor
        ));

    fun(&cluster,
        branch_history_controller.get_view(),
        broadcaster_business_card_watchable_variable.get_watchable(),
        &broadcaster,
        &initial_store,
        &initial_listener);
}

void run_in_thread_pool_with_broadcaster(
        boost::function< void(
            simple_mailbox_cluster_t *,
            boost::shared_ptr<semilattice_readwrite_view_t<branch_history_t<memcached_protocol_t> > >,
            clone_ptr_t<watchable_t<boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > > > >,
            boost::scoped_ptr<broadcaster_t<memcached_protocol_t> > *,
            test_store_t<memcached_protocol_t> *,
            boost::scoped_ptr<listener_t<memcached_protocol_t> > *
            )> fun)
{
    run_in_thread_pool(boost::bind(&run_with_broadcaster, fun));
}


/* `PartialBackfill` backfills only in a specific sub-region. */

void write_to_broadcaster(broadcaster_t<memcached_protocol_t> *broadcaster, const std::string& key, const std::string& value, order_token_t otok, signal_t *) {
    sarc_mutation_t set;
    set.key = store_key_t(key);
    set.data = data_buffer_t::create(value.size());
    memcpy(set.data->buf(), value.c_str(), value.size());
    set.flags = 123;
    set.exptime = 0;
    set.add_policy = add_policy_yes;
    set.replace_policy = replace_policy_yes;
    fake_fifo_enforcement_t enforce;
    memcached_protocol_t::write_t write(set, time(NULL), 12345);
    fifo_enforcer_sink_t::exit_write_t exiter(&enforce.sink, enforce.source.enter_write());
    class : public broadcaster_t<memcached_protocol_t>::ack_callback_t {
    public:
        bool on_ack(peer_id_t) {
            return true;
        }
    } ack_callback;
    cond_t non_interruptor;
    broadcaster->write(write, &exiter, &ack_callback, otok, &non_interruptor);
}

void run_partial_backfill_test(simple_mailbox_cluster_t *cluster,<commit_msg>Fix error introduced by merge.<commit_after>#include "unittest/gtest.hpp"

#include "clustering/immediate_consistency/branch/broadcaster.hpp"
#include "clustering/immediate_consistency/branch/listener.hpp"
#include "clustering/immediate_consistency/branch/replier.hpp"
#include "memcached/protocol.hpp"
#include "unittest/dummy_metadata_controller.hpp"
#include "unittest/clustering_utils.hpp"
#include "unittest/unittest_utils.hpp"

namespace unittest {

void run_with_broadcaster(
        boost::function< void(
            simple_mailbox_cluster_t *,
            boost::shared_ptr<semilattice_readwrite_view_t<branch_history_t<memcached_protocol_t> > >,
            clone_ptr_t<watchable_t<boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > > > >,
            boost::scoped_ptr<broadcaster_t<memcached_protocol_t> > *,
            test_store_t<memcached_protocol_t> *,
            boost::scoped_ptr<listener_t<memcached_protocol_t> > *
            )> fun)
{
    /* Set up a cluster so mailboxes can be created */
    simple_mailbox_cluster_t cluster;

    /* Set up metadata meeting-places */
    branch_history_t<memcached_protocol_t> initial_branch_history;
    dummy_semilattice_controller_t<branch_history_t<memcached_protocol_t> >
        branch_history_controller(initial_branch_history);

    /* Set up a broadcaster and initial listener */
    test_store_t<memcached_protocol_t> initial_store;
    cond_t interruptor;

    boost::scoped_ptr<broadcaster_t<memcached_protocol_t> > broadcaster(
        new broadcaster_t<memcached_protocol_t>(
            cluster.get_mailbox_manager(),
            branch_history_controller.get_view(),
            &initial_store.store,
            &interruptor
        ));

    // TODO: visit a psychiatrist
    watchable_variable_t<boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > > > broadcaster_business_card_watchable_variable(boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > >(boost::optional<broadcaster_business_card_t<memcached_protocol_t> >(broadcaster->get_business_card())));

    boost::scoped_ptr<listener_t<memcached_protocol_t> > initial_listener(
        new listener_t<memcached_protocol_t>(
            cluster.get_mailbox_manager(),
            broadcaster_business_card_watchable_variable.get_watchable(),
            branch_history_controller.get_view(),
            broadcaster.get(),
            &interruptor
        ));

    fun(&cluster,
        branch_history_controller.get_view(),
        broadcaster_business_card_watchable_variable.get_watchable(),
        &broadcaster,
        &initial_store,
        &initial_listener);
}

void run_in_thread_pool_with_broadcaster(
        boost::function< void(
            simple_mailbox_cluster_t *,
            boost::shared_ptr<semilattice_readwrite_view_t<branch_history_t<memcached_protocol_t> > >,
            clone_ptr_t<watchable_t<boost::optional<boost::optional<broadcaster_business_card_t<memcached_protocol_t> > > > >,
            boost::scoped_ptr<broadcaster_t<memcached_protocol_t> > *,
            test_store_t<memcached_protocol_t> *,
            boost::scoped_ptr<listener_t<memcached_protocol_t> > *
            )> fun)
{
    run_in_thread_pool(boost::bind(&run_with_broadcaster, fun));
}


/* `PartialBackfill` backfills only in a specific sub-region. */

static void do_nothing() { }

void write_to_broadcaster(broadcaster_t<memcached_protocol_t> *broadcaster, const std::string& key, const std::string& value, order_token_t otok, signal_t *) {
    sarc_mutation_t set;
    set.key = store_key_t(key);
    set.data = data_buffer_t::create(value.size());
    memcpy(set.data->buf(), value.c_str(), value.size());
    set.flags = 123;
    set.exptime = 0;
    set.add_policy = add_policy_yes;
    set.replace_policy = replace_policy_yes;
    fake_fifo_enforcement_t enforce;
    memcached_protocol_t::write_t write(set, time(NULL), 12345);
    fifo_enforcer_sink_t::exit_write_t exiter(&enforce.sink, enforce.source.enter_write());
    class : public broadcaster_t<memcached_protocol_t>::ack_callback_t {
    public:
        bool on_ack(peer_id_t) {
            return true;
        }
    } ack_callback;
    cond_t non_interruptor;
    broadcaster->write(write, &exiter, &ack_callback, otok, &non_interruptor, &do_nothing);
}

void run_partial_backfill_test(simple_mailbox_cluster_t *cluster,
{'fraction_non_alphanumeric': 0.09952229299363058, 'fraction_numerical': 0.0020473157415832575, 'mean_word_length': 4.300180831826402, 'pattern_counts': {'":': 0, '<': 101, '<?xml version=': 0, '>': 107, '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': '2536852', 'n_tokens_mistral': 2882, 'n_tokens_neox': 2773, 'n_words': 341}