X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/a4175b2e216a20b86cc872dea8a08005c60617a5..06282fd37814c4a9d53bca089b048709b368f5b3:/lock_client.cc

diff --git a/lock_client.cc b/lock_client.cc
index 11bc476..009e051 100644
--- a/lock_client.cc
+++ b/lock_client.cc
@@ -1,44 +1,212 @@
-// RPC stubs for clients to talk to lock_server
+// RPC stubs for clients to talk to lock_server, and cache the locks.
 
 #include "lock_client.h"
-#include "rpc/rpc.h"
 #include <arpa/inet.h>
 
-#include <sstream>
-#include <iostream>
-#include <stdio.h>
-
-lock_client::lock_client(std::string dst)
-{
-    sockaddr_in dstsock;
-    make_sockaddr(dst.c_str(), &dstsock);
-    cl = new rpcc(dstsock);
-    if (cl->bind() < 0) {
-        printf("lock_client: call bind\n");
+void lock_state::wait(lock & mutex_lock) {
+    auto self = std::this_thread::get_id();
+    c[self].wait(mutex_lock);
+    c.erase(self);
+}
+
+void lock_state::signal() {
+    // signal anyone
+    if (c.begin() != c.end())
+        c.begin()->second.notify_one();
+}
+
+void lock_state::signal(thread::id who) {
+    if (c.count(who))
+        c[who].notify_one();
+}
+
+in_port_t lock_client::last_port = 0;
+
+lock_state & lock_client::get_lock_state(lock_protocol::lockid_t lid) {
+    lock sl(lock_table_lock);
+    return lock_table[lid]; // creates the lock if it doesn't already exist
+}
+
+lock_client::lock_client(string xdst, lock_release_user *_lu) : lu(_lu), next_xid(0) {
+    cl = unique_ptr<rpcc>(new rpcc(xdst));
+    if (cl->bind() < 0)
+        LOG << "lock_client: call bind";
+
+    srandom((uint32_t)time(NULL)^last_port);
+    rlock_port = ((random()%32000) | (0x1 << 10));
+    id = "127.0.0.1:" + std::to_string(rlock_port);
+    last_port = rlock_port;
+    rlsrpc = unique_ptr<rpcs>(new rpcs(rlock_port));
+    rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this);
+    rlsrpc->reg(rlock_protocol::retry, &lock_client::retry_handler, this);
+    rsmc = unique_ptr<rsm_client>(new rsm_client(xdst));
+    releaser_thread = thread(&lock_client::releaser, this);
+    rlsrpc->start();
+}
+
+lock_client::~lock_client() {
+    release_fifo.enq(nothing<lock_protocol::lockid_t>());
+    releaser_thread.join();
+}
+
+void lock_client::releaser() {
+    while (1) {
+        maybe<lock_protocol::lockid_t> mlid;
+        release_fifo.deq(&mlid);
+
+        if (!mlid) {
+            LOG << "Releaser stopping";
+            break;
+        }
+
+        lock_protocol::lockid_t lid = mlid;
+        LOG << "Releaser: " << lid;
+
+        lock_state & st = get_lock_state(lid);
+        lock sl(st.m);
+        VERIFY(st.state == lock_state::locked && st.held_by == releaser_thread.get_id());
+        st.state = lock_state::releasing;
+        {
+            sl.unlock();
+            int r;
+            rsmc->call(lock_protocol::release, r, lid, id, st.xid);
+            if (lu)
+                lu->dorelease(lid);
+            sl.lock();
+        }
+        st.state = lock_state::none;
+        LOG << "Lock " << lid << ": none";
+        st.signal();
     }
 }
 
-int
-lock_client::stat(lock_protocol::lockid_t lid)
-{
+int lock_client::stat(lock_protocol::lockid_t lid) {
+    VERIFY(0);
     int r;
-    lock_protocol::status ret = cl->call(lock_protocol::stat, cl->id(), lid, r);
+    auto ret = (lock_protocol::status)cl->call(lock_protocol::stat, r, lid, id);
     VERIFY (ret == lock_protocol::OK);
     return r;
 }
 
-lock_protocol::status
-lock_client::acquire(lock_protocol::lockid_t lid)
-{
-    int r;
-    return cl->call(lock_protocol::acquire, cl->id(), lid, r);
+lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
+    lock_state & st = get_lock_state(lid);
+    lock sl(st.m);
+    auto self = std::this_thread::get_id();
+
+    // check for reentrancy
+    VERIFY(st.state != lock_state::locked || st.held_by != self);
+    VERIFY(std::find(st.wanted_by.begin(), st.wanted_by.end(), self)
+            == st.wanted_by.end());
+
+    st.wanted_by.push_back(self);
+
+    while (1) {
+        if (st.state != lock_state::free)
+            LOG << "Lock " << lid << ": not free";
+
+        if (st.state == lock_state::none || st.state == lock_state::retrying) {
+            if (st.state == lock_state::none) {
+                lock l(xid_mutex);
+                st.xid = next_xid++;
+            }
+            st.state = lock_state::acquiring;
+            LOG << "Lock " << lid << ": acquiring";
+            lock_protocol::status result;
+            {
+                sl.unlock();
+                int r;
+                result = (lock_protocol::status)rsmc->call(lock_protocol::acquire, r, lid, id, st.xid);
+                sl.lock();
+            }
+            LOG << "acquire returned " << result;
+            if (result == lock_protocol::OK) {
+                st.state = lock_state::free;
+                LOG << "Lock " << lid << ": free";
+            }
+        }
+
+        VERIFY(st.wanted_by.size() != 0);
+        if (st.state == lock_state::free) {
+            // is it for me?
+            auto front = st.wanted_by.front();
+            if (front == releaser_thread.get_id()) {
+                st.wanted_by.pop_front();
+                st.state = lock_state::locked;
+                st.held_by = releaser_thread.get_id();
+                LOG << "Queuing " << lid << " for release";
+                release_fifo.enq(just(lid));
+            } else if (front == self) {
+                st.wanted_by.pop_front();
+                st.state = lock_state::locked;
+                st.held_by = self;
+                break;
+            } else {
+                st.signal(front);
+            }
+        }
+
+        LOG << "waiting...";
+        st.wait(sl);
+        LOG << "wait ended";
+    }
+
+    LOG << "Lock " << lid << ": locked";
+    return lock_protocol::OK;
 }
 
-lock_protocol::status
-lock_client::release(lock_protocol::lockid_t lid)
-{
-    int r;
-    return cl->call(lock_protocol::release, cl->id(), lid, r);
+lock_protocol::status lock_client::release(lock_protocol::lockid_t lid) {
+    lock_state & st = get_lock_state(lid);
+    lock sl(st.m);
+    auto self = std::this_thread::get_id();
+    VERIFY(st.state == lock_state::locked && st.held_by == self);
+    st.state = lock_state::free;
+    LOG << "Lock " << lid << ": free";
+    if (st.wanted_by.size()) {
+        auto front = st.wanted_by.front();
+        if (front == releaser_thread.get_id()) {
+            st.state = lock_state::locked;
+            st.held_by = releaser_thread.get_id();
+            st.wanted_by.pop_front();
+            LOG << "Queuing " << lid << " for release";
+            release_fifo.enq(just(lid));
+        } else
+            st.signal(front);
+    }
+    LOG << "Finished signaling.";
+    return lock_protocol::OK;
+}
+
+rlock_protocol::status lock_client::revoke_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t xid) {
+    LOG << "Revoke handler " << lid << " " << xid;
+    lock_state & st = get_lock_state(lid);
+    lock sl(st.m);
+
+    if (st.state == lock_state::releasing || st.state == lock_state::none)
+        return rlock_protocol::OK;
+
+    if (st.state == lock_state::free &&
+        (st.wanted_by.size() == 0 || st.wanted_by.front() == releaser_thread.get_id())) {
+        // gimme
+        st.state = lock_state::locked;
+        st.held_by = releaser_thread.get_id();
+        if (st.wanted_by.size())
+            st.wanted_by.pop_front();
+        release_fifo.enq(just(lid));
+    } else {
+        // get in line
+        st.wanted_by.push_back(releaser_thread.get_id());
+    }
+    return rlock_protocol::OK;
+}
+
+rlock_protocol::status lock_client::retry_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t) {
+    lock_state & st = get_lock_state(lid);
+    lock sl(st.m);
+    VERIFY(st.state == lock_state::acquiring);
+    st.state = lock_state::retrying;
+    LOG << "Lock " << lid << ": none";
+    st.signal(); // only one thread needs to wake up
+    return rlock_protocol::OK;
 }
 
 t4_lock_client *t4_lock_client_new(const char *dst) {
@@ -54,10 +222,9 @@ t4_status t4_lock_client_acquire(t4_lock_client *client, t4_lockid_t lid) {
 }
 
 t4_status t4_lock_client_release(t4_lock_client *client, t4_lockid_t lid) {
-    return ((lock_client *)client)->acquire(lid);
+    return ((lock_client *)client)->release(lid);
 }
 
 t4_status t4_lock_client_stat(t4_lock_client *client, t4_lockid_t lid) {
     return ((lock_client *)client)->stat(lid);
 }
-