X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/869c0cc91d8f6b2bb80026616372d16450b64d9f..e0c49ff6ba44cf5b47ab91d58b67763f5a1c7a58:/paxos.cc diff --git a/paxos.cc b/paxos.cc index 2c7e79a..c7f2d1d 100644 --- a/paxos.cc +++ b/paxos.cc @@ -93,20 +93,20 @@ bool proposer_acceptor::prepare(unsigned instance, nodes_t & accepts, prepareres res; prop_t highest_n_a{0, ""}; for (auto i : nodes) { - auto r = rpcc::bind_cached(i); - if (!r) + auto cl = rpcc::bind_cached(i); + if (!cl) continue; - auto status = (paxos_protocol::status)r->call_timeout( - paxos_protocol::preparereq, milliseconds(100), res, me, instance, proposal); + int status = cl->call_timeout(paxos_protocol::preparereq, milliseconds(100), + res, me, instance, proposal); if (status == paxos_protocol::OK) { - if (res.oldinstance) { + LOG << "preparereq response type=" << res.type << " n_a=(" << res.n_a.n + << ", " << res.n_a.m << ") " << "v_a=\"" << res.v_a << "\""; + if (res.type == prepareres::oldinstance) { LOG << "commiting old instance!"; - commit(instance, res.v_a); + lock ml(acceptor_mutex); + commit(instance, res.v_a, ml); return false; - } - LOG << "preparereq responded with oldinstance=" << res.oldinstance << " accept=" << res.accept << " n_a=(" << res.n_a.n << ", " << res.n_a.m << ") " - << "v_a=\"" << res.v_a << "\""; - if (res.accept) { + } else if (res.type == prepareres::accept) { accepts.push_back(i); if (res.n_a >= highest_n_a) { LOG << "found a newer accepted proposal, \"" << res.v_a << "\", with number (" << res.n_a.n << ", " << res.n_a.m << ")"; @@ -121,50 +121,42 @@ bool proposer_acceptor::prepare(unsigned instance, nodes_t & accepts, void proposer_acceptor::accept(unsigned instance, nodes_t & accepts, const nodes_t & nodes, const value_t & v) { + bool accept = false; for (auto i : nodes) { - auto r = rpcc::bind_cached(i); - if (!r) - continue; - bool accept = false; - int status = r->call_timeout( - paxos_protocol::acceptreq, milliseconds(100), accept, me, instance, proposal, v); - if (status == paxos_protocol::OK && accept) - accepts.push_back(i); + if (auto cl = rpcc::bind_cached(i)) { + int status = cl->call_timeout(paxos_protocol::acceptreq, milliseconds(100), + accept, me, instance, proposal, v); + if (status == paxos_protocol::OK && accept) + accepts.push_back(i); + } } } void proposer_acceptor::decide(unsigned instance, const nodes_t & accepts, const value_t & v) { - for (auto i : accepts) { - auto r = rpcc::bind_cached(i); - if (!r) - continue; - int res = 0; - r->call_timeout(paxos_protocol::decidereq, milliseconds(100), res, me, instance, v); - } + int res = 0; + for (auto i : accepts) + if (auto cl = rpcc::bind_cached(i)) + cl->call_timeout(paxos_protocol::decidereq, milliseconds(100), res, me, instance, v); } paxos_protocol::status proposer_acceptor::preparereq(prepareres & r, const node_t &, unsigned instance, prop_t n) { LOG << "instance " << instance << " proposal (" << n.n << ", " << n.m << ")"; lock ml(acceptor_mutex); - r.oldinstance = false; - r.accept = false; - r.n_a = accepted; - r.v_a = accepted_value; if (instance <= instance_h) { LOG << "old instance " << instance << " has value " << values[instance]; - r.oldinstance = true; - r.v_a = values[instance]; + r = prepareres{prepareres::oldinstance, accepted, values[instance]}; } else if (n > promise) { LOG << "looks good to me"; promise = n; l.logprop(promise); - r.accept = true; + r = prepareres{prepareres::accept, accepted, accepted_value}; } else { LOG << "I totally rejected this request. Ha."; + r = prepareres{prepareres::reject, accepted, accepted_value}; } - LOG << "preparereq is responding with oldinstance=" << r.oldinstance << " accept=" << r.accept << " n_a=(" << r.n_a.n << ", " << r.n_a.m << ") " - << "v_a=\"" << r.v_a << "\""; + LOG << "preparereq is responding with oldinstance=" << r.oldinstance << " accept=" << r.accept + << " n_a=(" << r.n_a.n << ", " << r.n_a.m << ") " << "v_a=\"" << r.v_a << "\""; return paxos_protocol::OK; } @@ -172,17 +164,15 @@ paxos_protocol::status proposer_acceptor::acceptreq(bool & r, const node_t &, unsigned instance, prop_t n, const value_t & v) { lock ml(acceptor_mutex); r = false; - if (instance == instance_h + 1) { - if (n >= promise) { - accepted = n; - accepted_value = v; - l.logaccept(accepted, accepted_value); - r = true; - } - return paxos_protocol::OK; - } else { + if (instance != instance_h + 1) return paxos_protocol::ERR; + if (n >= promise) { + accepted = n; + accepted_value = v; + l.logaccept(accepted, accepted_value); + r = true; } + return paxos_protocol::OK; } paxos_protocol::status @@ -191,7 +181,7 @@ proposer_acceptor::decidereq(int &, const node_t &, unsigned instance, const val LOG << "decidereq for accepted instance " << instance << " (my instance " << instance_h << ") v=" << accepted_value; if (instance == instance_h + 1) { VERIFY(accepted_value == v); - commit(instance, accepted_value, ml); + commit(instance, v, ml); } else if (instance <= instance_h) { // we are ahead; ignore. } else { @@ -201,12 +191,9 @@ proposer_acceptor::decidereq(int &, const node_t &, unsigned instance, const val return paxos_protocol::OK; } -void proposer_acceptor::commit(unsigned instance, const value_t & value) { - lock ml(acceptor_mutex); - commit(instance, value, ml); -} - -void proposer_acceptor::commit(unsigned instance, const value_t & value, lock & pxs_mutex_lock) { +void proposer_acceptor::commit(unsigned instance, const value_t & value, lock & acceptor_mutex_lock) { + VERIFY(&value != &accepted_value); // eited by aliasing? + VERIFY(acceptor_mutex_lock); LOG << "instance=" << instance << " has v=" << value; if (instance > instance_h) { LOG << "highestacceptedinstance = " << instance; @@ -214,12 +201,11 @@ void proposer_acceptor::commit(unsigned instance, const value_t & value, lock & l.loginstance(instance, value); instance_h = instance; accepted = promise = {0, me}; - string v = value; // gaaahhh aliasing of value and accepted_value - accepted_value.clear(); // this wipes out "value", too + accepted_value.clear(); if (delegate) { - pxs_mutex_lock.unlock(); - delegate->paxos_commit(instance, v); - pxs_mutex_lock.lock(); + acceptor_mutex_lock.unlock(); + delegate->paxos_commit(instance, value); + acceptor_mutex_lock.lock(); } } }