X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/5fd8cc8409d0efadc07dfe8d6774ad9ff477663d..e478ac59e66e89cbc174e781ac715c8644539947:/lock_server.cc?ds=sidebyside

diff --git a/lock_server.cc b/lock_server.cc
index 3801814..81cd805 100644
--- a/lock_server.cc
+++ b/lock_server.cc
@@ -1,46 +1,186 @@
-// the lock server implementation
+// the caching lock server implementation
 
+#include "types.h"
 #include "lock_server.h"
-#include <sstream>
-#include <stdio.h>
 #include <unistd.h>
 #include <arpa/inet.h>
+#include "handle.h"
 
-lock_server::lock_server():
-    nacquire (0)
+lock_state::lock_state():
+    held(false)
 {
 }
 
-// caller must hold lock_lock
-mutex &
-lock_server::get_lock(lock_protocol::lockid_t lid) {
-    lock_lock.acquire();
-    // by the semantics of std::map, this will create
-    // the mutex if it doesn't already exist
-    mutex &l = locks[lid];
-    lock_lock.release();
-    return l;
+lock_state::lock_state(const lock_state &other) {
+    *this = other;
 }
 
-lock_protocol::status
-lock_server::stat(int clt, lock_protocol::lockid_t lid, int &r)
-{
-    lock_protocol::status ret = lock_protocol::OK;
-    printf("stat request from clt %d\n", clt);
-    r = nacquire;
-    return ret;
+lock_state& lock_state::operator=(const lock_state& o) {
+    held = o.held;
+    held_by = o.held_by;
+    wanted_by = o.wanted_by;
+    old_requests = o.old_requests;
+    return *this;
 }
 
-lock_protocol::status
-lock_server::acquire(int clt, lock_protocol::lockid_t lid, int &r)
-{
-    get_lock(lid).acquire();
+lock_state & lock_server::get_lock_state(lock_protocol::lockid_t lid) {
+    lock sl(lock_table_lock);
+    // this will create the lock if it doesn't already exist
+    return lock_table[lid];
+}
+
+lock_server::lock_server(rsm *r) : rsm_ (r) {
+    thread(&lock_server::revoker, this).detach();
+    thread(&lock_server::retryer, this).detach();
+    rsm_->set_state_transfer(this);
+}
+
+void lock_server::revoker() [[noreturn]] {
+    while (1) {
+        lock_protocol::lockid_t lid;
+        revoke_fifo.deq(&lid);
+        LOG("Revoking " << lid);
+        if (rsm_ && !rsm_->amiprimary())
+            continue;
+
+        lock_state &st = get_lock_state(lid);
+        holder_t held_by;
+        {
+            lock sl(st.m);
+            held_by = st.held_by;
+        }
+
+        rpcc *proxy = NULL;
+        // try a few times?
+        //int t=5;
+        //while (t-- && !proxy)
+        proxy = handle(held_by.first).safebind();
+        if (proxy) {
+            int r;
+            auto ret = (rlock_protocol::status)proxy->call(rlock_protocol::revoke, r, lid, held_by.second);
+            LOG("Revoke returned " << ret);
+        }
+    }
+}
+
+void lock_server::retryer() [[noreturn]] {
+    while (1) {
+        lock_protocol::lockid_t lid;
+        retry_fifo.deq(&lid);
+        if (rsm_ && !rsm_->amiprimary())
+            continue;
+
+        LOG("Sending retry for " << lid);
+        lock_state &st = get_lock_state(lid);
+        holder_t front;
+        {
+            lock sl(st.m);
+            if (st.wanted_by.empty())
+                continue;
+            front = st.wanted_by.front();
+        }
+
+        rpcc *proxy = NULL;
+        // try a few times?
+        //int t=5;
+        //while (t-- && !proxy)
+        proxy = handle(front.first).safebind();
+        if (proxy) {
+            int r;
+            auto ret = (rlock_protocol::status)proxy->call(rlock_protocol::retry, r, lid, front.second);
+            LOG("Retry returned " << ret);
+        }
+    }
+}
+
+int lock_server::acquire(int &, lock_protocol::lockid_t lid, string id, lock_protocol::xid_t xid) {
+    LOG("lid=" << lid << " client=" << id << "," << xid);
+    holder_t h = holder_t(id, xid);
+    lock_state &st = get_lock_state(lid);
+    lock sl(st.m);
+
+    // deal with duplicated requests
+    if (st.old_requests.count(id)) {
+        lock_protocol::xid_t old_xid = st.old_requests[id];
+        if (old_xid > xid)
+            return lock_protocol::RPCERR;
+        else if (old_xid == xid) {
+            if (st.held && st.held_by == h) {
+                LOG("Client " << id << " sent duplicate acquire xid=" << xid);
+                return lock_protocol::OK;
+            }
+        }
+    }
+
+    // grant the lock if it's available and I'm next in line
+    if (!st.held && (st.wanted_by.empty() || st.wanted_by.front() == h)) {
+        if (!st.wanted_by.empty())
+            st.wanted_by.pop_front();
+        st.old_requests[id] = xid;
+
+        st.held = true;
+        st.held_by = h;
+        LOG("Lock " << lid << " held by " << h.first);
+        if (st.wanted_by.size())
+            revoke_fifo.enq(lid);
+        return lock_protocol::OK;
+    }
+
+    // get in line
+    bool found = false;
+    for (auto p : st.wanted_by) {
+        if (p.first == id) {
+            // make sure client is obeying serialization
+            if (p.second != xid) {
+                LOG("Client " << id << " sent acquire xid=" << xid << " with in-progress xid=" << p.second);
+                return lock_protocol::RPCERR;
+            }
+            found = true;
+            break;
+        }
+    }
+    if (!found)
+        st.wanted_by.push_back(h);
+
+    LOG("wanted_by=" << st.wanted_by);
+
+    // send revoke if we're first in line
+    if (st.wanted_by.front() == h)
+        revoke_fifo.enq(lid);
+
+    return lock_protocol::RETRY;
+}
+
+int lock_server::release(int &, lock_protocol::lockid_t lid, callback_t id, lock_protocol::xid_t xid) {
+    LOG("lid=" << lid << " client=" << id << "," << xid);
+    lock_state &st = get_lock_state(lid);
+    lock sl(st.m);
+    if (st.held && st.held_by == holder_t(id, xid)) {
+        st.held = false;
+        LOG("Lock " << lid << " not held");
+    }
+    if (st.wanted_by.size())
+        retry_fifo.enq(lid);
     return lock_protocol::OK;
 }
 
-lock_protocol::status
-lock_server::release(int clt, lock_protocol::lockid_t lid, int &r)
-{
-    get_lock(lid).release();
+string lock_server::marshal_state() {
+    lock sl(lock_table_lock);
+    marshall rep;
+    rep << nacquire << lock_table;
+    return rep.content();
+}
+
+void lock_server::unmarshal_state(string state) {
+    lock sl(lock_table_lock);
+    unmarshall rep(state, false);
+    rep >> nacquire >> lock_table;
+}
+
+lock_protocol::status lock_server::stat(int &r, lock_protocol::lockid_t lid) {
+    LOG("stat request for " << lid);
+    VERIFY(0);
+    r = nacquire;
     return lock_protocol::OK;
 }
+