C++11y allows us to eliminate some boilerplate!

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

diff --git a/rpc/rpc.cc b/rpc/rpc.cc
index 00f6d2e..80ec124 100644 (file)
--- a/rpc/rpc.cc
+++ b/rpc/rpc.cc
@@ -60,17 +60,14 @@
 #include <unistd.h>
 #include <string.h>
 
-inline void set_rand_seed() {
-    auto now = time_point_cast<nanoseconds>(steady_clock::now());
-    srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid());
-}
+using std::list;
+using namespace std::chrono;
 
 static sockaddr_in make_sockaddr(const string & hostandport);
 
 rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
 {
-    set_rand_seed();
-    clt_nonce_ = (nonce_t)random();
+    clt_nonce_ = (nonce_t)global->random_generator();
 
     char *loss_env = getenv("RPC_LOSSY");
     if (loss_env)
@@ -82,7 +79,10 @@ rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
 // IMPORTANT: destruction should happen only when no external threads
 // are blocked inside rpcc or will use rpcc in the future
 rpcc::~rpcc() {
-    cancel();
+    lock ml(m_);
+    cancel(ml);
+
+    lock cl(chan_m_);
     IF_LEVEL(2) LOG << "delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1);
     chan_.reset();
     VERIFY(calls_.size() == 0);
@@ -101,9 +101,29 @@ int rpcc::bind(milliseconds to) {
     return ret;
 }
 
+shared_ptr<rpcc> rpcc::bind_cached(const string & destination) {
+    auto client = global->get_handle(destination);
+    lock cl = lock(client->bind_m_);
+    if (!client->bind_done_) {
+        LOG_NONMEMBER << "bind(\"" << destination << "\")";
+        int ret = client->bind(milliseconds(1000));
+        if (ret < 0) {
+            LOG_NONMEMBER << "bind failure! " << destination << " " << ret;
+            client.reset();
+        } else {
+            LOG_NONMEMBER << "bind succeeded " << destination;
+        }
+    }
+    return client;
+}
+
+void rpcc::unbind_cached(const string & destination) {
+    global->erase_handle(destination);
+}
+
 // Cancel all outstanding calls
-void rpcc::cancel(void) {
-    lock ml(m_);
+void rpcc::cancel(lock & m_lock) {
+    VERIFY(m_lock);
     if (calls_.size()) {
         LOG << "force callers to fail";
         for (auto & p : calls_) {
@@ -119,7 +139,7 @@ void rpcc::cancel(void) {
 
         destroy_wait_ = true;
         while (calls_.size () > 0)
-            destroy_wait_c_.wait(ml);
+            destroy_wait_c_.wait(m_lock);
 
         LOG << "done";
     }
@@ -186,7 +206,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
             lock cal(ca.m);
             while (!ca.done) {
                 IF_LEVEL(2) LOG << "wait";
-                if (ca.c.wait_until(cal, nextdeadline) == cv_status::timeout) {
+                if (ca.c.wait_until(cal, nextdeadline) == std::cv_status::timeout) {
                     IF_LEVEL(2) LOG << "timeout";
                     break;
                 }
@@ -236,7 +256,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
                     << ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret;
 
     // destruction of req automatically frees its buffer
-    return (ca.done? ca.intret : rpc_protocol::timeout_failure);
+    return ca.done ? ca.intret : rpc_protocol::timeout_failure;
 }
 
 void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
@@ -248,14 +268,9 @@ void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
         ch = chan_;
 }
 
-// PollMgr's thread is being used to
-// make this upcall from connection object to rpcc.
-// this funtion must not block.
-//
-// this function keeps no reference for connection *c
-bool
-rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
-{
+// Runs in poll_mgr's thread as an upcall from the connection object to the
+// rpcc.  Does not call blocking RPC handlers.
+bool rpcc::got_pdu(const shared_ptr<connection> &, const string & b) {
     unmarshall rep(b, true);
     rpc_protocol::reply_header h;
     rep.read_header(h);
@@ -311,8 +326,7 @@ compress:
 
 rpcs::rpcs(in_port_t p1) : port_(p1)
 {
-    set_rand_seed();
-    nonce_ = (nonce_t)random();
+    nonce_ = (nonce_t)global->random_generator();
     IF_LEVEL(2) LOG << "created with nonce " << nonce_;
 
     reg(rpc_protocol::bind, &rpcs::rpcbind, this);
@@ -373,7 +387,6 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
         if (procs_.count(proc) < 1) {
             LOG << "unknown proc 0x" << std::hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_;
             VERIFY(0);
-            return;
         }
 
         f = procs_[proc];
@@ -404,7 +417,7 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
 
     switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
         case NEW: // new request
-            rh.ret = (*f)(forward<unmarshall>(req), rep);
+            rh.ret = (*f)(std::forward<unmarshall>(req), rep);
             if (rh.ret == rpc_protocol::unmarshall_args_failure) {
                 LOG << "failed to unmarshall the arguments. You are "
                     << "probably calling RPC 0x" << std::hex << proc << " with the wrong "
@@ -555,6 +568,6 @@ static sockaddr_in make_sockaddr(const string & hostandport) {
         memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t));
         dst.sin_addr.s_addr = a.s_addr;
     }
-    dst.sin_port = hton((in_port_t)stoi(port));
+    dst.sin_port = hton((in_port_t)std::stoi(port));
     return dst;
 }