<commit_before>// Copyright 2010-2015 RethinkDB, all rights reserved.
#include "clustering/table_contract/coordinator.hpp"

#include "clustering/table_contract/cpu_sharding.hpp"

/* A `contract_ack_t` is not necessarily homogeneous. It may have different `version_t`s
for different regions, and a region with a single `version_t` may need to be split
further depending on the branch history. Since `calculate_contract()` assumes it's
processing a homogeneous input, we need to break the `contract_ack_t` into homogeneous
pieces. `contract_ack_frag_t` is like a homogeneous version of `contract_ack_t`; in place
of the `region_map_t<version_t>` it has a single `state_timestamp_t`. Use
`break_ack_into_fragments()` to convert a `contract_ack_t` into a
`region_map_t<contract_ack_frag_t>`. */

class contract_ack_frag_t {
public:
    contract_ack_t::state_t state;
    boost::optional<state_timestamp_t> version;
    boost::optional<branch_id_t> branch;
};

region_map_t<contract_ack_frag_t> break_ack_into_fragments(
        const region_t &region,
        const contract_ack_t &ack,
        const branch_id_t &branch,
        const branch_history_reader_t *raft_branch_history) {
    contract_ack_frag_t base_frag;
    base_frag.state = ack.state;
    base_frag.branch = ack.branch;
    if (!static_cast<bool>(ack.version)) {
        return region_map_t<contract_ack_frag_t>(region, base_frag);
    } else {
        branch_history_combiner_t combined_branch_history(
            raft_branch_history, &ack.branch_history);
        std::vector<std::pair<region_t, contract_ack_frag_t> > parts;
        for (const std::pair<region_t, version_t> &vers_pair : *ack.version) {
            region_map_t<version_t> points_on_canonical_branch =
                version_find_branch_common(
                    &combined_branch_history,
                    vers_pair.second,
                    branch,
                    vers_pair.first);
            for (const auto &can_pair : points_on_canonical_branch) {
                base_frag.version = boost::make_optional(can_pair.second.timestamp);
                parts.push_back(std::make_pair(can_pair.first, base_frag));
            }
        }
        return region_map_t<contract_ack_frag_t>(parts.begin(), parts.end());
    }
}

/* `calculate_contract()` calculates a new contract for a region. Whenever any of the
inputs changes, the coordinator will call `update_contract()` to compute a contract for
each range of keys. The new contract will often be the same as the old, in which case it
doesn't get a new contract ID. */
contract_t calculate_contract(
        /* The old contract that contains this region. */
        const contract_t &old_c,
        /* The user-specified configuration for the shard containing this region. */
        const table_config_t::shard_t &config,
        /* Contract acks from replicas regarding `old_c`. If a replica hasn't sent us an
        ack *specifically* for `old_c`, it won't appear in this map; we don't include
        acks for contracts that were in the same region before `old_c`. */
        const std::map<server_id_t, contract_ack_frag_t> &acks) {

    contract_t new_c = old_c;

    /* If there are new servers in `config.replicas`, add them to `c.replicas` */
    new_c.replicas.insert(config.replicas.begin(), config.replicas.end());

    /* If there is a mismatch between `config.replicas` and `c.voters`, then correct
    it */
    if (!static_cast<bool>(old_c.temp_voters) && old_c.voters != config.replicas) {
        size_t num_streaming = 0;
        for (const server_id_t &server : config.replicas) {
            auto it = acks.find(server);
            if (it != acks.end() &&
                    (it->second.state == contract_ack_t::state_t::secondary_streaming ||
                    (static_cast<bool>(old_c.primary) &&
                        old_c.primary->server == server))) {
                ++num_streaming;
            }
        }

        /* We don't want to initiate the change until a majority of the new replicas are
        already streaming, or else we'll lose write availability as soon as we set
        `temp_voters`. */
        if (num_streaming > config.replicas.size() / 2) {
            /* OK, we're ready to go */
            new_c.temp_voters = boost::make_optional(config.replicas);
        }
    }

    /* If we already initiated a voter change by setting `temp_voters`, it might be time
    to commit that change by setting `voters` to `temp_voters`. */
    if (static_cast<bool>(old_c.temp_voters)) {
        /* Before we change `voters`, we have to make sure that we'll preserve the
        invariant that every acked write is on a majority of `voters`. This is mostly the
        job of the primary; it will not report `primary_running` unless it is requiring
        acks from a majority of both `voters` and `temp_voters` before acking writes to
        the client, *and* it has ensured that every write that was acked before that
        policy was implemented has been backfilled to a majority of `temp_voters`. So we
        can't switch voters unless the primary reports `primary_running`. */
        if (static_cast<bool>(old_c.primary) &&
                acks.count(old_c.primary->server) == 1 &&
                acks.at(old_c.primary->server).state ==
                    contract_ack_t::state_t::primary_ready) {
            /* OK, it's safe to commit. */
            new_c.voters = *new_c.temp_voters;
            new_c.temp_voters = boost::none;
        }
    }

    /* If a server was removed from `config.replicas` and `c.voters` but it's still in
    `c.replicas`, then remove it. And if it's primary, then make it not be primary. */
    bool should_kill_primary = false;
    for (const server_id_t &server : old_c.replicas) {
        if (config.replicas.count(server) == 0 &&
                new_c.voters.count(server) == 0 &&
                (!static_cast<bool>(new_c.temp_voters) ||
                    new_c.temp_voters->count(server) == 0)) {
            new_c.replicas.erase(server);
            if (static_cast<bool>(old_c.primary) && old_c.primary->server == server) {
                /* Actual killing happens further down */
                should_kill_primary = true;
            }
        }
    }

    /* If we don't have a primary, choose a primary. Servers are not eligible to be a
    primary unless they are carrying every acked write. In addition, we must choose
    `config.primary_replica` if it is eligible. There will be at least one eligible
    server if and only if we have reports from a majority of `new_c.voters`. */
    if (!static_cast<bool>(old_c.primary)) {
        /* We have an invariant that every acked write must be on the path from the root
        of the branch history to `old_c.branch`. So we project each voter's state onto
        that path, then sort them by position along the path. Any voter that is at least
        as up to date, according to that metric, as more than half of the voters
        (including itself) is eligible. */

        /* First, collect the states from the servers, and sort them by how up-to-date
        they are. Note that we use the server ID as a secondary sorting key. This mean we
        tend to pick the same server if we run the algorithm twice; this helps to reduce
        unnecessary fragmentation. */
        std::vector<std::pair<state_timestamp_t, server_id_t> > sorted_candidates;
        for (const server_id_t &server : new_c.voters) {
            if (acks.count(server) == 1 && acks.at(server).state ==
                    contract_ack_t::state_t::secondary_need_primary) {
                sorted_candidates.push_back(
                    std::make_pair(*(acks.at(server).version), server));
            }
        }
        std::sort(sorted_candidates.begin(), sorted_candidates.end());

        /* Second, determine which servers are eligible to become primary. */
        std::vector<server_id_t> eligible_candidates;
        for (size_t i = 0; i < sorted_candidates.size(); ++i) {
            server_id_t server = sorted_candidates[i].second;
            /* `up_to_date_count` is the number of servers that `server` is at least as
            up-to-date as. We know `server` must be at least as up-to-date as itself and
            all of the servers that are earlier in the list. */
            size_t up_to_date_count = i + 1;
            /* If there are several servers with the same timestamp, they will appear
            together in the list. So `server` may be at least as up-to-date as some of
            the servers that appear after it in the list. */
            while (up_to_date_count < sorted_candidates.size() &&
                    sorted_candidates[up_to_date_count].first ==
                        sorted_candidates[i].first) {
                ++up_to_date_count;
            }
            /* OK, now `up_to_date_count` is the number of servers that this server is
            at least as up-to-date as. */
            if (up_to_date_count > new_c.voters.size() / 2) {
                eligible_candidates.push_back(server);
            }
        }

        if (!eligible_candidates.empty()) {
            contract_t::primary_t p;

            /* Select the primary. It's safe for us to pick any eligible candidate, but
            we should always pick `config.primary_replica` if it's available. Otherwise,
            we just pick the most up-to-date one. */
            auto it = std::find(eligible_candidates.begin(), eligible_candidates.end(),
                config.primary_replica);
            if (it != eligible_candidates.end()) {
                p.server = config.primary_replica;
            } else {
                /* `eligible_candidates` is ordered by how up-to-date they are */
                p.server = eligible_candidates.back();
            }

            /* RSI(raft): If `config.primary_replica` isn't connected or isn't ready, we
            should elect a different primary. If `config.primary_replica` is connected
            but just takes a little bit longer to reply to our contracts than the other
            replicas, we should wait for it to reply to our contracts and then elect it.
            Under the current implementation, we don't wait. This could lead to an
            awkward loop, where we elect the wrong primary, then un-elect it because we
            realize `config.primary_replica` is ready, and then re-elect the wrong
            primary instead of electing `config.primary_replica`. */

            new_c.primary = boost::make_optional(p);
        }
    }

    /* Sometimes we already have a primary, but we need to pick a different one. There
    are three such situations:
    - The existing primary is disconnected
    - The existing primary isn't `config.primary_replica`, and `config.primary_replica`
        is ready to take over the role
    - `config.primary_replica` isn't ready to take over the role, but the existing
        primary isn't even supposed to be a replica anymore.
    In the first situation, we'll simply remove `c.primary`. In the second and third
    situations, we'll first set `c.primary->warm_shutdown` to `true`, and then only
    once the primary acknowledges that, we'll remove `c.primary`. Either way, once the
    replicas acknowledge the contract in which we removed `c.primary`, the logic earlier
    in this function will select a new primary. Note that we can't go straight from the
    old primary to the new one; we need a majority of replicas to promise to stop
    receiving updates from the old primary before it's safe to elect a new one. */
    if (static_cast<bool>(old_c.primary)) {
        /* Note we already checked for the case where the old primary wasn't supposed to
        be a replica. If this is so, then `should_kill_primary` will already be set to
        `true`. */

        /* Check if we need to do an auto-failover. The precise form of this condition
        isn't important for correctness. If we do an auto-failover when the primary isn't
        actually dead, or don't do an auto-failover when the primary is actually dead,
        the worst that will happen is we'll lose availability.

        RSI(raft): Maybe wait until the primary stays down for a certain amount of time
        before failing over. */
        size_t voters_cant_reach_primary = 0;
        for (const server_id_t &server : new_c.voters) {
            if (acks.count(server) == 1 && acks.at(server).state ==
                    contract_ack_t::state_t::secondary_need_primary) {
                ++voters_cant_reach_primary;
            }
        }
        if (voters_cant_reach_primary > new_c.voters.size() / 2) {
            should_kill_primary = true;
        }

        if (should_kill_primary) {
            new_c.primary = boost::none;
        } else if (old_c.primary->server != config.primary_replica &&
                acks.count(config.primary_replica) == 1 &&
                acks.at(config.primary_replica).state ==
                    contract_ack_t::state_t::secondary_streaming) {
            /* The old primary is still a valid replica, but it isn't equal to
            `config.primary_replica`. So we have to do a hand-over to ensure that after
            we kill the primary, `config.primary_replica` will be a valid candidate. */

            if (old_c.primary->hand_over == config.primary_replica &&
                    acks.count(old_c.primary->server) == 1 &&
                    acks.at(old_c.primary->server).state ==
                        contract_ack_t::state_t::primary_ready) {
                /* We already did the hand over. Now it's safe to stop the old primary.
                */
                new_c.primary = boost::none;
            } else {
                new_c.primary->hand_over = boost::make_optional(config.primary_replica);
            }
        } else {
            /* We're sticking with the current primary, so `hand_over` should be empty.
            In the unlikely event that we were in the middle of a hand-over and then
            changed our minds, it might not be empty, so we clear it manually. */
            new_c.primary->hand_over = boost::none;
        }
    }

    /* Register a branch if a primary is asking us to */
    if (static_cast<bool>(old_c.primary) &&
            static_cast<bool>(new_c.primary) &&
            old_c.primary->server == new_c.primary->server &&<commit_msg>Clarify comment<commit_after>// Copyright 2010-2015 RethinkDB, all rights reserved.
#include "clustering/table_contract/coordinator.hpp"

#include "clustering/table_contract/cpu_sharding.hpp"

/* A `contract_ack_t` is not necessarily homogeneous. It may have different `version_t`s
for different regions, and a region with a single `version_t` may need to be split
further depending on the branch history. Since `calculate_contract()` assumes it's
processing a homogeneous input, we need to break the `contract_ack_t` into homogeneous
pieces. `contract_ack_frag_t` is like a homogeneous version of `contract_ack_t`; in place
of the `region_map_t<version_t>` it has a single `state_timestamp_t`. Use
`break_ack_into_fragments()` to convert a `contract_ack_t` into a
`region_map_t<contract_ack_frag_t>`. */

class contract_ack_frag_t {
public:
    contract_ack_t::state_t state;
    boost::optional<state_timestamp_t> version;
    boost::optional<branch_id_t> branch;
};

region_map_t<contract_ack_frag_t> break_ack_into_fragments(
        const region_t &region,
        const contract_ack_t &ack,
        const branch_id_t &branch,
        const branch_history_reader_t *raft_branch_history) {
    contract_ack_frag_t base_frag;
    base_frag.state = ack.state;
    base_frag.branch = ack.branch;
    if (!static_cast<bool>(ack.version)) {
        return region_map_t<contract_ack_frag_t>(region, base_frag);
    } else {
        branch_history_combiner_t combined_branch_history(
            raft_branch_history, &ack.branch_history);
        std::vector<std::pair<region_t, contract_ack_frag_t> > parts;
        for (const std::pair<region_t, version_t> &vers_pair : *ack.version) {
            region_map_t<version_t> points_on_canonical_branch =
                version_find_branch_common(
                    &combined_branch_history,
                    vers_pair.second,
                    branch,
                    vers_pair.first);
            for (const auto &can_pair : points_on_canonical_branch) {
                base_frag.version = boost::make_optional(can_pair.second.timestamp);
                parts.push_back(std::make_pair(can_pair.first, base_frag));
            }
        }
        return region_map_t<contract_ack_frag_t>(parts.begin(), parts.end());
    }
}

/* `calculate_contract()` calculates a new contract for a region. Whenever any of the
inputs changes, the coordinator will call `update_contract()` to compute a contract for
each range of keys. The new contract will often be the same as the old, in which case it
doesn't get a new contract ID. */
contract_t calculate_contract(
        /* The old contract that contains this region. */
        const contract_t &old_c,
        /* The user-specified configuration for the shard containing this region. */
        const table_config_t::shard_t &config,
        /* Contract acks from replicas regarding `old_c`. If a replica hasn't sent us an
        ack *specifically* for `old_c`, it won't appear in this map; we don't include
        acks for contracts that were in the same region before `old_c`. */
        const std::map<server_id_t, contract_ack_frag_t> &acks) {

    contract_t new_c = old_c;

    /* If there are new servers in `config.replicas`, add them to `c.replicas` */
    new_c.replicas.insert(config.replicas.begin(), config.replicas.end());

    /* If there is a mismatch between `config.replicas` and `c.voters`, then correct
    it */
    if (!static_cast<bool>(old_c.temp_voters) && old_c.voters != config.replicas) {
        size_t num_streaming = 0;
        for (const server_id_t &server : config.replicas) {
            auto it = acks.find(server);
            if (it != acks.end() &&
                    (it->second.state == contract_ack_t::state_t::secondary_streaming ||
                    (static_cast<bool>(old_c.primary) &&
                        old_c.primary->server == server))) {
                ++num_streaming;
            }
        }

        /* We don't want to initiate the change until a majority of the new replicas are
        already streaming, or else we'll lose write availability as soon as we set
        `temp_voters`. */
        if (num_streaming > config.replicas.size() / 2) {
            /* OK, we're ready to go */
            new_c.temp_voters = boost::make_optional(config.replicas);
        }
    }

    /* If we already initiated a voter change by setting `temp_voters`, it might be time
    to commit that change by setting `voters` to `temp_voters`. */
    if (static_cast<bool>(old_c.temp_voters)) {
        /* Before we change `voters`, we have to make sure that we'll preserve the
        invariant that every acked write is on a majority of `voters`. This is mostly the
        job of the primary; it will not report `primary_running` unless it is requiring
        acks from a majority of both `voters` and `temp_voters` before acking writes to
        the client, *and* it has ensured that every write that was acked before that
        policy was implemented has been backfilled to a majority of `temp_voters`. So we
        can't switch voters unless the primary reports `primary_running`. */
        if (static_cast<bool>(old_c.primary) &&
                acks.count(old_c.primary->server) == 1 &&
                acks.at(old_c.primary->server).state ==
                    contract_ack_t::state_t::primary_ready) {
            /* OK, it's safe to commit. */
            new_c.voters = *new_c.temp_voters;
            new_c.temp_voters = boost::none;
        }
    }

    /* If a server was removed from `config.replicas` and `c.voters` but it's still in
    `c.replicas`, then remove it. And if it's primary, then make it not be primary. */
    bool should_kill_primary = false;
    for (const server_id_t &server : old_c.replicas) {
        if (config.replicas.count(server) == 0 &&
                new_c.voters.count(server) == 0 &&
                (!static_cast<bool>(new_c.temp_voters) ||
                    new_c.temp_voters->count(server) == 0)) {
            new_c.replicas.erase(server);
            if (static_cast<bool>(old_c.primary) && old_c.primary->server == server) {
                /* Actual killing happens further down */
                should_kill_primary = true;
            }
        }
    }

    /* If we don't have a primary, choose a primary. Servers are not eligible to be a
    primary unless they are carrying every acked write. In addition, we must choose
    `config.primary_replica` if it is eligible. There will be at least one eligible
    server if and only if we have reports from a majority of `new_c.voters`. */
    if (!static_cast<bool>(old_c.primary)) {
        /* We have an invariant that every acked write must be on the path from the root
        of the branch history to `old_c.branch`. So we project each voter's state onto
        that path, then sort them by position along the path. Any voter that is at least
        as up to date, according to that metric, as more than half of the voters
        (including itself) is eligible. */

        /* First, collect the states from the servers, and sort them by how up-to-date
        they are. Note that we use the server ID as a secondary sorting key. This mean we
        tend to pick the same server if we run the algorithm twice; this helps to reduce
        unnecessary fragmentation. */
        std::vector<std::pair<state_timestamp_t, server_id_t> > sorted_candidates;
        for (const server_id_t &server : new_c.voters) {
            if (acks.count(server) == 1 && acks.at(server).state ==
                    contract_ack_t::state_t::secondary_need_primary) {
                sorted_candidates.push_back(
                    std::make_pair(*(acks.at(server).version), server));
            }
        }
        std::sort(sorted_candidates.begin(), sorted_candidates.end());

        /* Second, determine which servers are eligible to become primary. */
        std::vector<server_id_t> eligible_candidates;
        for (size_t i = 0; i < sorted_candidates.size(); ++i) {
            server_id_t server = sorted_candidates[i].second;
            /* `up_to_date_count` is the number of servers that `server` is at least as
            up-to-date as. We know `server` must be at least as up-to-date as itself and
            all of the servers that are earlier in the list. */
            size_t up_to_date_count = i + 1;
            /* If there are several servers with the same timestamp, they will appear
            together in the list. So `server` may be at least as up-to-date as some of
            the servers that appear after it in the list. */
            while (up_to_date_count < sorted_candidates.size() &&
                    sorted_candidates[up_to_date_count].first ==
                        sorted_candidates[i].first) {
                ++up_to_date_count;
            }
            /* OK, now `up_to_date_count` is the number of servers that this server is
            at least as up-to-date as. */
            if (up_to_date_count > new_c.voters.size() / 2) {
                eligible_candidates.push_back(server);
            }
        }

        if (!eligible_candidates.empty()) {
            contract_t::primary_t p;

            /* Select the primary. It's safe for us to pick any eligible candidate, but
            we should always pick `config.primary_replica` if it's available. Otherwise,
            we just pick the most up-to-date one. */
            auto it = std::find(eligible_candidates.begin(), eligible_candidates.end(),
                config.primary_replica);
            if (it != eligible_candidates.end()) {
                p.server = config.primary_replica;
            } else {
                /* `eligible_candidates` is ordered by how up-to-date they are */
                p.server = eligible_candidates.back();
            }

            /* RSI(raft): If `config.primary_replica` isn't connected or isn't ready, we
            should elect a different primary. If `config.primary_replica` is connected
            but just takes a little bit longer to reply to our contracts than the other
            replicas, we should wait for it to reply to our contracts and then elect it.
            Under the current implementation, we don't wait. This could lead to an
            awkward loop, where we elect the wrong primary, then un-elect it because we
            realize `config.primary_replica` is ready, and then re-elect the wrong
            primary instead of electing `config.primary_replica`. */

            new_c.primary = boost::make_optional(p);
        }
    }

    /* Sometimes we already have a primary, but we need to pick a different one. There
    are three such situations:
    - The existing primary is disconnected
    - The existing primary isn't `config.primary_replica`, and `config.primary_replica`
        is ready to take over the role
    - `config.primary_replica` isn't ready to take over the role, but the existing
        primary isn't even supposed to be a replica anymore.
    In the first situation, we'll simply remove `c.primary`. In the second and third
    situations, we'll first set `c.primary->warm_shutdown` to `true`, and then only
    once the primary acknowledges that, we'll remove `c.primary`. Either way, once the
    replicas acknowledge the contract in which we removed `c.primary`, the logic earlier
    in this function will select a new primary. Note that we can't go straight from the
    old primary to the new one; we need a majority of replicas to promise to stop
    receiving updates from the old primary before it's safe to elect a new one. */
    if (static_cast<bool>(old_c.primary)) {
        /* Note we already checked for the case where the old primary wasn't supposed to
        be a replica. If this is so, then `should_kill_primary` will already be set to
        `true`. */

        /* Check if we need to do an auto-failover. The precise form of this condition
        isn't important for correctness. If we do an auto-failover when the primary isn't
        actually dead, or don't do an auto-failover when the primary is actually dead,
        the worst that will happen is we'll lose availability.

        RSI(raft): Maybe wait until the primary stays down for a certain amount of time
        before failing over. */
        size_t voters_cant_reach_primary = 0;
        for (const server_id_t &server : new_c.voters) {
            if (acks.count(server) == 1 && acks.at(server).state ==
                    contract_ack_t::state_t::secondary_need_primary) {
                ++voters_cant_reach_primary;
            }
        }
        if (voters_cant_reach_primary > new_c.voters.size() / 2) {
            should_kill_primary = true;
        }

        if (should_kill_primary) {
            new_c.primary = boost::none;
        } else if (old_c.primary->server != config.primary_replica &&
                acks.count(config.primary_replica) == 1 &&
                acks.at(config.primary_replica).state ==
                    contract_ack_t::state_t::secondary_streaming) {
            /* The old primary is still a valid replica, but it isn't equal to
            `config.primary_replica`. So we have to do a hand-over to ensure that after
            we kill the primary, `config.primary_replica` will be a valid candidate. */

            if (old_c.primary->hand_over == config.primary_replica &&
                    acks.count(old_c.primary->server) == 1 &&
                    acks.at(old_c.primary->server).state ==
                        contract_ack_t::state_t::primary_ready) {
                /* We already did the hand over. Now it's safe to stop the old primary.
                The new primary will be started later, after a majority of the replicas
                acknowledge that they are no longer listening for writes from the old
                primary. */
                new_c.primary = boost::none;
            } else {
                new_c.primary->hand_over = boost::make_optional(config.primary_replica);
            }
        } else {
            /* We're sticking with the current primary, so `hand_over` should be empty.
            In the unlikely event that we were in the middle of a hand-over and then
            changed our minds, it might not be empty, so we clear it manually. */
            new_c.primary->hand_over = boost::none;
        }
    }

    /* Register a branch if a primary is asking us to */
    if (static_cast<bool>(old_c.primary) &&
            static_cast<bool>(new_c.primary) &&
            old_c.primary->server == new_c.primary->server &&
{'fraction_non_alphanumeric': 0.08590057612032981, 'fraction_numerical': 0.001586918273708904, 'mean_word_length': 2.3703057783978605, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 91, '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': '2484527', 'n_tokens_mistral': 8394, 'n_tokens_neox': 8073, 'n_words': 2819}