<commit_before>#ifndef __CLUSTERING_BACKFILLER_HPP__
#define __CLUSTERING_BACKFILLER_HPP__

#include "clustering/backfill_metadata.hpp"



#endif /* __CLUSTERING_BACKFILLER_HPP__ */<commit_msg>Backfiller<commit_after>#ifndef __CLUSTERING_BACKFILLER_HPP__
#define __CLUSTERING_BACKFILLER_HPP__

#include "clustering/backfill_metadata.hpp"

template<class protocol_t>
struct backfiller_t {

    backfiller_t(
            mailbox_cluster_t *c,
            typename protocol_t::store_t *s,
            metadata_view_t<boost::optional<backfiller_metadata_t<protocol_t> > > *md) :
        cluster(c),
        store(s),
        metadata(md)
    {
        /* Set up mailboxes */
        backfill_mailbox.reset(new backfiller_metadata_t<protocol_t>::backfill_mailbox_t(
            cluster, boost::bind(&backfiller_t::on_backfill, this, _1, _2, _3, _4)));
        cancel_backfill_mailbox.reset(new backfiller_metadata_t<protocol_t>::cancel_backfill_mailbox_t(
            cluster, boost::bind(&backfiller_t::on_cancel_backfill, this, _1)));

        /* Tell everybody we're open for business */
        backfiller_metadata_t<protocol_t> our_business_card;
        our_business_card.is_alive = true;
        our_business_card.backfill_mailbox = &backfill_mailbox;
        our_business_card.cancel_backfill_mailbox = &cancel_backfill_mailbox;
        metadata->join_metadata(boost::make_optional(our_business_card));
    }

    ~backfiller_t() {

        /* Tell everybody we're no longer open for business */
        backfiller_metadata_t<protocol_t> our_business_card;
        our_business_card.is_alive = false;
        metadata->join_metadata(boost::make_optional(our_business_card));

        /* Tear backfill down mailbox so no new backfills can begin */
        backfill_mailbox.reset();

        /* Cancel all existing backfills */
        global_interruptor.pulse();

        /* Wait for cancellation to take effect */
        drain_semaphore.drain();
    }

private:
    typedef backfiller_metadata_t<protocol_t>::backfill_session_id_t session_id_t;

    void on_backfill(
        session_id_t session_id,
        typename protocol_t::backfill_request_t request,
        async_mailbox_t<void(typename protocol_t::backfill_chunk_t)>::address_t chunk_cont,
        async_mailbox_t<void()>::address_t end_cont) {

        /* Acquire the drain semaphore so that the backfiller doesn't shut down
        while we're still running. */
        drain_semaphore_t::lock_t lock(&drain_semaphore);

        /* Set up a local interruptor cond and put it in the map so that this
        session can be interrupted if the backfillee decides to abort */
        cond_t local_interruptor;
        local_interruptors[session_id] = &local_interruptor;

        /* Set up a cond that gets pulsed if we're interrupted either way */
        cond_t any_interruption;
        cond_link_t listen_for_local_interrupts(&local_interruptor, &any_interruption);
        cond_link_t listen_for_global_interrupts(&global_interruptor, &any_interruption);

        /* Perform the backfill */
        bool success;
        try {
            store->send_backfill(
                request,
                boost::bind(&send, cluster, chunk_cont, _1),
                &any_interruption
                );
            success = true;
        } catch (interrupted_exc_t) {
            rassert(any_interruption.is_pulsed());
            success = false;
        }

        /* Send the confirmation */
        if (success) {
            send(cluster, end_cont);
        }

        /* Get rid of our local interruptor */
        local_interruptors.erase(session_id);
    }

    void on_cancel_backfill(session_id_t session_id) {

        std::map<session_id_t, cond_t *>::iterator it =
            local_interruptors.find(session_id);
        if (it != local_interruptors.end()) {
            (*it).second->pulse();
        }
    }

    mailbox_cluster_t *cluster;
    typename protocol_t::store_t *store;
    metadata_view_t<boost::optional<backfiller_metadata_t<protocol_t> > > *metadata;

    boost::scoped_ptr<backfiller_metadata_t<protocol_t>::backfill_mailbox_t> backfill_mailbox;
    boost::scoped_ptr<backfiller_metadata_t<protocol_t>::cancel_backfill_mailbox_t> cancel_backfill_mailbox;

    drain_semaphore_t drain_semaphore;
    cond_t global_interruptor;
    std::map<session_id_t, cond_t *> local_interruptors;
};

#endif /* __CLUSTERING_BACKFILLER_HPP__ */
{'fraction_non_alphanumeric': 0.08854754913033656, 'fraction_numerical': 0.0013553196295459679, 'mean_word_length': 3.1538461538461537, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 27, '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': '6934050', 'n_tokens_mistral': 1423, 'n_tokens_neox': 1381, 'n_words': 247}