Unverified Commit 676c73d2 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Check quorum (#37)

Implemented check-quorum
parent 933828d1
...@@ -68,6 +68,12 @@ typedef struct { ...@@ -68,6 +68,12 @@ typedef struct {
int election_timeout_rand; int election_timeout_rand;
int request_timeout; int request_timeout;
/* timer interval to check if we still have quorum */
long quorum_timeout;
/* latest quorum id for the previous quorum_timeout round */
raft_msg_id_t last_acked_msg_id;
/* what this node thinks is the node ID of the current leader, or NULL if /* what this node thinks is the node ID of the current leader, or NULL if
* there isn't a known current leader. */ * there isn't a known current leader. */
raft_node_t* current_leader; raft_node_t* current_leader;
...@@ -113,6 +119,8 @@ int raft_become_candidate(raft_server_t* me); ...@@ -113,6 +119,8 @@ int raft_become_candidate(raft_server_t* me);
void raft_randomize_election_timeout(raft_server_t* me_); void raft_randomize_election_timeout(raft_server_t* me_);
void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id);
/** /**
* @return 0 on error */ * @return 0 on error */
int raft_send_requestvote(raft_server_t* me, raft_node_t* node); int raft_send_requestvote(raft_server_t* me, raft_node_t* node);
......
...@@ -13,12 +13,9 @@ ...@@ -13,12 +13,9 @@
#include <stdio.h> #include <stdio.h>
#include <assert.h> #include <assert.h>
#include <stdint.h> #include <stdint.h>
/* for varags */
#include <stdarg.h> #include <stdarg.h>
#include "raft.h" #include "raft.h"
#include "raft_log.h"
#include "raft_private.h" #include "raft_private.h"
#ifndef min #ifndef min
...@@ -75,24 +72,37 @@ void raft_randomize_election_timeout(raft_server_t* me_) ...@@ -75,24 +72,37 @@ void raft_randomize_election_timeout(raft_server_t* me_)
raft_log(me_, NULL, "randomize election timeout to %d", me->election_timeout_rand); raft_log(me_, NULL, "randomize election timeout to %d", me->election_timeout_rand);
} }
void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
// Make sure that timeout is greater than 'randomized election timeout'
me->quorum_timeout = me->election_timeout * 2;
me->last_acked_msg_id = id;
}
raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
{ {
raft_server_private_t* me = raft_server_private_t* me = __raft_calloc(1, sizeof(raft_server_private_t));
(raft_server_private_t*)__raft_calloc(1, sizeof(raft_server_private_t));
if (!me) if (!me)
return NULL; return NULL;
me->current_term = 0; me->current_term = 0;
me->voted_for = -1; me->voted_for = -1;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
me->request_timeout = 200; me->request_timeout = 200;
me->election_timeout = 1000; me->election_timeout = 1000;
raft_update_quorum_meta((raft_server_t*)me, me->msg_id);
raft_randomize_election_timeout((raft_server_t*)me); raft_randomize_election_timeout((raft_server_t*)me);
me->log_impl = log_impl; me->log_impl = log_impl;
me->log = me->log_impl->init(me, log_arg); me->log = me->log_impl->init(me, log_arg);
if (!me->log) { if (!me->log) {
__raft_free(me); __raft_free(me);
return NULL; return NULL;
} }
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->current_leader = NULL;
...@@ -196,7 +206,9 @@ int raft_become_leader(raft_server_t* me_) ...@@ -196,7 +206,9 @@ int raft_become_leader(raft_server_t* me_)
} }
raft_set_state(me_, RAFT_STATE_LEADER); raft_set_state(me_, RAFT_STATE_LEADER);
raft_update_quorum_meta(me_, me->msg_id);
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
{ {
raft_node_t* node = me->nodes[i]; raft_node_t* node = me->nodes[i];
...@@ -258,6 +270,45 @@ void raft_become_follower(raft_server_t* me_) ...@@ -258,6 +270,45 @@ void raft_become_follower(raft_server_t* me_)
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
} }
static int msgid_cmp(const void *a, const void *b)
{
raft_msg_id_t va = *((raft_msg_id_t*) a);
raft_msg_id_t vb = *((raft_msg_id_t*) b);
return va > vb ? -1 : 1;
}
static raft_msg_id_t quorum_msg_id(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
raft_msg_id_t msg_ids[me->num_nodes];
int num_voters = 0;
for (int i = 0; i < me->num_nodes; i++) {
raft_node_t* node = me->nodes[i];
if (!raft_node_is_voting(node))
continue;
if (me->node == node) {
msg_ids[num_voters++] = me->msg_id;
} else {
msg_ids[num_voters++] = raft_node_get_last_acked_msgid(node);
}
}
assert(num_voters == raft_get_num_voting_nodes(me_));
/**
* Sort the acknowledged msg_ids in the descending order and return
* the median value. Median value means it's the highest msg_id
* acknowledged by the majority.
*/
qsort(msg_ids, num_voters, sizeof(raft_msg_id_t), msgid_cmp);
return msg_ids[num_voters / 2];
}
int raft_periodic(raft_server_t* me_, int msec_since_last_period) int raft_periodic(raft_server_t* me_, int msec_since_last_period)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -281,7 +332,36 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -281,7 +332,36 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
if (me->state == RAFT_STATE_LEADER) if (me->state == RAFT_STATE_LEADER)
{ {
if (me->request_timeout <= me->timeout_elapsed) if (me->request_timeout <= me->timeout_elapsed)
{
me->msg_id++;
raft_send_appendentries_all(me_); raft_send_appendentries_all(me_);
}
me->quorum_timeout -= msec_since_last_period;
if (me->quorum_timeout < 0)
{
/**
* Check-quorum implementation
*
* Periodically (every quorum_timeout), we enter this check to
* verify quorum exists. The current 'quorum msg id' should be
* greater than the last time we were here. It means we've got
* responses for append entry requests from the cluster, so we
* conclude that cluster is operational and quorum exists. In that
* case, we save the current quorum msg id and update the
* quorum_timeout timer. Otherwise, it means quorum does not exist.
* We should step down and become a follower.
*/
raft_msg_id_t quorum_id = quorum_msg_id(me_);
if (me->last_acked_msg_id == quorum_id)
{
raft_log(me_, NULL, "quorum does not exist, stepping down");
raft_become_follower(me_);
}
raft_update_quorum_meta(me_, quorum_id);
}
} }
else if (me->election_timeout_rand <= me->timeout_elapsed && else if (me->election_timeout_rand <= me->timeout_elapsed &&
/* Don't become the leader when building snapshots or bad things will /* Don't become the leader when building snapshots or bad things will
...@@ -1588,45 +1668,6 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len) ...@@ -1588,45 +1668,6 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
} }
} }
static int msgid_cmp(const void *a, const void *b)
{
raft_msg_id_t va = *((raft_msg_id_t*) a);
raft_msg_id_t vb = *((raft_msg_id_t*) b);
return va > vb ? -1 : 1;
}
raft_msg_id_t quorum_msg_id(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
raft_msg_id_t msg_ids[me->num_nodes];
int num_voters = 0;
for (int i = 0; i < me->num_nodes; i++) {
raft_node_t* node = me->nodes[i];
if (!raft_node_is_voting(node))
continue;
if (me->node == node) {
msg_ids[num_voters++] = me->msg_id;
} else {
msg_ids[num_voters++] = raft_node_get_last_acked_msgid(node);
}
}
assert(num_voters == raft_get_num_voting_nodes(me_));
/**
* Sort the acknowledged msg_ids in the descending order and return
* the median value. Median value means it's the highest msg_id
* acknowledged by the majority.
*/
qsort(msg_ids, num_voters, sizeof(raft_msg_id_t), msgid_cmp);
return msg_ids[num_voters / 2];
}
void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg) void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg)
{ {
raft_server_private_t* me = (raft_server_private_t*) me_; raft_server_private_t* me = (raft_server_private_t*) me_;
......
...@@ -7,22 +7,19 @@ ...@@ -7,22 +7,19 @@
* @author Willem Thiart himself@willemthiart.com * @author Willem Thiart himself@willemthiart.com
*/ */
#include <stdlib.h>
#include <string.h> #include <string.h>
#include <stdio.h>
#include <assert.h> #include <assert.h>
/* for varags */
#include <stdarg.h>
#include "raft.h" #include "raft.h"
#include "raft_log.h"
#include "raft_private.h" #include "raft_private.h"
void raft_set_election_timeout(raft_server_t* me_, int millisec) void raft_set_election_timeout(raft_server_t* me_, int millisec)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->election_timeout = millisec; me->election_timeout = millisec;
raft_update_quorum_meta(me_, me->last_acked_msg_id);
raft_randomize_election_timeout(me_); raft_randomize_election_timeout(me_);
} }
......
...@@ -3853,45 +3853,92 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc) ...@@ -3853,45 +3853,92 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1); raft_set_current_term(r, 1);
raft_set_election_timeout(r, 10000);
raft_set_request_timeout(r, 10000);
raft_become_leader(r); raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa"); __RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
raft_periodic(r, 1000); raft_periodic(r, 100);
raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val); raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 2000); raft_periodic(r, 200);
// Read request is pending as it requires two acks // Read request is pending as it requires two acks
CuAssertIntEquals(tc, 0, val); CuAssertIntEquals(tc, 0, val);
// Second node acknowledges reaq req // Second node acknowledges reaq req
raft_node_set_last_ack(raft_get_node(r, 2), 1, 1); raft_node_set_last_ack(raft_get_node(r, 2), 1, 1);
raft_periodic(r, 2000); raft_periodic(r, 200);
CuAssertIntEquals(tc, 1, val); CuAssertIntEquals(tc, 1, val);
val = 0; val = 0;
raft_add_node(r, NULL, 3, 0); raft_add_node(r, NULL, 3, 0);
raft_add_node(r, NULL, 4, 0); raft_add_node(r, NULL, 4, 0);
raft_periodic(r, 1000); raft_periodic(r, 100);
raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val); raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 2000); raft_periodic(r, 200);
// Read request is pending as it requires three acks // Read request is pending as it requires three acks
CuAssertIntEquals(tc, 0, val); CuAssertIntEquals(tc, 0, val);
// Second node acknowledges reaq req, // Second node acknowledges read req,
raft_node_set_last_ack(raft_get_node(r, 2), 2, 1); raft_node_set_last_ack(raft_get_node(r, 2), 2, 1);
raft_periodic(r, 2000); raft_periodic(r, 200);
CuAssertIntEquals(tc, 0, val); CuAssertIntEquals(tc, 0, val);
// Third node acknowledges reaq req // Third node acknowledges read req
raft_node_set_last_ack(raft_get_node(r, 3), 2, 1); raft_node_set_last_ack(raft_get_node(r, 3), 2, 1);
raft_periodic(r, 2000); raft_periodic(r, 200);
CuAssertIntEquals(tc, 1, val); CuAssertIntEquals(tc, 1, val);
} }
void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_set_request_timeout(r, 1000);
// Quorum timeout is twice the election timeout.
int quorum_timeout = raft_get_election_timeout(r) * 2;
raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 2, 1, "aaa");
raft_set_commit_idx(r, 2);
raft_periodic(r, 200);
CuAssertTrue(tc, raft_is_leader(r));
// Waiting more than quorum timeout will make leader step down.
raft_periodic(r, quorum_timeout + 1);
CuAssertTrue(tc, !raft_is_leader(r));
raft_node_set_last_ack(raft_get_node(r, 2), 1, 1);
raft_become_leader(r);
raft_periodic(r, 200);
CuAssertTrue(tc, raft_is_leader(r));
// Trigger new round of append entries
raft_periodic(r, raft_get_request_timeout(r) + 1);
CuAssertTrue(tc, raft_is_leader(r));
// If there is an ack from the follower, leader won't step down.
raft_node_set_last_ack(raft_get_node(r, 2), 2, 1);
raft_periodic(r, quorum_timeout);
CuAssertTrue(tc, raft_is_leader(r));
// No ack along quorum_timeout, leader will step down.
raft_periodic(r, quorum_timeout + 1);
CuAssertTrue(tc, !raft_is_leader(r));
}
int main(void) int main(void)
{ {
CuString *output = CuStringNew(); CuString *output = CuStringNew();
...@@ -4017,6 +4064,7 @@ int main(void) ...@@ -4017,6 +4064,7 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_read_action_callback); SUITE_ADD_TEST(suite, TestRaft_read_action_callback);
SUITE_ADD_TEST(suite, TestRaft_single_node_commits_noop); SUITE_ADD_TEST(suite, TestRaft_single_node_commits_noop);
SUITE_ADD_TEST(suite, TestRaft_quorum_msg_id_correctness); SUITE_ADD_TEST(suite, TestRaft_quorum_msg_id_correctness);
SUITE_ADD_TEST(suite, TestRaft_leader_steps_down_if_there_is_no_quorum);
CuSuiteRun(suite); CuSuiteRun(suite);
CuSuiteDetails(suite, output); CuSuiteDetails(suite, output);
......
...@@ -1067,10 +1067,7 @@ class RaftServer(object): ...@@ -1067,10 +1067,7 @@ class RaftServer(object):
change = ffi.from_handle(lib.raft_entry_getdata(ety)) change = ffi.from_handle(lib.raft_entry_getdata(ety))
server = self.network.id2server(change.node_id) server = self.network.id2server(change.node_id)
if ety.type == lib.RAFT_LOGTYPE_DEMOTE_NODE: if ety.type == lib.RAFT_LOGTYPE_REMOVE_NODE:
server.set_connection_status(NODE_CONNECTED)
elif ety.type == lib.RAFT_LOGTYPE_REMOVE_NODE:
pass pass
elif ety.type == lib.RAFT_LOGTYPE_ADD_NONVOTING_NODE: elif ety.type == lib.RAFT_LOGTYPE_ADD_NONVOTING_NODE:
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment