rsm_client.cc

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