RPC procedures are now identified via a struct containing a string name.

[invirt/third/libt4.git] / lock_client.cc

diff --git a/lock_client.cc b/lock_client.cc
index 7a44940..ca21d9d 100644 (file)
--- a/lock_client.cc
+++ b/lock_client.cc
@@ -4,7 +4,7 @@
 #include <arpa/inet.h>
 
 void lock_state::wait(lock & mutex_lock) {
 #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[self].wait(mutex_lock);
     c.erase(self);
 }

@@ -20,36 +20,30 @@ void lock_state::signal(thread::id who) {
         c[who].notify_one();
 }
 
         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 sl(lock_table_lock);
     return lock_table[lid]; // creates the lock if it doesn't already exist
 }
 
 lock_client::lock_client(string xdst, lock_release_user *_lu) : lu(_lu), next_xid(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, 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";
-
-    srandom((uint32_t)time(NULL)^last_port);
-    rlock_port = ((random()%32000) | (0x1 << 10));
-    id = "127.0.0.1:" + to_string(rlock_port);
-    last_port = rlock_port;
-    rlsrpc = unique_ptr<rpcs>(new rpcs(rlock_port));
+    rlock_port = std::uniform_int_distribution<in_port_t>(1024,32000+1024)(global->random_generator);
+    id = "127.0.0.1:" + std::to_string(rlock_port);
+    rlsrpc = std::make_unique<rpcs>(rlock_port);

     rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this);
     rlsrpc->reg(rlock_protocol::retry, &lock_client::retry_handler, this);
     rlsrpc->reg(rlock_protocol::revoke, &lock_client::revoke_handler, this);
     rlsrpc->reg(rlock_protocol::retry, &lock_client::retry_handler, this);

-    rsmc = unique_ptr<rsm_client>(new rsm_client(xdst));
+    rsmc = std::make_unique<rsm_client>(xdst);

     releaser_thread = thread(&lock_client::releaser, this);
     rlsrpc->start();
 }
 
     releaser_thread = thread(&lock_client::releaser, this);
     rlsrpc->start();
 }
 

+lock_client::~lock_client() {
+    release_fifo.enq(nothing<lock_protocol::lockid_t>());
+    releaser_thread.join();
+}
+

 void lock_client::releaser() {
 void lock_client::releaser() {

-    while (1) {
-        lock_protocol::lockid_t lid;
-        release_fifo.deq(&lid);
+    while (auto mlid = release_fifo.deq()) {
+        lock_protocol::lockid_t lid = mlid;

         LOG << "Releaser: " << lid;
 
         lock_state & st = get_lock_state(lid);
         LOG << "Releaser: " << lid;
 
         lock_state & st = get_lock_state(lid);

@@ -68,20 +62,13 @@ void lock_client::releaser() {
         LOG << "Lock " << lid << ": none";
         st.signal();
     }
         LOG << "Lock " << lid << ": none";
         st.signal();
     }

-}
-
-int lock_client::stat(lock_protocol::lockid_t lid) {
-    VERIFY(0);
-    int r;
-    auto ret = (lock_protocol::status)cl->call(lock_protocol::stat, r, lid, id);
-    VERIFY (ret == lock_protocol::OK);
-    return r;
+    LOG << "Releaser stopping";

 }
 
 lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
     lock_state & st = get_lock_state(lid);
     lock sl(st.m);
 }
 
 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);
 
     // check for reentrancy
     VERIFY(st.state != lock_state::locked || st.held_by != self);

@@ -124,7 +111,7 @@ lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
                 st.state = lock_state::locked;
                 st.held_by = releaser_thread.get_id();
                 LOG << "Queuing " << lid << " for release";
                 st.state = lock_state::locked;
                 st.held_by = releaser_thread.get_id();
                 LOG << "Queuing " << lid << " for release";

-                release_fifo.enq(lid);
+                release_fifo.enq(just(lid));

             } else if (front == self) {
                 st.wanted_by.pop_front();
                 st.state = lock_state::locked;
             } else if (front == self) {
                 st.wanted_by.pop_front();
                 st.state = lock_state::locked;

@@ -147,7 +134,7 @@ lock_protocol::status lock_client::acquire(lock_protocol::lockid_t lid) {
 lock_protocol::status lock_client::release(lock_protocol::lockid_t lid) {
     lock_state & st = get_lock_state(lid);
     lock sl(st.m);
 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";
     VERIFY(st.state == lock_state::locked && st.held_by == self);
     st.state = lock_state::free;
     LOG << "Lock " << lid << ": free";

@@ -158,7 +145,7 @@ lock_protocol::status lock_client::release(lock_protocol::lockid_t lid) {
             st.held_by = releaser_thread.get_id();
             st.wanted_by.pop_front();
             LOG << "Queuing " << lid << " for release";
             st.held_by = releaser_thread.get_id();
             st.wanted_by.pop_front();
             LOG << "Queuing " << lid << " for release";

-            release_fifo.enq(lid);
+            release_fifo.enq(just(lid));

         } else
             st.signal(front);
     }
         } else
             st.signal(front);
     }

@@ -181,7 +168,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();
         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());
     } else {
         // get in line
         st.wanted_by.push_back(releaser_thread.get_id());

@@ -214,7 +201,3 @@ t4_status t4_lock_client_acquire(t4_lock_client *client, t4_lockid_t lid) {
 t4_status t4_lock_client_release(t4_lock_client *client, t4_lockid_t lid) {
     return ((lock_client *)client)->release(lid);
 }
 t4_status t4_lock_client_release(t4_lock_client *client, t4_lockid_t lid) {
     return ((lock_client *)client)->release(lid);
 }

-
-t4_status t4_lock_client_stat(t4_lock_client *client, t4_lockid_t lid) {
-    return ((lock_client *)client)->stat(lid);
-}