X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/4b9798f44ae94deabf87dd534337b55259272950..f0dcb6b97d6d40f67698d1f71ac26970f1776f82:/lock_server.cc diff --git a/lock_server.cc b/lock_server.cc index 0f82080..90ad5b2 100644 --- a/lock_server.cc +++ b/lock_server.cc @@ -1,37 +1,186 @@ -#include "rpc/rpc.h" -#include -#include -#include +// the caching lock server implementation + +#include "lock_server.h" #include -#include "lock_server_cache_rsm.h" -#include "paxos.h" -#include "rsm.h" +#include +#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; +} +