From e02e529809fe6ed8333769c48a9bad8bcf1b22f9 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 19 Feb 2026 15:10:51 +0100 Subject: [PATCH] Version 1.0 --- quakey/src/quakey.c | 14 +- raft/client.c | 10 + raft/config.h | 2 +- raft/main.c | 10 + raft/node.c | 70 ++++--- raft/node.h | 1 + table.txt | 247 ------------------------- vsr/config.h | 2 +- vsr/main.c | 5 + vsr/node.c | 439 ++++++++++++++++++++++++-------------------- vsr/node.h | 3 + 11 files changed, 321 insertions(+), 482 deletions(-) delete mode 100644 table.txt diff --git a/quakey/src/quakey.c b/quakey/src/quakey.c index fd8d0c0..dac7698 100644 --- a/quakey/src/quakey.c +++ b/quakey/src/quakey.c @@ -2635,8 +2635,8 @@ static int change_target_partition(Sim *sim) host_pair_list_free(&sim->target_partition); sim->target_partition = list; - //SIM_TRACE(sim, "PARTITION: new target partition with %d broken links (%d buckets)", - // list.count, num_buckets); + SIM_TRACE(sim, "PARTITION: new target partition with %d broken links (%d buckets)", + list.count, num_buckets); return 0; } @@ -2735,8 +2735,8 @@ static bool break_link(Sim *sim) host_pair_list_add(&sim->partition, pair.a, pair.b); drop_conns_between_hosts(sim, pair.a, pair.b); - //SIM_TRACE(sim, "PARTITION: break link %s <-> %s (%d broken links)", - // pair.a->name, pair.b->name, sim->partition.count); + SIM_TRACE(sim, "PARTITION: break link %s <-> %s (%d broken links)", + pair.a->name, pair.b->name, sim->partition.count); return true; } @@ -2748,10 +2748,10 @@ static bool repair_link(Sim *sim) if (idx < 0) return false; - //HostPair pair = sim->partition.pairs[idx]; + HostPair pair = sim->partition.pairs[idx]; host_pair_list_remove_at(&sim->partition, idx); - //SIM_TRACE(sim, "PARTITION: repair link %s <-> %s (%d broken links)", - // pair.a->name, pair.b->name, sim->partition.count); + SIM_TRACE(sim, "PARTITION: repair link %s <-> %s (%d broken links)", + pair.a->name, pair.b->name, sim->partition.count); return true; } diff --git a/raft/client.c b/raft/client.c index c4cc9f1..af7bf7e 100644 --- a/raft/client.c +++ b/raft/client.c @@ -79,6 +79,16 @@ process_message(ClientState *state, return -1; memcpy(&redirect_message, msg.ptr, sizeof(redirect_message)); + // Ignore stale redirects from previous requests. A redirect + // from server A for request N-1 can arrive after the client + // has already moved on to request N with server B. Without + // this check, the stale redirect would falsely cancel the + // current request, causing the client and linearizability + // checker to believe the request was rejected even though + // the new server may have committed it. + if (redirect_message.request_id != state->request_id) + return 0; + if (redirect_message.leader_idx >= 0 && redirect_message.leader_idx < state->num_servers) { CLIENT_TRACE("Redirected to leader %d", redirect_message.leader_idx); state->current_leader = redirect_message.leader_idx; diff --git a/raft/config.h b/raft/config.h index 811b1ef..6d716a4 100644 --- a/raft/config.h +++ b/raft/config.h @@ -3,6 +3,6 @@ #define NODE_LIMIT 32 #define HEARTBEAT_INTERVAL_SEC 1 -#define PRIMARY_DEATH_TIMEOUT_SEC 2 +#define PRIMARY_DEATH_TIMEOUT_SEC 5 #endif // CONFIG_INCLUDED \ No newline at end of file diff --git a/raft/main.c b/raft/main.c index 1d04e56..10436bd 100644 --- a/raft/main.c +++ b/raft/main.c @@ -156,6 +156,9 @@ int main(int argc, char **argv) #ifdef MAIN_CLIENT #include +#include +#include +#include #include "client.h" @@ -163,6 +166,8 @@ int main(int argc, char **argv) int main(int argc, char **argv) { + srand((unsigned)time(NULL) ^ (unsigned)getpid()); + int ret; ClientState state; @@ -215,6 +220,9 @@ int main(int argc, char **argv) #ifdef MAIN_SERVER #include +#include +#include +#include #include "node.h" @@ -222,6 +230,8 @@ int main(int argc, char **argv) int main(int argc, char **argv) { + srand((unsigned)time(NULL) ^ (unsigned)getpid()); + int ret; NodeState state; diff --git a/raft/node.c b/raft/node.c index ddf9a59..b7f7f21 100644 --- a/raft/node.c +++ b/raft/node.c @@ -258,8 +258,18 @@ send_appended_response(NodeState *state, int conn_idx, bool success, int match_i return HR_OK; } -static HandlerResult send_redirect(NodeState *state, int conn_idx) +static HandlerResult send_redirect(NodeState *state, int conn_idx, ByteView msg) { + // Extract request_id from the client request so the client + // can discard stale redirects that arrive after a new request + // has already been sent (e.g. from a different server). + uint64_t request_id = 0; + if (msg.len == sizeof(RequestMessage)) { + RequestMessage req; + memcpy(&req, msg.ptr, sizeof(req)); + request_id = req.request_id; + } + RedirectMessage redirect_message = { .base = { .version = MESSAGE_VERSION, @@ -267,6 +277,7 @@ static HandlerResult send_redirect(NodeState *state, int conn_idx) .length = sizeof(RedirectMessage), }, .leader_idx = state->leader_idx, + .request_id = request_id, }; node_log(state, "SEND REDIRECT", "-> conn %d leader_idx=%d", @@ -290,9 +301,6 @@ static void apply_committed(NodeState *state) WALEntry *wal_entry = wal_peek_entry(&state->wal, state->last_applied); - node_log(state, "APPLY", "entry %d (term %lu)", - state->last_applied, (unsigned long)wal_entry->term); - KVStoreResult result = kvstore_update(&state->kvstore, wal_entry->oper); // Update the client table with the committed result for ALL roles. @@ -488,9 +496,12 @@ static HandlerResult step_down(NodeState *state, uint64_t new_term) (unsigned long)state->term_and_vote.term, (unsigned long)new_term); state->role = ROLE_FOLLOWER; + state->leader_idx = -1; - if (set_term_and_vote(state, new_term, -1) < 0) - return HR_IO_FAILURE; + if (new_term != state->term_and_vote.term) { + if (set_term_and_vote(state, new_term, -1) < 0) + return HR_IO_FAILURE; + } return HR_OK; } @@ -582,7 +593,7 @@ static bool remote_has_recent_state(NodeState *state, } static HandlerResult -process_request_vote_for_folloer(NodeState *state, int conn_idx, ByteView msg) +process_request_vote_for_follower(NodeState *state, int conn_idx, ByteView msg) { RequestVoteMessage request_vote_message; if (msg.len != sizeof(request_vote_message)) @@ -653,10 +664,10 @@ process_request_vote_for_candidate(NodeState *state, int conn_idx, ByteView msg) if (request_vote_message.term < state->term_and_vote.term) return send_vote_response(state, conn_idx, false, -1); - // Higher term: step down and consider the vote - HandlerResult hret = step_down(state, request_vote_message.term); - if (hret != HR_OK) - return hret; + { + HandlerResult hret = step_down(state, request_vote_message.term); + if (hret != HR_OK) return hret; + } if (remote_has_recent_state(state, request_vote_message.last_log_index, @@ -675,8 +686,9 @@ process_voted_for_candidate(NodeState *state, int conn_idx, ByteView msg) memcpy(&message, msg.ptr, sizeof(message)); // Local state is stale - if (message.term > state->term_and_vote.term) + if (message.term > state->term_and_vote.term) { return step_down(state, message.term); + } // Ignore votes from old terms if (message.term < state->term_and_vote.term) @@ -712,9 +724,10 @@ process_append_entries_for_candidate(NodeState *state, int conn_idx, ByteView ms return send_appended_response(state, conn_idx, false, -1); // A valid leader exists for this or a higher term; step down - HandlerResult hret = step_down(state, append_entries_message.term); - if (hret != HR_OK) - return hret; + { + HandlerResult hret = step_down(state, append_entries_message.term); + if (hret != HR_OK) return hret; + } WALEntry *entries = (WALEntry*) (msg.ptr + sizeof(AppendEntriesMessage)); return handle_append_entries(state, conn_idx, &append_entries_message, entries); @@ -763,9 +776,10 @@ process_request_vote_for_leader(NodeState *state, int conn_idx, ByteView msg) if (request_vote_message.term > state->term_and_vote.term) { - HandlerResult hret = step_down(state, request_vote_message.term); - if (hret != HR_OK) - return hret; + { + HandlerResult hret = step_down(state, request_vote_message.term); + if (hret != HR_OK) return hret; + } if (remote_has_recent_state(state, request_vote_message.last_log_index, @@ -789,9 +803,11 @@ process_append_entries_for_leader(NodeState *state, int conn_idx, ByteView msg) // Leader with a higher term exists? Step down if (append_entries_message.term > state->term_and_vote.term) { - HandlerResult hret = step_down(state, append_entries_message.term); - if (hret != HR_OK) - return hret; + + { + HandlerResult hret = step_down(state, append_entries_message.term); + if (hret != HR_OK) return hret; + } WALEntry *entries = (WALEntry*) (msg.ptr + sizeof(AppendEntriesMessage)); return handle_append_entries(state, conn_idx, &append_entries_message, entries); @@ -813,10 +829,7 @@ process_appended_for_leader(NodeState *state, int conn_idx, ByteView msg) // Our state is stale if (appended_message.term > state->term_and_vote.term) { - HandlerResult hret = step_down(state, appended_message.term); - if (hret != HR_OK) - return hret; - return HR_OK; + return step_down(state, appended_message.term); } int follower_idx = appended_message.sender_idx; @@ -964,11 +977,11 @@ process_message(NodeState *state, case ROLE_FOLLOWER: switch (type) { case MESSAGE_TYPE_REQUEST_VOTE: - return process_request_vote_for_folloer(state, conn_idx, msg); + return process_request_vote_for_follower(state, conn_idx, msg); case MESSAGE_TYPE_APPEND_ENTRIES: return process_append_entries_for_follower(state, conn_idx, msg); case MESSAGE_TYPE_REQUEST: - return send_redirect(state, conn_idx); + return send_redirect(state, conn_idx, msg); } return HR_OK; @@ -981,7 +994,7 @@ process_message(NodeState *state, case MESSAGE_TYPE_APPEND_ENTRIES: return process_append_entries_for_candidate(state, conn_idx, msg); case MESSAGE_TYPE_REQUEST: - return send_redirect(state, conn_idx); + return send_redirect(state, conn_idx, msg); } return HR_OK; } @@ -1074,6 +1087,7 @@ int node_init(void *state_, int argc, char **argv, string wal_file = S("raft.wal"); string term_and_vote_file = S("term_and_vote.wal"); + state->num_nodes = 0; bool self_addr_set = false; for (int i = 1; i < argc; i++) { if (!strcmp(argv[i], "--addr")) { diff --git a/raft/node.h b/raft/node.h index 75a25f7..b64bb3f 100644 --- a/raft/node.h +++ b/raft/node.h @@ -69,6 +69,7 @@ typedef struct { typedef struct { MessageHeader base; int leader_idx; + uint64_t request_id; } RedirectMessage; typedef enum { diff --git a/table.txt b/table.txt deleted file mode 100644 index 5b27274..0000000 --- a/table.txt +++ /dev/null @@ -1,247 +0,0 @@ -Consensus Protocol Simulation Results -Simulated time: 1200s per run, seeds 1-100 -3 nodes, 2 clients, fault injection enabled, network partitioning enabled - -Column descriptions: - Seed - RNG seed used for the simulation run - Proto - Consensus protocol (VSR or Raft) - Invocations - Number of client requests initiated - OK - Requests that completed successfully with a response - Timeouts - Requests that timed out without a response - Rejected - Requests explicitly rejected (e.g. sent to a non-leader) - Checks - Linearizability checks performed by the checker - Steps - Total simulation scheduling steps (ticks across all hosts) - -Seed Proto Invocations OK Timeouts Rejected Checks Steps ------------------------------------------------------------------------ - 1 VSR 2244 2206 33 5 2244 105648 - 1 Raft 3276 1700 368 1206 3274 363190 - 2 VSR 2883 2839 39 5 2883 126828 - 2 Raft 3394 2761 285 346 3392 171163 - 3 VSR 592 497 94 1 592 42548 - 3 Raft 2789 2358 330 99 2787 150699 - 4 VSR 1274 1231 37 6 1274 78222 - 4 Raft 2068 1362 464 240 2066 88436 - 5 VSR 1405 1372 31 1 1404 75712 - 5 Raft 2304 1630 416 256 2302 131192 - 6 VSR 2025 1975 49 0 2024 169653 - 6 Raft 2182 1244 447 489 2180 378921 - 7 VSR 955 856 98 1 955 56039 - 7 Raft 3433 1609 311 1511 3431 297830 - 8 VSR 720 636 79 4 719 46945 - 8 Raft 2498 1759 436 301 2496 123015 - 9 VSR 859 780 77 2 859 56003 - 9 Raft 2778 1816 362 599 2777 225590 - 10 VSR 1486 1401 84 1 1486 95773 - 10 Raft 3114 2093 321 698 3112 341220 - 11 VSR 456 302 153 1 456 33396 - 11 Raft 3441 1209 375 1855 3439 345485 - 12 VSR 1220 1128 88 2 1218 66620 - 12 Raft 2555 1064 458 1031 2553 370909 - 13 VSR 863 753 108 1 862 52065 - 13 Raft 1965 1210 494 259 1963 162897 - 14 VSR 645 524 115 5 644 44976 - 14 Raft 2498 1918 399 179 2496 138682 - 15 VSR 690 638 50 1 689 48727 - 15 Raft 2541 1871 381 287 2539 200612 - 16 VSR 944 835 106 1 942 52517 - 16 Raft 2369 1451 466 450 2367 465559 - 17 VSR 647 564 80 2 646 40272 - 17 Raft 2446 1846 399 199 2444 124084 - 18 VSR 392 310 76 4 390 35414 - 18 Raft 1955 923 527 503 1953 218171 - 19 VSR 1165 1068 97 0 1165 60283 - 19 Raft 2987 1849 363 773 2985 230511 - 20 VSR 2229 2164 62 2 2228 108081 - 20 Raft 2996 1742 381 871 2994 450695 - 21 VSR 1390 1341 44 5 1390 78704 - 21 Raft 2441 1485 412 542 2439 120767 - 22 VSR 1844 1758 81 5 1844 96250 - 22 Raft 3561 1897 306 1356 3559 212082 - 23 VSR 1393 1335 55 3 1393 79177 - 23 Raft 2524 1496 434 593 2523 222489 - 24 VSR 971 951 18 0 969 56647 - 24 Raft 2177 1606 412 157 2175 177816 - 25 VSR 458 376 77 5 458 35323 - 25 Raft 2507 1898 374 233 2505 139076 - 26 VSR 1177 1117 60 0 1177 70344 - 26 Raft 2583 1978 365 238 2581 186689 - 27 VSR 1042 937 104 1 1042 70667 - 27 Raft 2794 2237 355 200 2792 171007 - 28 VSR 1118 1067 43 6 1116 67368 - 28 Raft 2790 1870 357 561 2788 257457 - 29 VSR 683 610 73 0 683 45166 - 29 Raft 2557 1439 408 708 2555 110388 - 30 VSR 1202 1134 63 3 1200 64864 - 30 Raft 3312 2297 316 697 3310 406253 - 31 VSR 1466 1399 65 2 1466 83025 - 31 Raft 2818 1612 377 827 2816 311506 - 32 VSR 1207 1161 41 5 1207 68520 - 32 Raft 2676 1141 416 1117 2674 255671 - 33 VSR 1219 1134 83 1 1218 65923 - 33 Raft 3260 2844 274 140 3258 191445 - 34 VSR 1255 1229 24 1 1254 71120 - 34 Raft 3311 1372 405 1532 3309 428477 - 35 VSR 1197 1142 50 4 1196 73779 - 35 Raft 2435 1793 418 222 2433 147736 - 36 VSR 1560 1500 55 4 1559 89849 - 36 Raft 2110 1186 515 407 2108 310505 - 37 VSR 1487 1424 58 4 1486 148383 - 37 Raft 3223 1407 335 1479 3221 242814 - 38 VSR 1397 1349 42 5 1396 76827 - 38 Raft 2466 1260 441 763 2464 352797 - 39 VSR 1304 1249 53 1 1303 78321 - 39 Raft 3027 2062 343 620 3025 334793 - 40 VSR 1299 1270 25 2 1297 72474 - 40 Raft 2183 1213 479 489 2181 229529 - 41 VSR 1091 1040 41 8 1089 65193 - 41 Raft 2856 2333 339 182 2854 162726 - 42 VSR 496 410 84 1 495 36493 - 42 Raft 2421 1797 400 222 2419 163015 - 43 VSR 658 562 92 2 656 46387 - 43 Raft 2834 2134 362 336 2832 161248 - 44 VSR 1372 1318 49 4 1371 75563 - 44 Raft 2654 2076 370 206 2652 262333 - 45 VSR 257 127 129 0 256 33593 - 45 Raft 3146 1319 366 1459 3144 395077 - 46 VSR 1515 1458 53 2 1513 105709 - 46 Raft 3320 1456 382 1480 3318 444523 - 47 VSR 1057 1017 39 0 1056 61117 - 47 Raft 2243 1351 467 423 2241 210836 - 48 VSR 1195 1140 51 3 1194 69487 - 48 Raft 2378 1708 422 246 2376 148188 - 49 VSR 950 876 71 2 949 55394 - 49 Raft 2984 2444 309 229 2982 153760 - 50 VSR 964 875 88 1 964 57668 - 50 Raft 2889 2295 344 248 2887 182648 - 51 VSR 1302 1250 49 1 1300 74028 - 51 Raft 2363 1734 415 212 2361 127568 - 52 VSR 1205 1111 91 2 1204 64727 - 52 Raft 2905 2380 303 221 2904 170346 - 53 VSR 869 744 124 0 868 95545 - 53 Raft 2203 964 444 794 2202 306208 - 54 VSR 408 333 72 3 408 36481 - 54 Raft 2113 1409 434 268 2111 103004 - 55 VSR 494 443 50 1 494 40161 - 55 Raft 2618 1937 375 305 2617 239299 - 56 VSR 1300 1231 66 2 1299 70686 - 56 Raft 3401 2723 282 394 3399 194017 - 57 VSR 238 145 93 0 238 34783 - 57 Raft 1941 1255 451 233 1939 90352 - 58 VSR 922 854 63 5 922 54772 - 58 Raft 2173 1303 448 420 2171 225395 - 59 VSR 1941 1853 84 3 1940 103258 - 59 Raft 2501 1237 420 842 2499 214547 - 60 VSR 1766 1645 114 5 1764 86384 - 60 Raft 2909 1062 437 1408 2907 446530 - 61 VSR 1630 1557 68 4 1629 76775 - 61 Raft 2956 2258 333 363 2954 199725 - 62 VSR 1077 1023 51 2 1076 107542 - 62 Raft 3418 1993 303 1120 3416 178317 - 63 VSR 2158 2107 47 3 2157 107086 - 63 Raft 2245 1039 473 731 2243 331394 - 64 VSR 1804 1743 53 7 1803 101345 - 64 Raft 2804 2152 325 325 2802 206594 - 65 VSR 934 825 106 1 932 77299 - 65 Raft 2279 1090 486 701 2277 266669 - 66 VSR 798 726 69 2 797 59313 - 66 Raft 2105 1135 444 524 2103 246688 - 67 VSR 965 874 91 0 965 71117 - 67 Raft 2268 1598 439 229 2266 103555 - 68 VSR 438 327 109 1 437 54528 - 68 Raft 2787 1755 395 635 2785 347905 - 69 VSR 922 869 52 1 922 61234 - 69 Raft 2588 1752 425 409 2586 218784 - 70 VSR 720 607 110 3 720 47228 - 70 Raft 2424 1495 414 513 2422 255104 - 71 VSR 1067 982 81 4 1067 62847 - 71 Raft 2207 1583 398 224 2205 165271 - 72 VSR 673 561 110 0 671 48927 - 72 Raft 2927 2052 348 525 2925 145377 - 73 VSR 1361 1270 88 2 1360 70464 - 73 Raft 2798 1976 353 467 2796 131273 - 74 VSR 1891 1842 43 5 1890 104058 - 74 Raft 2176 1640 405 129 2174 123942 - 75 VSR 970 879 91 0 970 86026 - 75 Raft 2456 879 491 1084 2454 380910 - 76 VSR 851 737 111 2 850 55131 - 76 Raft 3226 1945 343 936 3224 158287 - 77 VSR 610 474 133 2 609 40458 - 77 Raft 2492 1882 387 221 2490 134293 - 78 VSR 1269 1232 31 6 1269 69782 - 78 Raft 3411 1575 329 1506 3410 238137 - 79 VSR 921 859 62 0 921 55960 - 79 Raft 3008 2516 335 155 3006 176544 - 80 VSR 1271 1230 36 5 1271 77391 - 80 Raft 2605 1033 426 1144 2603 287940 - 81 VSR 835 708 124 2 834 52299 - 81 Raft 2582 1845 361 374 2580 140619 - 82 VSR 1024 913 109 1 1023 58016 - 82 Raft 2921 2264 330 325 2919 143004 - 83 VSR 922 802 120 0 922 51388 - 83 Raft 2600 1885 412 301 2598 129191 - 84 VSR 1533 1480 52 1 1533 79724 - 84 Raft 2344 1753 440 149 2342 129605 - 85 VSR 1650 1588 58 4 1650 88562 - 85 Raft 3366 1419 355 1590 3364 379641 - 86 VSR 884 763 118 1 882 49372 - 86 Raft 2736 1258 457 1019 2734 552128 - 87 VSR 1072 1003 68 0 1071 59323 - 87 Raft 1875 968 506 399 1873 253771 - 88 VSR 1493 1437 52 4 1493 92016 - 88 Raft 1902 1285 445 170 1900 109923 - 89 VSR 1294 1239 53 0 1292 69379 - 89 Raft 1922 1204 471 245 1920 91616 - 90 VSR 1215 1119 93 2 1214 67086 - 90 Raft 3440 2057 296 1085 3438 284840 - 91 VSR 2176 2142 28 6 2176 108769 - 91 Raft 2606 2025 358 221 2604 136452 - 92 VSR 594 550 42 0 592 43714 - 92 Raft 2563 1504 432 625 2561 370793 - 93 VSR 775 699 76 0 775 67757 - 93 Raft 1990 826 509 653 1988 320932 - 94 VSR 546 455 88 3 546 35972 - 94 Raft 2449 1686 399 362 2447 216855 - 95 VSR 420 326 94 0 420 33058 - 95 Raft 2506 1891 379 234 2504 141706 - 96 VSR 583 509 70 4 583 40050 - 96 Raft 2307 1263 473 569 2305 325134 - 97 VSR 1991 1965 21 3 1989 95473 - 97 Raft 2679 1360 397 920 2677 281323 - 98 VSR 1198 1123 73 2 1198 78181 - 98 Raft 2550 1467 467 614 2548 416874 - 99 VSR 954 884 67 3 954 72020 - 99 Raft 2051 1356 481 212 2049 102556 - 100 VSR 1426 1319 105 1 1425 76369 - 100 Raft 2817 1402 440 973 2815 474134 ------------------------------------------------------------------------ - -Summary Statistics - -VSR (100 completed runs, 0 violations): - Avg invocations: 1132.8 - Avg ok: 1057.4 - Avg timeouts: 72.3 - Avg rejected: 2.3 - Avg steps: 69089.2 - Total invocations: 113278 - Total ok: 105742 - Total timeouts: 7226 - Total rejected: 234 - OK rate: 93.3% - Timeout rate: 6.4% - Rejection rate: 0.2% - -Raft (100 completed runs, 0 violations): - Avg invocations: 2645.8 - Avg ok: 1664.9 - Avg timeouts: 397.6 - Avg rejected: 581.4 - Avg steps: 234875.8 - Total invocations: 264582 - Total ok: 166491 - Total timeouts: 39755 - Total rejected: 58142 - OK rate: 62.9% - Timeout rate: 15.0% - Rejection rate: 22.0% diff --git a/vsr/config.h b/vsr/config.h index f188dfb..c5a517d 100644 --- a/vsr/config.h +++ b/vsr/config.h @@ -5,9 +5,9 @@ #define FUTURE_LIMIT 32 #define HEARTBEAT_INTERVAL_SEC 1 #define PRIMARY_DEATH_TIMEOUT_SEC 5 -#define VIEW_CHANGE_TIMEOUT_SEC 10 #define RECOVERY_TIMEOUT_SEC 3 #define RECOVERY_ATTEMPT_LIMIT 10 #define STATE_TRANSFER_TIMEOUT_SEC 2 +#define VIEW_CHANGE_TIMEOUT_SEC 20 #endif // CONFIG_INCLUDED \ No newline at end of file diff --git a/vsr/main.c b/vsr/main.c index 374c28b..eb29433 100644 --- a/vsr/main.c +++ b/vsr/main.c @@ -167,6 +167,9 @@ int main(int argc, char **argv) #ifdef MAIN_CLIENT #include +#include +#include +#include #include "client.h" @@ -174,6 +177,8 @@ int main(int argc, char **argv) int main(int argc, char **argv) { + srand((unsigned)time(NULL) ^ (unsigned)getpid()); + int ret; ClientState state; diff --git a/vsr/node.c b/vsr/node.c index dac696e..e7e4fe9 100644 --- a/vsr/node.c +++ b/vsr/node.c @@ -157,11 +157,6 @@ static void begin_state_transfer(NodeState *state, int sender_idx) static HandlerResult process_request(NodeState *state, int conn_idx, ByteView msg) { - // Don't accept new requests during a view change. - // The leader must complete the view change first. - if (state->status != STATUS_NORMAL) - return HR_OK; - RequestMessage request_message; if (msg.len != sizeof(request_message)) return HR_INVALID_MESSAGE; @@ -307,32 +302,9 @@ process_request(NodeState *state, int conn_idx, ByteView msg) return HR_OK; } -static void reply_to_client(NodeState *state, uint64_t client_id, +static void reply_to_client(NodeState *state, ClientTableEntry *table_entry, uint64_t request_id, KVStoreOper *oper, KVStoreResult result) { - // After a view change, the new leader inherits log - // entries but not the client table's pending state. - // A client table entry may exist from a previous view - // (when this node was leader before) with pending=false. - // Only reply if this leader received the original REQUEST. - ClientTableEntry *table_entry = client_table_find(&state->client_table, client_id); - if (table_entry == NULL) - return; - - if (!table_entry->pending) - return; - - // Verify this reply is for the request the client is actually - // waiting for. After a view change, the new leader inherits - // uncommitted log entries from the old view carrying a stale - // request_id. If the client has since sent a newer request, - // we must not confuse the old result with the new request. - if (table_entry->last_request_id != request_id) - return; - - table_entry->pending = false; - table_entry->last_result = result; - int conn_idx = tcp_index_from_tag(&state->tcp, table_entry->conn_tag); if (conn_idx < 0) return; @@ -351,7 +323,7 @@ static void reply_to_client(NodeState *state, uint64_t client_id, char result_buf[64]; kvstore_snprint_result(result_buf, sizeof(result_buf), result); node_log(state, "SEND REPLY", "client=%lu req=%lu key=%.16s %s", - client_id, request_id, oper->key, result_buf); + table_entry->client_id, request_id, oper->key, result_buf); } ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); @@ -375,9 +347,32 @@ static void advance_commit_index(NodeState *state, int target_index, bool send_r kvstore_snprint_result(result_buf, sizeof(result_buf), result); node_log(state, "APPLY", "idx=%d key=%.16s %s -> %s", state->commit_index-1, entry->oper.key, oper_buf, result_buf); } - if (send_replies) { - reply_to_client(state, entry->client_id, entry->request_id, &entry->oper, result); - } + + // After a view change, the new leader inherits log + // entries but not the client table's pending state. + // A client table entry may exist from a previous view + // (when this node was leader before) with pending=false. + // Only reply if this leader received the original REQUEST. + ClientTableEntry *table_entry = client_table_find(&state->client_table, entry->client_id); + if (table_entry == NULL) + continue; + + if (!table_entry->pending) + continue; + + // Verify this reply is for the request the client is actually + // waiting for. After a view change, the new leader inherits + // uncommitted log entries from the old view carrying a stale + // request_id. If the client has since sent a newer request, + // we must not confuse the old result with the new request. + if (table_entry->last_request_id != entry->request_id) + continue; + + table_entry->pending = false; + table_entry->last_result = result; + + if (send_replies) + reply_to_client(state, table_entry, entry->request_id, &entry->oper, result); } } @@ -429,7 +424,6 @@ process_prepare_ok(NodeState *state, int conn_idx, ByteView msg) return HR_OK; // Drop if (message.view_number > state->view_number) { - // TODO: When can this happen? state->view_number = message.view_number; begin_state_transfer(state, message.sender_idx); return HR_OK; @@ -485,6 +479,7 @@ complete_view_change_and_become_primary(NodeState *state) state->status = STATUS_NORMAL; state->last_normal_view = state->view_number; + node_log(state, "STATUS NORMAL", "became primary (view change complete)"); // Apply committed entries that haven't been executed yet. // The state machine has only been updated up to the old @@ -506,8 +501,6 @@ complete_view_change_and_become_primary(NodeState *state) add_vote(&entry->votes, self_idx(state)); } - node_log(state, "STATUS NORMAL", "became primary (view change complete)"); - BeginViewMessage begin_view_message = { .base = { .version = MESSAGE_VERSION, @@ -540,11 +533,13 @@ process_do_view_change(NodeState *state, int conn_idx, ByteView msg) message.sender_idx, message.view_number, message.old_view_number, message.op_number, message.commit_index); - // Only process if we're still doing a view change. - // After completing a view change, status becomes NORMAL and - // we must not re-process stale DO_VIEW_CHANGE messages. - if (state->status != STATUS_CHANGE_VIEW) - return HR_OK; + // TODO: This should trigger a view change in replicas running + // under normal operation + // + // VRR 4.2: A replica notices the need for a view change either based + // on its own timer, or because it receives a STARTVIEWCHANGE + // or DOVIEWCHANGE message for a view with a larger number than + // its own view-number. // Only process if the view matches what we're transitioning to if (message.view_number != state->view_number) @@ -556,13 +551,15 @@ process_do_view_change(NodeState *state, int conn_idx, ByteView msg) state->view_change_old_view = message.old_view_number; + LogEntry *entries = (LogEntry*) (msg.ptr + sizeof(DoViewChangeMessage)); + // Parse the variable-sized log from the message int num_entries = (msg.len - sizeof(DoViewChangeMessage)) / sizeof(LogEntry); if (num_entries != message.op_number) return HR_INVALID_MESSAGE; // Message size mismatch log_free(&state->view_change_log); - if (log_init_from_network(&state->view_change_log, msg.ptr + sizeof(DoViewChangeMessage), num_entries) < 0) + if (log_init_from_network(&state->view_change_log, entries, num_entries) < 0) return HR_OUT_OF_MEMORY; } @@ -591,9 +588,6 @@ process_recovery(NodeState *state, int conn_idx, ByteView msg) node_log(state, "RECV RECOVERY", "from=%d nonce=%lu", recovery_message.sender_idx, recovery_message.nonce); - if (state->status != STATUS_NORMAL) - return HR_OK; // Ignore message. - node_log(state, "SEND RECOVERY_RESP", "to=%d view=%lu is_primary=%s", recovery_message.sender_idx, state->view_number, is_leader(state) ? "yes" : "no"); @@ -648,10 +642,10 @@ perform_log_transfer_for_view_change(NodeState *state) .commit_index = state->commit_index, .sender_idx = self_idx(state), }; + send_to_peer_ex(state, leader_idx(state), &do_view_change_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); node_log(state, "SEND DO_VIEW_CHANGE", "to=%d view=%lu old_view=%lu log=%d commit=%d", leader_idx(state), state->view_number, state->last_normal_view, state->log.count, state->commit_index); - send_to_peer_ex(state, leader_idx(state), &do_view_change_message.base, state->log.entries, state->log.count * sizeof(LogEntry)); } // Clear the future array since we're changing views @@ -672,12 +666,6 @@ process_begin_view_change(NodeState *state, int conn_idx, ByteView msg) node_log(state, "RECV BEGIN_VIEW_CHG", "from=%d view=%lu", message.sender_idx, message.view_number); - if (state->status == STATUS_RECOVERY) - return HR_OK; - - assert(state->status == STATUS_NORMAL - || state->status == STATUS_CHANGE_VIEW); - // Ignore old messages if (message.view_number < state->view_number) return HR_OK; @@ -696,7 +684,6 @@ process_begin_view_change(NodeState *state, int conn_idx, ByteView msg) // to the view change state. if (message.view_number > state->view_number) { - // Send our own BeginViewChange to all peers BeginViewChangeMessage message_2 = { .base = { .version = MESSAGE_VERSION, @@ -706,19 +693,22 @@ process_begin_view_change(NodeState *state, int conn_idx, ByteView msg) .view_number = message.view_number, .sender_idx = self_idx(state), }; - node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%lu", message.view_number); broadcast_to_peers(state, &message_2.base); + node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%lu", message.view_number); clear_view_change_fields(state); state->view_number = message.view_number; - node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number); - state->status = STATUS_CHANGE_VIEW; state->heartbeat = state->now; + state->status = STATUS_CHANGE_VIEW; + node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number); } - add_vote(&state->view_change_begin_votes, message.sender_idx); + bool before = reached_quorum(state, state->view_change_begin_votes); + add_vote(&state->view_change_begin_votes, self_idx(state)); - if (reached_quorum(state, state->view_change_begin_votes)) { + add_vote(&state->view_change_begin_votes, message.sender_idx); + + if (!before && reached_quorum(state, state->view_change_begin_votes)) { HandlerResult ret = perform_log_transfer_for_view_change(state); if (ret != HR_OK) return ret; @@ -749,8 +739,10 @@ process_begin_view(NodeState *state, int conn_idx, ByteView msg) state->last_normal_view = state->view_number; node_log(state, "STATUS NORMAL", "new view=%lu (follower)", state->view_number); - // Replace the local log with the authoritative log from the primary int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(LogEntry); + assert(num_entries >= state->commit_index); + + // Replace the local log with the authoritative log from the primary log_free(&state->log); if (log_init_from_network(&state->log, msg.ptr + sizeof(BeginViewMessage), num_entries) < 0) return HR_OUT_OF_MEMORY; @@ -771,9 +763,9 @@ process_begin_view(NodeState *state, int conn_idx, ByteView msg) .log_index = state->log.count - 1, .view_number = state->view_number, }; + send_to_peer(state, leader_idx(state), &ok_msg.base); node_log(state, "SEND PREPARE_OK", "to=%d idx=%d key=%.16s", leader_idx(state), state->log.count - 1, state->log.entries[state->log.count - 1].oper.key); - send_to_peer(state, leader_idx(state), &ok_msg.base); } advance_commit_index(state, message.commit_index, false); @@ -819,6 +811,7 @@ process_get_state(NodeState *state, int conn_idx, ByteView msg) .view_number = state->view_number, .op_number = num_entries, .commit_index = state->commit_index, + .start_index = start, }; node_log(state, "SEND NEW_STATE", "to=%d entries=%d commit=%d", get_state_message.sender_idx, num_entries, state->commit_index); @@ -833,26 +826,37 @@ process_message_as_leader(NodeState *state, { switch (type) { case MESSAGE_TYPE_REQUEST: - return process_request(state, conn_idx, msg); + if (state->status == STATUS_NORMAL) + return process_request(state, conn_idx, msg); + break; case MESSAGE_TYPE_PREPARE_OK: - return process_prepare_ok(state, conn_idx, msg); + if (state->status == STATUS_NORMAL) + return process_prepare_ok(state, conn_idx, msg); + break; case MESSAGE_TYPE_DO_VIEW_CHANGE: - return process_do_view_change(state, conn_idx, msg); + if (state->status == STATUS_CHANGE_VIEW) + return process_do_view_change(state, conn_idx, msg); + break; case MESSAGE_TYPE_RECOVERY: - return process_recovery(state, conn_idx, msg); + if (state->status == STATUS_NORMAL) + return process_recovery(state, conn_idx, msg); + break; case MESSAGE_TYPE_BEGIN_VIEW_CHANGE: - return process_begin_view_change(state, conn_idx, msg); + if (state->status != STATUS_RECOVERY) + return process_begin_view_change(state, conn_idx, msg); + break; case MESSAGE_TYPE_BEGIN_VIEW: - return process_begin_view(state, conn_idx, msg); + if (state->status != STATUS_RECOVERY) + return process_begin_view(state, conn_idx, msg); + break; case MESSAGE_TYPE_RECOVERY_RESPONSE: - // Ignore message. - break; + return HR_OK; case MESSAGE_TYPE_GET_STATE: return process_get_state(state, conn_idx, msg); @@ -947,19 +951,20 @@ process_recovery_response(NodeState *state, ByteView msg) node_log(state, "RECV RECOVERY_RESP", "from=%d view=%lu commit=%d nonce=%lu", message.sender_idx, message.view_number, message.commit_index, message.nonce); - // 1. Only process responses if we are actually in the recovering state [cite: 237] - // 2. Ensure the nonce matches the one we sent to prevent accepting - // delayed responses from previous recovery attempts[cite: 253]. - if (state->status != STATUS_RECOVERY) - return HR_OK; - if (message.nonce != state->recovery_nonce) return HR_OK; + state->recovery_view = MAX(state->recovery_view, message.view_number); + if (should_store_recovery_log(state, message)) { + + LogEntry *entries = (LogEntry*) (msg.ptr + sizeof(RecoveryResponseMessage)); + int num_entries = message.op_number + 1; + + assert(num_entries == (int) ((msg.len - sizeof(RecoveryResponseMessage)) / sizeof(LogEntry))); + log_free(&state->recovery_log); - if (log_init_from_network(&state->recovery_log, msg.ptr + sizeof(RecoveryResponseMessage), - (msg.len - sizeof(RecoveryResponseMessage)) / sizeof(LogEntry)) < 0) + if (log_init_from_network(&state->recovery_log, entries, num_entries) < 0) return HR_OUT_OF_MEMORY; state->recovery_log_view = message.view_number; @@ -967,8 +972,6 @@ process_recovery_response(NodeState *state, ByteView msg) } add_vote(&state->recovery_votes, message.sender_idx); - state->recovery_view = MAX(state->recovery_view, message.view_number); - if (reached_quorum(state, state->recovery_votes) && received_recovery_primary(state)) { HandlerResult ret = complete_recovery(state); if (ret != HR_OK) @@ -1036,14 +1039,6 @@ static int process_future_list(NodeState *state) static HandlerResult process_prepare(NodeState *state, ByteView msg) { - // A recovering node must not process PREPAREs. It doesn't have - // the full log yet and adopting a view from a PREPARE would set - // last_normal_view to a high value with an empty log, which can - // cause a subsequent view change to select this empty log over - // the correct one from other replicas. - if (state->status == STATUS_RECOVERY) - return HR_OK; - PrepareMessage message; if (msg.len != sizeof(message)) return HR_INVALID_MESSAGE; @@ -1057,48 +1052,29 @@ process_prepare(NodeState *state, ByteView msg) message.view_number, message.oper.key, oper_buf); } + // VRR 4.1: If the sender is behind, the receiver drops the message if (message.view_number < state->view_number) - return HR_OK; // Stale peer - - if (message.view_number > state->view_number) { - - // The new leader has started a view we haven't seen. - // Adopt the new view and process the PREPARE normally. - - state->view_number = message.view_number; - - state->status = STATUS_NORMAL; - state->last_normal_view = state->view_number; - - state->view_change_begin_votes = 0; - state->num_future = 0; - state->state_transfer_pending = false; - } - - // A replica that has entered CHANGE_VIEW must not process PREPAREs - // for the current view. It has already committed to the view change - // by broadcasting BEGIN_VIEW_CHANGE and must wait for the view - // change to complete via BEGIN_VIEW. Processing PREPAREs in this - // state would advance the log and commit_index without updating - // last_normal_view, causing a stale old_view_number in any - // subsequent DO_VIEW_CHANGE and potentially leading the new leader - // to select a shorter log over this replica's longer one. - if (state->status == STATUS_CHANGE_VIEW) return HR_OK; + // VRR 4.1: If the sender is ahead, the replica performs a state + // transfer: it requests information it is missing from + // the other replicas and uses this information to bring + // itself up to date before processing the message + if (message.view_number > state->view_number) { + state->view_number = message.view_number; + if (state->num_future < FUTURE_LIMIT) + state->future[state->num_future++] = message; + begin_state_transfer(state, message.sender_idx); + return HR_OK; + } + if (message.log_index < state->log.count) return HR_OK; // Message refers to an old entry. Ignore. if (message.log_index > state->log.count) { - - // The prepare message for a previous log entry was not received yet. - // Buffer this message to process it later (drop if buffer full) - if (state->num_future < FUTURE_LIMIT) { + if (state->num_future < FUTURE_LIMIT) state->future[state->num_future++] = message; - } - - // Request missing entries from the primary via State Transfer - begin_state_transfer(state, leader_idx(state)); + begin_state_transfer(state, message.sender_idx); return HR_OK; } @@ -1122,8 +1098,9 @@ process_prepare(NodeState *state, ByteView msg) .log_index = state->log.count-1, .view_number = state->view_number, }; - node_log(state, "SEND PREPARE_OK", "to=%d idx=%d key=%.16s", message.sender_idx, state->log.count-1, message.oper.key); send_to_peer(state, message.sender_idx, &ok_message.base); + node_log(state, "SEND PREPARE_OK", "to=%d idx=%d key=%.16s", + message.sender_idx, state->log.count-1, message.oper.key); process_future_list(state); advance_commit_index(state, message.commit_index, false); @@ -1144,13 +1121,16 @@ process_commit(NodeState *state, int conn_idx, ByteView msg) node_log(state, "RECV COMMIT", "commit=%d", message.commit_index); - // In RECOVERY and VIEW_CHANGE we ignore COMMIT messages. - // It's important we don't reset the timer so that the - // recovery and view change mechanism can retry on timeout. - if (state->status != STATUS_NORMAL) + if (message.view_number < state->view_number) + return HR_OK; // Stale peer + + if (message.view_number > state->view_number) { + begin_state_transfer(state, message.sender_idx); return HR_OK; + } advance_commit_index(state, message.commit_index, false); + state->heartbeat = state->now; return HR_OK; } @@ -1179,10 +1159,17 @@ process_new_state(NodeState *state, int conn_idx, ByteView msg) if (num_entries == 0) return HR_OK; - // Append received entries to our log + // Append received entries to our log. + // The entries array is a suffix of the sender's log starting at + // global position start_index. We skip entries we already have. LogEntry *entries = (LogEntry *)((uint8_t *)msg.ptr + sizeof(NewStateMessage)); + int start_index = new_state_message.start_index; for (int i = 0; i < num_entries; i++) { + int global_idx = start_index + i; + if (global_idx < state->log.count) + continue; // Already have this entry + LogEntry entry = { .oper = entries[i].oper, .votes = 0, @@ -1204,9 +1191,9 @@ process_new_state(NodeState *state, int conn_idx, ByteView msg) .log_index = state->log.count - 1, .view_number = state->view_number, }; + send_to_peer(state, leader_idx(state), &prepare_ok_message.base); node_log(state, "SEND PREPARE_OK", "to=%d idx=%d key=%.16s", leader_idx(state), state->log.count - 1, state->log.entries[state->log.count - 1].oper.key); - send_to_peer(state, leader_idx(state), &prepare_ok_message.base); } process_future_list(state); @@ -1251,22 +1238,34 @@ process_message_as_replica(NodeState *state, break; case MESSAGE_TYPE_PREPARE: - return process_prepare(state, msg); + if (state->status == STATUS_NORMAL) + return process_prepare(state, msg); + break; case MESSAGE_TYPE_COMMIT: - return process_commit(state, conn_idx, msg); + if (state->status == STATUS_NORMAL) + return process_commit(state, conn_idx, msg); + break; case MESSAGE_TYPE_BEGIN_VIEW_CHANGE: - return process_begin_view_change(state, conn_idx, msg); + if (state->status != STATUS_RECOVERY) + return process_begin_view_change(state, conn_idx, msg); + break; case MESSAGE_TYPE_BEGIN_VIEW: - return process_begin_view(state, conn_idx, msg); + if (state->status != STATUS_RECOVERY) + return process_begin_view(state, conn_idx, msg); + break; case MESSAGE_TYPE_RECOVERY: - return process_recovery(state, conn_idx, msg); + if (state->status == STATUS_NORMAL) + return process_recovery(state, conn_idx, msg); + break; case MESSAGE_TYPE_RECOVERY_RESPONSE: - return process_recovery_response(state, msg); + if (state->status == STATUS_RECOVERY) + return process_recovery_response(state, msg); + break; case MESSAGE_TYPE_NEW_STATE: return process_new_state(state, conn_idx, msg); @@ -1337,13 +1336,12 @@ process_message(NodeState *state, if (existing < 0) { // No connection tagged with this peer yet, tag this one tcp_set_tag(&state->tcp, conn_idx, sender_idx, false); - } else if (existing != conn_idx) { - // A different (possibly stale) connection has this tag. - // Close the old one and tag the current one. - tcp_close(&state->tcp, existing); - tcp_set_tag(&state->tcp, conn_idx, sender_idx, false); } - // If existing == conn_idx, already tagged correctly + // If a different connection is already tagged for this peer, + // keep it. Closing it would also disconnect the peer end, + // which may be carrying data in the opposite direction (e.g. + // a DO_VIEW_CHANGE queued on the cross-connection). Stale + // connections are detected and cleaned up when sends fail. } } @@ -1493,6 +1491,7 @@ int node_init(void *state_, int argc, char **argv, if (previously_crashed) { node_log(state, "STATUS RECOVERY", "nonce=%lu (crash detected)", state->recovery_nonce); + // Broadcast RECOVERY to all peers to learn the current view RecoveryMessage recovery_message = { .base = { @@ -1503,8 +1502,9 @@ int node_init(void *state_, int argc, char **argv, .sender_idx = self_idx(state), .nonce = state->recovery_nonce, }; - node_log(state, "SEND RECOVERY", "to=* nonce=%lu", state->recovery_nonce); broadcast_to_peers(state, &recovery_message.base); + node_log(state, "SEND RECOVERY", "to=* nonce=%lu", state->recovery_nonce); + nearest_deadline(&deadline, state->recovery_time + RECOVERY_TIMEOUT_SEC * 1000000000ULL); } @@ -1571,13 +1571,13 @@ int node_tick(void *state_, void **ctxs, Time deadline = INVALID_TIME; if (state->status == STATUS_RECOVERY) { - // Recovery handling runs regardless of leader/replica position, // since a recovering node must not act as leader until it learns // the current view from its peers. Time recovery_deadline = state->recovery_time + RECOVERY_TIMEOUT_SEC * 1000000000ULL; if (recovery_deadline <= state->now) { node_log_simple(state, "TIMEOUT RECOVERY"); + RecoveryMessage recovery_message = { .base = { .version = MESSAGE_VERSION, @@ -1587,67 +1587,29 @@ int node_tick(void *state_, void **ctxs, .sender_idx = self_idx(state), .nonce = state->recovery_nonce, }; - node_log(state, "SEND RECOVERY", "to=* nonce=%lu", state->recovery_nonce); broadcast_to_peers(state, &recovery_message.base); + node_log(state, "SEND RECOVERY", "to=* nonce=%lu", state->recovery_nonce); + state->recovery_votes = 0; state->recovery_log_view = 0; state->recovery_time = state->now; + } else { nearest_deadline(&deadline, recovery_deadline); } + } else if (state->status == STATUS_CHANGE_VIEW) { - } else if (is_leader(state)) { + Time view_change_deadline = state->heartbeat + VIEW_CHANGE_TIMEOUT_SEC * 1000000000ULL; + if (view_change_deadline <= state->now) { - Time heartbeat_deadline = state->heartbeat + HEARTBEAT_INTERVAL_SEC * 1000000000ULL; - if (heartbeat_deadline <= state->now) { // TODO: check the time conversion here - CommitMessage commit_message = { - .base = { - .version = MESSAGE_VERSION, - .type = MESSAGE_TYPE_COMMIT, - .length = sizeof(CommitMessage), - }, - .commit_index = state->commit_index, - }; - node_log(state, "SEND COMMIT", "to=* commit=%d", state->commit_index); - broadcast_to_peers(state, &commit_message.base); + node_log_simple(state, "TIMEOUT CHANGE_VIEW"); + + clear_view_change_fields(state); + + add_vote(&state->view_change_begin_votes, self_idx(state)); + + state->view_number++; state->heartbeat = state->now; - } else { - nearest_deadline(&deadline, heartbeat_deadline); - } - - } else { - - // State transfer retry: if we're waiting for missing log entries - // and the timeout has elapsed, re-send GET_STATE to the primary. - if (state->state_transfer_pending) { - - Time st_deadline = state->state_transfer_time + STATE_TRANSFER_TIMEOUT_SEC * 1000000000ULL; - if (st_deadline <= state->now) { - node_log(state, "TIMEOUT STATE_TRANSFER", "op=%d", state->log.count); - GetStateMessage get_state_message = { - .base = { - .version = MESSAGE_VERSION, - .type = MESSAGE_TYPE_GET_STATE, - .length = sizeof(GetStateMessage), - }, - .view_number = state->view_number, - .op_number = state->log.count, - .sender_idx = self_idx(state), - }; - node_log(state, "SEND GET_STATE", "to=%d op=%d", leader_idx(state), state->log.count); - send_to_peer(state, leader_idx(state), &get_state_message.base); - state->state_transfer_time = state->now; - } else { - nearest_deadline(&deadline, st_deadline); - } - } - - int death_timeout = (state->status == STATUS_CHANGE_VIEW) - ? VIEW_CHANGE_TIMEOUT_SEC : PRIMARY_DEATH_TIMEOUT_SEC; - Time death_deadline = state->heartbeat + death_timeout * 1000000000ULL; - if (death_deadline <= state->now) { - - node_log_simple(state, "TIMEOUT PRIMARY_DEATH"); BeginViewChangeMessage begin_view_change_message = { .base = { @@ -1655,18 +1617,99 @@ int node_tick(void *state_, void **ctxs, .type = MESSAGE_TYPE_BEGIN_VIEW_CHANGE, .length = sizeof(BeginViewChangeMessage), }, - .view_number = state->view_number + 1, + .view_number = state->view_number, .sender_idx = self_idx(state), }; - node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%lu", state->view_number + 1); + node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%lu", state->view_number); broadcast_to_peers(state, &begin_view_change_message.base); - state->status = STATUS_CHANGE_VIEW; - node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number + 1); - state->heartbeat = state->now; + } else { + nearest_deadline(&deadline, view_change_deadline); + } + } else { + assert(state->status == STATUS_NORMAL); + + if (is_leader(state)) { + Time heartbeat_deadline = state->heartbeat + HEARTBEAT_INTERVAL_SEC * 1000000000ULL; + if (heartbeat_deadline <= state->now) { // TODO: check the time conversion here + + CommitMessage commit_message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_COMMIT, + .length = sizeof(CommitMessage), + }, + .view_number = state->view_number, + .sender_idx = self_idx(state), + .commit_index = state->commit_index, + }; + broadcast_to_peers(state, &commit_message.base); + node_log(state, "SEND COMMIT", "to=* commit=%d", state->commit_index); + + state->heartbeat = state->now; + + } else { + nearest_deadline(&deadline, heartbeat_deadline); + } + } else { + Time death_deadline = state->heartbeat + PRIMARY_DEATH_TIMEOUT_SEC * 1000000000ULL; + if (death_deadline <= state->now) { + + node_log_simple(state, "TIMEOUT PRIMARY_DEATH"); + + clear_view_change_fields(state); + + add_vote(&state->view_change_begin_votes, self_idx(state)); + + state->view_number++; + state->status = STATUS_CHANGE_VIEW; + state->heartbeat = state->now; + + BeginViewChangeMessage begin_view_change_message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_BEGIN_VIEW_CHANGE, + .length = sizeof(BeginViewChangeMessage), + }, + .view_number = state->view_number, + .sender_idx = self_idx(state), + }; + node_log(state, "SEND BEGIN_VIEW_CHG", "to=* view=%lu", state->view_number); + broadcast_to_peers(state, &begin_view_change_message.base); + + node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number); + + } else { + nearest_deadline(&deadline, death_deadline); + } + } + } + + // State transfer retry: if we're waiting for missing log entries + // and the timeout has elapsed, re-send GET_STATE to the primary. + if (state->state_transfer_pending) { + + Time st_deadline = state->state_transfer_time + STATE_TRANSFER_TIMEOUT_SEC * 1000000000ULL; + if (st_deadline <= state->now) { + node_log(state, "TIMEOUT STATE_TRANSFER", "op=%d", state->log.count); + + GetStateMessage get_state_message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_GET_STATE, + .length = sizeof(GetStateMessage), + }, + .view_number = state->view_number, + .op_number = state->log.count, + .sender_idx = self_idx(state), + }; + send_to_peer(state, leader_idx(state), &get_state_message.base); + node_log(state, "SEND GET_STATE", "to=%d op=%d", leader_idx(state), state->log.count); + + state->state_transfer_time = state->now; } else { - nearest_deadline(&deadline, death_deadline); + nearest_deadline(&deadline, st_deadline); } } diff --git a/vsr/node.h b/vsr/node.h index 228bbc5..cb8ccfa 100644 --- a/vsr/node.h +++ b/vsr/node.h @@ -64,6 +64,8 @@ typedef struct { typedef struct { MessageHeader base; + uint64_t view_number; + int sender_idx; int commit_index; } CommitMessage; @@ -125,6 +127,7 @@ typedef struct { uint64_t view_number; int op_number; // Number of log entries that follow int commit_index; + int start_index; // Global log index of the first entry in the suffix // Followed by: LogEntry log[op_number] } NewStateMessage;