1 // the caching lock server implementation
3 #include "lock_server.h"
7 lock_state::lock_state():
12 lock_state::lock_state(const lock_state & other) {
16 lock_state & lock_state::operator=(const lock_state & o) {
19 wanted_by = o.wanted_by;
20 old_requests = o.old_requests;
24 lock_state & lock_server::get_lock_state(lock_protocol::lockid_t lid) {
25 lock sl(lock_table_lock);
26 // this will create the lock if it doesn't already exist
27 return lock_table[lid];
30 lock_server::lock_server(rsm & r) : rsm_ (&r) {
31 thread(&lock_server::revoker, this).detach();
32 thread(&lock_server::retryer, this).detach();
33 r.set_state_transfer(this);
35 r.reg(lock_protocol::acquire, &lock_server::acquire, this);
36 r.reg(lock_protocol::release, &lock_server::release, this);
37 r.reg(lock_protocol::stat, &lock_server::stat, this);
40 void lock_server::revoker () {
42 lock_protocol::lockid_t lid;
43 revoke_fifo.deq(&lid);
44 LOG << "Revoking " << lid;
45 if (rsm_ && !rsm_->amiprimary())
48 lock_state & st = get_lock_state(lid);
50 holder_t held_by = st.held_by;
53 if (auto cl = rpcc::bind_cached(held_by.first)) {
55 auto ret = (rlock_protocol::status)cl->call(rlock_protocol::revoke, r, lid, held_by.second);
56 LOG << "Revoke returned " << ret;
61 void lock_server::retryer() {
63 lock_protocol::lockid_t lid;
65 if (rsm_ && !rsm_->amiprimary())
68 LOG << "Sending retry for " << lid;
69 lock_state & st = get_lock_state(lid);
73 if (st.wanted_by.empty())
75 front = st.wanted_by.front();
78 if (auto cl = rpcc::bind_cached(front.first)) {
80 auto ret = (rlock_protocol::status)cl->call(rlock_protocol::retry, r, lid, front.second);
81 LOG << "Retry returned " << ret;
86 lock_protocol::status lock_server::acquire(int &, lock_protocol::lockid_t lid, const callback_t & id, lock_protocol::xid_t xid) {
87 LOG << "lid=" << lid << " client=" << id << "," << xid;
88 holder_t h = holder_t(id, xid);
89 lock_state & st = get_lock_state(lid);
92 // deal with duplicated requests
93 if (st.old_requests.count(id)) {
94 lock_protocol::xid_t old_xid = st.old_requests[id];
96 return lock_protocol::RPCERR;
97 else if (old_xid == xid) {
98 if (st.held && st.held_by == h) {
99 LOG << "Client " << id << " sent duplicate acquire xid=" << xid;
100 return lock_protocol::OK;
105 // grant the lock if it's available and I'm next in line
106 if (!st.held && (st.wanted_by.empty() || st.wanted_by.front() == h)) {
107 if (!st.wanted_by.empty())
108 st.wanted_by.pop_front();
109 st.old_requests[id] = xid;
113 LOG << "Lock " << lid << " held by " << h.first;
114 if (st.wanted_by.size())
115 revoke_fifo.enq(lid);
116 return lock_protocol::OK;
121 for (auto p : st.wanted_by) {
123 // make sure client is obeying serialization
124 if (p.second != xid) {
125 LOG << "Client " << id << " sent acquire xid=" << xid << " with in-progress xid=" << p.second;
126 return lock_protocol::RPCERR;
133 st.wanted_by.push_back(h);
135 LOG << "wanted_by=" << st.wanted_by;
137 // send revoke if we're first in line
138 if (st.wanted_by.front() == h)
139 revoke_fifo.enq(lid);
141 return lock_protocol::RETRY;
144 lock_protocol::status lock_server::release(int &, lock_protocol::lockid_t lid, const callback_t & id, lock_protocol::xid_t xid) {
145 LOG << "lid=" << lid << " client=" << id << "," << xid;
146 lock_state & st = get_lock_state(lid);
148 if (st.held && st.held_by == holder_t(id, xid)) {
150 LOG << "Lock " << lid << " not held";
152 if (st.wanted_by.size())
154 return lock_protocol::OK;
157 string lock_server::marshal_state() {
158 lock sl(lock_table_lock);
159 return marshall(nacquire, lock_table).content();
162 void lock_server::unmarshal_state(const string & state) {
163 lock sl(lock_table_lock);
164 unmarshall(state, false, nacquire, lock_table);
167 lock_protocol::status lock_server::stat(int & r, lock_protocol::lockid_t lid, const callback_t &) {
168 LOG << "stat request for " << lid;
171 return lock_protocol::OK;