Unverified Commit bdd0797f authored by Shaya Potter's avatar Shaya Potter Committed by GitHub
Browse files

add a read_queue test to virtraft (#41)

* add a read_queue test to virtraft

every iteration we push a read_queue request and the handler and an we pass to it set a variable on calling.  we can use this to make sure that the read_queue doesn't get too far from the iteration.  i.e. we pass the leader's msg_id and can check to ensure that leader's msg_id doesn't get too far from the msg_id (variable) we see in read_queue test

this is analog to the current log applying deadlock test.

* implements msg_id checking in virtraft for verification of read_queue requests

When we pop an entry off the read_queue with the can_read flag, we verify across all nodes that the majority of nodes have accepted from the leader a msg_id past what this read_queue entry needs.

the problem with this is, that until now, msg_id was private to each server instance and has no relevance to the followre nodes except to include it back in response. 

we change that to have the followers store the max msg_id they've seen from their current leader

using that, in the read_queue_handler we can verify that the leader's voting nodes have a quorum past the msg_id variable that the handler will return to ensure that this read_queue handler call is correct.
parent e2deade5
...@@ -89,12 +89,19 @@ tests_full: ...@@ -89,12 +89,19 @@ tests_full:
.PHONY: test_virtraft .PHONY: test_virtraft
test_virtraft: test_virtraft:
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 1 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 1 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 1 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 7 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 1 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 7 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 1 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 7 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 1 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 2 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 2 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 2 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 3 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 3 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 3 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 4 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 4 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 4 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 5 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 5 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 5 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 6 -m 3 $(VIRTRAFT_OPTS) python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 6 -m 3 $(VIRTRAFT_OPTS)
python3 tests/virtraft2.py --servers 5 -i 20000 --compaction_rate 50 --drop_rate 5 -P 10 --seed 6 -m 3 --client_rate 0 $(VIRTRAFT_OPTS)
.PHONY: amalgamation .PHONY: amalgamation
amalgamation: amalgamation:
......
...@@ -196,7 +196,7 @@ typedef struct ...@@ -196,7 +196,7 @@ typedef struct
typedef struct typedef struct
{ {
/** the msg_id this response refers to */ /** the msg_id this response refers to */
unsigned long msg_id; raft_msg_id_t msg_id;
/** currentTerm, to force other leader/candidate to step down */ /** currentTerm, to force other leader/candidate to step down */
raft_term_t term; raft_term_t term;
......
...@@ -109,6 +109,10 @@ typedef struct { ...@@ -109,6 +109,10 @@ typedef struct {
* we're still the leader. * we're still the leader.
*/ */
raft_msg_id_t msg_id; raft_msg_id_t msg_id;
/*
* the maximum msg_id we've seen from our current leader. reset on term change
*/
raft_msg_id_t max_seen_msg_id;
raft_read_request_t *read_queue_head; raft_read_request_t *read_queue_head;
raft_read_request_t *read_queue_tail; raft_read_request_t *read_queue_tail;
} raft_server_private_t; } raft_server_private_t;
...@@ -166,4 +170,10 @@ extern void *(*raft_calloc)(size_t nmemb, size_t size); ...@@ -166,4 +170,10 @@ extern void *(*raft_calloc)(size_t nmemb, size_t size);
extern void *(*raft_realloc)(void *ptr, size_t size); extern void *(*raft_realloc)(void *ptr, size_t size);
extern void (*raft_free)(void *ptr); extern void (*raft_free)(void *ptr);
/* get the max message id this server has seen from its current leader, reset to 0 on term change */
raft_msg_id_t raft_get_max_seen_msg_id(raft_server_t* me_);
/* get the server's (primarily for leader) current msg_id */
raft_msg_id_t raft_get_msg_id(raft_server_t* me_);
#endif /* RAFT_PRIVATE_H_ */ #endif /* RAFT_PRIVATE_H_ */
...@@ -730,6 +730,7 @@ int raft_recv_appendentries( ...@@ -730,6 +730,7 @@ int raft_recv_appendentries(
} }
r->msg_id = ae->msg_id; r->msg_id = ae->msg_id;
r->prev_log_idx = ae->prev_log_idx; r->prev_log_idx = ae->prev_log_idx;
r->success = 0; r->success = 0;
...@@ -753,6 +754,10 @@ int raft_recv_appendentries( ...@@ -753,6 +754,10 @@ int raft_recv_appendentries(
goto out; goto out;
} }
if (me->max_seen_msg_id < ae->msg_id) {
me->max_seen_msg_id = ae->msg_id;
}
/* update current leader because ae->term is up to date */ /* update current leader because ae->term is up to date */
me->leader_id = ae->leader_id; me->leader_id = ae->leader_id;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
...@@ -810,6 +815,13 @@ int raft_recv_appendentries( ...@@ -810,6 +815,13 @@ int raft_recv_appendentries(
r->success = 1; r->success = 1;
r->current_idx = ae->prev_log_idx; r->current_idx = ae->prev_log_idx;
/* synchronize msg_id to leader. not really needed for raft, but needed for virtraft for msg_id to be increasing
* cluster wide so that can verify read_queue correctness easily. Otherwise, it be fine for msg_id to be unique to
* each raft_server_t.
*/
if (ae->msg_id > me->msg_id) {
me->msg_id = ae->msg_id;
}
/* 3. If an existing entry conflicts with a new one (same index /* 3. If an existing entry conflicts with a new one (same index
but different terms), delete the existing entry and all that but different terms), delete the existing entry and all that
......
...@@ -93,6 +93,7 @@ int raft_set_current_term(raft_server_t* me_, const raft_term_t term) ...@@ -93,6 +93,7 @@ int raft_set_current_term(raft_server_t* me_, const raft_term_t term)
} }
me->current_term = term; me->current_term = term;
me->voted_for = voted_for; me->voted_for = voted_for;
me->max_seen_msg_id = 0;
} }
return 0; return 0;
} }
...@@ -276,3 +277,15 @@ int raft_is_single_node_voting_cluster(raft_server_t *me_) ...@@ -276,3 +277,15 @@ int raft_is_single_node_voting_cluster(raft_server_t *me_)
return (1 == raft_get_num_voting_nodes(me_) && raft_node_is_voting(me->node)); return (1 == raft_get_num_voting_nodes(me_) && raft_node_is_voting(me->node));
} }
raft_msg_id_t raft_get_msg_id(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
return me->msg_id;
}
raft_msg_id_t raft_get_max_seen_msg_id(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
return me->max_seen_msg_id;
}
...@@ -48,6 +48,7 @@ def load(fname): ...@@ -48,6 +48,7 @@ def load(fname):
ffi.cdef('void *malloc(size_t __size);') ffi.cdef('void *malloc(size_t __size);')
ffi.cdef(load('include/raft.h')) ffi.cdef(load('include/raft.h'))
ffi.cdef(load('include/raft_private.h'))
ffi.cdef(load('include/raft_log.h')) ffi.cdef(load('include/raft_log.h'))
ffi.cdef('raft_entry_t *raft_entry_newdata(void *data);') ffi.cdef('raft_entry_t *raft_entry_newdata(void *data);')
......
...@@ -60,10 +60,10 @@ import coloredlogs ...@@ -60,10 +60,10 @@ import coloredlogs
import docopt import docopt
import logging import logging
import random import random
import re
import sys import sys
import terminaltables import terminaltables
import traceback import traceback
import os
from raft_cffi import ffi, lib from raft_cffi import ffi, lib
...@@ -222,6 +222,75 @@ def raft_notify_membership_event(raft, udata, node, ety, event_type): ...@@ -222,6 +222,75 @@ def raft_notify_membership_event(raft, udata, node, ety, event_type):
return ffi.from_handle(udata).notify_membership_event(node, ety, event_type) return ffi.from_handle(udata).notify_membership_event(node, ety, event_type)
def find_leader():
max_term = -1
leader = None
for s in net.servers:
if lib.raft_is_leader(s.raft):
term = lib.raft_get_current_term(s.raft)
if term > max_term:
max_term = term
leader = s
return leader
def get_voting_node_ids(leader):
voting_nodes_ids = []
for i in range(net.server_id + 1):
if i == 0:
continue
node = lib.raft_get_node(leader.raft, i)
if node == ffi.NULL:
continue
if lib.raft_node_is_voting(node) != 0:
voting_nodes_ids.append(i)
return voting_nodes_ids
def verify_read(arg):
leader = find_leader()
if leader is None:
logger.error("failed to find leader")
os._exit(-1)
voter_ids = get_voting_node_ids(leader)
num_nodes = len(voter_ids)
# primary verification logic. we always need to count more than the required, looking to see that for voters,
# the read_queue_id they have recorded is equal to our greater than the arg (i.e. the read_queue id this was sent on
required = int(num_nodes / 2) + 1
count = 0
leader_term = lib.raft_get_current_term(net.servers[leader.id-1].raft)
for i in voter_ids:
if net.servers[i-1] == leader:
count += 1
continue
msg_id = lib.raft_get_max_seen_msg_id(net.servers[i-1].raft)
follower_term = lib.raft_get_current_term(net.servers[i-1].raft)
if msg_id >= arg and leader_term == follower_term:
count += 1
if count < required:
logger.error(f"verify_read failed {count} < {required}")
os._exit(-1)
def handle_read_queue(arg, can_read):
val = int(ffi.cast("int", arg))
if can_read != 0:
verify_read(val)
logger.debug(f"handling read_request {val}")
net.last_seen_read_queue_msg_id = val
else:
logger.debug(f"ignoring read_request {val}")
def raft_log(raft, node, udata, buf): def raft_log(raft, node, udata, buf):
server = ffi.from_handle(lib.raft_get_udata(raft)) server = ffi.from_handle(lib.raft_get_udata(raft))
# if server.id in [1] or (node and node.id in [1]): # if server.id in [1] or (node and node.id in [1]):
...@@ -255,6 +324,7 @@ class Network(object): ...@@ -255,6 +324,7 @@ class Network(object):
self.partitions = set() self.partitions = set()
self.no_random_period = False self.no_random_period = False
self.duplex_partition = False self.duplex_partition = False
self.last_seen_read_queue_msg_id = -1
self.server_id = 0 self.server_id = 0
...@@ -292,6 +362,13 @@ class Network(object): ...@@ -292,6 +362,13 @@ class Network(object):
assert e == 0 or e == lib.RAFT_ERR_NOT_LEADER assert e == 0 or e == lib.RAFT_ERR_NOT_LEADER
break break
def push_read(self):
for sv in self.active_servers:
if lib.raft_is_leader(sv.raft):
logger.debug(f"adding a read_request to {sv.id}")
msg_id = lib.raft_get_msg_id(sv.raft) + 1
lib.raft_queue_read_request(sv.raft, sv.handle_read_queue, ffi.cast("void *", msg_id))
def id2server(self, id): def id2server(self, id):
for server in self.servers: for server in self.servers:
if server.id == id: if server.id == id:
...@@ -342,7 +419,7 @@ class Network(object): ...@@ -342,7 +419,7 @@ class Network(object):
server.periodic(self.random.randint(1, 100)) server.periodic(self.random.randint(1, 100))
# Deadlock detection # Deadlock detection
if self.latest_applied_log_idx != 0 and self.latest_applied_log_iteration + 5000 < self.iteration: if self.client_rate != 0 and self.latest_applied_log_idx != 0 and self.latest_applied_log_iteration + 5000 < self.iteration:
logger.error("deadlock detected iteration:{0} appliedidx:{1}\n".format( logger.error("deadlock detected iteration:{0} appliedidx:{1}\n".format(
self.latest_applied_log_iteration, self.latest_applied_log_iteration,
self.latest_applied_log_idx, self.latest_applied_log_idx,
...@@ -350,6 +427,14 @@ class Network(object): ...@@ -350,6 +427,14 @@ class Network(object):
self.diagnostic_info() self.diagnostic_info()
sys.exit(1) sys.exit(1)
leader = find_leader()
if leader and self.last_seen_read_queue_msg_id + 5000 < lib.raft_get_msg_id(self.servers[leader.id - 1].raft):
logger.error("deadlock detected in handling reads: last seen read queue msg_id {0} current at {1}, iter: {2}\n".format(
self.last_seen_read_queue_msg_id, lib.raft_get_msg_id(leader.raft), self.iteration,
))
self.diagnostic_info()
sys.exit(1)
# Count leadership changes # Count leadership changes
assert len(self.active_servers) > 0, self.diagnostic_info() assert len(self.active_servers) > 0, self.diagnostic_info()
leader = lib.raft_get_leader_id(self.active_servers[0].raft) leader = lib.raft_get_leader_id(self.active_servers[0].raft)
...@@ -369,6 +454,7 @@ class Network(object): ...@@ -369,6 +454,7 @@ class Network(object):
for partition in self.partitions: for partition in self.partitions:
# Partitions are in one direction # Partitions are in one direction
if partition[0] is sendor and partition[1] is sendee: if partition[0] is sendor and partition[1] is sendee:
logger.info(f"dropping message from {sendor.id} to {sendee.id}")
return return
# duplex partitions prevent both directions from sending traffic # duplex partitions prevent both directions from sending traffic
if self.duplex_partition and ( if self.duplex_partition and (
...@@ -789,6 +875,7 @@ class RaftServer(object): ...@@ -789,6 +875,7 @@ class RaftServer(object):
self.raft_node_has_sufficient_logs = ffi.callback("int(raft_server_t* raft, void *user_data, raft_node_t* node)", raft_node_has_sufficient_logs) self.raft_node_has_sufficient_logs = ffi.callback("int(raft_server_t* raft, void *user_data, raft_node_t* node)", raft_node_has_sufficient_logs)
self.raft_notify_membership_event = ffi.callback("void(raft_server_t* raft, void *user_data, raft_node_t* node, raft_entry_t* ety, raft_membership_e)", raft_notify_membership_event) self.raft_notify_membership_event = ffi.callback("void(raft_server_t* raft, void *user_data, raft_node_t* node, raft_entry_t* ety, raft_membership_e)", raft_notify_membership_event)
self.raft_log = ffi.callback("void(raft_server_t*, raft_node_id_t, void*, const char* buf)", raft_log) self.raft_log = ffi.callback("void(raft_server_t*, raft_node_id_t, void*, const char* buf)", raft_log)
self.handle_read_queue = ffi.callback("void(void *arg, int can_read)", handle_read_queue)
def recv_entry(self, ety): def recv_entry(self, ety):
# FIXME: leak # FIXME: leak
...@@ -1039,7 +1126,7 @@ class RaftServer(object): ...@@ -1039,7 +1126,7 @@ class RaftServer(object):
def entry_append(self, ety, ety_idx): def entry_append(self, ety, ety_idx):
try: try:
assert not self.fsm_log or self.fsm_log[-1].term <= ety.term assert not self.fsm_log or self.fsm_log[-1].term <= ety.term, f"node {self.id}'s term {self.fsm_log[-1].term} > to entry term {ety.term}"
except Exception as e: except Exception as e:
self.abort_exception = e self.abort_exception = e
# FIXME: consider returning RAFT_ERR_SHUTDOWN # FIXME: consider returning RAFT_ERR_SHUTDOWN
...@@ -1180,6 +1267,7 @@ if __name__ == '__main__': ...@@ -1180,6 +1267,7 @@ if __name__ == '__main__':
logger.addHandler(fh) logger.addHandler(fh)
logger.propagate = False logger.propagate = False
global net
net = Network(int(args['--seed'])) net = Network(int(args['--seed']))
net.dupe_rate = int(args['--dupe_rate']) net.dupe_rate = int(args['--dupe_rate'])
...@@ -1204,6 +1292,8 @@ if __name__ == '__main__': ...@@ -1204,6 +1292,8 @@ if __name__ == '__main__':
for i in range(0, int(args['--iterations'])): for i in range(0, int(args['--iterations'])):
net.iteration += 1 net.iteration += 1
try: try:
net.push_read()
net.poll_messages()
net.periodic() net.periodic()
net.poll_messages() net.poll_messages()
except: except:
......
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