rsm_client.cc

   1 #include "rsm_client.h"
   2 #include <arpa/inet.h>
   3 #include <unistd.h>
   4
   5 rsm_client::rsm_client(string dst) : primary(dst) {
   6     LOG << "create rsm_client";
   7     lock ml(rsm_client_mutex);
   8     VERIFY (init_members(ml));
   9     LOG << "done";
  10 }
  11
  12 void rsm_client::primary_failure(lock &) {
  13     primary = known_mems.back();
  14     known_mems.pop_back();
  15 }
  16
  17 rsm_protocol::status rsm_client::invoke(unsigned int proc, string & rep, const string & req) {
  18     lock ml(rsm_client_mutex);
  19     while (1) {
  20         LOG << "proc " << std::hex << proc << " primary " << primary;
  21         string prim = primary;
  22
  23         ml.unlock();
  24         auto cl = rpcc::bind_cached(prim);
  25         auto ret = rsm_client_protocol::OK;
  26         if (cl)
  27             ret = (rsm_client_protocol::status)cl->call_timeout(rsm_client_protocol::invoke, milliseconds(500), rep, proc, req);
  28         ml.lock();
  29
  30         if (!cl)
  31             goto prim_fail;
  32
  33         LOG << "proc " << std::hex << proc << " primary " << prim << " ret " << std::dec << ret;
  34         if (ret == rsm_client_protocol::OK)
  35             return rsm_protocol::OK;
  36         if (ret == rsm_client_protocol::BUSY) {
  37             LOG << "rsm is busy " << prim;
  38             std::this_thread::sleep_for(milliseconds(300));
  39             continue;
  40         }
  41         if (ret == rsm_client_protocol::NOTPRIMARY) {
  42             LOG << "primary " << prim << " isn't the primary--let's get a complete list of mems";
  43             if (init_members(ml))
  44                 continue;
  45         }
  46 prim_fail:
  47         LOG << "primary " << prim << " failed ret " << std::dec << ret;
  48         primary_failure(ml);
  49         LOG << "retry new primary " << prim;
  50     }
  51 }
  52
  53 bool rsm_client::init_members(lock & rsm_client_mutex_lock) {
  54     LOG << "get members!";
  55     string prim = primary;
  56     int ret = rsm_client_protocol::ERR;
  57     shared_ptr<rpcc> cl;
  58     {
  59         rsm_client_mutex_lock.unlock();
  60         cl = rpcc::bind_cached(prim);
  61         if (cl)
  62             ret = cl->call_timeout(rsm_client_protocol::members, milliseconds(100), known_mems, 0);
  63         rsm_client_mutex_lock.lock();
  64     }
  65     if (cl == 0 || ret != rsm_protocol::OK)
  66         return false;
  67     if (known_mems.size() < 1) {
  68         LOG << "do not know any members!";
  69         VERIFY(0);
  70     }
  71
  72     primary = known_mems.back();
  73     known_mems.pop_back();
  74
  75     LOG << "primary " << primary;
  76
  77     return true;
  78 }