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 unsigned 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         LOG("lock_client: call bind");
  57     }
  58
  59     srandom((uint32_t)time(NULL)^last_port);
  60     rlock_port = ((random()%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         next_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() [[noreturn]] {
  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             if (lu)
  94                 lu->dorelease(lid);
  95             sl.lock();
  96         }
  97         st.state = lock_state::none;
  98         LOG("Lock " << lid << ": none");
  99         st.signal();
 100     }
 101 }
 102
 103 int lock_client::stat(lock_protocol::lockid_t lid) {
 104     VERIFY(0);
 105     int r;
 106     lock_protocol::status ret = cl->call(lock_protocol::stat, r, cl->id(), lid);
 107     VERIFY (ret == lock_protocol::OK);
 108     return r;
 109 }
 110
 111 lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
 112     lock_state &st = get_lock_state(lid);
 113     lock sl(st.m);
 114     auto self = std::this_thread::get_id();
 115
 116     // check for reentrancy
 117     VERIFY(st.state != lock_state::locked || st.held_by != self);
 118     VERIFY(find(st.wanted_by.begin(), st.wanted_by.end(), self) == st.wanted_by.end());
 119
 120     st.wanted_by.push_back(self);
 121
 122     while (1) {
 123         if (st.state != lock_state::free)
 124             LOG("Lock " << lid << ": not free");
 125
 126         if (st.state == lock_state::none || st.state == lock_state::retrying) {
 127             if (st.state == lock_state::none) {
 128                 lock l(xid_mutex);
 129                 st.xid = next_xid++;
 130             }
 131             st.state = lock_state::acquiring;
 132             LOG("Lock " << lid << ": acquiring");
 133             lock_protocol::status result;
 134             {
 135                 sl.unlock();
 136                 int r;
 137                 result = rsmc->call(lock_protocol::acquire, r, lid, id, st.xid);
 138                 sl.lock();
 139             }
 140             LOG("acquire returned " << result);
 141             if (result == lock_protocol::OK) {
 142                 st.state = lock_state::free;
 143                 LOG("Lock " << lid << ": free");
 144             }
 145         }
 146
 147         VERIFY(st.wanted_by.size() != 0);
 148         if (st.state == lock_state::free) {
 149             // is it for me?
 150             auto front = st.wanted_by.front();
 151             if (front == releaser_thread.get_id()) {
 152                 st.wanted_by.pop_front();
 153                 st.state = lock_state::locked;
 154                 st.held_by = releaser_thread.get_id();
 155                 LOG("Queuing " << lid << " for release");
 156                 release_fifo.enq(lid);
 157             } else if (front == self) {
 158                 st.wanted_by.pop_front();
 159                 st.state = lock_state::locked;
 160                 st.held_by = self;
 161                 break;
 162             } else {
 163                 st.signal(front);
 164             }
 165         }
 166
 167         LOG("waiting...");
 168         st.wait();
 169         LOG("wait ended");
 170     }
 171
 172     LOG("Lock " << lid << ": locked");
 173     return lock_protocol::OK;
 174 }
 175
 176 lock_protocol::status lock_client::release(lock_protocol::lockid_t lid) {
 177     lock_state &st = get_lock_state(lid);
 178     lock sl(st.m);
 179     auto self = std::this_thread::get_id();
 180     VERIFY(st.state == lock_state::locked && st.held_by == self);
 181     st.state = lock_state::free;
 182     LOG("Lock " << lid << ": free");
 183     if (st.wanted_by.size()) {
 184         auto front = st.wanted_by.front();
 185         if (front == releaser_thread.get_id()) {
 186             st.state = lock_state::locked;
 187             st.held_by = releaser_thread.get_id();
 188             st.wanted_by.pop_front();
 189             LOG("Queuing " << lid << " for release");
 190             release_fifo.enq(lid);
 191         } else
 192             st.signal(front);
 193     }
 194     LOG("Finished signaling.");
 195     return lock_protocol::OK;
 196 }
 197
 198 rlock_protocol::status lock_client::revoke_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t xid) {
 199     LOG("Revoke handler " << lid << " " << xid);
 200     lock_state &st = get_lock_state(lid);
 201     lock sl(st.m);
 202
 203     if (st.state == lock_state::releasing || st.state == lock_state::none)
 204         return rlock_protocol::OK;
 205
 206     if (st.state == lock_state::free &&
 207         (st.wanted_by.size() == 0 || st.wanted_by.front() == releaser_thread.get_id())) {
 208         // gimme
 209         st.state = lock_state::locked;
 210         st.held_by = releaser_thread.get_id();
 211         if (st.wanted_by.size())
 212             st.wanted_by.pop_front();
 213         release_fifo.enq(lid);
 214     } else {
 215         // get in line
 216         st.wanted_by.push_back(releaser_thread.get_id());
 217     }
 218     return rlock_protocol::OK;
 219 }
 220
 221 rlock_protocol::status lock_client::retry_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t) {
 222     lock_state &st = get_lock_state(lid);
 223     lock sl(st.m);
 224     VERIFY(st.state == lock_state::acquiring);
 225     st.state = lock_state::retrying;
 226     LOG("Lock " << lid << ": none");
 227     st.signal(); // only one thread needs to wake up
 228     return rlock_protocol::OK;
 229 }
 230
 231 t4_lock_client *t4_lock_client_new(const char *dst) {
 232     return (t4_lock_client *)new lock_client(dst);
 233 }
 234
 235 void t4_lock_client_delete(t4_lock_client *client) {
 236     delete (lock_client *)client;
 237 }
 238
 239 t4_status t4_lock_client_acquire(t4_lock_client *client, t4_lockid_t lid) {
 240     return ((lock_client *)client)->acquire(lid);
 241 }
 242
 243 t4_status t4_lock_client_release(t4_lock_client *client, t4_lockid_t lid) {
 244     return ((lock_client *)client)->acquire(lid);
 245 }
 246
 247 t4_status t4_lock_client_stat(t4_lock_client *client, t4_lockid_t lid) {
 248     return ((lock_client *)client)->stat(lid);
 249 }