rsm_client.cc

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