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
   9
  10 rsm_client::rsm_client(std::string dst) {
  11     printf("create rsm_client\n");
  12     std::vector<std::string> mems;
  13
  14     pthread_mutex_init(&rsm_client_mutex, NULL);
  15     sockaddr_in dstsock;
  16     make_sockaddr(dst.c_str(), &dstsock);
  17     primary = dst;
  18
  19     {
  20         ScopedLock ml(&rsm_client_mutex);
  21         VERIFY (init_members());
  22     }
  23     printf("rsm_client: done\n");
  24 }
  25
  26 // Assumes caller holds rsm_client_mutex
  27 void rsm_client::primary_failure() {
  28     primary = known_mems.back();
  29     known_mems.pop_back();
  30 }
  31
  32 rsm_protocol::status rsm_client::invoke(int proc, std::string req, std::string &rep) {
  33     int ret;
  34     ScopedLock ml(&rsm_client_mutex);
  35     while (1) {
  36         printf("rsm_client::invoke proc %x primary %s\n", proc, primary.c_str());
  37         handle h(primary);
  38
  39         VERIFY(pthread_mutex_unlock(&rsm_client_mutex)==0);
  40         rpcc *cl = h.safebind();
  41         if (cl)
  42             ret = cl->call(rsm_client_protocol::invoke, proc, req, rep, rpcc::to(5000));
  43         VERIFY(pthread_mutex_lock(&rsm_client_mutex)==0);
  44
  45         if (!cl)
  46             goto prim_fail;
  47
  48         printf("rsm_client::invoke proc %x primary %s ret %d\n", proc,
  49                 primary.c_str(), ret);
  50         if (ret == rsm_client_protocol::OK)
  51             break;
  52         if (ret == rsm_client_protocol::BUSY) {
  53             printf("rsm is busy %s\n", primary.c_str());
  54             sleep(3);
  55             continue;
  56         }
  57         if (ret == rsm_client_protocol::NOTPRIMARY) {
  58             printf("primary %s isn't the primary--let's get a complete list of mems\n",
  59                     primary.c_str());
  60             if (init_members())
  61                 continue;
  62         }
  63 prim_fail:
  64         printf("primary %s failed ret %d\n", primary.c_str(), ret);
  65         primary_failure();
  66         printf ("rsm_client::invoke: retry new primary %s\n", primary.c_str());
  67     }
  68     return ret;
  69 }
  70
  71 bool rsm_client::init_members() {
  72     printf("rsm_client::init_members get members!\n");
  73     handle h(primary);
  74     VERIFY(pthread_mutex_unlock(&rsm_client_mutex)==0);
  75     int ret;
  76     rpcc *cl = h.safebind();
  77     if (cl) {
  78         ret = cl->call(rsm_client_protocol::members, 0, known_mems,
  79                 rpcc::to(1000));
  80     }
  81     VERIFY(pthread_mutex_lock(&rsm_client_mutex)==0);
  82     if (cl == 0 || ret != rsm_protocol::OK)
  83         return false;
  84     if (known_mems.size() < 1) {
  85         printf("rsm_client::init_members do not know any members!\n");
  86         VERIFY(0);
  87     }
  88
  89     primary = known_mems.back();
  90     known_mems.pop_back();
  91
  92     printf("rsm_client::init_members: primary %s\n", primary.c_str());
  93
  94     return true;
  95 }