X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/eeab3e6cade87c1fe0a5f3d93522e12ccb9ec2ab..ab9eee5d7f1fbe7a3fe6229d4a78136efb14371b:/rsm_client.cc diff --git a/rsm_client.cc b/rsm_client.cc index b68ef0c..b24f056 100644 --- a/rsm_client.cc +++ b/rsm_client.cc @@ -1,95 +1,75 @@ -#include "rsm_client.h" -#include +#include "include/rsm_client.h" #include -#include -#include #include -#include "lang/verify.h" -#include "lock.h" -#include "tprintf.h" -rsm_client::rsm_client(std::string dst) { - LOG("create rsm_client"); - std::vector mems; +using namespace std::chrono; - sockaddr_in dstsock; - make_sockaddr(dst.c_str(), &dstsock); - primary = dst; - - { - lock ml(rsm_client_mutex); - VERIFY (init_members()); - } - LOG("rsm_client: done"); -} - -// Assumes caller holds rsm_client_mutex -void rsm_client::primary_failure() { - primary = known_mems.back(); - known_mems.pop_back(); +rsm_client::rsm_client(string dst) : primary(dst) { + LOG << "create rsm_client"; + lock ml(rsm_client_mutex); + VERIFY (init_members(ml)); + LOG << "done"; } -rsm_protocol::status rsm_client::invoke(unsigned int proc, std::string &rep, const std::string &req) { - int ret = 0; +rsm_protocol::status rsm_client::invoke(rpc_protocol::proc_id_t proc, string & rep, const string & req) { lock ml(rsm_client_mutex); while (1) { - LOG("rsm_client::invoke proc " << std::hex << proc << " primary " << primary); - handle h(primary); + LOG << "proc " << std::hex << proc << " primary " << primary; + string prim = primary; ml.unlock(); - rpcc *cl = h.safebind(); + auto cl = rpcc::bind_cached(prim); + auto ret = rsm_client_protocol::OK; if (cl) - ret = cl->call_timeout(rsm_client_protocol::invoke, rpcc::to(5000), rep, proc, req); + ret = (rsm_client_protocol::status)cl->call_timeout(rsm_client_protocol::invoke, 500ms, rep, proc, req); ml.lock(); if (!cl) goto prim_fail; - LOG("rsm_client::invoke proc " << std::hex << proc << " primary " << primary << " ret " << std::dec << ret); + LOG << "proc " << std::hex << proc << " primary " << prim << " ret " << std::dec << ret; if (ret == rsm_client_protocol::OK) - break; + return rsm_protocol::OK; if (ret == rsm_client_protocol::BUSY) { - LOG("rsm is busy " << primary); - sleep(3); + LOG << "rsm is busy " << prim; + std::this_thread::sleep_for(300ms); continue; } if (ret == rsm_client_protocol::NOTPRIMARY) { - LOG("primary " << primary << " isn't the primary--let's get a complete list of mems"); - if (init_members()) + LOG << "primary " << prim << " isn't the primary--let's get a complete list of mems"; + if (init_members(ml)) continue; } prim_fail: - LOG("primary " << primary << " failed ret " << std::dec << ret); - primary_failure(); - LOG("rsm_client::invoke: retry new primary " << primary); + LOG << "primary " << prim << " failed ret " << std::dec << ret; + primary = known_mems.back(); + known_mems.pop_back(); + LOG << "retry new primary " << prim; } - return ret; } -bool rsm_client::init_members() { - LOG("rsm_client::init_members get members!"); - handle h(primary); +bool rsm_client::init_members(lock & rsm_client_mutex_lock) { + LOG << "get members!"; + string prim = primary; int ret = rsm_client_protocol::ERR; - rpcc *cl; + shared_ptr cl; { - adopt_lock ml(rsm_client_mutex); - ml.unlock(); - cl = h.safebind(); - if (cl) - ret = cl->call_timeout(rsm_client_protocol::members, rpcc::to(1000), known_mems, 0); - ml.lock(); + rsm_client_mutex_lock.unlock(); + if ((cl = rpcc::bind_cached(prim))) + ret = cl->call_timeout(rsm_client_protocol::members, 100ms, known_mems, 0); + rsm_client_mutex_lock.lock(); } - if (cl == 0 || ret != rsm_protocol::OK) + if (ret != rsm_protocol::OK) return false; - if (known_mems.size() < 1) { - LOG("rsm_client::init_members do not know any members!"); + if (!known_mems.size()) { + LOG << "do not know any members!"; VERIFY(0); } primary = known_mems.back(); known_mems.pop_back(); - LOG("rsm_client::init_members: primary " << primary); + LOG << "primary " << primary; return true; }