X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/f0dcb6b97d6d40f67698d1f71ac26970f1776f82..06282fd37814c4a9d53bca089b048709b368f5b3:/lock_client.cc diff --git a/lock_client.cc b/lock_client.cc index beca1cc..009e051 100644 --- a/lock_client.cc +++ b/lock_client.cc @@ -4,7 +4,7 @@ #include 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); } @@ -20,8 +20,6 @@ void lock_state::signal(thread::id who) { c[who].notify_one(); } -typedef map lock_map; - in_port_t lock_client::last_port = 0; lock_state & lock_client::get_lock_state(lock_protocol::lockid_t lid) { @@ -32,11 +30,11 @@ 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(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(new rpcs(rlock_port)); rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this); @@ -46,11 +44,23 @@ lock_client::lock_client(string xdst, lock_release_user *_lu) : lu(_lu), next_xi rlsrpc->start(); } +lock_client::~lock_client() { + release_fifo.enq(nothing()); + releaser_thread.join(); +} + void lock_client::releaser() { while (1) { - lock_protocol::lockid_t lid; - release_fifo.deq(&lid); - LOG("Releaser: " << lid); + maybe 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); @@ -65,7 +75,7 @@ void lock_client::releaser() { sl.lock(); } st.state = lock_state::none; - LOG("Lock " << lid << ": none"); + LOG << "Lock " << lid << ": none"; st.signal(); } } @@ -81,17 +91,18 @@ int lock_client::stat(lock_protocol::lockid_t lid) { 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) { @@ -99,7 +110,7 @@ lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) { st.xid = next_xid++; } st.state = lock_state::acquiring; - LOG("Lock " << lid << ": acquiring"); + LOG << "Lock " << lid << ": acquiring"; lock_protocol::status result; { sl.unlock(); @@ -107,10 +118,10 @@ lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) { 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"; } } @@ -122,8 +133,8 @@ lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) { 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; @@ -134,39 +145,39 @@ lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) { } } - 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); @@ -180,7 +191,7 @@ rlock_protocol::status lock_client::revoke_handler(int &, lock_protocol::lockid_ 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()); @@ -193,7 +204,7 @@ rlock_protocol::status lock_client::retry_handler(int &, lock_protocol::lockid_t 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; }