Implémentation de Raft en langage C (avec code)

Implémentation Raft en langage C (avec code)

Introduction

Cet article présente une implémentation simple de Raft. Si vous avez lu l'article Raft, il vous sera plus facile de lire cette implémentation de Raft, car les détails de l'implémentation sont décrits en détail dans l'article Raft. L'implémentation technique consiste essentiellement à réexprimer la description dans l'article Raft. en langage de programmation. C'est le plus grand avantage de Raft par rapport à Paxos, c'est-à-dire qu'il est facile à comprendre et à mettre en œuvre. L'implémentation Raft présentée dans cet article est codée en langage C. À l'exception de la fonction de compression des journaux, d'autres fonctionnalités sont implémentées. Le mécanisme de changement de membre est également relativement simple et ne prend en charge qu'un seul changement de configuration à la fois. Pour les principes de Raft, vous pouvez lire le document Raft et « Raft Understanding ».

2. Concepts de base du Raft

2.1 Statut

Le radeau a trois états : Leader, Candidat et Suiveur. Les transitions entre ces trois états sont représentées dans la figure ci-dessous. Seul le Leader a le droit de traiter les demandes des clients et de copier les journaux vers les Followers. Le candidat est un état intermédiaire entre le suiveur et le leader. Lorsqu'il n'y a pas de leader dans le cluster, le suiveur entre dans l'état candidat et lance un vote pour le cluster. Le suiveur qui obtient la majorité des voix deviendra le leader.

2.2 Messages

Afin d'améliorer la compréhensibilité du protocole, Raft a simplifié le paramétrage des types de messages et n'a que les deux requêtes suivantes. [indent]
requestVote lance une demande de vote. Demande du candidat lors du lancement du vote. Il est reçu et traité par les autres abonnés et candidats du cluster.
appendEntries Ajouter une demande de journal. La demande émise par le Leader lors de l'ajout de logs au Follower. [/indent]

2.3 Numéro de terme

Le terme numéro de terme est utilisé dans le protocole Raft pour indiquer l'ancienne et la nouvelle relation de temps. Cette valeur de terme dans le. Le mandat de chaque leader est immuable, il est absolument différent selon les différents leaders et augmente de façon monotone avec le temps. Si la durée de la requête A est supérieure à celle de la requête B, la requête B est obsolète.

3. Implémentation du radeau

3.1 Protocole

Introduisez d'abord quatre structures de données importantes, correspondant à celles mentionnées ci-dessus requestVote et appendEntries demande et réponse.

/** requestVote 请求投票
   * 竞选者Candidate去竞选Leader时发送给其它node的投票请求。
   * 其它Leader或者Candidate收到term比自己大的投票请求时，会自动变成Follower*/
typedef struct
{
    /** 当前任期号，通过任期号的大小与其它Candidate竞争Leader */
    int term;

    /** 竞选者的id */
    int candidate_id;

    /** 竞选者本地保存的最新一条日志的index */
    int last_log_idx;

    /** 竞选者本地保存的最新一条日志的任期号*/
    int last_log_term;
} msg_requestvote_t;


/** 投票请求的回复response.
  * 该response主要是给返回某个node是否接收了Candidate的投票请求. */
typedef struct
{
    /** node的任期号，Candidate根据投票结果和node的任期号来更新自己的任期号 */
    int term;

    /** 投票结果，如果node给Candidate投票则为true */
    int vote_granted;
} msg_requestvote_response_t;

/**  添加日志请求.
  * Follower可以从该消息中知道哪些日志可以安全地提交到状态机FSM中去。
  * Leader可以将该消息作为心跳消息定期发送。
  * 旧的Leader和Candidate收到该消息后可能会自动变成Follower */
typedef struct
{
    /** Leader当前的任期号 */
    int term;

    /** 最新日志的前一条日志的index，用于Follower确认与Leader的日志完全一致 */
    int prev_log_idx;

    /** 最新日志的前一条日志的任期号term */
    int prev_log_term;

    /** leader当前已经确认提交到状态机FSM的日志索引index，这意味着Follower也可以安全地将该索引index以前的日志提交 */
    int leader_commit;

    /** 这条添加日志消息携带的日志条数，该实现中最多只有一条 */
    int n_entries;

    /** 这条添加日志消息中携带的日志数组 */
    msg_entry_t* entries;
} msg_appendentries_t;

/** 添加日志回复.
 * 旧的Leader或Candidate收到该消息会变成Follower *／
typedef struct
{
    /** 当前任期号 */
    int term;

    /** node成功添加日志时返回ture，即prev_log_index和prev_log_term都比对成功。否则返回false */
    int success;

    /* 下面两个字段不是Raft论文中规定的字段:
    /* 用来优化日志追加过程，以加速日志的追加。Raft原文中的追加过程是一次只能追加一条日志*/

    /** 处理添加日志请求后本地的最大日志索引 */
    int current_idx;

    /** 从添加日志请求中接受的第一条日志索引 */
    int first_idx;
} msg_appendentries_response_t;

3.2 Deux abstractions importantes

raft_server_private_t Cette structure est l'abstraction dans l'implémentation de Raft Enregistre le statut et tout. données requises lors de l’exécution du protocole Raft.

typedef struct {
    /* 所有服务器比较固定的状态: */

    /* 服务器最后一次知道的任期号（初始化为 0，持续递增） */
    int current_term;

    /* 记录在当前分期内给哪个Candidate投过票，
       */
    int voted_for;

    /* 日志条目集；每一个条目包含一个用户状态机执行的指令，和收到时的任期号 */
    void* log;

    /* 变动比较频繁的变量: */

    /* 已知的最大的已经被提交的日志条目的索引值 */
    int commit_idx;

    /* 最后被应用到状态机的日志条目索引值（初始化为 0，持续递增） */
    int last_applied_idx;

    /* 三种状态：follower/leader/candidate */
    int state;

    /* 计时器，周期函数每次执行时会递增改值 */
    int timeout_elapsed;

    raft_node_t* nodes;
    int num_nodes;

    int election_timeout;
    int request_timeout;

    /* 保存Leader的信息，没有Leader时为NULL */
    raft_node_t* current_leader;

    /* callbacks，由调用该raft实现的调用者来实现，网络IO和持久存储
     * 都由调用者在callback中实现 */
    raft_cbs_t cb;
    void* udata;

    /* 自己的信息 */
    raft_node_t* node;

    /* 该raft实现每次只进行一个服务器的配置更改，该变量记录raft server
     * 是否正在进行配置更改*/
    int voting_cfg_change_log_idx;
} raft_server_private_t;

raft_node_private_t Le corps abstrait du nœud de machine dans le cluster, y compris les informations sur les autres machines qui doivent être enregistrées lors de l'exécution du protocole raft

typedef struct
{
    void* udata;  /*一般保存与其它机器的连接信息，由使用者决定怎么实现连接*/

    int next_idx; /*对于每一个服务器，需要发送给他的下一个日志条目的索引值（初始化为领导人最后索引值加一）*/
    int match_idx; /*对于每一个服务器，已经复制给他的日志的最高索引值*/

    int flags; /*有三种取值，是相或的关系 1:该机器有给我投票 2:该机器有投票权  3: 该机器有最新的日志*/

    int id; /*机器对应的id值，这个每台机器在全局都是唯一的*/
} raft_node_private_t;

3.3 Processus du protocole Raft

Fonction périodique Raft doit faire quelque chose périodiquement. Par exemple, le leader doit périodiquement ajouter des journaux à d'autres serveurs pour donner aux serveurs avec des journaux en retard. une chance de rattraper son retard ; tous les serveurs doivent appliquer périodiquement les journaux de validation confirmés à la machine d'état, etc.
La fonction raft_periodic est une fonction qui est appelée périodiquement dans l'implémentation de raft, et la période d'appel est de 1000 ms. La machine fera différentes choses dans cette fonction dans différents états. Leader synchronise périodiquement les journaux avec Follower. Le suiveur détecte périodiquement s'il n'a pas reçu un paquet de battements de cœur du leader dans un délai déterminé. Si tel est le cas, il devient candidat et commence à voter pour le leader. Le leader et le suiveur valideront périodiquement les journaux soumis à la machine à états FSM.

/** raft周期性执行的函数,实现raft中的定时器以及定期应用日志到状态机
  */
int raft_periodic(raft_server_t* me_, int msec_since_last_period)
{
    raft_server_private_t* me = (raft_server_private_t*)me_;

    /* 选举计时器；Follower每次收到Leader的心跳后会重置清0，Leader每次发送日志也会清0 */
    me->timeout_elapsed += msec_since_last_period;

    /* Leader周期性地向Follower同步日志 */
    if (me->state == RAFT_STATE_LEADER) 
    {
        if (me->request_timeout <= me->timeout_elapsed)
            raft_send_appendentries_all(me_);
    }
    /* Follower检测选举计时器是否超时 */
    else if (me->election_timeout <= me->timeout_elapsed)
    {
        if (1 < me->num_nodes)
            raft_election_start(me_);
    }

    /* 周期性地将已经确认commit的日志应用到状态机FSM */
    if (me->last_applied_idx < me->commit_idx)
        if (-1 == raft_apply_entry(me_))
            return -1;

    return 0;
}

Devenez candidat Candidat Chaque serveur du cluster dispose d'un minuteur d'élection Lorsqu'un serveur ne reçoit pas de battement de cœur du leader dans le délai d'expiration du minuteur, le cluster est considéré s'il y en a. s'il n'y a pas de leader sur le serveur ou si le leader est en panne, le serveur deviendra candidat, puis lancera un vote pour se présenter comme leader. La fonction raft_become_candidate suivante est la fonction permettant au serveur de devenir candidat. La fonction fait principalement ce qui suit. choses :

Incrémentez le numéro du mandat actuel (currentTerm)

Votez pour vous-même

Réinitialisez le délai d'attente des élections

Envoyer Le RPC demandant le vote est envoyé à tous les autres serveurs

/** Follower成为Candidate执行的函数
  */
void raft_become_candidate(raft_server_t* me_)
{
    raft_server_private_t* me = (raft_server_private_t*)me_;
    int i;

    /*自增当前的任期号；给自己投票，设置自己的状态为CANDIDATE*/
    raft_set_current_term(me_, raft_get_current_term(me_) + 1);
    for (i = 0; i < me->num_nodes; i++)
        raft_node_vote_for_me(me->nodes[i], 0);
    raft_vote(me_, me->node);
    me->current_leader = NULL;
    raft_set_state(me_, RAFT_STATE_CANDIDATE);

    /* 重置选举超时计时器。为了防止多个Candidate竞争，将下一次发起投票的时间间隔设置成随机值*/
    /* TODO: this should probably be lower */
    me->timeout_elapsed = rand() % me->election_timeout;

    /*发送请求投票的 RPC 给其他所有服务器*/
    for (i = 0; i < me->num_nodes; i++)
        if (me->node != me->nodes[i] && raft_node_is_voting(me->nodes[i]))
            raft_send_requestvote(me_, me->nodes[i]);
}

Traitement des demandes de vote La logique du traitement des demandes de vote est principalement de juger s'il faut accepter de voter. le jugement est le numéro de terme et les informations de journal dans la demande. Le niveau de fraîcheur et d'âge, et si vous avez voté pour d'autres serveurs avec le même numéro de terme. Si vous avez déjà voté, vous ne pouvez pas voter à nouveau. .

Si term > currentTerm, passez en mode Follower.
Le serveur qui reçoit la demande de vote ici peut être un Leader avec de mauvaises conditions de réseau ou un Candidat qui n'a pas encore eu le temps d'émettre une demande de vote. Après avoir reçu une demande avec un numéro de mandat plus récent que le sien, il doit le faire. devenez inconditionnellement Follower pour vous assurer qu'un seul leader existe

如果term daab63a7de571b85be82bb1122a6d0ecterm即为过时)

follower是否成功添加日志，如果添加失败，则减小发给follower的日志索引nextIndex再重试；如果添加成功则更新本地记录的follower日志信息，并检查日志是否最新，如果不是最新则继续发送添加日志请求。

新机器的日志添加，详见3.4节-- 成员变更

/** 处理添加日志请求回复
  * /
int raft_recv_appendentries_response(raft_server_t* me_,
                                     raft_node_t* node,
                                     msg_appendentries_response_t* r)
{
    raft_server_private_t* me = (raft_server_private_t*)me_;

    __log(me_, node,
          "received appendentries response %s ci:%d rci:%d 1stidx:%d",
          r->success == 1 ? "SUCCESS" : "fail",
          raft_get_current_idx(me_),
          r->current_idx,
          r->first_idx);

    /* 过时的回复 -- 忽略 */
    if (r->current_idx != 0 && r->current_idx <= raft_node_get_match_idx(node))
        return 0;

    /* oh~我不是Leader */
    if (!raft_is_leader(me_))
        return -1;

    /* 回复中的term比自己的要大，说明自己是一个过时的Leader，无条件转为Follower */
    if (me->current_term < r->term)
    {
        raft_set_current_term(me_, r->term);
        raft_become_follower(me_);
        return 0;
    }
    /* 过时的回复，网络状况不好时会出现 */
    else if (me->current_term != r->term)
        return 0;

    /* stop processing, this is a node we don&#39;t have in our configuration */
    if (!node)
        return 0;

    /* 由于日志不一致导致添加日志不成功*/
    if (0 == r->success)
    {
        assert(0 <= raft_node_get_next_idx(node));

        /* 将nextIdex减*/
        int next_idx = raft_node_get_next_idx(node);
        assert(0 <= next_idx);
        /* Follower的日志数量还远远少于Leader，将nextIdex设为回复中的current_idx+1和Leader
         * 当前索引中较小的一个，一般回复中的current_idx+1会比较小*/
        if (r->current_idx < next_idx - 1)
            raft_node_set_next_idx(node, min(r->current_idx + 1, raft_get_current_idx(me_)));
        /* Follower的日志数量和Leader差不多，但是比对前一条日志时失败，这种情况将next_idx减1
         * 重试*/
        else
            raft_node_set_next_idx(node, next_idx - 1);

        /* 使用更新后的nextIdx重新发送添加日志请求 */
        raft_send_appendentries(me_, node);
        return 0;
    }

    assert(r->current_idx <= raft_get_current_idx(me_));

    /* 下面处理添加日志请求的情况 */
    /* 更新本地记录的Follower的日志情况 */
    raft_node_set_next_idx(node, r->current_idx + 1);
    raft_node_set_match_idx(node, r->current_idx);

    /* 如果是新加入的机器，则判断它的日志是否是最新，如果达到了最新，则赋予它投票权，
     * 这里逻辑的详细解释在第3.4节 -- 成员变更*/
    if (!raft_node_is_voting(node) &&
        -1 == me->voting_cfg_change_log_idx &&
        raft_get_current_idx(me_) <= r->current_idx + 1 &&
        me->cb.node_has_sufficient_logs &&
        0 == raft_node_has_sufficient_logs(node)
        )
    {
        raft_node_set_has_sufficient_logs(node);
        me->cb.node_has_sufficient_logs(me_, me->udata, node);
    }

    /* 如果一条日志回复成功的数量超过一半，则将日志提交commit，即允许应用到状态机 */
    int votes = 1; /* include me */
    int point = r->current_idx;
    int i;
    for (i = 0; i < me->num_nodes; i++)
    {
        if (me->node == me->nodes[i] || !raft_node_is_voting(me->nodes[i]))
            continue;

        int match_idx = raft_node_get_match_idx(me->nodes[i]);

        if (0 < match_idx)
        {
            raft_entry_t* ety = raft_get_entry_from_idx(me_, match_idx);
            /*如果follower已经添加了索引大于等于r->current_idx的日志，则vote加1*/
            if (ety->term == me->current_term && point <= match_idx)
                votes++;
        }
    }

    /* 投票数大于所有服务器的一半，则将日志提交 */
    if (me->num_nodes / 2 < votes && raft_get_commit_idx(me_) < point)
        raft_set_commit_idx(me_, point);

    /* 如果follower的日志还没有最新，那么继续发送添加日志请求 */
    if (raft_get_entry_from_idx(me_, raft_node_get_next_idx(node)))
        raft_send_appendentries(me_, node);

    /* periodic applies committed entries lazily */

    return 0;
}

3.3 成员变更

成员的变更都是以日志的形式下发的。添加的新成员分两阶段进行，第一阶段中新成员没有有投票权，但是有接收日志的权力；当它的日志同步到最新后就进入到第二阶段，由Leader赋予投票权，从而成为集群中完整的一员。删除成员相对比较简单，所有服务器收到删除成员的日志后，立马将该成员的信息从本地抹除。

添加成员过程

1. 管理员向Leader发送添加成员命令
   

2. Leader添加一条 RAFT_LOGTYPE_ADD_NONVOTING_NODE日志，即添加没有投票权的服务器。该日志与其它普通日志一样同步给集群中其它服务器。收到该日志的服务器在本地保存该新成员的信息。
   

3. 当新成员的日志同步到最新后，Leader添加一条 RAFT_LOGTYPE_ADD_NODE日志，即有投票权的服务器，同样地，该日志与其它普通日志一样同步给集群中其它服务器。收到该日志的服务器在本地保存该新成员的信息，以后的投票活动会将新成员考虑进去。

删除成员过程

1. 管理员向Leader发送删除成员命令。
   

2. Leader添加一条 RAFT_LOGTYPE_REMOVE_NODE 日志，并跟普通日志一样同步给其它服务器。收到该日志的服务器立即将被成员信息从本地删除。

感谢大家的阅读，希望大家收益多多。

推荐教程：《C语言》

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!