X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/dfe8486473094c0769fd1922329c3f0dfd8f43c0..5987a5357ccfd818d540a7aa58eb498a1be07aa1:/rsm.cc diff --git a/rsm.cc b/rsm.cc index b93c701..bdeabaa 100644 --- a/rsm.cc +++ b/rsm.cc @@ -108,17 +108,17 @@ rsm::rsm(std::string _first, std::string _me) : commit_change(1); } rsmrpc = cfg->get_rpcs(); - rsmrpc->reg(rsm_client_protocol::invoke, this, &rsm::client_invoke); - rsmrpc->reg(rsm_client_protocol::members, this, &rsm::client_members); - rsmrpc->reg(rsm_protocol::invoke, this, &rsm::invoke); - rsmrpc->reg(rsm_protocol::transferreq, this, &rsm::transferreq); - rsmrpc->reg(rsm_protocol::transferdonereq, this, &rsm::transferdonereq); - rsmrpc->reg(rsm_protocol::joinreq, this, &rsm::joinreq); + rsmrpc->reg(rsm_client_protocol::invoke, &rsm::client_invoke, this); + rsmrpc->reg(rsm_client_protocol::members, &rsm::client_members, this); + rsmrpc->reg(rsm_protocol::invoke, &rsm::invoke, this); + rsmrpc->reg(rsm_protocol::transferreq, &rsm::transferreq, this); + rsmrpc->reg(rsm_protocol::transferdonereq, &rsm::transferdonereq, this); + rsmrpc->reg(rsm_protocol::joinreq, &rsm::joinreq, this); // tester must be on different port, otherwise it may partition itself testsvr = new rpcs(atoi(_me.c_str()) + 1); - testsvr->reg(rsm_test_protocol::net_repair, this, &rsm::test_net_repairreq); - testsvr->reg(rsm_test_protocol::breakpoint, this, &rsm::breakpointreq); + 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); @@ -189,12 +189,12 @@ bool rsm::sync_with_backups() { adopt_lock ml(rsm_mutex); ml.unlock(); { - // Make sure that the state of lock_server_cache_rsm is stable during + // Make sure that the state of lock_server is stable during // synchronization; otherwise, the primary's state may be more recent // than replicas after the synchronization. lock ml(invoke_mutex); // By acquiring and releasing the invoke_mutex once, we make sure that - // the state of lock_server_cache_rsm will not be changed until all + // the state of lock_server will not be changed until all // replicas are synchronized. The reason is that client_invoke arrives // after this point of time will see inviewchange == true, and returns // BUSY. @@ -239,8 +239,8 @@ bool rsm::statetransfer(std::string m) ml.unlock(); cl = h.safebind(); if (cl) { - ret = cl->call(rsm_protocol::transferreq, cfg->myaddr(), - last_myvs, vid_insync, r, rpcc::to(1000)); + ret = cl->call_timeout(rsm_protocol::transferreq, rpcc::to(1000), + r, cfg->myaddr(), last_myvs, vid_insync); } ml.lock(); } @@ -266,7 +266,7 @@ bool rsm::statetransferdone(std::string m) { bool done = false; if (cl) { int r; - rsm_protocol::status ret = cl->call(rsm_protocol::transferdonereq, cfg->myaddr(), vid_insync, r); + rsm_protocol::status ret = cl->call(rsm_protocol::transferdonereq, r, cfg->myaddr(), vid_insync); done = (ret == rsm_protocol::OK); } ml.lock(); @@ -287,8 +287,8 @@ bool rsm::join(std::string m) { ml.unlock(); cl = h.safebind(); if (cl != 0) { - ret = cl->call(rsm_protocol::joinreq, cfg->myaddr(), last_myvs, - r, rpcc::to(120000)); + ret = cl->call_timeout(rsm_protocol::joinreq, rpcc::to(120000), r, + cfg->myaddr(), last_myvs); } ml.lock(); } @@ -336,7 +336,7 @@ void rsm::execute(int procno, std::string req, std::string &r) { unmarshall args(req); marshall rep; std::string reps; - rsm_protocol::status ret = h->fn(args, rep); + rsm_protocol::status ret = (*h)(args, rep); marshall rep1; rep1 << ret; rep1 << rep.str(); @@ -349,7 +349,7 @@ void rsm::execute(int procno, std::string req, std::string &r) { // number, and invokes it on all members of the replicated state // machine. // -rsm_client_protocol::status rsm::client_invoke(int procno, std::string req, std::string &r) { +rsm_client_protocol::status rsm::client_invoke(std::string &r, int procno, std::string req) { LOG("rsm::client_invoke: procno 0x" << std::hex << procno); lock ml(invoke_mutex); std::vector m; @@ -383,7 +383,7 @@ rsm_client_protocol::status rsm::client_invoke(int procno, std::string req, std: return rsm_client_protocol::BUSY; rsm_protocol::status ret; int r; - ret = cl->call(rsm_protocol::invoke, procno, vs, req, r, rpcc::to(1000)); + ret = cl->call_timeout(rsm_protocol::invoke, rpcc::to(1000), r, procno, vs, req); LOG("Invoke returned " << ret); if (ret != rsm_protocol::OK) return rsm_client_protocol::BUSY; @@ -403,7 +403,7 @@ rsm_client_protocol::status rsm::client_invoke(int procno, std::string req, std: // the replica must execute requests in order (with no gaps) // according to requests' seqno -rsm_protocol::status rsm::invoke(int proc, viewstamp vs, std::string req, int &dummy) { +rsm_protocol::status rsm::invoke(int &, int proc, viewstamp vs, std::string req) { LOG("rsm::invoke: procno 0x" << std::hex << proc); lock ml(invoke_mutex); std::vector m; @@ -438,8 +438,8 @@ rsm_protocol::status rsm::invoke(int proc, viewstamp vs, std::string req, int &d /** * RPC handler: Send back the local node's state to the caller */ -rsm_protocol::status rsm::transferreq(std::string src, viewstamp last, unsigned vid, - rsm_protocol::transferres &r) { +rsm_protocol::status rsm::transferreq(rsm_protocol::transferres &r, std::string src, + viewstamp last, unsigned vid) { lock ml(rsm_mutex); int ret = rsm_protocol::OK; tprintf("transferreq from %s (%d,%d) vs (%d,%d)\n", src.c_str(), @@ -457,7 +457,7 @@ rsm_protocol::status rsm::transferreq(std::string src, viewstamp last, unsigned * RPC handler: Inform the local node (the primary) that node m has synchronized * for view vid */ -rsm_protocol::status rsm::transferdonereq(std::string m, unsigned vid, int &) { +rsm_protocol::status rsm::transferdonereq(int &, std::string m, unsigned vid) { lock ml(rsm_mutex); if (!insync || vid != vid_insync) return rsm_protocol::BUSY; @@ -470,7 +470,7 @@ rsm_protocol::status rsm::transferdonereq(std::string m, unsigned vid, int &) { // 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(std::string m, viewstamp last, rsm_protocol::joinres &r) { +rsm_protocol::status rsm::joinreq(rsm_protocol::joinres &r, std::string m, viewstamp last) { int ret = rsm_protocol::OK; lock ml(rsm_mutex); @@ -508,7 +508,7 @@ rsm_protocol::status rsm::joinreq(std::string m, viewstamp last, rsm_protocol::j * so the client can switch to a different primary * when it existing primary fails */ -rsm_client_protocol::status rsm::client_members(int i, std::vector &r) { +rsm_client_protocol::status rsm::client_members(std::vector &r, int i) { std::vector m; lock ml(rsm_mutex); cfg->get_view(vid_commit, m); @@ -568,7 +568,7 @@ void rsm::net_repair_wo(bool heal) { rsmrpc->set_reachable(heal); } -rsm_test_protocol::status rsm::test_net_repairreq(int heal, int &r) { +rsm_test_protocol::status rsm::test_net_repairreq(int &r, int heal) { lock ml(rsm_mutex); tprintf("rsm::test_net_repairreq: %d (dopartition %d, partitioned %d)\n", heal, dopartition, partitioned); @@ -607,7 +607,7 @@ void rsm::partition1() { } } -rsm_test_protocol::status rsm::breakpointreq(int b, int &r) { +rsm_test_protocol::status rsm::breakpointreq(int &r, int b) { r = rsm_test_protocol::OK; lock ml(rsm_mutex); tprintf("rsm::breakpointreq: %d\n", b);