X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/5a5c578e2e358a121cdb9234a6cb11c4ecfbf323..06282fd37814c4a9d53bca089b048709b368f5b3:/lock_client.cc?ds=sidebyside diff --git a/lock_client.cc b/lock_client.cc index 99dcb5b..009e051 100644 --- a/lock_client.cc +++ b/lock_client.cc @@ -20,38 +20,49 @@ void lock_state::signal(thread::id who) { c[who].notify_one(); } -typedef map lock_map; - -unsigned int lock_client::last_port = 0; +in_port_t lock_client::last_port = 0; lock_state & lock_client::get_lock_state(lock_protocol::lockid_t lid) { lock sl(lock_table_lock); return lock_table[lid]; // creates the lock if it doesn't already exist } -lock_client::lock_client(string xdst, class lock_release_user *_lu) : lu(_lu), next_xid(0) { - cl = new rpcc(xdst); +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; - rpcs *rlsrpc = new rpcs(rlock_port); + rlsrpc = unique_ptr(new rpcs(rlock_port)); rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this); rlsrpc->reg(rlock_protocol::retry, &lock_client::retry_handler, this); - rsmc = new rsm_client(xdst); + rsmc = unique_ptr(new rsm_client(xdst)); releaser_thread = thread(&lock_client::releaser, this); + rlsrpc->start(); +} + +lock_client::~lock_client() { + release_fifo.enq(nothing()); + releaser_thread.join(); } -void lock_client::releaser() [[noreturn]] { +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_state & st = get_lock_state(lid); lock sl(st.m); VERIFY(st.state == lock_state::locked && st.held_by == releaser_thread.get_id()); st.state = lock_state::releasing; @@ -64,7 +75,7 @@ void lock_client::releaser() [[noreturn]] { sl.lock(); } st.state = lock_state::none; - LOG("Lock " << lid << ": none"); + LOG << "Lock " << lid << ": none"; st.signal(); } } @@ -72,25 +83,26 @@ void lock_client::releaser() [[noreturn]] { int lock_client::stat(lock_protocol::lockid_t lid) { VERIFY(0); int r; - auto ret = (lock_protocol::status)cl->call(lock_protocol::stat, r, cl->id(), lid); + auto ret = (lock_protocol::status)cl->call(lock_protocol::stat, r, lid, id); VERIFY (ret == lock_protocol::OK); return r; } lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) { - lock_state &st = get_lock_state(lid); + lock_state & st = get_lock_state(lid); lock sl(st.m); 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) { @@ -98,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(); @@ -106,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"; } } @@ -121,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; @@ -133,40 +145,40 @@ 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_state & st = get_lock_state(lid); lock sl(st.m); 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); - lock_state &st = get_lock_state(lid); + LOG << "Revoke handler " << lid << " " << xid; + lock_state & st = get_lock_state(lid); lock sl(st.m); if (st.state == lock_state::releasing || st.state == lock_state::none) @@ -179,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()); @@ -188,11 +200,11 @@ rlock_protocol::status lock_client::revoke_handler(int &, lock_protocol::lockid_ } rlock_protocol::status lock_client::retry_handler(int &, lock_protocol::lockid_t lid, lock_protocol::xid_t) { - lock_state &st = get_lock_state(lid); + lock_state & st = get_lock_state(lid); 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; }