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

fix read queue test to be more accurate (#59)

* fix read queue test to be more accurate

1) each server's set of nodes, keeps the max msg_id its seen from that node when that node has been later
1a) because of this get_max_seen now operates on a server's nodes, not on the server itself b

2) virtraft encodes the leader id into the "arg" (with a user redefinable multiplier) it sends as part of the callback, so when we get the callback, we can know what leader its for and check its voters for correctness.
parent 36be2723
...@@ -172,9 +172,11 @@ extern void *(*raft_calloc)(size_t nmemb, size_t size); ...@@ -172,9 +172,11 @@ 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 */ /* update the max_seen_msg_id for this node */
raft_msg_id_t raft_get_max_seen_msg_id(raft_server_t* me_); void raft_node_update_max_seen_msg_id(raft_node_t *me_, raft_msg_id_t msg_id);
/* get the server's (primarily for leader) current msg_id */ /* get the max message id this server has seen from its the specified node */
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me_);
/* get the server's current msg_id */
raft_msg_id_t raft_get_msg_id(raft_server_t* me_); raft_msg_id_t raft_get_msg_id(raft_server_t* me_);
......
...@@ -36,6 +36,7 @@ typedef struct ...@@ -36,6 +36,7 @@ typedef struct
/* last AE heartbeat response received */ /* last AE heartbeat response received */
raft_term_t last_acked_term; raft_term_t last_acked_term;
raft_msg_id_t last_acked_msgid; raft_msg_id_t last_acked_msgid;
raft_msg_id_t max_seen_msgid;
} raft_node_private_t; } raft_node_private_t;
raft_node_t* raft_node_new(void* udata, raft_node_id_t id) raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
...@@ -207,3 +208,17 @@ raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_) ...@@ -207,3 +208,17 @@ raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_)
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
return me->last_acked_msgid; return me->last_acked_msgid;
} }
void raft_node_update_max_seen_msg_id(raft_node_t *me_, raft_msg_id_t msg_id)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (msg_id > me->max_seen_msgid) {
me->max_seen_msgid = msg_id;
}
}
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->max_seen_msgid;
}
\ No newline at end of file
...@@ -778,8 +778,8 @@ int raft_recv_appendentries( ...@@ -778,8 +778,8 @@ int raft_recv_appendentries(
goto out; goto out;
} }
if (me->max_seen_msg_id < ae->msg_id) { if (node != NULL) {
me->max_seen_msg_id = ae->msg_id; raft_node_update_max_seen_msg_id(node, ae->msg_id);
} }
/* update current leader because ae->term is up to date */ /* update current leader because ae->term is up to date */
......
...@@ -288,9 +288,3 @@ raft_msg_id_t raft_get_msg_id(raft_server_t* me_) ...@@ -288,9 +288,3 @@ raft_msg_id_t raft_get_msg_id(raft_server_t* me_)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
return me->msg_id; 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;
}
...@@ -20,7 +20,7 @@ Some chaos we generate: ...@@ -20,7 +20,7 @@ Some chaos we generate:
Usage: Usage:
virtraft --servers SERVERS [-d RATE] [-D RATE] [-c RATE] [-C RATE] [-m RATE] virtraft --servers SERVERS [-d RATE] [-D RATE] [-c RATE] [-C RATE] [-m RATE]
[-P RATE] [-s SEED] [-i ITERS] [-p] [--tsv] [-P RATE] [-s SEED] [-i ITERS] [-p] [--tsv] [--rqm MULTI]
[-q] [-v] [-l LEVEL] [-j] [-L LOGFILE] [--duplex_partition] [-q] [-v] [-l LEVEL] [-j] [-L LOGFILE] [--duplex_partition]
virtraft --version virtraft --version
virtraft --help virtraft --help
...@@ -41,6 +41,7 @@ Options: ...@@ -41,6 +41,7 @@ Options:
-i --iterations ITERS Number of iterations before the simulation ends -i --iterations ITERS Number of iterations before the simulation ends
[default: 1000] [default: 1000]
--tsv Output node status tab separated values at exit --tsv Output node status tab separated values at exit
--rqm MULTI multiplier to use for node_id in read_queue test [default: 10000000000]
-v --verbose Show debug logs -v --verbose Show debug logs
-j --json_output JSON output for diagnostics -j --json_output JSON output for diagnostics
-l --log_level LEVEL Set log level -l --log_level LEVEL Set log level
...@@ -253,10 +254,9 @@ def get_voting_node_ids(leader): ...@@ -253,10 +254,9 @@ def get_voting_node_ids(leader):
def verify_read(arg): def verify_read(arg):
leader = find_leader() node_id = int(arg / net.rqm)
if leader is None: arg = arg % net.rqm
logger.error("failed to find leader") leader = net.servers[node_id - 1]
os._exit(-1)
voter_ids = get_voting_node_ids(leader) voter_ids = get_voting_node_ids(leader)
num_nodes = len(voter_ids) num_nodes = len(voter_ids)
...@@ -266,16 +266,14 @@ def verify_read(arg): ...@@ -266,16 +266,14 @@ def verify_read(arg):
required = int(num_nodes / 2) + 1 required = int(num_nodes / 2) + 1
count = 0 count = 0
leader_term = lib.raft_get_current_term(net.servers[leader.id-1].raft)
for i in voter_ids: for i in voter_ids:
if net.servers[i-1] == leader: if net.servers[i-1] == leader:
count += 1 count += 1
continue continue
msg_id = lib.raft_get_max_seen_msg_id(net.servers[i-1].raft) node = lib.raft_get_node(net.servers[i-1].raft, leader.id)
follower_term = lib.raft_get_current_term(net.servers[i-1].raft) msg_id = lib.raft_node_get_max_seen_msg_id(node)
if msg_id >= arg and leader_term == follower_term: if msg_id >= arg:
count += 1 count += 1
if count < required: if count < required:
...@@ -284,7 +282,7 @@ def verify_read(arg): ...@@ -284,7 +282,7 @@ def verify_read(arg):
def handle_read_queue(arg, can_read): def handle_read_queue(arg, can_read):
val = int(ffi.cast("int", arg)) val = int(ffi.cast("long", arg))
if can_read != 0: if can_read != 0:
verify_read(val) verify_read(val)
logger.debug(f"handling read_request {val}") logger.debug(f"handling read_request {val}")
...@@ -327,6 +325,7 @@ class Network(object): ...@@ -327,6 +325,7 @@ class Network(object):
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.last_seen_read_queue_msg_id = -1
self.rqm = 10000000000
self.server_id = 0 self.server_id = 0
...@@ -367,9 +366,9 @@ class Network(object): ...@@ -367,9 +366,9 @@ class Network(object):
def push_read(self): def push_read(self):
for sv in self.active_servers: for sv in self.active_servers:
if lib.raft_is_leader(sv.raft): 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 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)) arg = sv.id * net.rqm + msg_id
lib.raft_queue_read_request(sv.raft, sv.handle_read_queue, ffi.cast("void *", arg))
def id2server(self, id): def id2server(self, id):
for server in self.servers: for server in self.servers:
...@@ -1280,6 +1279,7 @@ if __name__ == '__main__': ...@@ -1280,6 +1279,7 @@ if __name__ == '__main__':
net.partition_rate = int(args['--partition_rate']) net.partition_rate = int(args['--partition_rate'])
net.no_random_period = 1 == int(args['--no_random_period']) net.no_random_period = 1 == int(args['--no_random_period'])
net.duplex_partition = 1 == int(args['--duplex_partition']) net.duplex_partition = 1 == int(args['--duplex_partition'])
net.rqm = int(args['--rqm'])
net.num_of_servers = int(args['--servers']) net.num_of_servers = int(args['--servers'])
...@@ -1294,10 +1294,10 @@ if __name__ == '__main__': ...@@ -1294,10 +1294,10 @@ 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()
net.push_read()
net.poll_messages()
except: except:
# for server in net.servers: # for server in net.servers:
# print(server, [l.term for l in server.fsm_log]) # print(server, [l.term for l in server.fsm_log])
......
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