rsm_client.cc

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