<commit_before>/*
 * Common functionalities for centralized fault tolerant components.
 *
 * Copyright (C) 2018 Nextworks
 * Author: <NAME> <<EMAIL>>
 *
 * This file is part of rlite.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#include <vector>
#include <map>

#include "uipcp-normal-ceft.hpp"
#include "Raft.pb.h"

namespace rlite {

std::string CeftReplica::ReqVoteObjClass           = "raft_rv";
std::string CeftReplica::ReqVoteRespObjClass       = "raft_rv_r";
std::string CeftReplica::AppendEntriesObjClass     = "raft_ae";
std::string CeftReplica::AppendEntriesRespObjClass = "raft_ae_r";

int
CeftReplica::init(const std::list<raft::ReplicaId> &peers)
{
    set_verbosity(raft::RaftSM::kVerboseInfo);

    raft::RaftSMOutput out;
    if (RaftSM::init(peers, &out)) {
        UPE(rib->uipcp, "Failed to init Raft state machine for DFT\n");
        return -1;
    }
    UPI(rib->uipcp, "Raft replica initialized\n");
    return process_sm_output(std::move(out));
}
int
CeftReplica::process_sm_output(raft::RaftSMOutput out)
{
    int ret = 0;

    /* Convert raft protocol messages from the library format to the
     * gpb format. */
    for (auto &pair : out.output_messages) {
        raft::RaftMessage *msg = pair.second.get();
        const auto *rv  = dynamic_cast<const raft::RaftRequestVote *>(msg);
        const auto *rvr = dynamic_cast<const raft::RaftRequestVoteResp *>(msg);
        auto *ae        = dynamic_cast<raft::RaftAppendEntries *>(msg);
        const auto *aer =
            dynamic_cast<const raft::RaftAppendEntriesResp *>(msg);
        auto m = utils::make_unique<CDAPMessage>();
        std::unique_ptr<::google::protobuf::MessageLite> obj;
        std::string obj_class;

        if (rv) {
            auto mm = utils::make_unique<gpb::RaftRequestVote>();
            mm->set_term(rv->term);
            mm->set_candidate_id(rv->candidate_id);
            mm->set_last_log_index(rv->last_log_index);
            mm->set_last_log_term(rv->last_log_term);
            obj       = std::move(mm);
            obj_class = ReqVoteObjClass;
        } else if (rvr) {
            auto mm = utils::make_unique<gpb::RaftRequestVoteResp>();
            mm->set_term(rvr->term);
            mm->set_vote_granted(rvr->vote_granted);
            obj       = std::move(mm);
            obj_class = ReqVoteRespObjClass;
        } else if (ae) {
            auto mm = utils::make_unique<gpb::RaftAppendEntries>();
            mm->set_term(ae->term);
            mm->set_leader_id(ae->leader_id);
            mm->set_leader_commit(ae->leader_commit);
            mm->set_prev_log_index(ae->prev_log_index);
            mm->set_prev_log_term(ae->prev_log_term);
            for (const auto &p : ae->entries) {
                gpb::RaftLogEntry *ge = mm->add_entries();
                ge->set_term(p.first);
                ge->set_buffer(p.second.get(), CommandSize);
            }
            obj       = std::move(mm);
            obj_class = AppendEntriesObjClass;
        } else if (aer) {
            auto mm = utils::make_unique<gpb::RaftAppendEntriesResp>();
            mm->set_term(aer->term);
            mm->set_follower_id(aer->follower_id);
            mm->set_log_index(aer->log_index);
            mm->set_success(aer->success);
            obj       = std::move(mm);
            obj_class = AppendEntriesRespObjClass;
        } else {
            assert(false);
        }

        m->m_write(obj_class, RibObjName);
        ret |=
            rib->send_to_dst_node(std::move(m), pair.first, obj.get(), nullptr);
    }

    /* Here we make an assumption about the raft library, that is one
     * and only one timer is active at all times; so for example if the
     * heartbeat timer is active, the leader election timer can't be active.
     * As a result we have just one 'timer' class member (Replica::timer).
     * We first look at Stop commands, and then at Restart commands; this is
     * needed to prevent Stop commands to cancel the effect of Restart ones. */
    for (const auto &cmd : out.timer_commands) {
        if (cmd.action == raft::RaftTimerAction::Stop) {
            timer      = nullptr;
            timer_type = raft::RaftTimerType::Invalid;
            /* We can break, as more Stop commands won't have any effect. */
            break;
        }
    }
    for (const auto &cmd : out.timer_commands) {
        if (cmd.action == raft::RaftTimerAction::Restart) {
            /* The following assertion will fail if our assumption about
             * the unicity of the timer is not true. */
            assert(timer == nullptr || timer_type == cmd.type);
            timer = utils::make_unique<TimeoutEvent>(
                cmd.milliseconds, rib->uipcp, this,
                [](struct uipcp *uipcp, void *arg) {
                    auto replica = static_cast<CeftReplica *>(arg);
                    replica->timer->fired();
                    replica->process_timeout();
                });
            timer_type = cmd.type;
        }
    }

    return ret;
}

int
CeftReplica::process_timeout()
{
    std::lock_guard<std::mutex> guard(rib->mutex);
    raft::RaftSMOutput out;

    timer_expired(timer_type, &out);

    return process_sm_output(std::move(out));
}

/* Apply a command to the replicated state machine. We just pass the command
 * to the same multimap implementation used by the fully replicated DFT. */
int
CeftReplica::apply(raft::LogIndex index, raft::Term term,
                   const char *const serbuf)
{
    std::unique_ptr<CDAPMessage> rm;

    /* Check if the committe entry has an associated client request. */
    auto mit = pending.find(index);
    if (mit != pending.end()) {
        rm = std::move(mit->second->m);
    }

    /* Invoke the actual state machine update. This may modify the response (if
     * any) with the result of the update. */
    apply(serbuf, rm.get());

    if (mit != pending.end()) {
        /* We must check that the pending response really matches the committed
         * entry, i.e., that we were the leader that submitted the entry. This
         * is true if and only if the term with which this entry was committed
         * matches the one that we saved at submit time (when the pending
         * request was created). This safety property follows from the guarantee
         * that each term has one and only one leader. */
        if (mit->second->term == term) {
            /* Send the response to the client and flush the pending response.
             */
            int invoke_id = rm->invoke_id;
            rib->send_to_dst_addr(std::move(rm), mit->second->requestor_addr);
            UPD(rib->uipcp,
                "Pending response for index %u sent to client %llu "
                "(invoke_id=%d)\n",
                index, (long long unsigned)mit->second->requestor_addr,
                invoke_id);
        }
        pending.erase(index);<commit_msg>user: ceft: small cosmetic improvements<commit_after>/*
 * Common functionalities for centralized fault tolerant components.
 *
 * Copyright (C) 2018 Nextworks
 * Author: <NAME> <<EMAIL>>
 *
 * This file is part of rlite.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 */

#include <vector>
#include <map>

#include "uipcp-normal-ceft.hpp"
#include "Raft.pb.h"

namespace rlite {

std::string CeftReplica::ReqVoteObjClass           = "raft_rv";
std::string CeftReplica::ReqVoteRespObjClass       = "raft_rv_r";
std::string CeftReplica::AppendEntriesObjClass     = "raft_ae";
std::string CeftReplica::AppendEntriesRespObjClass = "raft_ae_r";

int
CeftReplica::init(const std::list<raft::ReplicaId> &peers)
{
    raft::RaftSMOutput out;

    set_verbosity(raft::RaftSM::kVerboseInfo);

    if (RaftSM::init(peers, &out)) {
        UPE(rib->uipcp, "Failed to init Raft state machine\n");
        return -1;
    }
    UPI(rib->uipcp, "Raft replica initialized\n");

    return process_sm_output(std::move(out));
}
int
CeftReplica::process_sm_output(raft::RaftSMOutput out)
{
    int ret = 0;

    /* Convert raft protocol messages from the library format to the
     * gpb format. */
    for (auto &pair : out.output_messages) {
        raft::RaftMessage *msg = pair.second.get();
        const auto *rv  = dynamic_cast<const raft::RaftRequestVote *>(msg);
        const auto *rvr = dynamic_cast<const raft::RaftRequestVoteResp *>(msg);
        auto *ae        = dynamic_cast<raft::RaftAppendEntries *>(msg);
        const auto *aer =
            dynamic_cast<const raft::RaftAppendEntriesResp *>(msg);
        auto m = utils::make_unique<CDAPMessage>();
        std::unique_ptr<::google::protobuf::MessageLite> obj;
        std::string obj_class;

        if (rv) {
            auto mm = utils::make_unique<gpb::RaftRequestVote>();
            mm->set_term(rv->term);
            mm->set_candidate_id(rv->candidate_id);
            mm->set_last_log_index(rv->last_log_index);
            mm->set_last_log_term(rv->last_log_term);
            obj       = std::move(mm);
            obj_class = ReqVoteObjClass;
        } else if (rvr) {
            auto mm = utils::make_unique<gpb::RaftRequestVoteResp>();
            mm->set_term(rvr->term);
            mm->set_vote_granted(rvr->vote_granted);
            obj       = std::move(mm);
            obj_class = ReqVoteRespObjClass;
        } else if (ae) {
            auto mm = utils::make_unique<gpb::RaftAppendEntries>();
            mm->set_term(ae->term);
            mm->set_leader_id(ae->leader_id);
            mm->set_leader_commit(ae->leader_commit);
            mm->set_prev_log_index(ae->prev_log_index);
            mm->set_prev_log_term(ae->prev_log_term);
            for (const auto &p : ae->entries) {
                gpb::RaftLogEntry *ge = mm->add_entries();
                ge->set_term(p.first);
                ge->set_buffer(p.second.get(), CommandSize);
            }
            obj       = std::move(mm);
            obj_class = AppendEntriesObjClass;
        } else if (aer) {
            auto mm = utils::make_unique<gpb::RaftAppendEntriesResp>();
            mm->set_term(aer->term);
            mm->set_follower_id(aer->follower_id);
            mm->set_log_index(aer->log_index);
            mm->set_success(aer->success);
            obj       = std::move(mm);
            obj_class = AppendEntriesRespObjClass;
        } else {
            assert(false);
        }

        m->m_write(obj_class, RibObjName);
        ret |=
            rib->send_to_dst_node(std::move(m), pair.first, obj.get(), nullptr);
    }

    /* Here we make an assumption about the raft library, that is one
     * and only one timer is active at all times; so for example if the
     * heartbeat timer is active, the leader election timer can't be active.
     * As a result we have just one 'timer' class member (Replica::timer).
     * We first look at Stop commands, and then at Restart commands; this is
     * needed to prevent Stop commands to cancel the effect of Restart ones. */
    for (const auto &cmd : out.timer_commands) {
        if (cmd.action == raft::RaftTimerAction::Stop) {
            timer      = nullptr;
            timer_type = raft::RaftTimerType::Invalid;
            /* We can break, as more Stop commands won't have any effect. */
            break;
        }
    }
    for (const auto &cmd : out.timer_commands) {
        if (cmd.action == raft::RaftTimerAction::Restart) {
            /* The following assertion will fail if our assumption about
             * the unicity of the timer is not true. */
            assert(timer == nullptr || timer_type == cmd.type);
            timer = utils::make_unique<TimeoutEvent>(
                cmd.milliseconds, rib->uipcp, this,
                [](struct uipcp *uipcp, void *arg) {
                    auto replica = static_cast<CeftReplica *>(arg);
                    replica->timer->fired();
                    replica->process_timeout();
                });
            timer_type = cmd.type;
        }
    }

    return ret;
}

int
CeftReplica::process_timeout()
{
    std::lock_guard<std::mutex> guard(rib->mutex);
    raft::RaftSMOutput out;

    timer_expired(timer_type, &out);

    return process_sm_output(std::move(out));
}

/* Apply a command to the replicated state machine. */
int
CeftReplica::apply(raft::LogIndex index, raft::Term term,
                   const char *const serbuf)
{
    std::unique_ptr<CDAPMessage> rm;

    /* Check if the committed entry has an associated client request. */
    auto mit = pending.find(index);
    if (mit != pending.end()) {
        rm = std::move(mit->second->m);
    }

    /* Invoke the actual state machine update. This may modify the response (if
     * any) with the result of the update. */
    apply(serbuf, rm.get());

    if (mit != pending.end()) {
        /* We must check that the pending response really matches the committed
         * entry, i.e., that we were the leader that submitted the entry. This
         * is true if and only if the term with which this entry was committed
         * matches the one that we saved at submit time (when the pending
         * request was created). This safety property follows from the guarantee
         * that each term has one and only one leader. */
        if (mit->second->term == term) {
            /* Send the response to the client and flush the pending response.
             */
            int invoke_id = rm->invoke_id;
            rib->send_to_dst_addr(std::move(rm), mit->second->requestor_addr);
            UPD(rib->uipcp,
                "Pending response for index %u sent to client %llu "
                "(invoke_id=%d)\n",
                index, (long long unsigned)mit->second->requestor_addr,
                invoke_id);
        }
        pending.erase(index);
{'fraction_non_alphanumeric': 0.11191216350287718, 'fraction_numerical': 0.002513393742972419, 'mean_word_length': 2.420814479638009, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 151, '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': '4803464', 'n_tokens_mistral': 4526, 'n_tokens_neox': 4410, 'n_words': 1287}