-#include "rsm_client.h"
-#include <vector>
+#include "include/rsm_client.h"
#include <arpa/inet.h>
-#include <stdio.h>
-#include <handle.h>
#include <unistd.h>
-#include "lang/verify.h"
-#include "lock.h"
-#include "tprintf.h"
-rsm_client::rsm_client(std::string dst) {
- LOG("create rsm_client");
- std::vector<std::string> 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<rpcc> 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;
}