#include <arpa/inet.h>
void lock_state::wait(lock & mutex_lock) {
- auto self = this_thread::get_id();
+ auto self = std::this_thread::get_id();
c[self].wait(mutex_lock);
c.erase(self);
}
c[who].notify_one();
}
-typedef map<lock_protocol::lockid_t, lock_state> lock_map;
-
in_port_t lock_client::last_port = 0;
lock_state & lock_client::get_lock_state(lock_protocol::lockid_t lid) {
lock_client::lock_client(string xdst, lock_release_user *_lu) : lu(_lu), next_xid(0) {
cl = unique_ptr<rpcc>(new rpcc(xdst));
if (cl->bind() < 0)
- LOG("lock_client: call bind");
+ LOG << "lock_client: call bind";
srandom((uint32_t)time(NULL)^last_port);
rlock_port = ((random()%32000) | (0x1 << 10));
- id = "127.0.0.1:" + to_string(rlock_port);
+ id = "127.0.0.1:" + std::to_string(rlock_port);
last_port = rlock_port;
rlsrpc = unique_ptr<rpcs>(new rpcs(rlock_port));
rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this);
rlsrpc->start();
}
+lock_client::~lock_client() {
+ release_fifo.enq(nothing<lock_protocol::lockid_t>());
+ releaser_thread.join();
+}
+
void lock_client::releaser() {
while (1) {
- lock_protocol::lockid_t lid;
- release_fifo.deq(&lid);
- LOG("Releaser: " << lid);
+ maybe<lock_protocol::lockid_t> mlid;
+ release_fifo.deq(&mlid);
+
+ if (!mlid) {
+ LOG << "Releaser stopping";
+ break;
+ }
+
+ lock_protocol::lockid_t lid = mlid;
+ LOG << "Releaser: " << lid;
lock_state & st = get_lock_state(lid);
lock sl(st.m);
sl.lock();
}
st.state = lock_state::none;
- LOG("Lock " << lid << ": none");
+ LOG << "Lock " << lid << ": none";
st.signal();
}
}
lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
lock_state & st = get_lock_state(lid);
lock sl(st.m);
- auto self = this_thread::get_id();
+ auto self = std::this_thread::get_id();
// check for reentrancy
VERIFY(st.state != lock_state::locked || st.held_by != self);
- VERIFY(find(st.wanted_by.begin(), st.wanted_by.end(), self) == st.wanted_by.end());
+ VERIFY(std::find(st.wanted_by.begin(), st.wanted_by.end(), self)
+ == st.wanted_by.end());
st.wanted_by.push_back(self);
while (1) {
if (st.state != lock_state::free)
- LOG("Lock " << lid << ": not free");
+ LOG << "Lock " << lid << ": not free";
if (st.state == lock_state::none || st.state == lock_state::retrying) {
if (st.state == lock_state::none) {
st.xid = next_xid++;
}
st.state = lock_state::acquiring;
- LOG("Lock " << lid << ": acquiring");
+ LOG << "Lock " << lid << ": acquiring";
lock_protocol::status result;
{
sl.unlock();
result = (lock_protocol::status)rsmc->call(lock_protocol::acquire, r, lid, id, st.xid);
sl.lock();
}
- LOG("acquire returned " << result);
+ LOG << "acquire returned " << result;
if (result == lock_protocol::OK) {
st.state = lock_state::free;
- LOG("Lock " << lid << ": free");
+ LOG << "Lock " << lid << ": free";
}
}
st.wanted_by.pop_front();
st.state = lock_state::locked;
st.held_by = releaser_thread.get_id();
- LOG("Queuing " << lid << " for release");
- release_fifo.enq(lid);
+ LOG << "Queuing " << lid << " for release";
+ release_fifo.enq(just(lid));
} else if (front == self) {
st.wanted_by.pop_front();
st.state = lock_state::locked;
}
}
- LOG("waiting...");
+ LOG << "waiting...";
st.wait(sl);
- LOG("wait ended");
+ LOG << "wait ended";
}
- LOG("Lock " << lid << ": locked");
+ LOG << "Lock " << lid << ": locked";
return lock_protocol::OK;
}
lock_protocol::status lock_client::release(lock_protocol::lockid_t lid) {
lock_state & st = get_lock_state(lid);
lock sl(st.m);
- auto self = this_thread::get_id();
+ auto self = std::this_thread::get_id();
VERIFY(st.state == lock_state::locked && st.held_by == self);
st.state = lock_state::free;
- LOG("Lock " << lid << ": free");
+ LOG << "Lock " << lid << ": free";
if (st.wanted_by.size()) {
auto front = st.wanted_by.front();
if (front == releaser_thread.get_id()) {
st.state = lock_state::locked;
st.held_by = releaser_thread.get_id();
st.wanted_by.pop_front();
- LOG("Queuing " << lid << " for release");
- release_fifo.enq(lid);
+ LOG << "Queuing " << lid << " for release";
+ release_fifo.enq(just(lid));
} else
st.signal(front);
}
- LOG("Finished signaling.");
+ LOG << "Finished signaling.";
return lock_protocol::OK;
}
rlock_protocol::status lock_client::revoke_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t xid) {
- LOG("Revoke handler " << lid << " " << xid);
+ LOG << "Revoke handler " << lid << " " << xid;
lock_state & st = get_lock_state(lid);
lock sl(st.m);
st.held_by = releaser_thread.get_id();
if (st.wanted_by.size())
st.wanted_by.pop_front();
- release_fifo.enq(lid);
+ release_fifo.enq(just(lid));
} else {
// get in line
st.wanted_by.push_back(releaser_thread.get_id());
lock sl(st.m);
VERIFY(st.state == lock_state::acquiring);
st.state = lock_state::retrying;
- LOG("Lock " << lid << ": none");
+ LOG << "Lock " << lid << ": none";
st.signal(); // only one thread needs to wake up
return rlock_protocol::OK;
}