X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/4b9798f44ae94deabf87dd534337b55259272950..e478ac59e66e89cbc174e781ac715c8644539947:/lock_server.cc diff --git a/lock_server.cc b/lock_server.cc index 0f82080..81cd805 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 "types.h" +#include "lock_server.h" #include -#include "lock_server_cache_rsm.h" -#include "paxos.h" -#include "rsm.h" +#include +#include "handle.h" -// Main loop of lock_server +lock_state::lock_state(): + held(false) +{ +} -char tprintf_thread_prefix = 's'; +lock_state::lock_state(const lock_state &other) { + *this = other; +} -int -main(int argc, char *argv[]) -{ - setvbuf(stdout, NULL, _IONBF, 0); - setvbuf(stderr, NULL, _IONBF, 0); +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; +} - srandom(getpid()); +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]; +} - if(argc != 3){ - fprintf(stderr, "Usage: %s [master:]port [me:]port\n", argv[0]); - exit(1); +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); + } } +} - rsm rsm(argv[1], argv[2]); - lock_server_cache_rsm ls(&rsm); - rsm.set_state_transfer(&ls); +void lock_server::retryer() [[noreturn]] { + while (1) { + lock_protocol::lockid_t lid; + retry_fifo.deq(&lid); + if (rsm_ && !rsm_->amiprimary()) + continue; - 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("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(); + } - while(1) - sleep(1000); + 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; +} + +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; +} +