lock_client.cc

   1 // RPC stubs for clients to talk to lock_server, and cache the locks.
   2
   3 #include "lock_client.h"
   4 #include "rpc/rpc.h"
   5 #include <sstream>
   6 #include <iostream>
   7 #include <algorithm>
   8 #include <stdio.h>
   9 #include "tprintf.h"
  10 #include <arpa/inet.h>
  11
  12 #include "rsm_client.h"
  13 #include "lock.h"
  14
  15 using std::ostringstream;
  16
  17 lock_state::lock_state():
  18     state(none)
  19 {
  20 }
  21
  22 void lock_state::wait() {
  23     auto self = std::this_thread::get_id();
  24     {
  25         adopt_lock ml(m);
  26         c[self].wait(ml);
  27     }
  28     c.erase(self);
  29 }
  30
  31 void lock_state::signal() {
  32     // signal anyone
  33     if (c.begin() != c.end())
  34         c.begin()->second.notify_one();
  35 }
  36
  37 void lock_state::signal(std::thread::id who) {
  38     if (c.count(who))
  39         c[who].notify_one();
  40 }
  41
  42 int lock_client::last_port = 0;
  43
  44 lock_state & lock_client::get_lock_state(lock_protocol::lockid_t lid) {
  45     lock sl(lock_table_lock);
  46     // by the semantics of std::map, this will create
  47     // the lock if it doesn't already exist
  48     return lock_table[lid];
  49 }
  50
  51 lock_client::lock_client(string xdst, class lock_release_user *_lu) : lu(_lu) {
  52     sockaddr_in dstsock;
  53     make_sockaddr(xdst.c_str(), &dstsock);
  54     cl = new rpcc(dstsock);
  55     if (cl->bind() < 0) {
  56         printf("lock_client: call bind\n");
  57     }
  58
  59     srand(time(NULL)^last_port);
  60     rlock_port = ((rand()%32000) | (0x1 << 10));
  61     const char *hname;
  62     // VERIFY(gethostname(hname, 100) == 0);
  63     hname = "127.0.0.1";
  64     ostringstream host;
  65     host << hname << ":" << rlock_port;
  66     id = host.str();
  67     last_port = rlock_port;
  68     rpcs *rlsrpc = new rpcs(rlock_port);
  69     rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this);
  70     rlsrpc->reg(rlock_protocol::retry, &lock_client::retry_handler, this);
  71     {
  72         lock sl(xid_mutex);
  73         xid = 0;
  74     }
  75     rsmc = new rsm_client(xdst);
  76     releaser_thread = std::thread(&lock_client::releaser, this);
  77 }
  78
  79 void lock_client::releaser() {
  80     while (1) {
  81         lock_protocol::lockid_t lid;
  82         release_fifo.deq(&lid);
  83         LOG("Releaser: " << lid);
  84
  85         lock_state &st = get_lock_state(lid);
  86         lock sl(st.m);
  87         VERIFY(st.state == lock_state::locked && st.held_by == releaser_thread.get_id());
  88         st.state = lock_state::releasing;
  89         {
  90             sl.unlock();
  91             int r;
  92             rsmc->call(lock_protocol::release, r, lid, id, st.xid);
  93             sl.lock();
  94         }
  95         st.state = lock_state::none;
  96         LOG("Lock " << lid << ": none");
  97         st.signal();
  98     }
  99 }
 100
 101 int lock_client::stat(lock_protocol::lockid_t lid) {
 102     VERIFY(0);
 103     int r;
 104     lock_protocol::status ret = cl->call(lock_protocol::stat, r, cl->id(), lid);
 105     VERIFY (ret == lock_protocol::OK);
 106     return r;
 107 }
 108
 109 lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
 110     lock_state &st = get_lock_state(lid);
 111     lock sl(st.m);
 112     auto self = std::this_thread::get_id();
 113
 114     // check for reentrancy
 115     VERIFY(st.state != lock_state::locked || st.held_by != self);
 116     VERIFY(find(st.wanted_by.begin(), st.wanted_by.end(), self) == st.wanted_by.end());
 117
 118     st.wanted_by.push_back(self);
 119
 120     while (1) {
 121         if (st.state != lock_state::free)
 122             LOG("Lock " << lid << ": not free");
 123
 124         if (st.state == lock_state::none || st.state == lock_state::retrying) {
 125             if (st.state == lock_state::none) {
 126                 lock sl(xid_mutex);
 127                 st.xid = xid++;
 128             }
 129             st.state = lock_state::acquiring;
 130             LOG("Lock " << lid << ": acquiring");
 131             lock_protocol::status result;
 132             {
 133                 sl.unlock();
 134                 int r;
 135                 result = rsmc->call(lock_protocol::acquire, r, lid, id, st.xid);
 136                 sl.lock();
 137             }
 138             LOG("acquire returned " << result);
 139             if (result == lock_protocol::OK) {
 140                 st.state = lock_state::free;
 141                 LOG("Lock " << lid << ": free");
 142             }
 143         }
 144
 145         VERIFY(st.wanted_by.size() != 0);
 146         if (st.state == lock_state::free) {
 147             // is it for me?
 148             auto front = st.wanted_by.front();
 149             if (front == releaser_thread.get_id()) {
 150                 st.wanted_by.pop_front();
 151                 st.state = lock_state::locked;
 152                 st.held_by = releaser_thread.get_id();
 153                 LOG("Queuing " << lid << " for release");
 154                 release_fifo.enq(lid);
 155             } else if (front == self) {
 156                 st.wanted_by.pop_front();
 157                 st.state = lock_state::locked;
 158                 st.held_by = self;
 159                 break;
 160             } else {
 161                 st.signal(front);
 162             }
 163         }
 164
 165         LOG("waiting...");
 166         st.wait();
 167         LOG("wait ended");
 168     }
 169
 170     LOG("Lock " << lid << ": locked");
 171     return lock_protocol::OK;
 172 }
 173
 174 lock_protocol::status lock_client::release(lock_protocol::lockid_t lid) {
 175     lock_state &st = get_lock_state(lid);
 176     lock sl(st.m);
 177     auto self = std::this_thread::get_id();
 178     VERIFY(st.state == lock_state::locked && st.held_by == self);
 179     st.state = lock_state::free;
 180     LOG("Lock " << lid << ": free");
 181     if (st.wanted_by.size()) {
 182         auto front = st.wanted_by.front();
 183         if (front == releaser_thread.get_id()) {
 184             st.state = lock_state::locked;
 185             st.held_by = releaser_thread.get_id();
 186             st.wanted_by.pop_front();
 187             LOG("Queuing " << lid << " for release");
 188             release_fifo.enq(lid);
 189         } else
 190             st.signal(front);
 191     }
 192     LOG("Finished signaling.");
 193     return lock_protocol::OK;
 194 }
 195
 196 rlock_protocol::status lock_client::revoke_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t xid) {
 197     LOG("Revoke handler " << lid << " " << xid);
 198     lock_state &st = get_lock_state(lid);
 199     lock sl(st.m);
 200
 201     if (st.state == lock_state::releasing || st.state == lock_state::none)
 202         return rlock_protocol::OK;
 203
 204     if (st.state == lock_state::free &&
 205         (st.wanted_by.size() == 0 || st.wanted_by.front() == releaser_thread.get_id())) {
 206         // gimme
 207         st.state = lock_state::locked;
 208         st.held_by = releaser_thread.get_id();
 209         if (st.wanted_by.size())
 210             st.wanted_by.pop_front();
 211         release_fifo.enq(lid);
 212     } else {
 213         // get in line
 214         st.wanted_by.push_back(releaser_thread.get_id());
 215     }
 216     return rlock_protocol::OK;
 217 }
 218
 219 rlock_protocol::status lock_client::retry_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t xid) {
 220     lock_state &st = get_lock_state(lid);
 221     lock sl(st.m);
 222     VERIFY(st.state == lock_state::acquiring);
 223     st.state = lock_state::retrying;
 224     LOG("Lock " << lid << ": none");
 225     st.signal(); // only one thread needs to wake up
 226     return rlock_protocol::OK;
 227 }
 228
 229 t4_lock_client *t4_lock_client_new(const char *dst) {
 230     return (t4_lock_client *)new lock_client(dst);
 231 }
 232
 233 void t4_lock_client_delete(t4_lock_client *client) {
 234     delete (lock_client *)client;
 235 }
 236
 237 t4_status t4_lock_client_acquire(t4_lock_client *client, t4_lockid_t lid) {
 238     return ((lock_client *)client)->acquire(lid);
 239 }
 240
 241 t4_status t4_lock_client_release(t4_lock_client *client, t4_lockid_t lid) {
 242     return ((lock_client *)client)->acquire(lid);
 243 }
 244
 245 t4_status t4_lock_client_stat(t4_lock_client *client, t4_lockid_t lid) {
 246     return ((lock_client *)client)->stat(lid);
 247 }