Global destructor clean-ups and python test fixes

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

diff --git a/rpc/rpc.cc b/rpc/rpc.cc
index 6985498..de33675 100644 (file)
--- a/rpc/rpc.cc
+++ b/rpc/rpc.cc
@@ -60,6 +60,9 @@
 #include <unistd.h>
 #include <string.h>
 
 #include <unistd.h>
 #include <string.h>
 

+using std::list;
+using namespace std::chrono;
+

 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());
 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());

@@ -76,27 +79,27 @@ rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
     if (loss_env)
         lossytest_ = atoi(loss_env);
 
     if (loss_env)
         lossytest_ = atoi(loss_env);
 

-    IF_LEVEL(2) LOG("cltn_nonce is " << clt_nonce_ << " lossy " << lossytest_);
+    IF_LEVEL(2) LOG << "cltn_nonce is " << clt_nonce_ << " lossy " << lossytest_;

 }
 
 // IMPORTANT: destruction should happen only when no external threads
 // are blocked inside rpcc or will use rpcc in the future
 rpcc::~rpcc() {
     cancel();
 }
 
 // IMPORTANT: destruction should happen only when no external threads
 // are blocked inside rpcc or will use rpcc in the future
 rpcc::~rpcc() {
     cancel();

-    IF_LEVEL(2) LOG("delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1));
+    IF_LEVEL(2) LOG << "delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1);

     chan_.reset();
     VERIFY(calls_.size() == 0);
 }
 
 int rpcc::bind(milliseconds to) {
     chan_.reset();
     VERIFY(calls_.size() == 0);
 }
 
 int rpcc::bind(milliseconds to) {

-    nonce_t r;
+    nonce_t r = 0;

     rpc_protocol::status ret = call_timeout(rpc_protocol::bind, to, r);
     if (ret == 0) {
         lock ml(m_);
         bind_done_ = true;
         srv_nonce_ = r;
     } else {
     rpc_protocol::status ret = call_timeout(rpc_protocol::bind, to, r);
     if (ret == 0) {
         lock ml(m_);
         bind_done_ = true;
         srv_nonce_ = r;
     } else {

-        IF_LEVEL(2) LOG("bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret);
+        IF_LEVEL(2) LOG << "bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret;

     }
     return ret;
 }
     }
     return ret;
 }

@@ -105,11 +108,11 @@ int rpcc::bind(milliseconds to) {
 void rpcc::cancel(void) {
     lock ml(m_);
     if (calls_.size()) {
 void rpcc::cancel(void) {
     lock ml(m_);
     if (calls_.size()) {

-        LOG("force callers to fail");
+        LOG << "force callers to fail";

         for (auto & p : calls_) {
             caller *ca = p.second;
 
         for (auto & p : calls_) {
             caller *ca = p.second;
 

-            IF_LEVEL(2) LOG("force caller to fail");
+            IF_LEVEL(2) LOG << "force caller to fail";

 
             lock cl(ca->m);
             ca->done = true;
 
             lock cl(ca->m);
             ca->done = true;

@@ -121,7 +124,7 @@ void rpcc::cancel(void) {
         while (calls_.size () > 0)
             destroy_wait_c_.wait(ml);
 
         while (calls_.size () > 0)
             destroy_wait_c_.wait(ml);
 

-        LOG("done");
+        LOG << "done";

     }
 }
 
     }
 }
 

@@ -133,7 +136,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
         lock ml(m_);
 
         if ((proc != rpc_protocol::bind.id && !bind_done_) || (proc == rpc_protocol::bind.id && bind_done_)) {
         lock ml(m_);
 
         if ((proc != rpc_protocol::bind.id && !bind_done_) || (proc == rpc_protocol::bind.id && bind_done_)) {

-            IF_LEVEL(1) LOG("rpcc has not been bound to dst or binding twice");
+            IF_LEVEL(1) LOG << "rpcc has not been bound to dst or binding twice";

             return rpc_protocol::bind_failure;
         }
 
             return rpc_protocol::bind_failure;
         }
 

@@ -143,7 +146,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
         ca.xid = xid_++;
         calls_[ca.xid] = &ca;
 
         ca.xid = xid_++;
         calls_[ca.xid] = &ca;
 

-        req.pack_header(rpc_protocol::request_header{
+        req.write_header(rpc_protocol::request_header{

                 ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
                 });
         xid_rep = xid_rep_window_.front();
                 ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
                 });
         xid_rep = xid_rep_window_.front();

@@ -172,27 +175,27 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
                         ch->send(forgot.buf);
                     ch->send(req);
                 }
                         ch->send(forgot.buf);
                     ch->send(req);
                 }

-                else IF_LEVEL(1) LOG("not reachable");
-                IF_LEVEL(2) LOG(clt_nonce_ << " just sent req proc " << hex << proc <<
-                                " xid " << dec << ca.xid << " clt_nonce " << clt_nonce_);
+                else IF_LEVEL(1) LOG << "not reachable";
+                IF_LEVEL(2) LOG << clt_nonce_ << " just sent req proc " << std::hex << proc
+                                << " xid " << std::dec << ca.xid << " clt_nonce " << clt_nonce_;

             }
             transmit = false; // only send once on a given channel
         }
 
             }
             transmit = false; // only send once on a given channel
         }
 

-        auto nextdeadline = min(steady_clock::now() + curr_to, finaldeadline);
+        auto nextdeadline = std::min(steady_clock::now() + curr_to, finaldeadline);

         curr_to *= 2;
 
         {
             lock cal(ca.m);
             while (!ca.done) {
         curr_to *= 2;
 
         {
             lock cal(ca.m);
             while (!ca.done) {

-                IF_LEVEL(2) LOG("wait");
-                if (ca.c.wait_until(cal, nextdeadline) == cv_status::timeout) {
-                    IF_LEVEL(2) LOG("timeout");
+                IF_LEVEL(2) LOG << "wait";
+                if (ca.c.wait_until(cal, nextdeadline) == std::cv_status::timeout) {
+                    IF_LEVEL(2) LOG << "timeout";

                     break;
                 }
             }
             if (ca.done) {
                     break;
                 }
             }
             if (ca.done) {

-                IF_LEVEL(2) LOG("reply received");
+                IF_LEVEL(2) LOG << "reply received";

                 break;
             }
         }
                 break;
             }
         }

@@ -231,9 +234,9 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
 
     lock cal(ca.m);
 
 
     lock cal(ca.m);
 

-    IF_LEVEL(2) LOG(clt_nonce_ << " call done for req proc " << hex << proc <<
-                    " xid " << dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":" <<
-                    ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret);
+    IF_LEVEL(2) LOG << clt_nonce_ << " call done for req proc " << std::hex << proc
+                    << " xid " << std::dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":"
+                    << 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);
 
     // destruction of req automatically frees its buffer
     return (ca.done? ca.intret : rpc_protocol::timeout_failure);

@@ -258,10 +261,10 @@ rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
 {
     unmarshall rep(b, true);
     rpc_protocol::reply_header h;
 {
     unmarshall rep(b, true);
     rpc_protocol::reply_header h;

-    rep.unpack_header(h);
+    rep.read_header(h);

 
     if (!rep.ok()) {
 
     if (!rep.ok()) {

-        IF_LEVEL(1) LOG("unmarshall header failed!!!");
+        IF_LEVEL(1) LOG << "unmarshall header failed!!!";

         return true;
     }
 
         return true;
     }
 

@@ -270,7 +273,7 @@ rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
     update_xid_rep(h.xid, ml);
 
     if (calls_.find(h.xid) == calls_.end()) {
     update_xid_rep(h.xid, ml);
 
     if (calls_.find(h.xid) == calls_.end()) {

-        IF_LEVEL(2) LOG("xid " << h.xid << " no pending request");
+        IF_LEVEL(2) LOG << "xid " << h.xid << " no pending request";

         return true;
     }
     caller *ca = calls_[h.xid];
         return true;
     }
     caller *ca = calls_[h.xid];

@@ -280,7 +283,7 @@ rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
         *ca->rep = b;
         ca->intret = h.ret;
         if (ca->intret < 0) {
         *ca->rep = b;
         ca->intret = h.ret;
         if (ca->intret < 0) {

-            IF_LEVEL(2) LOG("RPC reply error for xid " << h.xid << " intret " << ca->intret);
+            IF_LEVEL(2) LOG << "RPC reply error for xid " << h.xid << " intret " << ca->intret;

         }
         ca->done = 1;
     }
         }
         ca->done = 1;
     }

@@ -313,7 +316,7 @@ rpcs::rpcs(in_port_t p1) : port_(p1)
 {
     set_rand_seed();
     nonce_ = (nonce_t)random();
 {
     set_rand_seed();
     nonce_ = (nonce_t)random();

-    IF_LEVEL(2) LOG("created with nonce " << nonce_);
+    IF_LEVEL(2) LOG << "created with nonce " << nonce_;

 
     reg(rpc_protocol::bind, &rpcs::rpcbind, this);
 }
 
     reg(rpc_protocol::bind, &rpcs::rpcbind, this);
 }

@@ -331,7 +334,7 @@ rpcs::~rpcs() {
 
 bool rpcs::got_pdu(const shared_ptr<connection> & c, const string & b) {
     if (!reachable_) {
 
 bool rpcs::got_pdu(const shared_ptr<connection> & c, const string & b) {
     if (!reachable_) {

-        IF_LEVEL(1) LOG("not reachable");
+        IF_LEVEL(1) LOG << "not reachable";

         return true;
     }
 
         return true;
     }
 

@@ -342,26 +345,26 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
     unmarshall req(buf, true);
 
     rpc_protocol::request_header h;
     unmarshall req(buf, true);
 
     rpc_protocol::request_header h;

-    req.unpack_header(h);
+    req.read_header(h);

     proc_id_t proc = h.proc;
 
     if (!req.ok()) {
     proc_id_t proc = h.proc;
 
     if (!req.ok()) {

-        IF_LEVEL(1) LOG("unmarshall header failed");
+        IF_LEVEL(1) LOG << "unmarshall header failed";

         return;
     }
 
         return;
     }
 

-    IF_LEVEL(2) LOG("rpc " << h.xid << " (proc " << hex << proc << ", last_rep " <<
-                    dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce);
+    IF_LEVEL(2) LOG << "rpc " << h.xid << " (proc " << std::hex << proc << ", last_rep "
+                    << std::dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce;

 
     marshall rep;
     rpc_protocol::reply_header rh{h.xid,0};
 
     // is client sending to an old instance of server?
     if (h.srv_nonce != 0 && h.srv_nonce != nonce_) {
 
     marshall rep;
     rpc_protocol::reply_header rh{h.xid,0};
 
     // is client sending to an old instance of server?
     if (h.srv_nonce != 0 && h.srv_nonce != nonce_) {

-        IF_LEVEL(2) LOG("rpc for an old server instance " << h.srv_nonce <<
-                        " (current " << nonce_ << ") proc " << hex << h.proc);
+        IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce
+                        << " (current " << nonce_ << ") proc " << std::hex << h.proc;

         rh.ret = rpc_protocol::oldsrv_failure;
         rh.ret = rpc_protocol::oldsrv_failure;

-        rep.pack_header(rh);
+        rep.write_header(rh);

         c->send(rep);
         return;
     }
         c->send(rep);
         return;
     }

@@ -371,7 +374,7 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
     {
         lock pl(procs_m_);
         if (procs_.count(proc) < 1) {
     {
         lock pl(procs_m_);
         if (procs_.count(proc) < 1) {

-            LOG("unknown proc 0x" << hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_);
+            LOG << "unknown proc 0x" << std::hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_;

             VERIFY(0);
             return;
         }
             VERIFY(0);
             return;
         }

@@ -386,8 +389,8 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
         if (reply_window_.find(h.clt_nonce) == reply_window_.end()) {
             VERIFY (reply_window_[h.clt_nonce].size() == 0); // create
             reply_window_[h.clt_nonce].push_back(reply_t(-1)); // store starting reply xid
         if (reply_window_.find(h.clt_nonce) == reply_window_.end()) {
             VERIFY (reply_window_[h.clt_nonce].size() == 0); // create
             reply_window_[h.clt_nonce].push_back(reply_t(-1)); // store starting reply xid

-            IF_LEVEL(2) LOG("new client " << h.clt_nonce << " xid " << h.xid <<
-                            " chan " << c->fd << ", total clients " << (reply_window_.size()-1));
+            IF_LEVEL(2) LOG << "new client " << h.clt_nonce << " xid " << h.xid
+                            << " chan " << c->fd << ", total clients " << (reply_window_.size()-1);

         }
     }
 
         }
     }
 

@@ -404,20 +407,21 @@ 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
 
     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) {
             if (rh.ret == rpc_protocol::unmarshall_args_failure) {

-                LOG("failed to unmarshall the arguments. You are " <<
-                    "probably calling RPC 0x" << hex << proc << " with the wrong " <<
-                    "types of arguments.");
+                LOG << "failed to unmarshall the arguments. You are "
+                    << "probably calling RPC 0x" << std::hex << proc << " with the wrong "
+                    << "types of arguments.";

                 VERIFY(0);
             }
             VERIFY(rh.ret >= 0);
 
                 VERIFY(0);
             }
             VERIFY(rh.ret >= 0);
 

-            rep.pack_header(rh);
+            rep.write_header(rh);

             b1 = rep;
 
             b1 = rep;
 

-            IF_LEVEL(2) LOG("sending and saving reply of size " << b1.size() << " for rpc " <<
-                            h.xid << ", proc " << hex << proc << " ret " << dec << rh.ret << ", clt " << h.clt_nonce);
+            IF_LEVEL(2) LOG << "sending and saving reply of size " << b1.size() << " for rpc "
+                            << h.xid << ", proc " << std::hex << proc << " ret " << std::dec
+                            << rh.ret << ", clt " << h.clt_nonce;

 
             add_reply(h.clt_nonce, h.xid, b1);
 
 
             add_reply(h.clt_nonce, h.xid, b1);
 

@@ -436,9 +440,9 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
             c->send(b1);
             break;
         case FORGOTTEN: // very old request and we don't have the response anymore
             c->send(b1);
             break;
         case FORGOTTEN: // very old request and we don't have the response anymore

-            IF_LEVEL(2) LOG("very old request " << h.xid << " from " << h.clt_nonce);
+            IF_LEVEL(2) LOG << "very old request " << h.xid << " from " << h.clt_nonce;

             rh.ret = rpc_protocol::atmostonce_failure;
             rh.ret = rpc_protocol::atmostonce_failure;

-            rep.pack_header(rh);
+            rep.write_header(rh);

             c->send(rep);
             break;
     }
             c->send(rep);
             break;
     }

@@ -515,7 +519,7 @@ void rpcs::add_reply(nonce_t clt_nonce, xid_t xid, const string & b) {
     for (it++; it != l.end() && it->xid < xid; it++);
     // there should already be an entry, so whine if there isn't
     if (it == l.end() || it->xid != xid) {
     for (it++; it != l.end() && it->xid < xid; it++);
     // there should already be an entry, so whine if there isn't
     if (it == l.end() || it->xid != xid) {

-        LOG("Could not find reply struct in add_reply");
+        LOG << "Could not find reply struct in add_reply";

         l.insert(it, reply_t(xid, b));
     } else {
         *it = reply_t(xid, b);
         l.insert(it, reply_t(xid, b));
     } else {
         *it = reply_t(xid, b);

@@ -523,7 +527,7 @@ void rpcs::add_reply(nonce_t clt_nonce, xid_t xid, const string & b) {
 }
 
 rpc_protocol::status rpcs::rpcbind(nonce_t & r) {
 }
 
 rpc_protocol::status rpcs::rpcbind(nonce_t & r) {

-    IF_LEVEL(2) LOG("called return nonce " << nonce_);
+    IF_LEVEL(2) LOG << "called return nonce " << nonce_;

     r = nonce_;
     return 0;
 }
     r = nonce_;
     return 0;
 }

@@ -548,12 +552,12 @@ static sockaddr_in make_sockaddr(const string & hostandport) {
         struct hostent *hp = gethostbyname(host.c_str());
 
         if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) {
         struct hostent *hp = gethostbyname(host.c_str());
 
         if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) {

-            LOG_NONMEMBER("cannot find host name " << host);
+            LOG_NONMEMBER << "cannot find host name " << host;

             exit(1);
         }
         memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t));
         dst.sin_addr.s_addr = a.s_addr;
     }
             exit(1);
         }
         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;
 }
     return dst;
 }