2 // Replicated state machine implementation with a primary and several
3 // backups. The primary receives requests, assigns each a view stamp (a
4 // vid, and a sequence number) in the order of reception, and forwards
5 // them to all backups. A backup executes requests in the order that
6 // the primary stamps them and replies with an OK to the primary. The
7 // primary executes the request after it receives OKs from all backups,
8 // and sends the reply back to the client.
10 // The config module will tell the RSM about a new view. If the
11 // primary in the previous view is a member of the new view, then it
12 // will stay the primary. Otherwise, the smallest numbered node of
13 // the previous view will be the new primary. In either case, the new
14 // primary will be a node from the previous view. The configuration
15 // module constructs the sequence of views for the RSM and the RSM
16 // ensures there will be always one primary, who was a member of the
19 // When a new node starts, the recovery thread is in charge of joining
20 // the RSM. It will collect the internal RSM state from the primary;
21 // the primary asks the config module to add the new node and returns
22 // to the joining the internal RSM state (e.g., paxos log). Since
23 // there is only one primary, all joins happen in well-defined total
26 // The recovery thread also runs during a view change (e.g, when a node
27 // has failed). After a failure some of the backups could have
28 // processed a request that the primary has not, but those results are
29 // not visible to clients (since the primary responds). If the
30 // primary of the previous view is in the current view, then it will
31 // be the primary and its state is authoritive: the backups download
32 // from the primary the current state. A primary waits until all
33 // backups have downloaded the state. Once the RSM is in sync, the
34 // primary accepts requests again from clients. If one of the backups
35 // is the new primary, then its state is authoritative. In either
36 // scenario, the next view uses a node as primary that has the state
37 // resulting from processing all acknowledged client requests.
38 // Therefore, if the nodes sync up before processing the next request,
39 // the next view will have the correct state.
41 // While the RSM in a view change (i.e., a node has failed, a new view
42 // has been formed, but the sync hasn't completed), another failure
43 // could happen, which complicates a view change. During syncing the
44 // primary or backups can timeout, and initiate another Paxos round.
45 // There are 2 variables that RSM uses to keep track in what state it
47 // - inviewchange: a node has failed and the RSM is performing a view change
48 // - insync: this node is syncing its state
50 // If inviewchange is false and a node is the primary, then it can
51 // process client requests. If it is true, clients are told to retry
52 // later again. While inviewchange is true, the RSM may go through several
53 // member list changes, one by one. After a member list
54 // change completes, the nodes tries to sync. If the sync complets,
55 // the view change completes (and inviewchange is set to false). If
56 // the sync fails, the node may start another member list change
57 // (inviewchange = true and insync = false).
59 // The implementation should be used only with servers that run all
60 // requests run to completion; in particular, a request shouldn't
61 // block. If a request blocks, the backup won't respond to the
62 // primary, and the primary won't execute the request. A request may
63 // send an RPC to another host, but the RPC should be a one-way
64 // message to that host; the backup shouldn't do anything based on the
65 // response or execute after the response, because it is not
66 // guaranteed that all backup will receive the same response and
67 // execute in the same order.
69 // The implementation can be viewed as a layered system:
70 // RSM module ---- in charge of replication
71 // config module ---- in charge of view management
72 // Paxos module ---- in charge of running Paxos to agree on a value
74 // Each module has threads and internal locks. Furthermore, a thread
75 // may call down through the layers (e.g., to run Paxos's proposer).
76 // When Paxos's acceptor accepts a new value for an instance, a thread
77 // will invoke an upcall to inform higher layers of the new value.
78 // The rule is that a module releases its internal locks before it
79 // upcalls, but can keep its locks when calling down.
84 #include <sys/types.h>
90 #include "lang/verify.h"
91 #include "rsm_client.h"
94 rsm::rsm(std::string _first, std::string _me) :
95 stf(0), primary(_first), insync (false), inviewchange (true), vid_commit(0),
96 partitioned (false), dopartition(false), break1(false), break2(false)
103 cfg = new config(_first, _me, this);
106 // Commit the first view here. We can not have acceptor::acceptor
107 // do the commit, since at that time this->cfg is not initialized
110 rsmrpc = cfg->get_rpcs();
111 rsmrpc->reg(rsm_client_protocol::invoke, &rsm::client_invoke, this);
112 rsmrpc->reg(rsm_client_protocol::members, &rsm::client_members, this);
113 rsmrpc->reg(rsm_protocol::invoke, &rsm::invoke, this);
114 rsmrpc->reg(rsm_protocol::transferreq, &rsm::transferreq, this);
115 rsmrpc->reg(rsm_protocol::transferdonereq, &rsm::transferdonereq, this);
116 rsmrpc->reg(rsm_protocol::joinreq, &rsm::joinreq, this);
118 // tester must be on different port, otherwise it may partition itself
119 testsvr = new rpcs((uint32_t)std::stoi(_me) + 1);
120 testsvr->reg(rsm_test_protocol::net_repair, &rsm::test_net_repairreq, this);
121 testsvr->reg(rsm_test_protocol::breakpoint, &rsm::breakpointreq, this);
125 std::thread(&rsm::recovery, this).detach();
129 void rsm::reg1(int proc, handler *h) {
134 // The recovery thread runs this function
135 void rsm::recovery() [[noreturn]] {
140 while (!cfg->ismember(cfg->myaddr(), vid_commit)) {
141 // XXX iannucci 2013/09/15 -- I don't understand whether accessing
142 // cfg->view_id in this manner involves a race. I suspect not.
144 LOG("recovery: joined");
145 commit_change_wo(cfg->view_id());
148 std::this_thread::sleep_for(std::chrono::seconds(30)); // XXX make another node in cfg primary?
152 vid_insync = vid_commit;
153 LOG("recovery: sync vid_insync " << vid_insync);
154 if (primary == cfg->myaddr()) {
155 r = sync_with_backups();
157 r = sync_with_primary();
159 LOG("recovery: sync done");
161 // If there was a commited viewchange during the synchronization, restart
163 if (vid_insync != vid_commit)
167 myvs.vid = vid_commit;
169 inviewchange = false;
171 LOG("recovery: go to sleep " << insync << " " << inviewchange);
172 recovery_cond.wait(ml);
177 std::ostream & operator<<(std::ostream &o, const std::vector<A> &d) {
179 for (typename std::vector<A>::const_iterator i=d.begin(); i!=d.end(); i++) {
188 bool rsm::sync_with_backups() {
189 adopt_lock ml(rsm_mutex);
192 // Make sure that the state of lock_server is stable during
193 // synchronization; otherwise, the primary's state may be more recent
194 // than replicas after the synchronization.
195 lock ml2(invoke_mutex);
196 // By acquiring and releasing the invoke_mutex once, we make sure that
197 // the state of lock_server will not be changed until all
198 // replicas are synchronized. The reason is that client_invoke arrives
199 // after this point of time will see inviewchange == true, and returns
203 // Start accepting synchronization request (statetransferreq) now!
205 cfg->get_view(vid_insync, backups);
206 backups.erase(find(backups.begin(), backups.end(), cfg->myaddr()));
207 LOG("rsm::sync_with_backups " << backups);
214 bool rsm::sync_with_primary() {
215 // Remember the primary of vid_insync
216 std::string m = primary;
217 while (vid_insync == vid_commit) {
218 if (statetransfer(m))
221 return statetransferdone(m);
226 * Call to transfer state from m to the local node.
227 * Assumes that rsm_mutex is already held.
229 bool rsm::statetransfer(std::string m)
231 rsm_protocol::transferres r;
234 tprintf("rsm::statetransfer: contact %s w. my last_myvs(%d,%d)\n",
235 m.c_str(), last_myvs.vid, last_myvs.seqno);
238 adopt_lock ml(rsm_mutex);
242 ret = cl->call_timeout(rsm_protocol::transferreq, rpcc::to(1000),
243 r, cfg->myaddr(), last_myvs, vid_insync);
247 if (cl == 0 || ret != rsm_protocol::OK) {
248 tprintf("rsm::statetransfer: couldn't reach %s %lx %d\n", m.c_str(),
249 (long unsigned) cl, ret);
252 if (stf && last_myvs != r.last) {
253 stf->unmarshal_state(r.state);
256 tprintf("rsm::statetransfer transfer from %s success, vs(%d,%d)\n",
257 m.c_str(), last_myvs.vid, last_myvs.seqno);
261 bool rsm::statetransferdone(std::string m) {
262 adopt_lock ml(rsm_mutex);
265 rpcc *cl = h.safebind();
269 rsm_protocol::status ret = cl->call(rsm_protocol::transferdonereq, r, cfg->myaddr(), vid_insync);
270 done = (ret == rsm_protocol::OK);
277 bool rsm::join(std::string m) {
280 rsm_protocol::joinres r;
282 LOG("rsm::join: " << m << " mylast (" << last_myvs.vid << "," << last_myvs.seqno << ")");
285 adopt_lock ml(rsm_mutex);
289 ret = cl->call_timeout(rsm_protocol::joinreq, rpcc::to(120000), r,
290 cfg->myaddr(), last_myvs);
295 if (cl == 0 || ret != rsm_protocol::OK) {
296 LOG("rsm::join: couldn't reach " << m << " " << std::hex << cl << " " << std::dec << ret);
299 LOG("rsm::join: succeeded " << r.log);
305 * Config informs rsm whenever it has successfully
306 * completed a view change
308 void rsm::commit_change(unsigned vid) {
310 commit_change_wo(vid);
311 if (cfg->ismember(cfg->myaddr(), vid_commit))
315 void rsm::commit_change_wo(unsigned vid) {
316 if (vid <= vid_commit)
318 tprintf("commit_change: new view (%d) last vs (%d,%d) %s insync %d\n",
319 vid, last_myvs.vid, last_myvs.seqno, primary.c_str(), insync);
323 recovery_cond.notify_one();
324 sync_cond.notify_one();
325 if (cfg->ismember(cfg->myaddr(), vid_commit))
330 void rsm::execute(int procno, std::string req, std::string &r) {
331 tprintf("execute\n");
332 handler *h = procs[procno];
334 unmarshall args(req);
337 rsm_protocol::status ret = (*h)(args, rep);
345 // Clients call client_invoke to invoke a procedure on the replicated state
346 // machine: the primary receives the request, assigns it a sequence
347 // number, and invokes it on all members of the replicated state
350 rsm_client_protocol::status rsm::client_invoke(std::string &r, int procno, std::string req) {
351 LOG("rsm::client_invoke: procno 0x" << std::hex << procno);
352 lock ml(invoke_mutex);
353 std::vector<std::string> m;
358 LOG("Checking for inviewchange");
360 return rsm_client_protocol::BUSY;
361 LOG("Checking for primacy");
362 myaddr = cfg->myaddr();
363 if (primary != myaddr)
364 return rsm_client_protocol::NOTPRIMARY;
365 LOG("Assigning a viewstamp");
366 cfg->get_view(vid_commit, m);
367 // assign the RPC the next viewstamp number
372 // send an invoke RPC to all slaves in the current view with a timeout of 1 second
373 LOG("Invoking slaves");
374 for (unsigned i = 0; i < m.size(); i++) {
375 if (m[i] != myaddr) {
376 // if invoke on slave fails, return rsm_client_protocol::BUSY
378 LOG("Sending invoke to " << m[i]);
379 rpcc *cl = h.safebind();
381 return rsm_client_protocol::BUSY;
382 rsm_protocol::status ret;
384 ret = cl->call_timeout(rsm_protocol::invoke, rpcc::to(1000), ignored_rval, procno, vs, req);
385 LOG("Invoke returned " << ret);
386 if (ret != rsm_protocol::OK)
387 return rsm_client_protocol::BUSY;
392 execute(procno, req, r);
394 return rsm_client_protocol::OK;
398 // The primary calls the internal invoke at each member of the
399 // replicated state machine
401 // the replica must execute requests in order (with no gaps)
402 // according to requests' seqno
404 rsm_protocol::status rsm::invoke(int &, int proc, viewstamp vs, std::string req) {
405 LOG("rsm::invoke: procno 0x" << std::hex << proc);
406 lock ml(invoke_mutex);
407 std::vector<std::string> m;
411 // check if !inviewchange
412 LOG("Checking for view change");
414 return rsm_protocol::ERR;
416 LOG("Checking for slave status");
417 myaddr = cfg->myaddr();
418 if (primary == myaddr)
419 return rsm_protocol::ERR;
420 cfg->get_view(vid_commit, m);
421 if (find(m.begin(), m.end(), myaddr) == m.end())
422 return rsm_protocol::ERR;
423 // check sequence number
424 LOG("Checking sequence number");
426 return rsm_protocol::ERR;
430 execute(proc, req, r);
433 return rsm_protocol::OK;
437 * RPC handler: Send back the local node's state to the caller
439 rsm_protocol::status rsm::transferreq(rsm_protocol::transferres &r, std::string src,
440 viewstamp last, unsigned vid) {
442 int ret = rsm_protocol::OK;
443 tprintf("transferreq from %s (%d,%d) vs (%d,%d)\n", src.c_str(),
444 last.vid, last.seqno, last_myvs.vid, last_myvs.seqno);
445 if (!insync || vid != vid_insync) {
446 return rsm_protocol::BUSY;
448 if (stf && last != last_myvs)
449 r.state = stf->marshal_state();
455 * RPC handler: Inform the local node (the primary) that node m has synchronized
458 rsm_protocol::status rsm::transferdonereq(int &, std::string m, unsigned vid) {
460 if (!insync || vid != vid_insync)
461 return rsm_protocol::BUSY;
462 backups.erase(find(backups.begin(), backups.end(), m));
464 sync_cond.notify_one();
465 return rsm_protocol::OK;
468 // a node that wants to join an RSM as a server sends a
469 // joinreq to the RSM's current primary; this is the
470 // handler for that RPC.
471 rsm_protocol::status rsm::joinreq(rsm_protocol::joinres &r, std::string m, viewstamp last) {
472 int ret = rsm_protocol::OK;
475 tprintf("joinreq: src %s last (%d,%d) mylast (%d,%d)\n", m.c_str(),
476 last.vid, last.seqno, last_myvs.vid, last_myvs.seqno);
477 if (cfg->ismember(m, vid_commit)) {
478 tprintf("joinreq: is still a member\n");
480 } else if (cfg->myaddr() != primary) {
481 tprintf("joinreq: busy\n");
482 ret = rsm_protocol::BUSY;
484 // We cache vid_commit to avoid adding m to a view which already contains
485 // m due to race condition
486 unsigned vid_cache = vid_commit;
490 succ = cfg->add(m, vid_cache);
493 if (cfg->ismember(m, cfg->view_id())) {
495 tprintf("joinreq: ret %d log %s\n:", ret, r.log.c_str());
497 tprintf("joinreq: failed; proposer couldn't add %d\n", succ);
498 ret = rsm_protocol::BUSY;
505 * RPC handler: Send back all the nodes this local knows about to client
506 * so the client can switch to a different primary
507 * when it existing primary fails
509 rsm_client_protocol::status rsm::client_members(std::vector<std::string> &r, int) {
510 std::vector<std::string> m;
512 cfg->get_view(vid_commit, m);
513 m.push_back(primary);
515 tprintf("rsm::client_members return %s m %s\n", print_members(m).c_str(),
517 return rsm_client_protocol::OK;
520 // if primary is member of new view, that node is primary
521 // otherwise, the lowest number node of the previous view.
522 // caller should hold rsm_mutex
523 void rsm::set_primary(unsigned vid) {
524 std::vector<std::string> c, p;
525 cfg->get_view(vid, c);
526 cfg->get_view(vid - 1, p);
527 VERIFY (c.size() > 0);
529 if (isamember(primary,c)) {
530 tprintf("set_primary: primary stays %s\n", primary.c_str());
534 VERIFY(p.size() > 0);
535 for (unsigned i = 0; i < p.size(); i++) {
536 if (isamember(p[i], c)) {
538 tprintf("set_primary: primary is %s\n", primary.c_str());
545 bool rsm::amiprimary() {
547 return primary == cfg->myaddr() && !inviewchange;
553 // Simulate partitions
555 // assumes caller holds rsm_mutex
556 void rsm::net_repair_wo(bool heal) {
557 std::vector<std::string> m;
558 cfg->get_view(vid_commit, m);
559 for (unsigned i = 0; i < m.size(); i++) {
560 if (m[i] != cfg->myaddr()) {
562 tprintf("rsm::net_repair_wo: %s %d\n", m[i].c_str(), heal);
563 if (h.safebind()) h.safebind()->set_reachable(heal);
566 rsmrpc->set_reachable(heal);
569 rsm_test_protocol::status rsm::test_net_repairreq(int &r, int heal) {
571 tprintf("rsm::test_net_repairreq: %d (dopartition %d, partitioned %d)\n",
572 heal, dopartition, partitioned);
580 r = rsm_test_protocol::OK;
584 // simulate failure at breakpoint 1 and 2
586 void rsm::breakpoint1() {
588 tprintf("Dying at breakpoint 1 in rsm!\n");
593 void rsm::breakpoint2() {
595 tprintf("Dying at breakpoint 2 in rsm!\n");
600 void rsm::partition1() {
602 net_repair_wo(false);
608 rsm_test_protocol::status rsm::breakpointreq(int &r, int b) {
609 r = rsm_test_protocol::OK;
611 tprintf("rsm::breakpointreq: %d\n", b);
612 if (b == 1) break1 = true;
613 else if (b == 2) break2 = true;
614 else if (b == 3 || b == 4) cfg->breakpoint(b);
615 else r = rsm_test_protocol::ERR;