X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/5d99dbf06a14904944f5593c63705934bdfdcfb7..4e881433f37417ccbda89c09ffdf936855d462d4:/rsm.cc diff --git a/rsm.cc b/rsm.cc index 843418a..956f45d 100644 --- a/rsm.cc +++ b/rsm.cc @@ -78,24 +78,16 @@ // The rule is that a module releases its internal locks before it // upcalls, but can keep its locks when calling down. -#include -#include - -#include "types.h" -#include "handle.h" #include "rsm.h" +#include "handle.h" #include "rsm_client.h" +#include -rsm::rsm(string _first, string _me) : +rsm::rsm(const string & _first, const string & _me) : stf(0), primary(_first), insync (false), inviewchange (true), vid_commit(0), partitioned (false), dopartition(false), break1(false), break2(false) { - last_myvs.vid = 0; - last_myvs.seqno = 0; - myvs = last_myvs; - myvs.seqno = 1; - - cfg = new config(_first, _me, this); + cfg = unique_ptr(new config(_first, _me, this)); if (_first == _me) { // Commit the first view here. We can not have acceptor::acceptor @@ -111,23 +103,25 @@ rsm::rsm(string _first, string _me) : rsmrpc->reg(rsm_protocol::joinreq, &rsm::joinreq, this); // tester must be on different port, otherwise it may partition itself - testsvr = new rpcs((uint32_t)stoi(_me) + 1); + testsvr = unique_ptr(new rpcs((in_port_t)stoi(_me) + 1)); testsvr->reg(rsm_test_protocol::net_repair, &rsm::test_net_repairreq, this); testsvr->reg(rsm_test_protocol::breakpoint, &rsm::breakpointreq, this); +} - { - lock ml(rsm_mutex); - thread(&rsm::recovery, this).detach(); - } +void rsm::start() { + lock ml(rsm_mutex); + rsmrpc->start(); + testsvr->start(); + thread(&rsm::recovery, this).detach(); } -void rsm::reg1(int proc, handler *h) { +void rsm::reg1(rpc_protocol::proc_id_t proc, handler *h) { lock ml(rsm_mutex); procs[proc] = h; } // The recovery thread runs this function -void rsm::recovery() [[noreturn]] { +void rsm::recovery() { bool r = true; lock ml(rsm_mutex); @@ -136,22 +130,22 @@ void rsm::recovery() [[noreturn]] { // XXX iannucci 2013/09/15 -- I don't understand whether accessing // cfg->view_id in this manner involves a race. I suspect not. if (join(primary, ml)) { - LOG("recovery: joined"); + LOG("joined"); commit_change(cfg->view_id(), ml); } else { ml.unlock(); - this_thread::sleep_for(seconds(30)); // XXX make another node in cfg primary? + this_thread::sleep_for(seconds(3)); // XXX make another node in cfg primary? ml.lock(); } } vid_insync = vid_commit; - LOG("recovery: sync vid_insync " << vid_insync); + LOG("sync vid_insync " << vid_insync); if (primary == cfg->myaddr()) { r = sync_with_backups(ml); } else { r = sync_with_primary(ml); } - LOG("recovery: sync done"); + LOG("sync done"); // If there was a commited viewchange during the synchronization, restart // the recovery @@ -163,7 +157,7 @@ void rsm::recovery() [[noreturn]] { myvs.seqno = 1; inviewchange = false; } - LOG("recovery: go to sleep " << insync << " " << inviewchange); + LOG("go to sleep " << insync << " " << inviewchange); recovery_cond.wait(ml); } } @@ -186,7 +180,7 @@ bool rsm::sync_with_backups(lock & rsm_mutex_lock) { insync = true; cfg->get_view(vid_insync, backups); backups.erase(find(backups.begin(), backups.end(), cfg->myaddr())); - LOG("rsm::sync_with_backups " << backups); + LOG("backups " << backups); sync_cond.wait(rsm_mutex_lock); insync = false; return true; @@ -204,39 +198,39 @@ bool rsm::sync_with_primary(lock & rsm_mutex_lock) { } -/** - * Call to transfer state from m to the local node. - * Assumes that rsm_mutex is already held. - */ -bool rsm::statetransfer(string m, lock & rsm_mutex_lock) +// +// Call to transfer state from m to the local node. +// Assumes that rsm_mutex is already held. +// +bool rsm::statetransfer(const string & m, lock & rsm_mutex_lock) { rsm_protocol::transferres r; handle h(m); int ret = 0; - LOG("rsm::statetransfer: contact " << m << " w. my last_myvs(" << last_myvs.vid << "," << last_myvs.seqno << ")"); + LOG("contact " << m << " w. my last_myvs(" << last_myvs.vid << "," << last_myvs.seqno << ")"); rpcc *cl; { rsm_mutex_lock.unlock(); cl = h.safebind(); if (cl) { - ret = cl->call_timeout(rsm_protocol::transferreq, rpcc::to(1000), + ret = cl->call_timeout(rsm_protocol::transferreq, milliseconds(100), r, cfg->myaddr(), last_myvs, vid_insync); } rsm_mutex_lock.lock(); } if (cl == 0 || ret != rsm_protocol::OK) { - LOG("rsm::statetransfer: couldn't reach " << m << " " << hex << cl << " " << dec << ret); + LOG("couldn't reach " << m << " " << hex << cl << " " << dec << ret); return false; } if (stf && last_myvs != r.last) { stf->unmarshal_state(r.state); } last_myvs = r.last; - LOG("rsm::statetransfer transfer from " << m << " success, vs(" << last_myvs.vid << "," << last_myvs.seqno << ")"); + LOG("transfer from " << m << " success, vs(" << last_myvs.vid << "," << last_myvs.seqno << ")"); return true; } -bool rsm::statetransferdone(string m, lock & rsm_mutex_lock) { +bool rsm::statetransferdone(const string & m, lock & rsm_mutex_lock) { rsm_mutex_lock.unlock(); handle h(m); rpcc *cl = h.safebind(); @@ -251,36 +245,36 @@ bool rsm::statetransferdone(string m, lock & rsm_mutex_lock) { } -bool rsm::join(string m, lock & rsm_mutex_lock) { +bool rsm::join(const string & m, lock & rsm_mutex_lock) { handle h(m); int ret = 0; string log; - LOG("rsm::join: " << m << " mylast (" << last_myvs.vid << "," << last_myvs.seqno << ")"); + LOG("contacting " << m << " mylast (" << last_myvs.vid << "," << last_myvs.seqno << ")"); rpcc *cl; { rsm_mutex_lock.unlock(); cl = h.safebind(); if (cl != 0) { - ret = cl->call_timeout(rsm_protocol::joinreq, rpcc::to(120000), log, + ret = cl->call_timeout(rsm_protocol::joinreq, milliseconds(12000), log, cfg->myaddr(), last_myvs); } rsm_mutex_lock.lock(); } if (cl == 0 || ret != rsm_protocol::OK) { - LOG("rsm::join: couldn't reach " << m << " " << hex << cl << " " << dec << ret); + LOG("couldn't reach " << m << " " << hex << cl << " " << dec << ret); return false; } - LOG("rsm::join: succeeded " << log); + LOG("succeeded " << log); cfg->restore(log); return true; } -/* - * Config informs rsm whenever it has successfully - * completed a view change - */ +// +// Config informs rsm whenever it has successfully +// completed a view change +// void rsm::commit_change(unsigned vid) { lock ml(rsm_mutex); commit_change(vid, ml); @@ -291,7 +285,7 @@ void rsm::commit_change(unsigned vid) { void rsm::commit_change(unsigned vid, lock &) { if (vid <= vid_commit) return; - LOG("commit_change: new view (" << vid << ") last vs (" << last_myvs.vid << "," << + LOG("new view (" << vid << ") last vs (" << last_myvs.vid << "," << last_myvs.seqno << ") " << primary << " insync " << insync); vid_commit = vid; inviewchange = true; @@ -303,18 +297,14 @@ void rsm::commit_change(unsigned vid, lock &) { } -void rsm::execute(int procno, string req, string &r) { +void rsm::execute(rpc_protocol::proc_id_t procno, const string & req, string & r) { LOG("execute"); handler *h = procs[procno]; VERIFY(h); unmarshall args(req, false); marshall rep; - string reps; auto ret = (rsm_protocol::status)(*h)(args, rep); - marshall rep1; - rep1 << ret; - rep1 << rep.content(); - r = rep1.content(); + r = marshall{ret, rep.content()}.content(); } // @@ -323,8 +313,8 @@ void rsm::execute(int procno, string req, string &r) { // number, and invokes it on all members of the replicated state // machine. // -rsm_client_protocol::status rsm::client_invoke(string &r, int procno, string req) { - LOG("rsm::client_invoke: procno 0x" << hex << procno); +rsm_client_protocol::status rsm::client_invoke(string & r, rpc_protocol::proc_id_t procno, const string & req) { + LOG("invoke procno 0x" << hex << procno); lock ml(invoke_mutex); vector m; string myaddr; @@ -356,7 +346,7 @@ rsm_client_protocol::status rsm::client_invoke(string &r, int procno, string req if (!cl) return rsm_client_protocol::BUSY; int ignored_rval; - auto ret = (rsm_protocol::status)cl->call_timeout(rsm_protocol::invoke, rpcc::to(1000), ignored_rval, procno, vs, req); + auto ret = (rsm_protocol::status)cl->call_timeout(rsm_protocol::invoke, milliseconds(100), ignored_rval, procno, vs, req); LOG("Invoke returned " << ret); if (ret != rsm_protocol::OK) return rsm_client_protocol::BUSY; @@ -366,6 +356,9 @@ rsm_client_protocol::status rsm::client_invoke(string &r, int procno, string req } } execute(procno, req, r); + for (size_t i=0; i m; string myaddr; @@ -409,10 +402,10 @@ rsm_protocol::status rsm::invoke(int &, int proc, viewstamp vs, string req) { return rsm_protocol::OK; } -/** - * RPC handler: Send back the local node's state to the caller - */ -rsm_protocol::status rsm::transferreq(rsm_protocol::transferres &r, string src, +// +// RPC handler: Send back the local node's state to the caller +// +rsm_protocol::status rsm::transferreq(rsm_protocol::transferres &r, const string & src, viewstamp last, unsigned vid) { lock ml(rsm_mutex); LOG("transferreq from " << src << " (" << last.vid << "," << last.seqno << ") vs (" << @@ -425,11 +418,11 @@ rsm_protocol::status rsm::transferreq(rsm_protocol::transferres &r, string src, return rsm_protocol::OK; } -/** - * RPC handler: Inform the local node (the primary) that node m has synchronized - * for view vid - */ -rsm_protocol::status rsm::transferdonereq(int &, string m, unsigned vid) { +// +// RPC handler: Inform the local node (the primary) that node m has synchronized +// for view vid +// +rsm_protocol::status rsm::transferdonereq(int &, const string & m, unsigned vid) { lock ml(rsm_mutex); if (!insync || vid != vid_insync) return rsm_protocol::BUSY; @@ -442,7 +435,7 @@ rsm_protocol::status rsm::transferdonereq(int &, string m, unsigned vid) { // a node that wants to join an RSM as a server sends a // joinreq to the RSM's current primary; this is the // handler for that RPC. -rsm_protocol::status rsm::joinreq(string & log, string m, viewstamp last) { +rsm_protocol::status rsm::joinreq(string & log, const string & m, viewstamp last) { auto ret = rsm_protocol::OK; lock ml(rsm_mutex); @@ -476,18 +469,17 @@ rsm_protocol::status rsm::joinreq(string & log, string m, viewstamp last) { return ret; } -/* - * RPC handler: Send back all the nodes this local knows about to client - * so the client can switch to a different primary - * when it existing primary fails - */ +// +// RPC handler: Responds with the list of known nodes for fall-back on a +// primary failure +// rsm_client_protocol::status rsm::client_members(vector &r, int) { vector m; lock ml(rsm_mutex); cfg->get_view(vid_commit, m); m.push_back(primary); r = m; - LOG("rsm::client_members return " << m << " m " << primary); + LOG("return " << m << " m " << primary); return rsm_client_protocol::OK; } @@ -501,7 +493,7 @@ void rsm::set_primary(unsigned vid) { VERIFY (c.size() > 0); if (isamember(primary,c)) { - LOG("set_primary: primary stays " << primary); + LOG("primary stays " << primary); return; } @@ -509,7 +501,7 @@ void rsm::set_primary(unsigned vid) { for (unsigned i = 0; i < p.size(); i++) { if (isamember(p[i], c)) { primary = p[i]; - LOG("set_primary: primary is " << primary); + LOG("primary is " << primary); return; } } @@ -522,18 +514,15 @@ bool rsm::amiprimary() { } -// Testing server - -// Simulate partitions +// Test RPCs -- simulate partitions and failures -// assumes caller holds rsm_mutex -void rsm::net_repair(bool heal, lock &) { +void rsm::net_repair(bool heal, lock &/*rsm_mutex_lock*/) { vector m; cfg->get_view(vid_commit, m); for (unsigned i = 0; i < m.size(); i++) { if (m[i] != cfg->myaddr()) { handle h(m[i]); - LOG("rsm::net_repair: " << m[i] << " " << heal); + LOG("member " << m[i] << " " << heal); if (h.safebind()) h.safebind()->set_reachable(heal); } } @@ -542,17 +531,14 @@ void rsm::net_repair(bool heal, lock &) { rsm_test_protocol::status rsm::test_net_repairreq(rsm_test_protocol::status &r, int heal) { lock ml(rsm_mutex); - LOG("rsm::test_net_repairreq: " << heal << " (dopartition " << + LOG("heal " << heal << " (dopartition " << dopartition << ", partitioned " << partitioned << ")"); - if (heal) { + if (heal) net_repair(heal, ml); - partitioned = false; - } else { + else dopartition = true; - partitioned = false; - } - r = rsm_test_protocol::OK; - return r; + partitioned = false; + return r = rsm_test_protocol::OK; } // simulate failure at breakpoint 1 and 2 @@ -582,7 +568,7 @@ void rsm::partition1(lock & rsm_mutex_lock) { rsm_test_protocol::status rsm::breakpointreq(rsm_test_protocol::status &r, int b) { r = rsm_test_protocol::OK; lock ml(rsm_mutex); - LOG("rsm::breakpointreq: " << b); + LOG("breakpoint " << b); if (b == 1) break1 = true; else if (b == 2) break2 = true; else if (b == 3 || b == 4) cfg->breakpoint(b);