Cosmetic improvements.

[invirt/third/libt4.git] / lock_server.cc

diff --git a/lock_server.cc b/lock_server.cc
index 0f82080..90ad5b2 100644 (file)
--- a/lock_server.cc
+++ b/lock_server.cc
@@ -1,37 +1,186 @@
-#include "rpc/rpc.h"
-#include <arpa/inet.h>
-#include <stdlib.h>
-#include <stdio.h>
+// the caching lock server implementation
+
+#include "lock_server.h"

 #include <unistd.h>
 #include <unistd.h>

-#include "lock_server_cache_rsm.h"
-#include "paxos.h"
-#include "rsm.h"
+#include <arpa/inet.h>
+#include "handle.h"
+
+lock_state::lock_state():
+    held(false)
+{
+}

 
 

-// Main loop of lock_server
+lock_state::lock_state(const lock_state & other) {
+    *this = other;
+}

 
 

-char tprintf_thread_prefix = 's';
+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;
+}

 
 

-int
-main(int argc, char *argv[])
-{
-    setvbuf(stdout, NULL, _IONBF, 0);
-    setvbuf(stderr, NULL, _IONBF, 0);
+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();
+    r.set_state_transfer(this);
+
+    r.reg(lock_protocol::acquire, &lock_server::acquire, this);
+    r.reg(lock_protocol::release, &lock_server::release, this);
+    r.reg(lock_protocol::stat, &lock_server::stat, this);
+}
+
+void lock_server::revoker () {
+    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() {
+    while (1) {
+        lock_protocol::lockid_t lid;
+        retry_fifo.deq(&lid);
+        if (rsm_ && !rsm_->amiprimary())
+            continue;

 
 

-    srandom(getpid());
+        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();
+        }

 
 

-    if(argc != 3){
-        fprintf(stderr, "Usage: %s [master:]port [me:]port\n", argv[0]);
-        exit(1);
+        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);
+        }

     }
     }

+}
+
+lock_protocol::status lock_server::acquire(int &, lock_protocol::lockid_t lid, const callback_t & 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;

 
 

-    rsm rsm(argv[1], argv[2]);
-    lock_server_cache_rsm ls(&rsm);
-    rsm.set_state_transfer(&ls);
+        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);

 
 

-    rsm.reg(lock_protocol::acquire, &lock_server_cache_rsm::acquire, &ls);
-    rsm.reg(lock_protocol::release, &lock_server_cache_rsm::release, &ls);
-    rsm.reg(lock_protocol::stat, &lock_server_cache_rsm::stat, &ls);
+    LOG("wanted_by=" << st.wanted_by);

 
 

-    while(1)
-        sleep(1000);
+    // send revoke if we're first in line
+    if (st.wanted_by.front() == h)
+        revoke_fifo.enq(lid);
+
+    return lock_protocol::RETRY;

 }
 }

+
+lock_protocol::status lock_server::release(int &, lock_protocol::lockid_t lid, const 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;
+}
+
+string lock_server::marshal_state() {
+    lock sl(lock_table_lock);
+    return marshall(nacquire, lock_table).content();
+}
+
+void lock_server::unmarshal_state(const string & state) {
+    lock sl(lock_table_lock);
+    unmarshall(state, false, nacquire, lock_table);
+}
+
+lock_protocol::status lock_server::stat(int & r, lock_protocol::lockid_t lid, const callback_t &) {
+    LOG("stat request for " << lid);
+    VERIFY(0);
+    r = nacquire;
+    return lock_protocol::OK;
+}
+